refactor: move letzshop endpoints to marketplace module and add vendor service tests
Move letzshop-related functionality from tenancy to marketplace module: - Move admin letzshop routes to marketplace/routes/api/admin_letzshop.py - Move letzshop schemas to marketplace/schemas/letzshop.py - Remove letzshop code from tenancy module (admin_vendors, vendor_service) - Update model exports and imports Add comprehensive unit tests for vendor services: - test_company_service.py: Company management operations - test_platform_service.py: Platform management operations - test_vendor_domain_service.py: Vendor domain operations - test_vendor_team_service.py: Vendor team management Update module definitions: - billing, messaging, payments: Minor definition updates Add architecture proposals documentation: - Module dependency redesign session notes - Decouple modules implementation plan - Module decoupling proposal Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -177,65 +177,8 @@ class TestAdminService:
|
||||
exception = exc_info.value
|
||||
assert exception.error_code == "VENDOR_NOT_FOUND"
|
||||
|
||||
# Marketplace Import Jobs Tests
|
||||
def test_get_marketplace_import_jobs_no_filters(
|
||||
self, db, test_marketplace_import_job
|
||||
):
|
||||
"""Test getting marketplace import jobs without filters"""
|
||||
result = self.service.get_marketplace_import_jobs(db, skip=0, limit=10)
|
||||
|
||||
assert len(result) >= 1
|
||||
# Find our test job in the results
|
||||
test_job = next(
|
||||
(job for job in result if job.job_id == test_marketplace_import_job.id),
|
||||
None,
|
||||
)
|
||||
assert test_job is not None
|
||||
assert (
|
||||
test_job.marketplace.lower()
|
||||
== test_marketplace_import_job.marketplace.lower()
|
||||
)
|
||||
assert test_job.status == test_marketplace_import_job.status
|
||||
|
||||
def test_get_marketplace_import_jobs_with_marketplace_filter(
|
||||
self, db, test_marketplace_import_job
|
||||
):
|
||||
"""Test filtering marketplace import jobs by marketplace"""
|
||||
result = self.service.get_marketplace_import_jobs(
|
||||
db, marketplace=test_marketplace_import_job.marketplace, skip=0, limit=10
|
||||
)
|
||||
|
||||
assert len(result) >= 1
|
||||
for job in result:
|
||||
assert (
|
||||
test_marketplace_import_job.marketplace.lower()
|
||||
in job.marketplace.lower()
|
||||
)
|
||||
|
||||
def test_get_marketplace_import_jobs_with_status_filter(
|
||||
self, db, test_marketplace_import_job
|
||||
):
|
||||
"""Test filtering marketplace import jobs by status"""
|
||||
result = self.service.get_marketplace_import_jobs(
|
||||
db, status=test_marketplace_import_job.status, skip=0, limit=10
|
||||
)
|
||||
|
||||
assert len(result) >= 1
|
||||
for job in result:
|
||||
assert job.status == test_marketplace_import_job.status
|
||||
|
||||
def test_get_marketplace_import_jobs_pagination(
|
||||
self, db, test_marketplace_import_job
|
||||
):
|
||||
"""Test marketplace import jobs pagination"""
|
||||
result_page1 = self.service.get_marketplace_import_jobs(db, skip=0, limit=1)
|
||||
result_page2 = self.service.get_marketplace_import_jobs(db, skip=1, limit=1)
|
||||
|
||||
assert len(result_page1) >= 0
|
||||
assert len(result_page2) >= 0
|
||||
|
||||
if len(result_page1) > 0 and len(result_page2) > 0:
|
||||
assert result_page1[0].job_id != result_page2[0].job_id
|
||||
# NOTE: Marketplace Import Jobs tests have been moved to the marketplace module.
|
||||
# See tests/unit/services/test_marketplace_import_job_service.py
|
||||
|
||||
# Statistics Tests
|
||||
def test_get_user_statistics(self, db, test_user, test_admin):
|
||||
|
||||
355
tests/unit/services/test_company_service.py
Normal file
355
tests/unit/services/test_company_service.py
Normal file
@@ -0,0 +1,355 @@
|
||||
# tests/unit/services/test_company_service.py
|
||||
"""Unit tests for CompanyService."""
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
from app.modules.tenancy.exceptions import CompanyNotFoundException, UserNotFoundException
|
||||
from app.modules.tenancy.services.company_service import company_service
|
||||
from app.modules.tenancy.models import Company
|
||||
from app.modules.tenancy.schemas.company import (
|
||||
CompanyCreate,
|
||||
CompanyUpdate,
|
||||
CompanyTransferOwnership,
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# FIXTURES
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def unverified_company(db, test_user):
|
||||
"""Create an unverified test company."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
company = Company(
|
||||
name=f"Unverified Company {unique_id}",
|
||||
owner_user_id=test_user.id,
|
||||
contact_email=f"unverified{unique_id}@company.com",
|
||||
is_active=True,
|
||||
is_verified=False,
|
||||
)
|
||||
db.add(company)
|
||||
db.commit()
|
||||
db.refresh(company)
|
||||
return company
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def inactive_company(db, test_user):
|
||||
"""Create an inactive test company."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
company = Company(
|
||||
name=f"Inactive Company {unique_id}",
|
||||
owner_user_id=test_user.id,
|
||||
contact_email=f"inactive{unique_id}@company.com",
|
||||
is_active=False,
|
||||
is_verified=True,
|
||||
)
|
||||
db.add(company)
|
||||
db.commit()
|
||||
db.refresh(company)
|
||||
return company
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# CREATE TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestCompanyServiceCreate:
|
||||
"""Test suite for company creation."""
|
||||
|
||||
def test_create_company_with_new_owner(self, db):
|
||||
"""Test creating a company with a new owner user."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
company_data = CompanyCreate(
|
||||
name=f"New Company {unique_id}",
|
||||
owner_email=f"newowner{unique_id}@example.com",
|
||||
contact_email=f"contact{unique_id}@company.com",
|
||||
description="A new test company",
|
||||
)
|
||||
|
||||
company, owner, temp_password = company_service.create_company_with_owner(
|
||||
db, company_data
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert company is not None
|
||||
assert company.name == f"New Company {unique_id}"
|
||||
assert company.is_active is True
|
||||
assert company.is_verified is False
|
||||
assert owner is not None
|
||||
assert owner.email == f"newowner{unique_id}@example.com"
|
||||
assert temp_password is not None # New user gets temp password
|
||||
|
||||
def test_create_company_with_existing_owner(self, db, test_user):
|
||||
"""Test creating a company with an existing user as owner."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
company_data = CompanyCreate(
|
||||
name=f"Existing Owner Company {unique_id}",
|
||||
owner_email=test_user.email,
|
||||
contact_email=f"contact{unique_id}@company.com",
|
||||
)
|
||||
|
||||
company, owner, temp_password = company_service.create_company_with_owner(
|
||||
db, company_data
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert company is not None
|
||||
assert owner.id == test_user.id
|
||||
assert temp_password is None # Existing user doesn't get new password
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# READ TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestCompanyServiceRead:
|
||||
"""Test suite for reading companies."""
|
||||
|
||||
def test_get_company_by_id_success(self, db, test_company):
|
||||
"""Test getting a company by ID."""
|
||||
company = company_service.get_company_by_id(db, test_company.id)
|
||||
|
||||
assert company is not None
|
||||
assert company.id == test_company.id
|
||||
assert company.name == test_company.name
|
||||
|
||||
def test_get_company_by_id_not_found(self, db):
|
||||
"""Test getting a non-existent company raises exception."""
|
||||
with pytest.raises(CompanyNotFoundException) as exc_info:
|
||||
company_service.get_company_by_id(db, 99999)
|
||||
|
||||
assert "99999" in str(exc_info.value)
|
||||
|
||||
def test_get_companies_paginated(self, db, test_company, other_company):
|
||||
"""Test getting paginated list of companies."""
|
||||
companies, total = company_service.get_companies(db, skip=0, limit=10)
|
||||
|
||||
assert len(companies) >= 2
|
||||
assert total >= 2
|
||||
|
||||
def test_get_companies_with_search(self, db, test_company):
|
||||
"""Test searching companies by name."""
|
||||
# Get the unique part of the company name
|
||||
search_term = test_company.name.split()[0] # "Test"
|
||||
|
||||
companies, total = company_service.get_companies(
|
||||
db, skip=0, limit=100, search=search_term
|
||||
)
|
||||
|
||||
assert len(companies) >= 1
|
||||
assert any(c.id == test_company.id for c in companies)
|
||||
|
||||
def test_get_companies_filter_by_active(self, db, test_company, inactive_company):
|
||||
"""Test filtering companies by active status."""
|
||||
active_companies, _ = company_service.get_companies(
|
||||
db, skip=0, limit=100, is_active=True
|
||||
)
|
||||
inactive_companies, _ = company_service.get_companies(
|
||||
db, skip=0, limit=100, is_active=False
|
||||
)
|
||||
|
||||
assert all(c.is_active for c in active_companies)
|
||||
assert all(not c.is_active for c in inactive_companies)
|
||||
|
||||
def test_get_companies_filter_by_verified(self, db, test_company, unverified_company):
|
||||
"""Test filtering companies by verified status."""
|
||||
verified_companies, _ = company_service.get_companies(
|
||||
db, skip=0, limit=100, is_verified=True
|
||||
)
|
||||
unverified_companies, _ = company_service.get_companies(
|
||||
db, skip=0, limit=100, is_verified=False
|
||||
)
|
||||
|
||||
assert all(c.is_verified for c in verified_companies)
|
||||
assert all(not c.is_verified for c in unverified_companies)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# UPDATE TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestCompanyServiceUpdate:
|
||||
"""Test suite for updating companies."""
|
||||
|
||||
def test_update_company_success(self, db, test_company):
|
||||
"""Test updating company information."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
update_data = CompanyUpdate(
|
||||
name=f"Updated Company {unique_id}",
|
||||
description="Updated description",
|
||||
)
|
||||
|
||||
updated = company_service.update_company(db, test_company.id, update_data)
|
||||
db.commit()
|
||||
|
||||
assert updated.name == f"Updated Company {unique_id}"
|
||||
assert updated.description == "Updated description"
|
||||
|
||||
def test_update_company_partial(self, db, test_company):
|
||||
"""Test partial update of company."""
|
||||
original_name = test_company.name
|
||||
update_data = CompanyUpdate(description="Only description updated")
|
||||
|
||||
updated = company_service.update_company(db, test_company.id, update_data)
|
||||
db.commit()
|
||||
|
||||
assert updated.name == original_name # Name unchanged
|
||||
assert updated.description == "Only description updated"
|
||||
|
||||
def test_update_company_not_found(self, db):
|
||||
"""Test updating non-existent company raises exception."""
|
||||
update_data = CompanyUpdate(name="New Name")
|
||||
|
||||
with pytest.raises(CompanyNotFoundException):
|
||||
company_service.update_company(db, 99999, update_data)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# DELETE TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestCompanyServiceDelete:
|
||||
"""Test suite for deleting companies."""
|
||||
|
||||
def test_delete_company_success(self, db, test_user):
|
||||
"""Test deleting a company."""
|
||||
# Create a company to delete
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
company = Company(
|
||||
name=f"To Delete {unique_id}",
|
||||
owner_user_id=test_user.id,
|
||||
contact_email=f"delete{unique_id}@company.com",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(company)
|
||||
db.commit()
|
||||
company_id = company.id
|
||||
|
||||
# Delete it
|
||||
company_service.delete_company(db, company_id)
|
||||
db.commit()
|
||||
|
||||
# Verify it's gone
|
||||
with pytest.raises(CompanyNotFoundException):
|
||||
company_service.get_company_by_id(db, company_id)
|
||||
|
||||
def test_delete_company_not_found(self, db):
|
||||
"""Test deleting non-existent company raises exception."""
|
||||
with pytest.raises(CompanyNotFoundException):
|
||||
company_service.delete_company(db, 99999)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# TOGGLE TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestCompanyServiceToggle:
|
||||
"""Test suite for toggling company status."""
|
||||
|
||||
def test_toggle_verification_on(self, db, unverified_company):
|
||||
"""Test setting verification to True."""
|
||||
result = company_service.toggle_verification(db, unverified_company.id, True)
|
||||
db.commit()
|
||||
|
||||
assert result.is_verified is True
|
||||
|
||||
def test_toggle_verification_off(self, db, test_company):
|
||||
"""Test setting verification to False."""
|
||||
result = company_service.toggle_verification(db, test_company.id, False)
|
||||
db.commit()
|
||||
|
||||
assert result.is_verified is False
|
||||
|
||||
def test_toggle_active_on(self, db, inactive_company):
|
||||
"""Test setting active status to True."""
|
||||
result = company_service.toggle_active(db, inactive_company.id, True)
|
||||
db.commit()
|
||||
|
||||
assert result.is_active is True
|
||||
|
||||
def test_toggle_active_off(self, db, test_company):
|
||||
"""Test setting active status to False."""
|
||||
result = company_service.toggle_active(db, test_company.id, False)
|
||||
db.commit()
|
||||
|
||||
assert result.is_active is False
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# OWNERSHIP TRANSFER TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestCompanyServiceOwnershipTransfer:
|
||||
"""Test suite for company ownership transfer."""
|
||||
|
||||
def test_transfer_ownership_success(self, db, test_company, other_user):
|
||||
"""Test successful ownership transfer."""
|
||||
original_owner_id = test_company.owner_user_id
|
||||
transfer_data = CompanyTransferOwnership(
|
||||
new_owner_user_id=other_user.id,
|
||||
transfer_reason="Testing ownership transfer",
|
||||
)
|
||||
|
||||
company, old_owner, new_owner = company_service.transfer_ownership(
|
||||
db, test_company.id, transfer_data
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert company.owner_user_id == other_user.id
|
||||
assert old_owner.id == original_owner_id
|
||||
assert new_owner.id == other_user.id
|
||||
|
||||
def test_transfer_ownership_to_same_owner_fails(self, db, test_company, test_user):
|
||||
"""Test transfer to same owner raises error."""
|
||||
transfer_data = CompanyTransferOwnership(
|
||||
new_owner_user_id=test_user.id,
|
||||
transfer_reason="This should fail",
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
company_service.transfer_ownership(db, test_company.id, transfer_data)
|
||||
|
||||
assert "current owner" in str(exc_info.value).lower()
|
||||
|
||||
def test_transfer_ownership_to_nonexistent_user_fails(self, db, test_company):
|
||||
"""Test transfer to non-existent user raises exception."""
|
||||
transfer_data = CompanyTransferOwnership(
|
||||
new_owner_user_id=99999,
|
||||
transfer_reason="This should fail",
|
||||
)
|
||||
|
||||
with pytest.raises(UserNotFoundException):
|
||||
company_service.transfer_ownership(db, test_company.id, transfer_data)
|
||||
|
||||
def test_transfer_ownership_nonexistent_company_fails(self, db, other_user):
|
||||
"""Test transfer on non-existent company raises exception."""
|
||||
transfer_data = CompanyTransferOwnership(
|
||||
new_owner_user_id=other_user.id,
|
||||
transfer_reason="This should fail",
|
||||
)
|
||||
|
||||
with pytest.raises(CompanyNotFoundException):
|
||||
company_service.transfer_ownership(db, 99999, transfer_data)
|
||||
297
tests/unit/services/test_platform_service.py
Normal file
297
tests/unit/services/test_platform_service.py
Normal file
@@ -0,0 +1,297 @@
|
||||
# tests/unit/services/test_platform_service.py
|
||||
"""Unit tests for PlatformService."""
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
from app.modules.tenancy.exceptions import PlatformNotFoundException
|
||||
from app.modules.tenancy.services.platform_service import platform_service, PlatformStats
|
||||
from app.modules.tenancy.models import Platform, VendorPlatform
|
||||
from app.modules.cms.models import ContentPage
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# FIXTURES
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def inactive_platform(db):
|
||||
"""Create an inactive test platform."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
platform = Platform(
|
||||
code=f"inactive_{unique_id}",
|
||||
name=f"Inactive Platform {unique_id}",
|
||||
description="An inactive test platform",
|
||||
path_prefix=f"inactive{unique_id}",
|
||||
is_active=False,
|
||||
is_public=False,
|
||||
default_language="en",
|
||||
supported_languages=["en"],
|
||||
)
|
||||
db.add(platform)
|
||||
db.commit()
|
||||
db.refresh(platform)
|
||||
return platform
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def platform_with_vendor(db, test_platform, test_vendor):
|
||||
"""Create a vendor-platform assignment."""
|
||||
vendor_platform = VendorPlatform(
|
||||
vendor_id=test_vendor.id,
|
||||
platform_id=test_platform.id,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(vendor_platform)
|
||||
db.commit()
|
||||
return test_platform
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def platform_with_pages(db, test_platform):
|
||||
"""Create a platform with content pages."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
|
||||
# Platform marketing page (published)
|
||||
platform_page = ContentPage(
|
||||
platform_id=test_platform.id,
|
||||
vendor_id=None,
|
||||
slug=f"platform-page-{unique_id}",
|
||||
page_type="marketing",
|
||||
is_platform_page=True,
|
||||
is_published=True,
|
||||
)
|
||||
db.add(platform_page)
|
||||
|
||||
# Vendor default page (draft)
|
||||
default_page = ContentPage(
|
||||
platform_id=test_platform.id,
|
||||
vendor_id=None,
|
||||
slug=f"vendor-default-{unique_id}",
|
||||
page_type="about",
|
||||
is_platform_page=False,
|
||||
is_published=False,
|
||||
)
|
||||
db.add(default_page)
|
||||
|
||||
db.commit()
|
||||
return test_platform
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# GET PLATFORM TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestPlatformServiceGet:
|
||||
"""Test suite for getting platforms."""
|
||||
|
||||
def test_get_platform_by_code_success(self, db, test_platform):
|
||||
"""Test getting a platform by code."""
|
||||
platform = platform_service.get_platform_by_code(db, test_platform.code)
|
||||
|
||||
assert platform is not None
|
||||
assert platform.id == test_platform.id
|
||||
assert platform.code == test_platform.code
|
||||
|
||||
def test_get_platform_by_code_not_found(self, db):
|
||||
"""Test getting a non-existent platform raises exception."""
|
||||
with pytest.raises(PlatformNotFoundException) as exc_info:
|
||||
platform_service.get_platform_by_code(db, "nonexistent_platform")
|
||||
|
||||
assert "nonexistent_platform" in str(exc_info.value)
|
||||
|
||||
def test_get_platform_by_code_optional_found(self, db, test_platform):
|
||||
"""Test optional get returns platform when found."""
|
||||
platform = platform_service.get_platform_by_code_optional(db, test_platform.code)
|
||||
|
||||
assert platform is not None
|
||||
assert platform.id == test_platform.id
|
||||
|
||||
def test_get_platform_by_code_optional_not_found(self, db):
|
||||
"""Test optional get returns None when not found."""
|
||||
platform = platform_service.get_platform_by_code_optional(db, "nonexistent")
|
||||
|
||||
assert platform is None
|
||||
|
||||
def test_get_platform_by_id_success(self, db, test_platform):
|
||||
"""Test getting a platform by ID."""
|
||||
platform = platform_service.get_platform_by_id(db, test_platform.id)
|
||||
|
||||
assert platform is not None
|
||||
assert platform.code == test_platform.code
|
||||
|
||||
def test_get_platform_by_id_not_found(self, db):
|
||||
"""Test getting a non-existent platform by ID raises exception."""
|
||||
with pytest.raises(PlatformNotFoundException):
|
||||
platform_service.get_platform_by_id(db, 99999)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# LIST PLATFORMS TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestPlatformServiceList:
|
||||
"""Test suite for listing platforms."""
|
||||
|
||||
def test_list_platforms_active_only(self, db, test_platform, inactive_platform):
|
||||
"""Test listing only active platforms."""
|
||||
platforms = platform_service.list_platforms(db, include_inactive=False)
|
||||
|
||||
platform_ids = [p.id for p in platforms]
|
||||
assert test_platform.id in platform_ids
|
||||
assert inactive_platform.id not in platform_ids
|
||||
|
||||
def test_list_platforms_include_inactive(self, db, test_platform, inactive_platform):
|
||||
"""Test listing all platforms including inactive."""
|
||||
platforms = platform_service.list_platforms(db, include_inactive=True)
|
||||
|
||||
platform_ids = [p.id for p in platforms]
|
||||
assert test_platform.id in platform_ids
|
||||
assert inactive_platform.id in platform_ids
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# COUNT TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestPlatformServiceCounts:
|
||||
"""Test suite for platform counts."""
|
||||
|
||||
def test_get_vendor_count_zero(self, db, test_platform):
|
||||
"""Test vendor count is zero when no vendors assigned."""
|
||||
count = platform_service.get_vendor_count(db, test_platform.id)
|
||||
|
||||
assert count == 0
|
||||
|
||||
def test_get_vendor_count_with_vendors(self, db, platform_with_vendor):
|
||||
"""Test vendor count with assigned vendors."""
|
||||
count = platform_service.get_vendor_count(db, platform_with_vendor.id)
|
||||
|
||||
assert count >= 1
|
||||
|
||||
def test_get_platform_pages_count(self, db, platform_with_pages):
|
||||
"""Test platform pages count."""
|
||||
count = platform_service.get_platform_pages_count(db, platform_with_pages.id)
|
||||
|
||||
assert count >= 1
|
||||
|
||||
def test_get_vendor_defaults_count(self, db, platform_with_pages):
|
||||
"""Test vendor defaults count."""
|
||||
count = platform_service.get_vendor_defaults_count(db, platform_with_pages.id)
|
||||
|
||||
assert count >= 1
|
||||
|
||||
def test_get_published_pages_count(self, db, platform_with_pages):
|
||||
"""Test published pages count."""
|
||||
count = platform_service.get_published_pages_count(db, platform_with_pages.id)
|
||||
|
||||
assert count >= 1
|
||||
|
||||
def test_get_draft_pages_count(self, db, platform_with_pages):
|
||||
"""Test draft pages count."""
|
||||
count = platform_service.get_draft_pages_count(db, platform_with_pages.id)
|
||||
|
||||
assert count >= 1
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# PLATFORM STATS TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestPlatformServiceStats:
|
||||
"""Test suite for platform statistics."""
|
||||
|
||||
def test_get_platform_stats(self, db, platform_with_pages, test_vendor):
|
||||
"""Test getting comprehensive platform statistics."""
|
||||
# Add a vendor to the platform
|
||||
vendor_platform = VendorPlatform(
|
||||
vendor_id=test_vendor.id,
|
||||
platform_id=platform_with_pages.id,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(vendor_platform)
|
||||
db.commit()
|
||||
|
||||
stats = platform_service.get_platform_stats(db, platform_with_pages)
|
||||
|
||||
assert isinstance(stats, PlatformStats)
|
||||
assert stats.platform_id == platform_with_pages.id
|
||||
assert stats.platform_code == platform_with_pages.code
|
||||
assert stats.platform_name == platform_with_pages.name
|
||||
assert stats.vendor_count >= 1
|
||||
assert stats.platform_pages_count >= 1
|
||||
assert stats.vendor_defaults_count >= 1
|
||||
|
||||
def test_get_platform_stats_empty_platform(self, db, test_platform):
|
||||
"""Test stats for a platform with no content."""
|
||||
stats = platform_service.get_platform_stats(db, test_platform)
|
||||
|
||||
assert stats.vendor_count == 0
|
||||
assert stats.platform_pages_count == 0
|
||||
assert stats.vendor_defaults_count == 0
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# UPDATE TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestPlatformServiceUpdate:
|
||||
"""Test suite for updating platforms."""
|
||||
|
||||
def test_update_platform_name(self, db, test_platform):
|
||||
"""Test updating platform name."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
new_name = f"Updated Platform {unique_id}"
|
||||
|
||||
updated = platform_service.update_platform(
|
||||
db, test_platform, {"name": new_name}
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert updated.name == new_name
|
||||
|
||||
def test_update_platform_multiple_fields(self, db, test_platform):
|
||||
"""Test updating multiple platform fields."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
update_data = {
|
||||
"name": f"Multi Update {unique_id}",
|
||||
"description": "Updated description",
|
||||
"is_public": False,
|
||||
}
|
||||
|
||||
updated = platform_service.update_platform(db, test_platform, update_data)
|
||||
db.commit()
|
||||
|
||||
assert updated.name == f"Multi Update {unique_id}"
|
||||
assert updated.description == "Updated description"
|
||||
assert updated.is_public is False
|
||||
|
||||
def test_update_platform_ignores_invalid_fields(self, db, test_platform):
|
||||
"""Test that invalid fields are ignored during update."""
|
||||
original_name = test_platform.name
|
||||
update_data = {
|
||||
"nonexistent_field": "value",
|
||||
"name": original_name, # Keep same name
|
||||
}
|
||||
|
||||
# Should not raise an error
|
||||
updated = platform_service.update_platform(db, test_platform, update_data)
|
||||
|
||||
assert updated.name == original_name
|
||||
420
tests/unit/services/test_vendor_domain_service.py
Normal file
420
tests/unit/services/test_vendor_domain_service.py
Normal file
@@ -0,0 +1,420 @@
|
||||
# tests/unit/services/test_vendor_domain_service.py
|
||||
"""Unit tests for VendorDomainService."""
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app.modules.tenancy.exceptions import (
|
||||
DNSVerificationException,
|
||||
DomainAlreadyVerifiedException,
|
||||
DomainNotVerifiedException,
|
||||
DomainVerificationFailedException,
|
||||
MaxDomainsReachedException,
|
||||
VendorDomainAlreadyExistsException,
|
||||
VendorDomainNotFoundException,
|
||||
VendorNotFoundException,
|
||||
)
|
||||
from app.modules.tenancy.models import VendorDomain
|
||||
from app.modules.tenancy.schemas.vendor_domain import VendorDomainCreate, VendorDomainUpdate
|
||||
from app.modules.tenancy.services.vendor_domain_service import vendor_domain_service
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# FIXTURES
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_domain(db, test_vendor):
|
||||
"""Create a test domain for a vendor."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
domain = VendorDomain(
|
||||
vendor_id=test_vendor.id,
|
||||
domain=f"test{unique_id}.example.com",
|
||||
is_primary=False,
|
||||
is_active=False,
|
||||
is_verified=False,
|
||||
verification_token=f"token_{unique_id}",
|
||||
ssl_status="pending",
|
||||
)
|
||||
db.add(domain)
|
||||
db.commit()
|
||||
db.refresh(domain)
|
||||
return domain
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def verified_domain(db, test_vendor):
|
||||
"""Create a verified domain for a vendor."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
domain = VendorDomain(
|
||||
vendor_id=test_vendor.id,
|
||||
domain=f"verified{unique_id}.example.com",
|
||||
is_primary=False,
|
||||
is_active=True,
|
||||
is_verified=True,
|
||||
verified_at=datetime.now(UTC),
|
||||
verification_token=f"verified_token_{unique_id}",
|
||||
ssl_status="active",
|
||||
)
|
||||
db.add(domain)
|
||||
db.commit()
|
||||
db.refresh(domain)
|
||||
return domain
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def primary_domain(db, test_vendor):
|
||||
"""Create a primary domain for a vendor."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
domain = VendorDomain(
|
||||
vendor_id=test_vendor.id,
|
||||
domain=f"primary{unique_id}.example.com",
|
||||
is_primary=True,
|
||||
is_active=True,
|
||||
is_verified=True,
|
||||
verified_at=datetime.now(UTC),
|
||||
verification_token=f"primary_token_{unique_id}",
|
||||
ssl_status="active",
|
||||
)
|
||||
db.add(domain)
|
||||
db.commit()
|
||||
db.refresh(domain)
|
||||
return domain
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# ADD DOMAIN TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestVendorDomainServiceAdd:
|
||||
"""Test suite for adding vendor domains."""
|
||||
|
||||
def test_add_domain_success(self, db, test_vendor):
|
||||
"""Test successfully adding a domain to a vendor."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
domain_data = VendorDomainCreate(
|
||||
domain=f"newdomain{unique_id}.example.com",
|
||||
is_primary=False,
|
||||
)
|
||||
|
||||
result = vendor_domain_service.add_domain(db, test_vendor.id, domain_data)
|
||||
db.commit()
|
||||
|
||||
assert result is not None
|
||||
assert result.vendor_id == test_vendor.id
|
||||
assert result.domain == f"newdomain{unique_id}.example.com"
|
||||
assert result.is_primary is False
|
||||
assert result.is_verified is False
|
||||
assert result.is_active is False
|
||||
assert result.verification_token is not None
|
||||
|
||||
def test_add_domain_as_primary(self, db, test_vendor, primary_domain):
|
||||
"""Test adding a domain as primary unsets other primary domains."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
domain_data = VendorDomainCreate(
|
||||
domain=f"newprimary{unique_id}.example.com",
|
||||
is_primary=True,
|
||||
)
|
||||
|
||||
result = vendor_domain_service.add_domain(db, test_vendor.id, domain_data)
|
||||
db.commit()
|
||||
db.refresh(primary_domain)
|
||||
|
||||
assert result.is_primary is True
|
||||
assert primary_domain.is_primary is False
|
||||
|
||||
def test_add_domain_vendor_not_found(self, db):
|
||||
"""Test adding domain to non-existent vendor raises exception."""
|
||||
domain_data = VendorDomainCreate(
|
||||
domain="test.example.com",
|
||||
is_primary=False,
|
||||
)
|
||||
|
||||
with pytest.raises(VendorNotFoundException):
|
||||
vendor_domain_service.add_domain(db, 99999, domain_data)
|
||||
|
||||
def test_add_domain_already_exists(self, db, test_vendor, test_domain):
|
||||
"""Test adding a domain that already exists raises exception."""
|
||||
domain_data = VendorDomainCreate(
|
||||
domain=test_domain.domain,
|
||||
is_primary=False,
|
||||
)
|
||||
|
||||
with pytest.raises(VendorDomainAlreadyExistsException):
|
||||
vendor_domain_service.add_domain(db, test_vendor.id, domain_data)
|
||||
|
||||
def test_add_domain_max_limit_reached(self, db, test_vendor):
|
||||
"""Test adding domain when max limit reached raises exception."""
|
||||
# Create max domains
|
||||
for i in range(vendor_domain_service.max_domains_per_vendor):
|
||||
domain = VendorDomain(
|
||||
vendor_id=test_vendor.id,
|
||||
domain=f"domain{i}_{uuid.uuid4().hex[:6]}.example.com",
|
||||
verification_token=f"token_{i}_{uuid.uuid4().hex[:6]}",
|
||||
)
|
||||
db.add(domain)
|
||||
db.commit()
|
||||
|
||||
domain_data = VendorDomainCreate(
|
||||
domain="onemore.example.com",
|
||||
is_primary=False,
|
||||
)
|
||||
|
||||
with pytest.raises(MaxDomainsReachedException):
|
||||
vendor_domain_service.add_domain(db, test_vendor.id, domain_data)
|
||||
|
||||
def test_add_domain_reserved_subdomain(self, db, test_vendor):
|
||||
"""Test adding a domain with reserved subdomain raises exception.
|
||||
|
||||
Note: Reserved subdomain validation happens in Pydantic schema first.
|
||||
"""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
VendorDomainCreate(
|
||||
domain="admin.example.com",
|
||||
is_primary=False,
|
||||
)
|
||||
|
||||
assert "reserved subdomain" in str(exc_info.value).lower()
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# GET DOMAINS TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestVendorDomainServiceGet:
|
||||
"""Test suite for getting vendor domains."""
|
||||
|
||||
def test_get_vendor_domains_success(self, db, test_vendor, test_domain, verified_domain):
|
||||
"""Test getting all domains for a vendor."""
|
||||
domains = vendor_domain_service.get_vendor_domains(db, test_vendor.id)
|
||||
|
||||
assert len(domains) >= 2
|
||||
domain_ids = [d.id for d in domains]
|
||||
assert test_domain.id in domain_ids
|
||||
assert verified_domain.id in domain_ids
|
||||
|
||||
def test_get_vendor_domains_empty(self, db, test_vendor):
|
||||
"""Test getting domains for vendor with no domains."""
|
||||
domains = vendor_domain_service.get_vendor_domains(db, test_vendor.id)
|
||||
|
||||
# May have domains from other fixtures, so just check it returns a list
|
||||
assert isinstance(domains, list)
|
||||
|
||||
def test_get_vendor_domains_vendor_not_found(self, db):
|
||||
"""Test getting domains for non-existent vendor raises exception."""
|
||||
with pytest.raises(VendorNotFoundException):
|
||||
vendor_domain_service.get_vendor_domains(db, 99999)
|
||||
|
||||
def test_get_domain_by_id_success(self, db, test_domain):
|
||||
"""Test getting a domain by ID."""
|
||||
domain = vendor_domain_service.get_domain_by_id(db, test_domain.id)
|
||||
|
||||
assert domain is not None
|
||||
assert domain.id == test_domain.id
|
||||
assert domain.domain == test_domain.domain
|
||||
|
||||
def test_get_domain_by_id_not_found(self, db):
|
||||
"""Test getting non-existent domain raises exception."""
|
||||
with pytest.raises(VendorDomainNotFoundException):
|
||||
vendor_domain_service.get_domain_by_id(db, 99999)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# UPDATE DOMAIN TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestVendorDomainServiceUpdate:
|
||||
"""Test suite for updating vendor domains."""
|
||||
|
||||
def test_update_domain_set_primary(self, db, test_domain, primary_domain):
|
||||
"""Test setting a domain as primary."""
|
||||
update_data = VendorDomainUpdate(is_primary=True)
|
||||
|
||||
# First verify the domain (required for activation)
|
||||
test_domain.is_verified = True
|
||||
db.commit()
|
||||
|
||||
result = vendor_domain_service.update_domain(db, test_domain.id, update_data)
|
||||
db.commit()
|
||||
db.refresh(primary_domain)
|
||||
|
||||
assert result.is_primary is True
|
||||
assert primary_domain.is_primary is False
|
||||
|
||||
def test_update_domain_activate_verified(self, db, verified_domain):
|
||||
"""Test activating a verified domain."""
|
||||
verified_domain.is_active = False
|
||||
db.commit()
|
||||
|
||||
update_data = VendorDomainUpdate(is_active=True)
|
||||
|
||||
result = vendor_domain_service.update_domain(db, verified_domain.id, update_data)
|
||||
db.commit()
|
||||
|
||||
assert result.is_active is True
|
||||
|
||||
def test_update_domain_activate_unverified_fails(self, db, test_domain):
|
||||
"""Test activating an unverified domain raises exception."""
|
||||
update_data = VendorDomainUpdate(is_active=True)
|
||||
|
||||
with pytest.raises(DomainNotVerifiedException):
|
||||
vendor_domain_service.update_domain(db, test_domain.id, update_data)
|
||||
|
||||
def test_update_domain_not_found(self, db):
|
||||
"""Test updating non-existent domain raises exception."""
|
||||
update_data = VendorDomainUpdate(is_primary=True)
|
||||
|
||||
with pytest.raises(VendorDomainNotFoundException):
|
||||
vendor_domain_service.update_domain(db, 99999, update_data)
|
||||
|
||||
def test_update_domain_deactivate(self, db, verified_domain):
|
||||
"""Test deactivating a domain."""
|
||||
update_data = VendorDomainUpdate(is_active=False)
|
||||
|
||||
result = vendor_domain_service.update_domain(db, verified_domain.id, update_data)
|
||||
db.commit()
|
||||
|
||||
assert result.is_active is False
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# DELETE DOMAIN TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestVendorDomainServiceDelete:
|
||||
"""Test suite for deleting vendor domains."""
|
||||
|
||||
def test_delete_domain_success(self, db, test_vendor):
|
||||
"""Test successfully deleting a domain."""
|
||||
# Create a domain to delete
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
domain = VendorDomain(
|
||||
vendor_id=test_vendor.id,
|
||||
domain=f"todelete{unique_id}.example.com",
|
||||
verification_token=f"delete_token_{unique_id}",
|
||||
)
|
||||
db.add(domain)
|
||||
db.commit()
|
||||
domain_id = domain.id
|
||||
domain_name = domain.domain
|
||||
|
||||
result = vendor_domain_service.delete_domain(db, domain_id)
|
||||
db.commit()
|
||||
|
||||
assert domain_name in result
|
||||
|
||||
# Verify it's gone
|
||||
with pytest.raises(VendorDomainNotFoundException):
|
||||
vendor_domain_service.get_domain_by_id(db, domain_id)
|
||||
|
||||
def test_delete_domain_not_found(self, db):
|
||||
"""Test deleting non-existent domain raises exception."""
|
||||
with pytest.raises(VendorDomainNotFoundException):
|
||||
vendor_domain_service.delete_domain(db, 99999)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# VERIFY DOMAIN TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestVendorDomainServiceVerify:
|
||||
"""Test suite for domain verification."""
|
||||
|
||||
@patch("dns.resolver.resolve")
|
||||
def test_verify_domain_success(self, mock_resolve, db, test_domain):
|
||||
"""Test successful domain verification."""
|
||||
# Mock DNS response
|
||||
mock_txt = MagicMock()
|
||||
mock_txt.to_text.return_value = f'"{test_domain.verification_token}"'
|
||||
mock_resolve.return_value = [mock_txt]
|
||||
|
||||
domain, message = vendor_domain_service.verify_domain(db, test_domain.id)
|
||||
db.commit()
|
||||
|
||||
assert domain.is_verified is True
|
||||
assert domain.verified_at is not None
|
||||
assert "verified successfully" in message.lower()
|
||||
|
||||
def test_verify_domain_already_verified(self, db, verified_domain):
|
||||
"""Test verifying already verified domain raises exception."""
|
||||
with pytest.raises(DomainAlreadyVerifiedException):
|
||||
vendor_domain_service.verify_domain(db, verified_domain.id)
|
||||
|
||||
def test_verify_domain_not_found(self, db):
|
||||
"""Test verifying non-existent domain raises exception."""
|
||||
with pytest.raises(VendorDomainNotFoundException):
|
||||
vendor_domain_service.verify_domain(db, 99999)
|
||||
|
||||
@patch("dns.resolver.resolve")
|
||||
def test_verify_domain_token_not_found(self, mock_resolve, db, test_domain):
|
||||
"""Test verification fails when token not found in DNS."""
|
||||
# Mock DNS response with wrong token
|
||||
mock_txt = MagicMock()
|
||||
mock_txt.to_text.return_value = '"wrong_token"'
|
||||
mock_resolve.return_value = [mock_txt]
|
||||
|
||||
with pytest.raises(DomainVerificationFailedException) as exc_info:
|
||||
vendor_domain_service.verify_domain(db, test_domain.id)
|
||||
|
||||
assert "token not found" in str(exc_info.value).lower()
|
||||
|
||||
@patch("dns.resolver.resolve")
|
||||
def test_verify_domain_dns_nxdomain(self, mock_resolve, db, test_domain):
|
||||
"""Test verification fails when DNS record doesn't exist."""
|
||||
import dns.resolver
|
||||
|
||||
mock_resolve.side_effect = dns.resolver.NXDOMAIN()
|
||||
|
||||
with pytest.raises(DomainVerificationFailedException):
|
||||
vendor_domain_service.verify_domain(db, test_domain.id)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# VERIFICATION INSTRUCTIONS TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestVendorDomainServiceInstructions:
|
||||
"""Test suite for verification instructions."""
|
||||
|
||||
def test_get_verification_instructions(self, db, test_domain):
|
||||
"""Test getting verification instructions."""
|
||||
instructions = vendor_domain_service.get_verification_instructions(
|
||||
db, test_domain.id
|
||||
)
|
||||
|
||||
assert instructions["domain"] == test_domain.domain
|
||||
assert instructions["verification_token"] == test_domain.verification_token
|
||||
assert "instructions" in instructions
|
||||
assert "txt_record" in instructions
|
||||
assert instructions["txt_record"]["type"] == "TXT"
|
||||
assert instructions["txt_record"]["name"] == "_wizamart-verify"
|
||||
assert "common_registrars" in instructions
|
||||
|
||||
def test_get_verification_instructions_not_found(self, db):
|
||||
"""Test getting instructions for non-existent domain raises exception."""
|
||||
with pytest.raises(VendorDomainNotFoundException):
|
||||
vendor_domain_service.get_verification_instructions(db, 99999)
|
||||
@@ -1,5 +1,9 @@
|
||||
# tests/unit/services/test_vendor_service.py
|
||||
"""Unit tests for VendorService following the application's exception patterns."""
|
||||
"""Unit tests for VendorService following the application's exception patterns.
|
||||
|
||||
Note: Product catalog operations (add_product_to_catalog, get_products) have been
|
||||
moved to app.modules.catalog.services. See test_product_service.py for those tests.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
|
||||
@@ -12,12 +16,9 @@ from app.modules.tenancy.exceptions import (
|
||||
VendorAlreadyExistsException,
|
||||
VendorNotFoundException,
|
||||
)
|
||||
from app.modules.marketplace.exceptions import MarketplaceProductNotFoundException
|
||||
from app.modules.catalog.exceptions import ProductAlreadyExistsException
|
||||
from app.modules.tenancy.services.vendor_service import VendorService
|
||||
from app.modules.tenancy.models import Company
|
||||
from app.modules.tenancy.models import Vendor
|
||||
from app.modules.catalog.schemas import ProductCreate
|
||||
from app.modules.tenancy.schemas.vendor import VendorCreate
|
||||
|
||||
|
||||
@@ -354,98 +355,8 @@ class TestVendorService:
|
||||
|
||||
assert vendor.is_active is False
|
||||
|
||||
# ==================== add_product_to_catalog Tests ====================
|
||||
|
||||
def test_add_product_to_vendor_success(self, db, test_vendor, unique_product):
|
||||
"""Test successfully adding product to vendor."""
|
||||
from app.modules.marketplace.models import MarketplaceProduct
|
||||
|
||||
# Re-query objects to avoid session issues
|
||||
vendor = db.query(Vendor).filter(Vendor.id == test_vendor.id).first()
|
||||
mp = (
|
||||
db.query(MarketplaceProduct)
|
||||
.filter(MarketplaceProduct.id == unique_product.id)
|
||||
.first()
|
||||
)
|
||||
|
||||
product_data = ProductCreate(
|
||||
marketplace_product_id=mp.id,
|
||||
price=15.99,
|
||||
is_featured=True,
|
||||
)
|
||||
|
||||
product = self.service.add_product_to_catalog(db, vendor, product_data)
|
||||
db.commit()
|
||||
|
||||
assert product is not None
|
||||
assert product.vendor_id == vendor.id
|
||||
assert product.marketplace_product_id == mp.id
|
||||
|
||||
def test_add_product_to_vendor_product_not_found(self, db, test_vendor):
|
||||
"""Test adding non-existent product to vendor fails."""
|
||||
product_data = ProductCreate(
|
||||
marketplace_product_id=99999, # Non-existent ID
|
||||
price=15.99,
|
||||
)
|
||||
|
||||
with pytest.raises(MarketplaceProductNotFoundException) as exc_info:
|
||||
self.service.add_product_to_catalog(db, test_vendor, product_data)
|
||||
|
||||
exception = exc_info.value
|
||||
assert exception.status_code == 404
|
||||
assert exception.error_code == "PRODUCT_NOT_FOUND"
|
||||
|
||||
def test_add_product_to_vendor_already_exists(self, db, test_vendor, test_product):
|
||||
"""Test adding product that's already in vendor fails."""
|
||||
# Re-query to get fresh instances
|
||||
from app.modules.catalog.models import Product
|
||||
|
||||
vendor = db.query(Vendor).filter(Vendor.id == test_vendor.id).first()
|
||||
product = db.query(Product).filter(Product.id == test_product.id).first()
|
||||
|
||||
product_data = ProductCreate(
|
||||
marketplace_product_id=product.marketplace_product_id,
|
||||
price=15.99,
|
||||
)
|
||||
|
||||
with pytest.raises(ProductAlreadyExistsException) as exc_info:
|
||||
self.service.add_product_to_catalog(db, vendor, product_data)
|
||||
|
||||
exception = exc_info.value
|
||||
assert exception.status_code == 409
|
||||
assert exception.error_code == "PRODUCT_ALREADY_EXISTS"
|
||||
|
||||
# ==================== get_products Tests ====================
|
||||
|
||||
def test_get_products_owner_access(self, db, test_user, test_vendor, test_product):
|
||||
"""Test vendor owner can get vendor products."""
|
||||
# Re-query vendor to get fresh instance
|
||||
vendor = db.query(Vendor).filter(Vendor.id == test_vendor.id).first()
|
||||
products, total = self.service.get_products(db, vendor, test_user)
|
||||
|
||||
assert total >= 1
|
||||
assert len(products) >= 1
|
||||
|
||||
def test_get_products_access_denied(self, db, test_user, inactive_vendor):
|
||||
"""Test non-owner cannot access unverified vendor products."""
|
||||
# Re-query vendor to get fresh instance
|
||||
vendor = db.query(Vendor).filter(Vendor.id == inactive_vendor.id).first()
|
||||
with pytest.raises(UnauthorizedVendorAccessException) as exc_info:
|
||||
self.service.get_products(db, vendor, test_user)
|
||||
|
||||
exception = exc_info.value
|
||||
assert exception.status_code == 403
|
||||
assert exception.error_code == "UNAUTHORIZED_VENDOR_ACCESS"
|
||||
|
||||
def test_get_products_with_filters(self, db, test_user, test_vendor, test_product):
|
||||
"""Test getting vendor products with various filters."""
|
||||
# Re-query vendor to get fresh instance
|
||||
vendor = db.query(Vendor).filter(Vendor.id == test_vendor.id).first()
|
||||
# Test active only filter
|
||||
products, total = self.service.get_products(
|
||||
db, vendor, test_user, active_only=True
|
||||
)
|
||||
assert all(p.is_active for p in products)
|
||||
# NOTE: add_product_to_catalog and get_products tests have been moved to
|
||||
# test_product_service.py since those methods are now in the catalog module.
|
||||
|
||||
# ==================== Helper Method Tests ====================
|
||||
|
||||
|
||||
506
tests/unit/services/test_vendor_team_service.py
Normal file
506
tests/unit/services/test_vendor_team_service.py
Normal file
@@ -0,0 +1,506 @@
|
||||
# tests/unit/services/test_vendor_team_service.py
|
||||
"""Unit tests for VendorTeamService."""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.modules.tenancy.exceptions import (
|
||||
CannotRemoveOwnerException,
|
||||
InvalidInvitationTokenException,
|
||||
TeamInvitationAlreadyAcceptedException,
|
||||
TeamMemberAlreadyExistsException,
|
||||
UserNotFoundException,
|
||||
)
|
||||
from app.modules.tenancy.models import Role, User, Vendor, VendorUser, VendorUserType
|
||||
from app.modules.tenancy.services.vendor_team_service import vendor_team_service
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# FIXTURES
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def team_vendor(db, test_company):
|
||||
"""Create a vendor for team tests."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
vendor = Vendor(
|
||||
company_id=test_company.id,
|
||||
vendor_code=f"TEAMVENDOR_{unique_id.upper()}",
|
||||
subdomain=f"teamvendor{unique_id.lower()}",
|
||||
name=f"Team Vendor {unique_id}",
|
||||
is_active=True,
|
||||
is_verified=True,
|
||||
)
|
||||
db.add(vendor)
|
||||
db.commit()
|
||||
db.refresh(vendor)
|
||||
return vendor
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def vendor_owner(db, team_vendor, test_user):
|
||||
"""Create an owner for the team vendor."""
|
||||
vendor_user = VendorUser(
|
||||
vendor_id=team_vendor.id,
|
||||
user_id=test_user.id,
|
||||
user_type=VendorUserType.OWNER.value,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(vendor_user)
|
||||
db.commit()
|
||||
db.refresh(vendor_user)
|
||||
return vendor_user
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def team_member(db, team_vendor, other_user, auth_manager):
|
||||
"""Create a team member for the vendor."""
|
||||
# Create a role first
|
||||
role = Role(
|
||||
vendor_id=team_vendor.id,
|
||||
name="staff",
|
||||
permissions=["orders.view", "products.view"],
|
||||
)
|
||||
db.add(role)
|
||||
db.commit()
|
||||
|
||||
vendor_user = VendorUser(
|
||||
vendor_id=team_vendor.id,
|
||||
user_id=other_user.id,
|
||||
user_type=VendorUserType.TEAM_MEMBER.value,
|
||||
role_id=role.id,
|
||||
is_active=True,
|
||||
invitation_accepted_at=datetime.utcnow(),
|
||||
)
|
||||
db.add(vendor_user)
|
||||
db.commit()
|
||||
db.refresh(vendor_user)
|
||||
return vendor_user
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def pending_invitation(db, team_vendor, test_user, auth_manager):
|
||||
"""Create a pending invitation."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
|
||||
# Create new user for invitation
|
||||
new_user = User(
|
||||
email=f"pending_{unique_id}@example.com",
|
||||
username=f"pending_{unique_id}",
|
||||
hashed_password=auth_manager.hash_password("temppass"),
|
||||
role="vendor",
|
||||
is_active=False,
|
||||
)
|
||||
db.add(new_user)
|
||||
db.commit()
|
||||
|
||||
# Create role
|
||||
role = Role(
|
||||
vendor_id=team_vendor.id,
|
||||
name="support",
|
||||
permissions=["support.view"],
|
||||
)
|
||||
db.add(role)
|
||||
db.commit()
|
||||
|
||||
# Create pending vendor user
|
||||
vendor_user = VendorUser(
|
||||
vendor_id=team_vendor.id,
|
||||
user_id=new_user.id,
|
||||
user_type=VendorUserType.TEAM_MEMBER.value,
|
||||
role_id=role.id,
|
||||
invited_by=test_user.id,
|
||||
invitation_token=f"pending_token_{unique_id}",
|
||||
invitation_sent_at=datetime.utcnow(),
|
||||
is_active=False,
|
||||
)
|
||||
db.add(vendor_user)
|
||||
db.commit()
|
||||
db.refresh(vendor_user)
|
||||
return vendor_user
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def expired_invitation(db, team_vendor, test_user, auth_manager):
|
||||
"""Create an expired invitation."""
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
|
||||
# Create new user for invitation
|
||||
new_user = User(
|
||||
email=f"expired_{unique_id}@example.com",
|
||||
username=f"expired_{unique_id}",
|
||||
hashed_password=auth_manager.hash_password("temppass"),
|
||||
role="vendor",
|
||||
is_active=False,
|
||||
)
|
||||
db.add(new_user)
|
||||
db.commit()
|
||||
|
||||
# Create role
|
||||
role = Role(
|
||||
vendor_id=team_vendor.id,
|
||||
name="viewer",
|
||||
permissions=["read_only"],
|
||||
)
|
||||
db.add(role)
|
||||
db.commit()
|
||||
|
||||
# Create expired vendor user (sent 8 days ago, expires in 7)
|
||||
vendor_user = VendorUser(
|
||||
vendor_id=team_vendor.id,
|
||||
user_id=new_user.id,
|
||||
user_type=VendorUserType.TEAM_MEMBER.value,
|
||||
role_id=role.id,
|
||||
invited_by=test_user.id,
|
||||
invitation_token=f"expired_token_{unique_id}",
|
||||
invitation_sent_at=datetime.utcnow() - timedelta(days=8),
|
||||
is_active=False,
|
||||
)
|
||||
db.add(vendor_user)
|
||||
db.commit()
|
||||
db.refresh(vendor_user)
|
||||
return vendor_user
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# INVITE TEAM MEMBER TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestVendorTeamServiceInvite:
|
||||
"""Test suite for inviting team members."""
|
||||
|
||||
@patch("app.modules.billing.services.subscription_service.SubscriptionService.check_team_limit")
|
||||
def test_invite_new_user(self, mock_check_limit, db, team_vendor, test_user):
|
||||
"""Test inviting a new user to the team."""
|
||||
mock_check_limit.return_value = None # No limit reached
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
email = f"newmember_{unique_id}@example.com"
|
||||
|
||||
result = vendor_team_service.invite_team_member(
|
||||
db=db,
|
||||
vendor=team_vendor,
|
||||
inviter=test_user,
|
||||
email=email,
|
||||
role_name="staff",
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert result is not None
|
||||
assert result["email"] == email
|
||||
assert result["invitation_token"] is not None
|
||||
assert result["role"] == "staff"
|
||||
assert result["existing_user"] is False
|
||||
|
||||
@patch("app.modules.billing.services.subscription_service.SubscriptionService.check_team_limit")
|
||||
def test_invite_existing_user(self, mock_check_limit, db, team_vendor, test_user, other_user):
|
||||
"""Test inviting an existing user to the team."""
|
||||
mock_check_limit.return_value = None
|
||||
|
||||
result = vendor_team_service.invite_team_member(
|
||||
db=db,
|
||||
vendor=team_vendor,
|
||||
inviter=test_user,
|
||||
email=other_user.email,
|
||||
role_name="manager",
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert result is not None
|
||||
assert result["email"] == other_user.email
|
||||
assert result["invitation_token"] is not None
|
||||
|
||||
@patch("app.modules.billing.services.subscription_service.SubscriptionService.check_team_limit")
|
||||
def test_invite_already_member_raises_error(
|
||||
self, mock_check_limit, db, team_vendor, test_user, team_member
|
||||
):
|
||||
"""Test inviting an existing member raises exception."""
|
||||
mock_check_limit.return_value = None
|
||||
|
||||
with pytest.raises(TeamMemberAlreadyExistsException):
|
||||
vendor_team_service.invite_team_member(
|
||||
db=db,
|
||||
vendor=team_vendor,
|
||||
inviter=test_user,
|
||||
email=team_member.user.email,
|
||||
role_name="staff",
|
||||
)
|
||||
|
||||
@patch("app.modules.billing.services.subscription_service.SubscriptionService.check_team_limit")
|
||||
def test_invite_with_custom_permissions(self, mock_check_limit, db, team_vendor, test_user):
|
||||
"""Test inviting with custom permissions."""
|
||||
mock_check_limit.return_value = None
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
email = f"custom_{unique_id}@example.com"
|
||||
custom_perms = ["orders.view", "orders.edit", "products.view"]
|
||||
|
||||
result = vendor_team_service.invite_team_member(
|
||||
db=db,
|
||||
vendor=team_vendor,
|
||||
inviter=test_user,
|
||||
email=email,
|
||||
role_name="custom_role",
|
||||
custom_permissions=custom_perms,
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert result is not None
|
||||
assert result["role"] == "custom_role"
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# ACCEPT INVITATION TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestVendorTeamServiceAccept:
|
||||
"""Test suite for accepting invitations."""
|
||||
|
||||
def test_accept_invitation_success(self, db, pending_invitation):
|
||||
"""Test accepting a valid invitation."""
|
||||
result = vendor_team_service.accept_invitation(
|
||||
db=db,
|
||||
invitation_token=pending_invitation.invitation_token,
|
||||
password="newpassword123",
|
||||
first_name="John",
|
||||
last_name="Doe",
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert result is not None
|
||||
assert result["user"].is_active is True
|
||||
assert result["user"].first_name == "John"
|
||||
assert result["user"].last_name == "Doe"
|
||||
|
||||
def test_accept_invitation_invalid_token(self, db):
|
||||
"""Test accepting with invalid token raises exception."""
|
||||
with pytest.raises(InvalidInvitationTokenException):
|
||||
vendor_team_service.accept_invitation(
|
||||
db=db,
|
||||
invitation_token="invalid_token_12345",
|
||||
password="password123",
|
||||
)
|
||||
|
||||
def test_accept_invitation_already_accepted(self, db, team_member):
|
||||
"""Test accepting an already accepted invitation raises exception."""
|
||||
# team_member already has invitation_accepted_at set
|
||||
with pytest.raises(InvalidInvitationTokenException):
|
||||
vendor_team_service.accept_invitation(
|
||||
db=db,
|
||||
invitation_token="some_token", # team_member has no token
|
||||
password="password123",
|
||||
)
|
||||
|
||||
def test_accept_invitation_expired(self, db, expired_invitation):
|
||||
"""Test accepting an expired invitation raises exception."""
|
||||
with pytest.raises(InvalidInvitationTokenException) as exc_info:
|
||||
vendor_team_service.accept_invitation(
|
||||
db=db,
|
||||
invitation_token=expired_invitation.invitation_token,
|
||||
password="password123",
|
||||
)
|
||||
|
||||
assert "expired" in str(exc_info.value).lower()
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# REMOVE TEAM MEMBER TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestVendorTeamServiceRemove:
|
||||
"""Test suite for removing team members."""
|
||||
|
||||
def test_remove_team_member_success(self, db, team_vendor, team_member):
|
||||
"""Test removing a team member."""
|
||||
result = vendor_team_service.remove_team_member(
|
||||
db=db,
|
||||
vendor=team_vendor,
|
||||
user_id=team_member.user_id,
|
||||
)
|
||||
db.commit()
|
||||
db.refresh(team_member)
|
||||
|
||||
assert result is True
|
||||
assert team_member.is_active is False
|
||||
|
||||
def test_remove_owner_raises_error(self, db, team_vendor, vendor_owner):
|
||||
"""Test removing owner raises exception."""
|
||||
with pytest.raises(CannotRemoveOwnerException):
|
||||
vendor_team_service.remove_team_member(
|
||||
db=db,
|
||||
vendor=team_vendor,
|
||||
user_id=vendor_owner.user_id,
|
||||
)
|
||||
|
||||
def test_remove_nonexistent_user_raises_error(self, db, team_vendor):
|
||||
"""Test removing non-existent user raises exception."""
|
||||
with pytest.raises(UserNotFoundException):
|
||||
vendor_team_service.remove_team_member(
|
||||
db=db,
|
||||
vendor=team_vendor,
|
||||
user_id=99999,
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# UPDATE MEMBER ROLE TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestVendorTeamServiceUpdateRole:
|
||||
"""Test suite for updating member roles."""
|
||||
|
||||
def test_update_role_success(self, db, team_vendor, team_member):
|
||||
"""Test updating a member's role."""
|
||||
result = vendor_team_service.update_member_role(
|
||||
db=db,
|
||||
vendor=team_vendor,
|
||||
user_id=team_member.user_id,
|
||||
new_role_name="manager",
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert result is not None
|
||||
assert result.role.name == "manager"
|
||||
|
||||
def test_update_owner_role_raises_error(self, db, team_vendor, vendor_owner):
|
||||
"""Test updating owner's role raises exception."""
|
||||
with pytest.raises(CannotRemoveOwnerException):
|
||||
vendor_team_service.update_member_role(
|
||||
db=db,
|
||||
vendor=team_vendor,
|
||||
user_id=vendor_owner.user_id,
|
||||
new_role_name="staff",
|
||||
)
|
||||
|
||||
def test_update_role_with_custom_permissions(self, db, team_vendor, team_member):
|
||||
"""Test updating role with custom permissions."""
|
||||
custom_perms = ["orders.view", "orders.edit", "reports.view"]
|
||||
|
||||
result = vendor_team_service.update_member_role(
|
||||
db=db,
|
||||
vendor=team_vendor,
|
||||
user_id=team_member.user_id,
|
||||
new_role_name="analyst",
|
||||
custom_permissions=custom_perms,
|
||||
)
|
||||
db.commit()
|
||||
|
||||
assert result.role.name == "analyst"
|
||||
assert result.role.permissions == custom_perms
|
||||
|
||||
def test_update_nonexistent_user_raises_error(self, db, team_vendor):
|
||||
"""Test updating non-existent user raises exception."""
|
||||
with pytest.raises(UserNotFoundException):
|
||||
vendor_team_service.update_member_role(
|
||||
db=db,
|
||||
vendor=team_vendor,
|
||||
user_id=99999,
|
||||
new_role_name="staff",
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# GET TEAM MEMBERS TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestVendorTeamServiceGetMembers:
|
||||
"""Test suite for getting team members."""
|
||||
|
||||
def test_get_team_members(self, db, team_vendor, vendor_owner, team_member):
|
||||
"""Test getting all team members."""
|
||||
members = vendor_team_service.get_team_members(db, team_vendor)
|
||||
|
||||
assert len(members) >= 2
|
||||
user_ids = [m["id"] for m in members]
|
||||
assert vendor_owner.user_id in user_ids
|
||||
assert team_member.user_id in user_ids
|
||||
|
||||
def test_get_team_members_excludes_inactive(
|
||||
self, db, team_vendor, vendor_owner, team_member
|
||||
):
|
||||
"""Test getting only active team members."""
|
||||
# Deactivate team_member
|
||||
team_member.is_active = False
|
||||
db.commit()
|
||||
|
||||
members = vendor_team_service.get_team_members(
|
||||
db, team_vendor, include_inactive=False
|
||||
)
|
||||
|
||||
user_ids = [m["id"] for m in members]
|
||||
assert vendor_owner.user_id in user_ids
|
||||
assert team_member.user_id not in user_ids
|
||||
|
||||
def test_get_team_members_includes_inactive(
|
||||
self, db, team_vendor, vendor_owner, team_member
|
||||
):
|
||||
"""Test getting all members including inactive."""
|
||||
# Deactivate team_member
|
||||
team_member.is_active = False
|
||||
db.commit()
|
||||
|
||||
members = vendor_team_service.get_team_members(
|
||||
db, team_vendor, include_inactive=True
|
||||
)
|
||||
|
||||
user_ids = [m["id"] for m in members]
|
||||
assert vendor_owner.user_id in user_ids
|
||||
assert team_member.user_id in user_ids
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# GET VENDOR ROLES TESTS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.tenancy
|
||||
class TestVendorTeamServiceGetRoles:
|
||||
"""Test suite for getting vendor roles."""
|
||||
|
||||
def test_get_vendor_roles_existing(self, db, team_vendor, team_member):
|
||||
"""Test getting roles when they exist."""
|
||||
# team_member fixture creates a role
|
||||
roles = vendor_team_service.get_vendor_roles(db, team_vendor.id)
|
||||
|
||||
assert len(roles) >= 1
|
||||
role_names = [r["name"] for r in roles]
|
||||
assert "staff" in role_names
|
||||
|
||||
def test_get_vendor_roles_creates_defaults(self, db, team_vendor):
|
||||
"""Test default roles are created if none exist."""
|
||||
roles = vendor_team_service.get_vendor_roles(db, team_vendor.id)
|
||||
db.commit()
|
||||
|
||||
assert len(roles) >= 5 # Default roles
|
||||
role_names = [r["name"] for r in roles]
|
||||
assert "manager" in role_names
|
||||
assert "staff" in role_names
|
||||
assert "support" in role_names
|
||||
assert "viewer" in role_names
|
||||
assert "marketing" in role_names
|
||||
|
||||
def test_get_vendor_roles_returns_permissions(self, db, team_vendor):
|
||||
"""Test roles include permissions."""
|
||||
roles = vendor_team_service.get_vendor_roles(db, team_vendor.id)
|
||||
|
||||
for role in roles:
|
||||
assert "permissions" in role
|
||||
assert isinstance(role["permissions"], list)
|
||||
Reference in New Issue
Block a user