sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(billing): complete billing module — fix tier change, platform support, merchant portal

Browse Source 
- Fix admin tier change: resolve tier_code→tier_id in update_subscription(),
  delegate to billing_service.change_tier() for Stripe-connected subs
- Add platform support to admin tiers page: platform column, filter dropdown,
  platform selector in create/edit modal, platform_name in tier API response
- Filter used platforms in create subscription modal on merchant detail page
- Enrich merchant portal API responses with tier code, tier_name, platform_name
- Add eager-load of platform relationship in get_merchant_subscription()
- Remove stale store_name/store_code references from merchant templates
- Add merchant tier change endpoint (POST /change-tier) and tier selector UI
  replacing broken requestUpgrade() button
- Fix subscription detail link to use platform_id instead of sub.id

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-02-10 20:49:48 +01:00
parent 0b37274140
commit d1fe3584ff
54 changed files with 222 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
app/modules/tenancy/static/admin/js/merchant-detail.js
View File

@@ -156,11 +156,16 @@ function adminMerchantDetail() {
}
},
// Open create subscription modal
// Open create subscription modal (only show platforms without existing subscriptions)
async openCreateSubscriptionModal() {
const firstPlatformId = this.platforms.length > 0 ? this.platforms[0].id : null;
this.createForm = { platform_id: firstPlatformId, tier_code: 'essential', status: 'trial', trial_days: 14, is_annual: false };
await this.loadTiers(firstPlatformId);
const usedPlatformIds = this.subscriptions.map(e => e.platform_id);
const available = this.platforms.filter(p => !usedPlatformIds.includes(p.id));
if (available.length === 0) {
Utils.showToast('All platforms already have subscriptions', 'info');
return;
}
this.createForm = { platform_id: available[0].id, tier_code: 'essential', status: 'trial', trial_days: 14, is_annual: false };
await this.loadTiers(available[0].id);
this.showCreateSubscriptionModal = true;
},

2
app/modules/tenancy/templates/tenancy/admin/merchant-detail.html
View File

@@ -320,7 +320,7 @@
<select x-model="createForm.platform_id"
@change="onCreatePlatformChange()"
class="w-full px-3 py-2 text-sm border border-gray-300 rounded-lg dark:border-gray-600 dark:bg-gray-700 dark:text-gray-300 focus:outline-none focus:ring-2 focus:ring-purple-500">
<template x-for="p in platforms" :key="p.id">
<template x-for="p in platforms.filter(p => !subscriptions.some(s => s.platform_id === p.id))" :key="p.id">
<option :value="p.id" x-text="p.name"></option>
</template>
</select>

219
app/modules/tenancy/tests/integration/test_merchant_domains_api.py Normal file
View File

@@ -0,0 +1,219 @@
# tests/integration/api/v1/admin/test_merchant_domains.py
"""Integration tests for admin merchant domain management endpoints.
Tests the /api/v1/admin/merchants/{id}/domains/* and
/api/v1/admin/merchants/domains/merchant/* endpoints.
"""
import uuid
from datetime import UTC, datetime
import pytest
from app.modules.tenancy.models.merchant_domain import MerchantDomain
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
class TestAdminMerchantDomainsAPI:
"""Test admin merchant domain management endpoints."""
def test_create_merchant_domain(self, client, admin_headers, test_merchant):
"""Test POST create merchant domain returns 201-equivalent (200 with data)."""
unique_id = str(uuid.uuid4())[:8]
response = client.post(
f"/api/v1/admin/merchants/{test_merchant.id}/domains",
headers=admin_headers,
json={
"domain": f"newmerch{unique_id}.example.com",
"is_primary": True,
},
)
assert response.status_code == 200
data = response.json()
assert data["domain"] == f"newmerch{unique_id}.example.com"
assert data["merchant_id"] == test_merchant.id
assert data["is_primary"] is True
assert data["is_verified"] is False
assert data["is_active"] is False
assert data["verification_token"] is not None
def test_create_duplicate_domain(
self, client, admin_headers, test_merchant, db
):
"""Test POST create duplicate domain returns 409 conflict."""
unique_id = str(uuid.uuid4())[:8]
domain_name = f"dup{unique_id}.example.com"
# Create existing domain
existing = MerchantDomain(
merchant_id=test_merchant.id,
domain=domain_name,
verification_token=f"dup_{unique_id}",
)
db.add(existing)
db.commit()
response = client.post(
f"/api/v1/admin/merchants/{test_merchant.id}/domains",
headers=admin_headers,
json={"domain": domain_name, "is_primary": True},
)
assert response.status_code == 409
data = response.json()
assert data["error_code"] == "MERCHANT_DOMAIN_ALREADY_EXISTS"
def test_create_domain_invalid_format(
self, client, admin_headers, test_merchant
):
"""Test POST create domain with invalid format returns 422."""
response = client.post(
f"/api/v1/admin/merchants/{test_merchant.id}/domains",
headers=admin_headers,
json={"domain": "notadomain", "is_primary": True},
)
assert response.status_code == 422
def test_create_domain_merchant_not_found(self, client, admin_headers):
"""Test POST create domain for non-existent merchant returns 404."""
response = client.post(
"/api/v1/admin/merchants/99999/domains",
headers=admin_headers,
json={"domain": "test.example.com", "is_primary": True},
)
assert response.status_code == 404
def test_list_merchant_domains(
self, client, admin_headers, test_merchant, db
):
"""Test GET list merchant domains returns 200."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"list{unique_id}.example.com",
verification_token=f"list_{unique_id}",
)
db.add(domain)
db.commit()
response = client.get(
f"/api/v1/admin/merchants/{test_merchant.id}/domains",
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
domain_names = [d["domain"] for d in data["domains"]]
assert f"list{unique_id}.example.com" in domain_names
def test_get_domain_detail(self, client, admin_headers, test_merchant, db):
"""Test GET domain detail returns 200."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"detail{unique_id}.example.com",
verification_token=f"det_{unique_id}",
)
db.add(domain)
db.commit()
db.refresh(domain)
response = client.get(
f"/api/v1/admin/merchants/domains/merchant/{domain.id}",
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["id"] == domain.id
assert data["domain"] == f"detail{unique_id}.example.com"
def test_update_domain(self, client, admin_headers, test_merchant, db):
"""Test PUT update domain returns 200."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"update{unique_id}.example.com",
is_verified=True,
verified_at=datetime.now(UTC),
is_active=False,
is_primary=False,
verification_token=f"upd_{unique_id}",
)
db.add(domain)
db.commit()
db.refresh(domain)
response = client.put(
f"/api/v1/admin/merchants/domains/merchant/{domain.id}",
headers=admin_headers,
json={"is_active": True, "is_primary": True},
)
assert response.status_code == 200
data = response.json()
assert data["is_active"] is True
assert data["is_primary"] is True
def test_update_activate_unverified_domain(
self, client, admin_headers, test_merchant, db
):
"""Test PUT activate unverified domain returns 400."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"unver{unique_id}.example.com",
is_verified=False,
is_active=False,
verification_token=f"unv_{unique_id}",
)
db.add(domain)
db.commit()
db.refresh(domain)
response = client.put(
f"/api/v1/admin/merchants/domains/merchant/{domain.id}",
headers=admin_headers,
json={"is_active": True},
)
assert response.status_code == 400
data = response.json()
assert data["error_code"] == "DOMAIN_NOT_VERIFIED"
def test_delete_domain(self, client, admin_headers, test_merchant, db):
"""Test DELETE domain returns 200."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"del{unique_id}.example.com",
verification_token=f"del_{unique_id}",
)
db.add(domain)
db.commit()
db.refresh(domain)
response = client.delete(
f"/api/v1/admin/merchants/domains/merchant/{domain.id}",
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert "deleted" in data["message"].lower()
assert data["merchant_id"] == test_merchant.id
def test_non_admin_access(self, client, auth_headers, test_merchant):
"""Test non-admin access returns 403."""
response = client.get(
f"/api/v1/admin/merchants/{test_merchant.id}/domains",
headers=auth_headers,
)
assert response.status_code == 403

271
app/modules/tenancy/tests/unit/test_admin_platform_model.py Normal file
View File

@@ -0,0 +1,271 @@
# tests/unit/models/database/test_admin_platform.py
"""
Unit tests for AdminPlatform model.
Tests the admin-platform junction table model and its relationships.
"""
import pytest
from sqlalchemy.exc import IntegrityError
from app.modules.tenancy.models import AdminPlatform
@pytest.mark.unit
@pytest.mark.database
@pytest.mark.admin
class TestAdminPlatformModel:
"""Test AdminPlatform model creation and constraints."""
def test_create_admin_platform_assignment(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test creating an admin platform assignment."""
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(assignment)
assert assignment.id is not None
assert assignment.user_id == test_platform_admin.id
assert assignment.platform_id == test_platform.id
assert assignment.is_active is True
assert assignment.assigned_by_user_id == test_super_admin.id
assert assignment.assigned_at is not None
def test_admin_platform_unique_constraint(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test that an admin can only be assigned to a platform once."""
# Create first assignment
assignment1 = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment1)
db.commit()
# Try to create duplicate assignment
assignment2 = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment2)
with pytest.raises(IntegrityError):
db.commit()
def test_admin_platform_cascade_delete_user(
self, db, auth_manager, test_platform, test_super_admin
):
"""Test that deleting user cascades to admin platform assignments."""
from app.modules.tenancy.models import User
# Create a temporary admin
temp_admin = User(
email="temp_admin@example.com",
username="temp_admin",
hashed_password=auth_manager.hash_password("temppass"),
role="admin",
is_active=True,
is_super_admin=False,
)
db.add(temp_admin)
db.flush()
# Create assignment
assignment = AdminPlatform(
user_id=temp_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
assignment_id = assignment.id
# Delete user - should cascade to assignment
db.delete(temp_admin)
db.commit()
# Verify assignment is gone
remaining = db.query(AdminPlatform).filter(AdminPlatform.id == assignment_id).first()
assert remaining is None
def test_admin_platform_relationships(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test AdminPlatform relationships are loaded correctly."""
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(assignment)
# Test relationships
assert assignment.user is not None
assert assignment.user.id == test_platform_admin.id
assert assignment.platform is not None
assert assignment.platform.id == test_platform.id
assert assignment.assigned_by is not None
assert assignment.assigned_by.id == test_super_admin.id
def test_admin_platform_properties(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test AdminPlatform computed properties."""
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(assignment)
# Test properties
assert assignment.platform_code == test_platform.code
assert assignment.platform_name == test_platform.name
def test_admin_platform_repr(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test AdminPlatform string representation."""
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(assignment)
repr_str = repr(assignment)
assert "AdminPlatform" in repr_str
assert str(test_platform_admin.id) in repr_str
assert str(test_platform.id) in repr_str
@pytest.mark.unit
@pytest.mark.database
@pytest.mark.admin
class TestUserAdminMethods:
"""Test User model admin-related methods."""
def test_is_super_admin_user_true(self, db, test_super_admin):
"""Test is_super_admin_user property for super admin."""
assert test_super_admin.is_super_admin_user is True
def test_is_super_admin_user_false_for_platform_admin(self, db, test_platform_admin):
"""Test is_super_admin_user property for platform admin."""
assert test_platform_admin.is_super_admin_user is False
def test_is_platform_admin_true(self, db, test_platform_admin):
"""Test is_platform_admin property for platform admin."""
assert test_platform_admin.is_platform_admin is True
def test_is_platform_admin_false_for_super_admin(self, db, test_super_admin):
"""Test is_platform_admin property for super admin."""
assert test_super_admin.is_platform_admin is False
def test_can_access_platform_super_admin(self, db, test_super_admin, test_platform):
"""Test that super admin can access any platform."""
assert test_super_admin.can_access_platform(test_platform.id) is True
def test_can_access_platform_assigned(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test that platform admin can access assigned platform."""
# Create assignment
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(test_platform_admin)
assert test_platform_admin.can_access_platform(test_platform.id) is True
def test_can_access_platform_not_assigned(
self, db, test_platform_admin, test_platform
):
"""Test that platform admin cannot access unassigned platform."""
# No assignment created
assert test_platform_admin.can_access_platform(test_platform.id) is False
def test_can_access_platform_inactive_assignment(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test that platform admin cannot access platform with inactive assignment."""
# Create inactive assignment
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=False, # Inactive
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(test_platform_admin)
assert test_platform_admin.can_access_platform(test_platform.id) is False
def test_get_accessible_platform_ids_super_admin(self, db, test_super_admin):
"""Test get_accessible_platform_ids returns None for super admin."""
result = test_super_admin.get_accessible_platform_ids()
assert result is None # None means all platforms
def test_get_accessible_platform_ids_platform_admin(
self, db, test_platform_admin, test_platform, another_platform, test_super_admin
):
"""Test get_accessible_platform_ids returns correct list for platform admin."""
# Create assignments for both platforms
assignment1 = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
assignment2 = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=another_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add_all([assignment1, assignment2])
db.commit()
db.refresh(test_platform_admin)
result = test_platform_admin.get_accessible_platform_ids()
assert len(result) == 2
assert test_platform.id in result
assert another_platform.id in result
def test_get_accessible_platform_ids_no_assignments(self, db, test_platform_admin):
"""Test get_accessible_platform_ids returns empty list when no assignments."""
result = test_platform_admin.get_accessible_platform_ids()
assert result == []
def test_get_accessible_platform_ids_store_user(self, db, test_store_user):
"""Test get_accessible_platform_ids returns empty list for non-admin."""
result = test_store_user.get_accessible_platform_ids()
assert result == []

464
app/modules/tenancy/tests/unit/test_admin_platform_service.py Normal file
View File

@@ -0,0 +1,464 @@
# tests/unit/services/test_admin_platform_service.py
"""
Unit tests for AdminPlatformService.
Tests the admin platform assignment service operations.
"""
import pytest
from app.exceptions import ValidationException
from app.modules.tenancy.exceptions import AdminOperationException, CannotModifySelfException
from app.modules.tenancy.services.admin_platform_service import AdminPlatformService
@pytest.mark.unit
@pytest.mark.admin
class TestAdminPlatformServiceAssign:
"""Test AdminPlatformService.assign_admin_to_platform."""
def test_assign_admin_to_platform_success(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test successfully assigning an admin to a platform."""
service = AdminPlatformService()
assignment = service.assign_admin_to_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
assert assignment is not None
assert assignment.user_id == test_platform_admin.id
assert assignment.platform_id == test_platform.id
assert assignment.is_active is True
assert assignment.assigned_by_user_id == test_super_admin.id
def test_assign_admin_user_not_found(self, db, test_platform, test_super_admin):
"""Test assigning non-existent user raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.assign_admin_to_platform(
db=db,
admin_user_id=99999,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
assert "User not found" in str(exc.value)
def test_assign_admin_not_admin_role(
self, db, test_store_user, test_platform, test_super_admin
):
"""Test assigning non-admin user raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.assign_admin_to_platform(
db=db,
admin_user_id=test_store_user.id,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
assert "must be an admin" in str(exc.value)
def test_assign_super_admin_raises_error(
self, db, test_super_admin, test_platform
):
"""Test assigning super admin raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.assign_admin_to_platform(
db=db,
admin_user_id=test_super_admin.id,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
assert "Super admins don't need platform assignments" in str(exc.value)
def test_assign_platform_not_found(
self, db, test_platform_admin, test_super_admin
):
"""Test assigning to non-existent platform raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.assign_admin_to_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=99999,
assigned_by_user_id=test_super_admin.id,
)
assert "Platform not found" in str(exc.value)
def test_assign_admin_already_assigned(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test assigning already assigned admin raises error."""
service = AdminPlatformService()
# First assignment
service.assign_admin_to_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
db.commit()
# Try to assign again
with pytest.raises(AdminOperationException) as exc:
service.assign_admin_to_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
assert "already assigned" in str(exc.value)
def test_reactivate_inactive_assignment(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test reactivating an inactive assignment."""
from app.modules.tenancy.models import AdminPlatform
service = AdminPlatformService()
# Create inactive assignment directly
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=False,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
# Assign again - should reactivate
result = service.assign_admin_to_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
assert result.is_active is True
@pytest.mark.unit
@pytest.mark.admin
class TestAdminPlatformServiceRemove:
"""Test AdminPlatformService.remove_admin_from_platform."""
def test_remove_admin_from_platform_success(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test successfully removing an admin from a platform."""
from app.modules.tenancy.models import AdminPlatform
service = AdminPlatformService()
# Create assignment first
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
# Remove
service.remove_admin_from_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=test_platform.id,
removed_by_user_id=test_super_admin.id,
)
db.commit()
db.refresh(assignment)
assert assignment.is_active is False
def test_remove_admin_not_assigned(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test removing non-existent assignment raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.remove_admin_from_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=test_platform.id,
removed_by_user_id=test_super_admin.id,
)
assert "not assigned" in str(exc.value)
@pytest.mark.unit
@pytest.mark.admin
class TestAdminPlatformServiceQueries:
"""Test AdminPlatformService query methods."""
def test_get_platforms_for_admin(
self, db, test_platform_admin, test_platform, another_platform, test_super_admin
):
"""Test getting platforms for an admin."""
from app.modules.tenancy.models import AdminPlatform
service = AdminPlatformService()
# Create assignments
for platform in [test_platform, another_platform]:
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
platforms = service.get_platforms_for_admin(db, test_platform_admin.id)
assert len(platforms) == 2
platform_ids = [p.id for p in platforms]
assert test_platform.id in platform_ids
assert another_platform.id in platform_ids
def test_get_platforms_for_admin_no_assignments(self, db, test_platform_admin):
"""Test getting platforms when no assignments exist."""
service = AdminPlatformService()
platforms = service.get_platforms_for_admin(db, test_platform_admin.id)
assert platforms == []
def test_get_admins_for_platform(
self, db, test_platform_admin, test_platform, test_super_admin, auth_manager
):
"""Test getting admins for a platform."""
from app.modules.tenancy.models import AdminPlatform
from app.modules.tenancy.models import User
service = AdminPlatformService()
# Create another platform admin
another_admin = User(
email="another_padmin@example.com",
username="another_padmin",
hashed_password=auth_manager.hash_password("pass"),
role="admin",
is_active=True,
is_super_admin=False,
)
db.add(another_admin)
db.flush()
# Create assignments for both admins
for admin in [test_platform_admin, another_admin]:
assignment = AdminPlatform(
user_id=admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
admins = service.get_admins_for_platform(db, test_platform.id)
assert len(admins) == 2
admin_ids = [a.id for a in admins]
assert test_platform_admin.id in admin_ids
assert another_admin.id in admin_ids
def test_get_admin_assignments(
self, db, test_platform_admin, test_platform, another_platform, test_super_admin
):
"""Test getting admin assignments with platform details."""
from app.modules.tenancy.models import AdminPlatform
service = AdminPlatformService()
# Create assignments
for platform in [test_platform, another_platform]:
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
assignments = service.get_admin_assignments(db, test_platform_admin.id)
assert len(assignments) == 2
# Verify platform relationship is loaded
for assignment in assignments:
assert assignment.platform is not None
assert assignment.platform.code is not None
@pytest.mark.unit
@pytest.mark.admin
class TestAdminPlatformServiceSuperAdmin:
"""Test AdminPlatformService super admin operations."""
def test_toggle_super_admin_promote(
self, db, test_platform_admin, test_super_admin
):
"""Test promoting admin to super admin."""
service = AdminPlatformService()
result = service.toggle_super_admin(
db=db,
user_id=test_platform_admin.id,
is_super_admin=True,
current_admin_id=test_super_admin.id,
)
db.commit()
assert result.is_super_admin is True
def test_toggle_super_admin_demote(
self, db, test_super_admin, auth_manager
):
"""Test demoting super admin to platform admin."""
from app.modules.tenancy.models import User
service = AdminPlatformService()
# Create another super admin to demote
another_super = User(
email="another_super@example.com",
username="another_super",
hashed_password=auth_manager.hash_password("pass"),
role="admin",
is_active=True,
is_super_admin=True,
)
db.add(another_super)
db.commit()
result = service.toggle_super_admin(
db=db,
user_id=another_super.id,
is_super_admin=False,
current_admin_id=test_super_admin.id,
)
db.commit()
assert result.is_super_admin is False
def test_toggle_super_admin_cannot_demote_self(self, db, test_super_admin):
"""Test that super admin cannot demote themselves."""
service = AdminPlatformService()
with pytest.raises(CannotModifySelfException):
service.toggle_super_admin(
db=db,
user_id=test_super_admin.id,
is_super_admin=False,
current_admin_id=test_super_admin.id,
)
def test_toggle_super_admin_user_not_found(self, db, test_super_admin):
"""Test toggling non-existent user raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.toggle_super_admin(
db=db,
user_id=99999,
is_super_admin=True,
current_admin_id=test_super_admin.id,
)
assert "User not found" in str(exc.value)
def test_toggle_super_admin_not_admin(
self, db, test_store_user, test_super_admin
):
"""Test toggling non-admin user raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.toggle_super_admin(
db=db,
user_id=test_store_user.id,
is_super_admin=True,
current_admin_id=test_super_admin.id,
)
assert "must be an admin" in str(exc.value)
@pytest.mark.unit
@pytest.mark.admin
class TestAdminPlatformServiceCreatePlatformAdmin:
"""Test AdminPlatformService.create_platform_admin."""
def test_create_platform_admin_success(
self, db, test_platform, another_platform, test_super_admin
):
"""Test creating a new platform admin with assignments."""
service = AdminPlatformService()
user, assignments = service.create_platform_admin(
db=db,
email="new_padmin@example.com",
username="new_padmin",
password="securepass123",
platform_ids=[test_platform.id, another_platform.id],
created_by_user_id=test_super_admin.id,
first_name="New",
last_name="Admin",
)
db.commit()
assert user is not None
assert user.email == "new_padmin@example.com"
assert user.username == "new_padmin"
assert user.role == "admin"
assert user.is_super_admin is False
assert user.first_name == "New"
assert user.last_name == "Admin"
assert len(assignments) == 2
def test_create_platform_admin_duplicate_email(
self, db, test_platform, test_super_admin, test_platform_admin
):
"""Test creating platform admin with duplicate email fails."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.create_platform_admin(
db=db,
email=test_platform_admin.email, # Duplicate
username="unique_username",
password="securepass123",
platform_ids=[test_platform.id],
created_by_user_id=test_super_admin.id,
)
assert "Email already exists" in str(exc.value)
def test_create_platform_admin_duplicate_username(
self, db, test_platform, test_super_admin, test_platform_admin
):
"""Test creating platform admin with duplicate username fails."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.create_platform_admin(
db=db,
email="unique@example.com",
username=test_platform_admin.username, # Duplicate
password="securepass123",
platform_ids=[test_platform.id],
created_by_user_id=test_super_admin.id,
)
assert "Username already exists" in str(exc.value)

275
app/modules/tenancy/tests/unit/test_merchant_domain.py Normal file
View File

@@ -0,0 +1,275 @@
# tests/unit/models/test_merchant_domain.py
"""Unit tests for MerchantDomain model and related model properties."""
import uuid
from datetime import UTC, datetime
import pytest
from app.modules.tenancy.models.merchant_domain import MerchantDomain
from app.modules.tenancy.models.store_domain import StoreDomain
# =============================================================================
# MODEL TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestMerchantDomainModel:
"""Test suite for MerchantDomain model."""
def test_create_merchant_domain(self, db, test_merchant):
"""Test creating a MerchantDomain with required fields."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"test{unique_id}.example.com",
is_primary=True,
verification_token=f"token_{unique_id}",
)
db.add(domain)
db.commit()
db.refresh(domain)
assert domain.id is not None
assert domain.merchant_id == test_merchant.id
assert domain.is_primary is True
assert domain.is_active is True # default
assert domain.is_verified is False # default
assert domain.ssl_status == "pending" # default
assert domain.verified_at is None
assert domain.platform_id is None
def test_merchant_domain_defaults(self, db, test_merchant):
"""Test default values for MerchantDomain."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"defaults{unique_id}.example.com",
verification_token=f"dtoken_{unique_id}",
)
db.add(domain)
db.commit()
db.refresh(domain)
assert domain.is_primary is True
assert domain.is_active is True
assert domain.is_verified is False
assert domain.ssl_status == "pending"
def test_merchant_domain_repr(self, db, test_merchant):
"""Test string representation."""
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain="repr.example.com",
)
assert "repr.example.com" in repr(domain)
assert str(test_merchant.id) in repr(domain)
def test_merchant_domain_full_url(self):
"""Test full_url property."""
domain = MerchantDomain(domain="test.example.com")
assert domain.full_url == "https://test.example.com"
def test_normalize_domain_removes_protocol(self):
"""Test normalize_domain strips protocols."""
assert MerchantDomain.normalize_domain("https://example.com") == "example.com"
assert MerchantDomain.normalize_domain("http://example.com") == "example.com"
def test_normalize_domain_removes_trailing_slash(self):
"""Test normalize_domain strips trailing slashes."""
assert MerchantDomain.normalize_domain("example.com/") == "example.com"
def test_normalize_domain_lowercases(self):
"""Test normalize_domain converts to lowercase."""
assert MerchantDomain.normalize_domain("EXAMPLE.COM") == "example.com"
def test_unique_domain_constraint(self, db, test_merchant):
"""Test that domain must be unique across all merchant domains."""
unique_id = str(uuid.uuid4())[:8]
domain_name = f"unique{unique_id}.example.com"
domain1 = MerchantDomain(
merchant_id=test_merchant.id,
domain=domain_name,
verification_token=f"t1_{unique_id}",
)
db.add(domain1)
db.commit()
domain2 = MerchantDomain(
merchant_id=test_merchant.id,
domain=domain_name,
verification_token=f"t2_{unique_id}",
)
db.add(domain2)
with pytest.raises(Exception): # IntegrityError
db.commit()
db.rollback()
# =============================================================================
# MERCHANT.primary_domain PROPERTY TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestMerchantPrimaryDomain:
"""Test Merchant.primary_domain property."""
def test_primary_domain_returns_active_verified_primary(self, db, test_merchant):
"""Test primary_domain returns domain when active, verified, and primary."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.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"pt_{unique_id}",
)
db.add(domain)
db.commit()
db.refresh(test_merchant)
assert test_merchant.primary_domain == f"primary{unique_id}.example.com"
def test_primary_domain_returns_none_when_no_domains(self, db, test_merchant):
"""Test primary_domain returns None when merchant has no domains."""
db.refresh(test_merchant)
# Fresh merchant without any domains added in this test
# Need to check if it may have domains from other fixtures
# Just verify the property works without error
result = test_merchant.primary_domain
assert result is None or isinstance(result, str)
def test_primary_domain_returns_none_when_inactive(self, db, test_merchant):
"""Test primary_domain returns None when domain is inactive."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"inactive{unique_id}.example.com",
is_primary=True,
is_active=False,
is_verified=True,
verified_at=datetime.now(UTC),
verification_token=f"it_{unique_id}",
)
db.add(domain)
db.commit()
db.refresh(test_merchant)
assert test_merchant.primary_domain is None
def test_primary_domain_returns_none_when_unverified(self, db, test_merchant):
"""Test primary_domain returns None when domain is unverified."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"unverified{unique_id}.example.com",
is_primary=True,
is_active=True,
is_verified=False,
verification_token=f"ut_{unique_id}",
)
db.add(domain)
db.commit()
db.refresh(test_merchant)
assert test_merchant.primary_domain is None
# =============================================================================
# STORE.effective_domain PROPERTY TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestStoreEffectiveDomain:
"""Test Store.effective_domain inheritance chain."""
def test_effective_domain_returns_store_domain_when_present(self, db, test_store):
"""Test effective_domain returns store's own custom domain (highest priority)."""
unique_id = str(uuid.uuid4())[:8]
store_domain = StoreDomain(
store_id=test_store.id,
domain=f"storeover{unique_id}.example.com",
is_primary=True,
is_active=True,
is_verified=True,
verified_at=datetime.now(UTC),
verification_token=f"sd_{unique_id}",
)
db.add(store_domain)
db.commit()
db.refresh(test_store)
assert test_store.effective_domain == f"storeover{unique_id}.example.com"
def test_effective_domain_returns_merchant_domain_when_no_store_domain(
self, db, test_store, test_merchant
):
"""Test effective_domain returns merchant domain when no store domain."""
unique_id = str(uuid.uuid4())[:8]
merchant_domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"merchant{unique_id}.example.com",
is_primary=True,
is_active=True,
is_verified=True,
verified_at=datetime.now(UTC),
verification_token=f"md_{unique_id}",
)
db.add(merchant_domain)
db.commit()
db.refresh(test_store)
db.refresh(test_merchant)
assert test_store.effective_domain == f"merchant{unique_id}.example.com"
def test_effective_domain_returns_subdomain_fallback(self, db, test_store):
"""Test effective_domain returns subdomain fallback when no custom domains."""
db.refresh(test_store)
# With no store or merchant domains, should fall back to subdomain
result = test_store.effective_domain
assert test_store.subdomain in result
def test_effective_domain_store_domain_overrides_merchant_domain(
self, db, test_store, test_merchant
):
"""Test that store domain takes priority over merchant domain."""
unique_id = str(uuid.uuid4())[:8]
# Add merchant domain
merchant_domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"merchantpri{unique_id}.example.com",
is_primary=True,
is_active=True,
is_verified=True,
verified_at=datetime.now(UTC),
verification_token=f"mpri_{unique_id}",
)
db.add(merchant_domain)
# Add store domain (should take priority)
store_domain = StoreDomain(
store_id=test_store.id,
domain=f"storepri{unique_id}.example.com",
is_primary=True,
is_active=True,
is_verified=True,
verified_at=datetime.now(UTC),
verification_token=f"spri_{unique_id}",
)
db.add(store_domain)
db.commit()
db.refresh(test_store)
db.refresh(test_merchant)
assert test_store.effective_domain == f"storepri{unique_id}.example.com"

526
app/modules/tenancy/tests/unit/test_merchant_domain_service.py Normal file
View File

@@ -0,0 +1,526 @@
# tests/unit/services/test_merchant_domain_service.py
"""Unit tests for MerchantDomainService."""
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,
MerchantDomainAlreadyExistsException,
MerchantDomainNotFoundException,
MerchantNotFoundException,
)
from app.modules.tenancy.models.merchant_domain import MerchantDomain
from app.modules.tenancy.models.store_domain import StoreDomain
from app.modules.tenancy.schemas.merchant_domain import (
MerchantDomainCreate,
MerchantDomainUpdate,
)
from app.modules.tenancy.services.merchant_domain_service import (
merchant_domain_service,
)
# =============================================================================
# ADD DOMAIN TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestMerchantDomainServiceAdd:
"""Test suite for adding merchant domains."""
def test_add_domain_success(self, db, test_merchant):
"""Test successfully adding a domain to a merchant."""
unique_id = str(uuid.uuid4())[:8]
domain_data = MerchantDomainCreate(
domain=f"newmerchant{unique_id}.example.com",
is_primary=True,
)
result = merchant_domain_service.add_domain(
db, test_merchant.id, domain_data
)
db.commit()
assert result is not None
assert result.merchant_id == test_merchant.id
assert result.domain == f"newmerchant{unique_id}.example.com"
assert result.is_primary is True
assert result.is_verified is False
assert result.is_active is False
assert result.verification_token is not None
def test_add_domain_merchant_not_found(self, db):
"""Test adding domain to non-existent merchant raises exception."""
domain_data = MerchantDomainCreate(
domain="test.example.com",
is_primary=True,
)
with pytest.raises(MerchantNotFoundException):
merchant_domain_service.add_domain(db, 99999, domain_data)
def test_add_domain_already_exists_as_merchant_domain(
self, db, test_merchant
):
"""Test adding a domain that already exists as MerchantDomain raises exception."""
unique_id = str(uuid.uuid4())[:8]
existing = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"existing{unique_id}.example.com",
verification_token=f"ext_{unique_id}",
)
db.add(existing)
db.commit()
domain_data = MerchantDomainCreate(
domain=f"existing{unique_id}.example.com",
is_primary=True,
)
with pytest.raises(MerchantDomainAlreadyExistsException):
merchant_domain_service.add_domain(
db, test_merchant.id, domain_data
)
def test_add_domain_already_exists_as_store_domain(
self, db, test_merchant, test_store
):
"""Test adding a domain that already exists as StoreDomain raises exception."""
unique_id = str(uuid.uuid4())[:8]
sd = StoreDomain(
store_id=test_store.id,
domain=f"storeexist{unique_id}.example.com",
verification_token=f"se_{unique_id}",
)
db.add(sd)
db.commit()
domain_data = MerchantDomainCreate(
domain=f"storeexist{unique_id}.example.com",
is_primary=True,
)
with pytest.raises(MerchantDomainAlreadyExistsException):
merchant_domain_service.add_domain(
db, test_merchant.id, domain_data
)
def test_add_domain_max_limit_reached(self, db, test_merchant):
"""Test adding domain when max limit reached raises exception."""
for i in range(merchant_domain_service.max_domains_per_merchant):
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"limit{i}_{uuid.uuid4().hex[:6]}.example.com",
verification_token=f"lim_{i}_{uuid.uuid4().hex[:6]}",
)
db.add(domain)
db.commit()
domain_data = MerchantDomainCreate(
domain="onemore.example.com",
is_primary=True,
)
with pytest.raises(MaxDomainsReachedException):
merchant_domain_service.add_domain(
db, test_merchant.id, domain_data
)
def test_add_domain_reserved_subdomain(self):
"""Test adding a domain with reserved subdomain is rejected by schema."""
with pytest.raises(ValidationError) as exc_info:
MerchantDomainCreate(
domain="admin.example.com",
is_primary=True,
)
assert "reserved subdomain" in str(exc_info.value).lower()
def test_add_domain_sets_primary_unsets_others(self, db, test_merchant):
"""Test adding a primary domain unsets other primary domains."""
unique_id = str(uuid.uuid4())[:8]
first = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"first{unique_id}.example.com",
is_primary=True,
verification_token=f"f_{unique_id}",
)
db.add(first)
db.commit()
domain_data = MerchantDomainCreate(
domain=f"second{unique_id}.example.com",
is_primary=True,
)
result = merchant_domain_service.add_domain(
db, test_merchant.id, domain_data
)
db.commit()
db.refresh(first)
assert result.is_primary is True
assert first.is_primary is False
# =============================================================================
# GET DOMAINS TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestMerchantDomainServiceGet:
"""Test suite for getting merchant domains."""
def test_get_merchant_domains_success(self, db, test_merchant):
"""Test getting all domains for a merchant."""
unique_id = str(uuid.uuid4())[:8]
d1 = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"get1{unique_id}.example.com",
verification_token=f"g1_{unique_id}",
)
d2 = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"get2{unique_id}.example.com",
verification_token=f"g2_{unique_id}",
is_primary=False,
)
db.add_all([d1, d2])
db.commit()
domains = merchant_domain_service.get_merchant_domains(
db, test_merchant.id
)
domain_names = [d.domain for d in domains]
assert f"get1{unique_id}.example.com" in domain_names
assert f"get2{unique_id}.example.com" in domain_names
def test_get_merchant_domains_merchant_not_found(self, db):
"""Test getting domains for non-existent merchant raises exception."""
with pytest.raises(MerchantNotFoundException):
merchant_domain_service.get_merchant_domains(db, 99999)
def test_get_domain_by_id_success(self, db, test_merchant):
"""Test getting a domain by ID."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"byid{unique_id}.example.com",
verification_token=f"bi_{unique_id}",
)
db.add(domain)
db.commit()
db.refresh(domain)
result = merchant_domain_service.get_domain_by_id(db, domain.id)
assert result.id == domain.id
def test_get_domain_by_id_not_found(self, db):
"""Test getting non-existent domain raises exception."""
with pytest.raises(MerchantDomainNotFoundException):
merchant_domain_service.get_domain_by_id(db, 99999)
# =============================================================================
# UPDATE DOMAIN TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestMerchantDomainServiceUpdate:
"""Test suite for updating merchant domains."""
def test_update_domain_set_primary(self, db, test_merchant):
"""Test setting a domain as primary."""
unique_id = str(uuid.uuid4())[:8]
d1 = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"upd1{unique_id}.example.com",
is_primary=True,
is_verified=True,
verified_at=datetime.now(UTC),
verification_token=f"u1_{unique_id}",
)
d2 = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"upd2{unique_id}.example.com",
is_primary=False,
is_verified=True,
verified_at=datetime.now(UTC),
verification_token=f"u2_{unique_id}",
)
db.add_all([d1, d2])
db.commit()
update_data = MerchantDomainUpdate(is_primary=True)
result = merchant_domain_service.update_domain(
db, d2.id, update_data
)
db.commit()
db.refresh(d1)
assert result.is_primary is True
assert d1.is_primary is False
def test_update_domain_activate_unverified_fails(self, db, test_merchant):
"""Test activating an unverified domain raises exception."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"unvact{unique_id}.example.com",
is_verified=False,
is_active=False,
verification_token=f"ua_{unique_id}",
)
db.add(domain)
db.commit()
update_data = MerchantDomainUpdate(is_active=True)
with pytest.raises(DomainNotVerifiedException):
merchant_domain_service.update_domain(db, domain.id, update_data)
def test_update_domain_activate_verified(self, db, test_merchant):
"""Test activating a verified domain succeeds."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"veract{unique_id}.example.com",
is_verified=True,
verified_at=datetime.now(UTC),
is_active=False,
verification_token=f"va_{unique_id}",
)
db.add(domain)
db.commit()
update_data = MerchantDomainUpdate(is_active=True)
result = merchant_domain_service.update_domain(
db, domain.id, update_data
)
db.commit()
assert result.is_active is True
def test_update_domain_deactivate(self, db, test_merchant):
"""Test deactivating a domain."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"deact{unique_id}.example.com",
is_verified=True,
verified_at=datetime.now(UTC),
is_active=True,
verification_token=f"da_{unique_id}",
)
db.add(domain)
db.commit()
update_data = MerchantDomainUpdate(is_active=False)
result = merchant_domain_service.update_domain(
db, domain.id, update_data
)
db.commit()
assert result.is_active is False
def test_update_domain_not_found(self, db):
"""Test updating non-existent domain raises exception."""
update_data = MerchantDomainUpdate(is_primary=True)
with pytest.raises(MerchantDomainNotFoundException):
merchant_domain_service.update_domain(db, 99999, update_data)
# =============================================================================
# DELETE DOMAIN TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestMerchantDomainServiceDelete:
"""Test suite for deleting merchant domains."""
def test_delete_domain_success(self, db, test_merchant):
"""Test successfully deleting a domain."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"todel{unique_id}.example.com",
verification_token=f"td_{unique_id}",
)
db.add(domain)
db.commit()
domain_id = domain.id
result = merchant_domain_service.delete_domain(db, domain_id)
db.commit()
assert "deleted successfully" in result
with pytest.raises(MerchantDomainNotFoundException):
merchant_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(MerchantDomainNotFoundException):
merchant_domain_service.delete_domain(db, 99999)
# =============================================================================
# VERIFY DOMAIN TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestMerchantDomainServiceVerify:
"""Test suite for merchant domain verification."""
@patch("dns.resolver.resolve")
def test_verify_domain_success(self, mock_resolve, db, test_merchant):
"""Test successful domain verification."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"verify{unique_id}.example.com",
is_verified=False,
verification_token=f"vt_{unique_id}",
)
db.add(domain)
db.commit()
mock_txt = MagicMock()
mock_txt.to_text.return_value = f'"vt_{unique_id}"'
mock_resolve.return_value = [mock_txt]
result_domain, message = merchant_domain_service.verify_domain(
db, domain.id
)
db.commit()
assert result_domain.is_verified is True
assert result_domain.verified_at is not None
assert "verified successfully" in message.lower()
def test_verify_domain_already_verified(self, db, test_merchant):
"""Test verifying already verified domain raises exception."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"alrver{unique_id}.example.com",
is_verified=True,
verified_at=datetime.now(UTC),
verification_token=f"av_{unique_id}",
)
db.add(domain)
db.commit()
with pytest.raises(DomainAlreadyVerifiedException):
merchant_domain_service.verify_domain(db, domain.id)
def test_verify_domain_not_found(self, db):
"""Test verifying non-existent domain raises exception."""
with pytest.raises(MerchantDomainNotFoundException):
merchant_domain_service.verify_domain(db, 99999)
@patch("dns.resolver.resolve")
def test_verify_domain_token_not_found(
self, mock_resolve, db, test_merchant
):
"""Test verification fails when token not found in DNS."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"tnf{unique_id}.example.com",
is_verified=False,
verification_token=f"tnf_{unique_id}",
)
db.add(domain)
db.commit()
mock_txt = MagicMock()
mock_txt.to_text.return_value = '"wrong_token"'
mock_resolve.return_value = [mock_txt]
with pytest.raises(DomainVerificationFailedException) as exc_info:
merchant_domain_service.verify_domain(db, 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_merchant
):
"""Test verification fails when DNS record doesn't exist."""
import dns.resolver
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"nxdom{unique_id}.example.com",
is_verified=False,
verification_token=f"nx_{unique_id}",
)
db.add(domain)
db.commit()
mock_resolve.side_effect = dns.resolver.NXDOMAIN()
with pytest.raises(DomainVerificationFailedException):
merchant_domain_service.verify_domain(db, domain.id)
# =============================================================================
# VERIFICATION INSTRUCTIONS TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestMerchantDomainServiceInstructions:
"""Test suite for verification instructions."""
def test_get_verification_instructions(self, db, test_merchant):
"""Test getting verification instructions."""
unique_id = str(uuid.uuid4())[:8]
domain = MerchantDomain(
merchant_id=test_merchant.id,
domain=f"instr{unique_id}.example.com",
verification_token=f"inst_{unique_id}",
)
db.add(domain)
db.commit()
instructions = merchant_domain_service.get_verification_instructions(
db, domain.id
)
assert instructions["domain"] == f"instr{unique_id}.example.com"
assert instructions["verification_token"] == f"inst_{unique_id}"
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(MerchantDomainNotFoundException):
merchant_domain_service.get_verification_instructions(db, 99999)

420
app/modules/tenancy/tests/unit/test_store_domain_service.py Normal file
View File

@@ -0,0 +1,420 @@
# tests/unit/services/test_store_domain_service.py
"""Unit tests for StoreDomainService."""
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,
StoreDomainAlreadyExistsException,
StoreDomainNotFoundException,
StoreNotFoundException,
)
from app.modules.tenancy.models import StoreDomain
from app.modules.tenancy.schemas.store_domain import StoreDomainCreate, StoreDomainUpdate
from app.modules.tenancy.services.store_domain_service import store_domain_service
# =============================================================================
# FIXTURES
# =============================================================================
@pytest.fixture
def test_domain(db, test_store):
"""Create a test domain for a store."""
unique_id = str(uuid.uuid4())[:8]
domain = StoreDomain(
store_id=test_store.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_store):
"""Create a verified domain for a store."""
unique_id = str(uuid.uuid4())[:8]
domain = StoreDomain(
store_id=test_store.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_store):
"""Create a primary domain for a store."""
unique_id = str(uuid.uuid4())[:8]
domain = StoreDomain(
store_id=test_store.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 TestStoreDomainServiceAdd:
"""Test suite for adding store domains."""
def test_add_domain_success(self, db, test_store):
"""Test successfully adding a domain to a store."""
unique_id = str(uuid.uuid4())[:8]
domain_data = StoreDomainCreate(
domain=f"newdomain{unique_id}.example.com",
is_primary=False,
)
result = store_domain_service.add_domain(db, test_store.id, domain_data)
db.commit()
assert result is not None
assert result.store_id == test_store.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_store, primary_domain):
"""Test adding a domain as primary unsets other primary domains."""
unique_id = str(uuid.uuid4())[:8]
domain_data = StoreDomainCreate(
domain=f"newprimary{unique_id}.example.com",
is_primary=True,
)
result = store_domain_service.add_domain(db, test_store.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_store_not_found(self, db):
"""Test adding domain to non-existent store raises exception."""
domain_data = StoreDomainCreate(
domain="test.example.com",
is_primary=False,
)
with pytest.raises(StoreNotFoundException):
store_domain_service.add_domain(db, 99999, domain_data)
def test_add_domain_already_exists(self, db, test_store, test_domain):
"""Test adding a domain that already exists raises exception."""
domain_data = StoreDomainCreate(
domain=test_domain.domain,
is_primary=False,
)
with pytest.raises(StoreDomainAlreadyExistsException):
store_domain_service.add_domain(db, test_store.id, domain_data)
def test_add_domain_max_limit_reached(self, db, test_store):
"""Test adding domain when max limit reached raises exception."""
# Create max domains
for i in range(store_domain_service.max_domains_per_store):
domain = StoreDomain(
store_id=test_store.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 = StoreDomainCreate(
domain="onemore.example.com",
is_primary=False,
)
with pytest.raises(MaxDomainsReachedException):
store_domain_service.add_domain(db, test_store.id, domain_data)
def test_add_domain_reserved_subdomain(self, db, test_store):
"""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:
StoreDomainCreate(
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 TestStoreDomainServiceGet:
"""Test suite for getting store domains."""
def test_get_store_domains_success(self, db, test_store, test_domain, verified_domain):
"""Test getting all domains for a store."""
domains = store_domain_service.get_store_domains(db, test_store.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_store_domains_empty(self, db, test_store):
"""Test getting domains for store with no domains."""
domains = store_domain_service.get_store_domains(db, test_store.id)
# May have domains from other fixtures, so just check it returns a list
assert isinstance(domains, list)
def test_get_store_domains_store_not_found(self, db):
"""Test getting domains for non-existent store raises exception."""
with pytest.raises(StoreNotFoundException):
store_domain_service.get_store_domains(db, 99999)
def test_get_domain_by_id_success(self, db, test_domain):
"""Test getting a domain by ID."""
domain = store_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(StoreDomainNotFoundException):
store_domain_service.get_domain_by_id(db, 99999)
# =============================================================================
# UPDATE DOMAIN TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestStoreDomainServiceUpdate:
"""Test suite for updating store domains."""
def test_update_domain_set_primary(self, db, test_domain, primary_domain):
"""Test setting a domain as primary."""
update_data = StoreDomainUpdate(is_primary=True)
# First verify the domain (required for activation)
test_domain.is_verified = True
db.commit()
result = store_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 = StoreDomainUpdate(is_active=True)
result = store_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 = StoreDomainUpdate(is_active=True)
with pytest.raises(DomainNotVerifiedException):
store_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 = StoreDomainUpdate(is_primary=True)
with pytest.raises(StoreDomainNotFoundException):
store_domain_service.update_domain(db, 99999, update_data)
def test_update_domain_deactivate(self, db, verified_domain):
"""Test deactivating a domain."""
update_data = StoreDomainUpdate(is_active=False)
result = store_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 TestStoreDomainServiceDelete:
"""Test suite for deleting store domains."""
def test_delete_domain_success(self, db, test_store):
"""Test successfully deleting a domain."""
# Create a domain to delete
unique_id = str(uuid.uuid4())[:8]
domain = StoreDomain(
store_id=test_store.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 = store_domain_service.delete_domain(db, domain_id)
db.commit()
assert domain_name in result
# Verify it's gone
with pytest.raises(StoreDomainNotFoundException):
store_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(StoreDomainNotFoundException):
store_domain_service.delete_domain(db, 99999)
# =============================================================================
# VERIFY DOMAIN TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestStoreDomainServiceVerify:
"""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 = store_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):
store_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(StoreDomainNotFoundException):
store_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:
store_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):
store_domain_service.verify_domain(db, test_domain.id)
# =============================================================================
# VERIFICATION INSTRUCTIONS TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestStoreDomainServiceInstructions:
"""Test suite for verification instructions."""
def test_get_verification_instructions(self, db, test_domain):
"""Test getting verification instructions."""
instructions = store_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(StoreDomainNotFoundException):
store_domain_service.get_verification_instructions(db, 99999)

138
app/modules/tenancy/tests/unit/test_store_model.py Normal file
View File

@@ -0,0 +1,138 @@
# tests/unit/models/database/test_store.py
"""Unit tests for Store database model."""
import pytest
from sqlalchemy.exc import IntegrityError
from app.modules.tenancy.models import Store
@pytest.mark.unit
@pytest.mark.database
class TestStoreModel:
"""Test Store model."""
def test_store_creation(self, db, test_merchant):
"""Test Store model creation with merchant relationship."""
store = Store(
merchant_id=test_merchant.id,
store_code="DBTEST",
subdomain="dbtest",
name="Database Test Store",
description="Testing store model",
contact_email="contact@dbtest.com",
contact_phone="+1234567890",
business_address="123 Test Street",
is_active=True,
is_verified=False,
)
db.add(store)
db.commit()
db.refresh(store)
assert store.id is not None
assert store.store_code == "DBTEST"
assert store.subdomain == "dbtest"
assert store.name == "Database Test Store"
assert store.merchant_id == test_merchant.id
assert store.contact_email == "contact@dbtest.com"
assert store.is_active is True
assert store.is_verified is False
assert store.created_at is not None
def test_store_with_letzshop_urls(self, db, test_merchant):
"""Test Store model with multi-language Letzshop URLs."""
store = Store(
merchant_id=test_merchant.id,
store_code="MULTILANG",
subdomain="multilang",
name="Multi-Language Store",
letzshop_csv_url_fr="https://example.com/feed_fr.csv",
letzshop_csv_url_en="https://example.com/feed_en.csv",
letzshop_csv_url_de="https://example.com/feed_de.csv",
is_active=True,
)
db.add(store)
db.commit()
db.refresh(store)
assert store.letzshop_csv_url_fr == "https://example.com/feed_fr.csv"
assert store.letzshop_csv_url_en == "https://example.com/feed_en.csv"
assert store.letzshop_csv_url_de == "https://example.com/feed_de.csv"
def test_store_code_uniqueness(self, db, test_merchant):
"""Test store_code unique constraint."""
store1 = Store(
merchant_id=test_merchant.id,
store_code="UNIQUE",
subdomain="unique1",
name="Unique Store 1",
)
db.add(store1)
db.commit()
# Duplicate store_code should raise error
with pytest.raises(IntegrityError):
store2 = Store(
merchant_id=test_merchant.id,
store_code="UNIQUE",
subdomain="unique2",
name="Unique Store 2",
)
db.add(store2)
db.commit()
def test_subdomain_uniqueness(self, db, test_merchant):
"""Test subdomain unique constraint."""
store1 = Store(
merchant_id=test_merchant.id,
store_code="STORE1",
subdomain="testsubdomain",
name="Store 1",
)
db.add(store1)
db.commit()
# Duplicate subdomain should raise error
with pytest.raises(IntegrityError):
store2 = Store(
merchant_id=test_merchant.id,
store_code="STORE2",
subdomain="testsubdomain",
name="Store 2",
)
db.add(store2)
db.commit()
def test_store_default_values(self, db, test_merchant):
"""Test Store model default values."""
store = Store(
merchant_id=test_merchant.id,
store_code="DEFAULTS",
subdomain="defaults",
name="Default Store",
)
db.add(store)
db.commit()
db.refresh(store)
assert store.is_active is True # Default
assert store.is_verified is False # Default
def test_store_merchant_relationship(self, db, test_merchant):
"""Test Store-Merchant relationship."""
store = Store(
merchant_id=test_merchant.id,
store_code="RELTEST",
subdomain="reltest",
name="Relationship Test Store",
)
db.add(store)
db.commit()
db.refresh(store)
assert store.merchant is not None
assert store.merchant.id == test_merchant.id
assert store.merchant.name == test_merchant.name

316
app/modules/tenancy/tests/unit/test_store_schema.py Normal file
View File

@@ -0,0 +1,316 @@
# tests/unit/models/schema/test_store.py
"""Unit tests for store Pydantic schemas."""
import pytest
from pydantic import ValidationError
from app.modules.tenancy.schemas.store import (
StoreCreate,
StoreDetailResponse,
StoreListResponse,
StoreResponse,
StoreSummary,
StoreUpdate,
)
@pytest.mark.unit
@pytest.mark.schema
class TestStoreCreateSchema:
"""Test StoreCreate schema validation."""
def test_valid_store_create(self):
"""Test valid store creation data."""
store = StoreCreate(
merchant_id=1,
store_code="TECHSTORE",
subdomain="techstore",
name="Tech Store",
)
assert store.merchant_id == 1
assert store.store_code == "TECHSTORE"
assert store.subdomain == "techstore"
assert store.name == "Tech Store"
def test_merchant_id_required(self):
"""Test merchant_id is required."""
with pytest.raises(ValidationError) as exc_info:
StoreCreate(
store_code="TECHSTORE",
subdomain="techstore",
name="Tech Store",
)
assert "merchant_id" in str(exc_info.value).lower()
def test_merchant_id_must_be_positive(self):
"""Test merchant_id must be > 0."""
with pytest.raises(ValidationError) as exc_info:
StoreCreate(
merchant_id=0,
store_code="TECHSTORE",
subdomain="techstore",
name="Tech Store",
)
assert "merchant_id" in str(exc_info.value).lower()
def test_store_code_required(self):
"""Test store_code is required."""
with pytest.raises(ValidationError) as exc_info:
StoreCreate(
merchant_id=1,
subdomain="techstore",
name="Tech Store",
)
assert "store_code" in str(exc_info.value).lower()
def test_store_code_uppercase_normalized(self):
"""Test store_code is normalized to uppercase."""
store = StoreCreate(
merchant_id=1,
store_code="techstore",
subdomain="techstore",
name="Tech Store",
)
assert store.store_code == "TECHSTORE"
def test_store_code_min_length(self):
"""Test store_code minimum length (2)."""
with pytest.raises(ValidationError) as exc_info:
StoreCreate(
merchant_id=1,
store_code="T",
subdomain="techstore",
name="Tech Store",
)
assert "store_code" in str(exc_info.value).lower()
def test_subdomain_required(self):
"""Test subdomain is required."""
with pytest.raises(ValidationError) as exc_info:
StoreCreate(
merchant_id=1,
store_code="TECHSTORE",
name="Tech Store",
)
assert "subdomain" in str(exc_info.value).lower()
def test_subdomain_uppercase_invalid(self):
"""Test subdomain with uppercase is invalid (validated before normalization)."""
with pytest.raises(ValidationError) as exc_info:
StoreCreate(
merchant_id=1,
store_code="TECHSTORE",
subdomain="TechStore",
name="Tech Store",
)
assert "subdomain" in str(exc_info.value).lower()
def test_subdomain_valid_format(self):
"""Test subdomain with valid format."""
store = StoreCreate(
merchant_id=1,
store_code="TECHSTORE",
subdomain="tech-store-123",
name="Tech Store",
)
assert store.subdomain == "tech-store-123"
def test_subdomain_invalid_format(self):
"""Test subdomain with invalid characters raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
StoreCreate(
merchant_id=1,
store_code="TECHSTORE",
subdomain="tech_store!",
name="Tech Store",
)
assert "subdomain" in str(exc_info.value).lower()
def test_name_required(self):
"""Test name is required."""
with pytest.raises(ValidationError) as exc_info:
StoreCreate(
merchant_id=1,
store_code="TECHSTORE",
subdomain="techstore",
)
assert "name" in str(exc_info.value).lower()
def test_name_min_length(self):
"""Test name minimum length (2)."""
with pytest.raises(ValidationError) as exc_info:
StoreCreate(
merchant_id=1,
store_code="TECHSTORE",
subdomain="techstore",
name="T",
)
assert "name" in str(exc_info.value).lower()
def test_optional_fields(self):
"""Test optional fields."""
store = StoreCreate(
merchant_id=1,
store_code="TECHSTORE",
subdomain="techstore",
name="Tech Store",
description="Best tech store",
letzshop_csv_url_fr="https://example.com/fr.csv",
contact_email="contact@techstore.com",
website="https://techstore.com",
)
assert store.description == "Best tech store"
assert store.letzshop_csv_url_fr == "https://example.com/fr.csv"
assert store.contact_email == "contact@techstore.com"
@pytest.mark.unit
@pytest.mark.schema
class TestStoreUpdateSchema:
"""Test StoreUpdate schema validation."""
def test_partial_update(self):
"""Test partial update with only some fields."""
update = StoreUpdate(name="New Tech Store")
assert update.name == "New Tech Store"
assert update.subdomain is None
assert update.is_active is None
def test_empty_update_is_valid(self):
"""Test empty update is valid."""
update = StoreUpdate()
assert update.model_dump(exclude_unset=True) == {}
def test_subdomain_normalized_to_lowercase(self):
"""Test subdomain is normalized to lowercase."""
update = StoreUpdate(subdomain="NewSubdomain")
assert update.subdomain == "newsubdomain"
def test_subdomain_stripped(self):
"""Test subdomain is stripped of whitespace."""
update = StoreUpdate(subdomain=" newsubdomain ")
assert update.subdomain == "newsubdomain"
def test_name_min_length(self):
"""Test name minimum length (2)."""
with pytest.raises(ValidationError):
StoreUpdate(name="X")
def test_is_active_update(self):
"""Test is_active can be updated."""
update = StoreUpdate(is_active=False)
assert update.is_active is False
def test_is_verified_update(self):
"""Test is_verified can be updated."""
update = StoreUpdate(is_verified=True)
assert update.is_verified is True
def test_reset_contact_to_merchant_flag(self):
"""Test reset_contact_to_merchant flag."""
update = StoreUpdate(reset_contact_to_merchant=True)
assert update.reset_contact_to_merchant is True
@pytest.mark.unit
@pytest.mark.schema
class TestStoreResponseSchema:
"""Test StoreResponse schema."""
def test_from_dict(self):
"""Test creating response from dict."""
from datetime import datetime
data = {
"id": 1,
"store_code": "TECHSTORE",
"subdomain": "techstore",
"name": "Tech Store",
"description": "Best tech store",
"merchant_id": 1,
"letzshop_csv_url_fr": None,
"letzshop_csv_url_en": None,
"letzshop_csv_url_de": None,
"is_active": True,
"is_verified": False,
"created_at": datetime.now(),
"updated_at": datetime.now(),
}
response = StoreResponse(**data)
assert response.id == 1
assert response.store_code == "TECHSTORE"
assert response.is_active is True
@pytest.mark.unit
@pytest.mark.schema
class TestStoreDetailResponseSchema:
"""Test StoreDetailResponse schema."""
def test_from_dict(self):
"""Test creating detail response from dict."""
from datetime import datetime
data = {
"id": 1,
"store_code": "TECHSTORE",
"subdomain": "techstore",
"name": "Tech Store",
"description": None,
"merchant_id": 1,
"letzshop_csv_url_fr": None,
"letzshop_csv_url_en": None,
"letzshop_csv_url_de": None,
"is_active": True,
"is_verified": True,
"created_at": datetime.now(),
"updated_at": datetime.now(),
# Additional detail fields
"merchant_name": "Tech Corp",
"owner_email": "owner@techcorp.com",
"owner_username": "owner",
"contact_email": "contact@techstore.com",
"contact_email_inherited": False,
}
response = StoreDetailResponse(**data)
assert response.merchant_name == "Tech Corp"
assert response.owner_email == "owner@techcorp.com"
assert response.contact_email_inherited is False
@pytest.mark.unit
@pytest.mark.schema
class TestStoreListResponseSchema:
"""Test StoreListResponse schema."""
def test_valid_list_response(self):
"""Test valid list response structure."""
response = StoreListResponse(
stores=[],
total=0,
skip=0,
limit=10,
)
assert response.stores == []
assert response.total == 0
@pytest.mark.unit
@pytest.mark.schema
class TestStoreSummarySchema:
"""Test StoreSummary schema."""
def test_from_dict(self):
"""Test creating summary from dict."""
data = {
"id": 1,
"store_code": "TECHSTORE",
"subdomain": "techstore",
"name": "Tech Store",
"merchant_id": 1,
"is_active": True,
}
summary = StoreSummary(**data)
assert summary.id == 1
assert summary.store_code == "TECHSTORE"
assert summary.is_active is True

639
app/modules/tenancy/tests/unit/test_store_service.py Normal file
View File

@@ -0,0 +1,639 @@
# tests/unit/services/test_store_service.py
"""Unit tests for StoreService 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
import pytest
from app.exceptions import ValidationException
from app.modules.tenancy.exceptions import (
InvalidStoreDataException,
UnauthorizedStoreAccessException,
StoreAlreadyExistsException,
StoreNotFoundException,
)
from app.modules.tenancy.services.store_service import StoreService
from app.modules.tenancy.models import Merchant
from app.modules.tenancy.models import Store
from app.modules.tenancy.schemas.store import StoreCreate
@pytest.fixture
def admin_merchant(db, test_admin):
"""Create a test merchant for admin."""
unique_id = str(uuid.uuid4())[:8]
merchant = Merchant(
name=f"Admin Merchant {unique_id}",
owner_user_id=test_admin.id,
contact_email=f"admin{unique_id}@merchant.com",
is_active=True,
is_verified=True,
)
db.add(merchant)
db.commit()
db.refresh(merchant)
return merchant
# Note: other_merchant fixture is defined in tests/fixtures/store_fixtures.py
@pytest.mark.unit
@pytest.mark.stores
class TestStoreService:
"""Test suite for StoreService following the application's exception patterns."""
def setup_method(self):
"""Setup method following the same pattern as admin service tests."""
self.service = StoreService()
# ==================== create_store Tests ====================
def test_create_store_success(self, db, test_user, test_merchant):
"""Test successful store creation."""
unique_id = str(uuid.uuid4())[:8]
store_data = StoreCreate(
merchant_id=test_merchant.id,
store_code=f"NEWSTORE_{unique_id}",
subdomain=f"newstore{unique_id.lower()}",
name=f"New Test Store {unique_id}",
description="A new test store",
)
store = self.service.create_store(db, store_data, test_user)
db.commit()
assert store is not None
assert store.store_code == f"NEWSTORE_{unique_id}".upper()
assert store.merchant_id == test_merchant.id
assert store.is_verified is False # Regular user creates unverified store
def test_create_store_admin_auto_verify(self, db, test_admin, admin_merchant):
"""Test admin creates verified store automatically."""
unique_id = str(uuid.uuid4())[:8]
store_data = StoreCreate(
merchant_id=admin_merchant.id,
store_code=f"ADMINSTORE_{unique_id}",
subdomain=f"adminstore{unique_id.lower()}",
name=f"Admin Test Store {unique_id}",
)
store = self.service.create_store(db, store_data, test_admin)
db.commit()
assert store.is_verified is True # Admin creates verified store
def test_create_store_duplicate_code(
self, db, test_user, test_merchant, test_store
):
"""Test store creation fails with duplicate store code."""
store_data = StoreCreate(
merchant_id=test_merchant.id,
store_code=test_store.store_code,
subdomain="duplicatesub",
name="Duplicate Name",
)
with pytest.raises(StoreAlreadyExistsException) as exc_info:
self.service.create_store(db, store_data, test_user)
exception = exc_info.value
assert exception.status_code == 409
assert exception.error_code == "STORE_ALREADY_EXISTS"
assert test_store.store_code.upper() in exception.message
def test_create_store_missing_merchant_id(self, db, test_user):
"""Test store creation fails without merchant_id."""
# StoreCreate requires merchant_id, so this should raise ValidationError
# from Pydantic before reaching service
with pytest.raises(Exception): # Pydantic ValidationError
StoreCreate(
store_code="NOMERCHANT",
subdomain="nomerchant",
name="No Merchant Store",
)
def test_create_store_unauthorized_user(self, db, test_user, other_merchant):
"""Test store creation fails when user doesn't own merchant."""
unique_id = str(uuid.uuid4())[:8]
store_data = StoreCreate(
merchant_id=other_merchant.id, # Not owned by test_user
store_code=f"UNAUTH_{unique_id}",
subdomain=f"unauth{unique_id.lower()}",
name=f"Unauthorized Store {unique_id}",
)
with pytest.raises(UnauthorizedStoreAccessException) as exc_info:
self.service.create_store(db, store_data, test_user)
exception = exc_info.value
assert exception.status_code == 403
assert exception.error_code == "UNAUTHORIZED_STORE_ACCESS"
def test_create_store_invalid_merchant_id(self, db, test_user):
"""Test store creation fails with non-existent merchant."""
unique_id = str(uuid.uuid4())[:8]
store_data = StoreCreate(
merchant_id=99999, # Non-existent merchant
store_code=f"BADMERCHANT_{unique_id}",
subdomain=f"badmerchant{unique_id.lower()}",
name=f"Bad Merchant Store {unique_id}",
)
with pytest.raises(InvalidStoreDataException) as exc_info:
self.service.create_store(db, store_data, test_user)
exception = exc_info.value
assert exception.status_code == 422
assert exception.error_code == "INVALID_STORE_DATA"
assert "merchant_id" in exception.details.get("field", "")
# ==================== get_stores Tests ====================
def test_get_stores_regular_user(
self, db, test_user, test_store, inactive_store
):
"""Test regular user can only see active verified stores and own stores."""
stores, total = self.service.get_stores(db, test_user, skip=0, limit=100)
store_codes = [store.store_code for store in stores]
assert test_store.store_code in store_codes
# Inactive store should not be visible to regular user
assert inactive_store.store_code not in store_codes
def test_get_stores_admin_user(
self, db, test_admin, test_store, inactive_store, verified_store
):
"""Test admin user can see all stores with filters."""
stores, total = self.service.get_stores(
db, test_admin, active_only=False, verified_only=False
)
store_codes = [store.store_code for store in stores]
assert test_store.store_code in store_codes
assert inactive_store.store_code in store_codes
assert verified_store.store_code in store_codes
def test_get_stores_pagination(self, db, test_admin):
"""Test store pagination."""
stores, total = self.service.get_stores(
db, test_admin, skip=0, limit=5, active_only=False
)
assert len(stores) <= 5
def test_get_stores_database_error(self, db, test_user, monkeypatch):
"""Test get stores handles database errors gracefully."""
def mock_query(*args):
raise Exception("Database query failed")
monkeypatch.setattr(db, "query", mock_query)
with pytest.raises(ValidationException) as exc_info:
self.service.get_stores(db, test_user)
exception = exc_info.value
assert exception.error_code == "VALIDATION_ERROR"
assert "Failed to retrieve stores" in exception.message
# ==================== get_store_by_code Tests ====================
def test_get_store_by_code_owner_access(self, db, test_user, test_store):
"""Test store owner can access their own store."""
store = self.service.get_store_by_code(
db, test_store.store_code.lower(), test_user
)
assert store is not None
assert store.id == test_store.id
def test_get_store_by_code_admin_access(self, db, test_admin, test_store):
"""Test admin can access any store."""
store = self.service.get_store_by_code(
db, test_store.store_code.lower(), test_admin
)
assert store is not None
assert store.id == test_store.id
def test_get_store_by_code_not_found(self, db, test_user):
"""Test store not found raises proper exception."""
with pytest.raises(StoreNotFoundException) as exc_info:
self.service.get_store_by_code(db, "NONEXISTENT", test_user)
exception = exc_info.value
assert exception.status_code == 404
assert exception.error_code == "STORE_NOT_FOUND"
def test_get_store_by_code_access_denied(self, db, test_user, inactive_store):
"""Test regular user cannot access unverified store they don't own."""
with pytest.raises(UnauthorizedStoreAccessException) as exc_info:
self.service.get_store_by_code(db, inactive_store.store_code, test_user)
exception = exc_info.value
assert exception.status_code == 403
assert exception.error_code == "UNAUTHORIZED_STORE_ACCESS"
# ==================== get_store_by_id Tests ====================
def test_get_store_by_id_success(self, db, test_store):
"""Test getting store by ID."""
store = self.service.get_store_by_id(db, test_store.id)
assert store is not None
assert store.id == test_store.id
assert store.store_code == test_store.store_code
def test_get_store_by_id_not_found(self, db):
"""Test getting non-existent store by ID."""
with pytest.raises(StoreNotFoundException) as exc_info:
self.service.get_store_by_id(db, 99999)
exception = exc_info.value
assert exception.status_code == 404
assert exception.error_code == "STORE_NOT_FOUND"
# ==================== get_active_store_by_code Tests ====================
def test_get_active_store_by_code_success(self, db, test_store):
"""Test getting active store by code (public access)."""
store = self.service.get_active_store_by_code(db, test_store.store_code)
assert store is not None
assert store.id == test_store.id
assert store.is_active is True
def test_get_active_store_by_code_inactive(self, db, inactive_store):
"""Test getting inactive store fails."""
with pytest.raises(StoreNotFoundException):
self.service.get_active_store_by_code(db, inactive_store.store_code)
def test_get_active_store_by_code_not_found(self, db):
"""Test getting non-existent store fails."""
with pytest.raises(StoreNotFoundException):
self.service.get_active_store_by_code(db, "NONEXISTENT")
# ==================== toggle_verification Tests ====================
def test_toggle_verification_verify(self, db, inactive_store):
"""Test toggling verification on."""
original_verified = inactive_store.is_verified
store, message = self.service.toggle_verification(db, inactive_store.id)
db.commit()
assert store.is_verified != original_verified
assert "verified" in message.lower()
def test_toggle_verification_unverify(self, db, verified_store):
"""Test toggling verification off."""
store, message = self.service.toggle_verification(db, verified_store.id)
db.commit()
assert store.is_verified is False
assert "unverified" in message.lower()
def test_toggle_verification_not_found(self, db):
"""Test toggle verification on non-existent store."""
with pytest.raises(StoreNotFoundException):
self.service.toggle_verification(db, 99999)
# ==================== toggle_status Tests ====================
def test_toggle_status_deactivate(self, db, test_store):
"""Test toggling active status off."""
store, message = self.service.toggle_status(db, test_store.id)
db.commit()
assert store.is_active is False
assert "inactive" in message.lower()
def test_toggle_status_activate(self, db, inactive_store):
"""Test toggling active status on."""
store, message = self.service.toggle_status(db, inactive_store.id)
db.commit()
assert store.is_active is True
assert "active" in message.lower()
def test_toggle_status_not_found(self, db):
"""Test toggle status on non-existent store."""
with pytest.raises(StoreNotFoundException):
self.service.toggle_status(db, 99999)
# ==================== set_verification / set_status Tests ====================
def test_set_verification_to_true(self, db, inactive_store):
"""Test setting verification to true."""
store, message = self.service.set_verification(db, inactive_store.id, True)
db.commit()
assert store.is_verified is True
def test_set_verification_to_false(self, db, verified_store):
"""Test setting verification to false."""
store, message = self.service.set_verification(db, verified_store.id, False)
db.commit()
assert store.is_verified is False
def test_set_status_to_active(self, db, inactive_store):
"""Test setting status to active."""
store, message = self.service.set_status(db, inactive_store.id, True)
db.commit()
assert store.is_active is True
def test_set_status_to_inactive(self, db, test_store):
"""Test setting status to inactive."""
store, message = self.service.set_status(db, test_store.id, False)
db.commit()
assert store.is_active is False
# 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 ====================
def test_store_code_exists(self, db, test_store):
"""Test _store_code_exists helper method."""
assert self.service._store_code_exists(db, test_store.store_code) is True
assert self.service._store_code_exists(db, "NONEXISTENT") is False
def test_can_access_store_admin(self, db, test_admin, test_store):
"""Test admin can always access store."""
# Re-query store to get fresh instance
store = db.query(Store).filter(Store.id == test_store.id).first()
assert self.service._can_access_store(store, test_admin) is True
def test_can_access_store_active_verified(self, db, test_user, verified_store):
"""Test any user can access active verified store."""
# Re-query store to get fresh instance
store = db.query(Store).filter(Store.id == verified_store.id).first()
assert self.service._can_access_store(store, test_user) is True
@pytest.mark.unit
@pytest.mark.stores
class TestStoreServiceExceptionDetails:
"""Additional tests focusing specifically on exception structure and details."""
def setup_method(self):
self.service = StoreService()
def test_exception_to_dict_structure(
self, db, test_user, test_store, test_merchant
):
"""Test that exceptions can be properly serialized to dict for API responses."""
store_data = StoreCreate(
merchant_id=test_merchant.id,
store_code=test_store.store_code,
subdomain="duplicate",
name="Duplicate",
)
with pytest.raises(StoreAlreadyExistsException) as exc_info:
self.service.create_store(db, store_data, test_user)
exception = exc_info.value
exception_dict = exception.to_dict()
# Verify structure matches expected API response format
assert "error_code" in exception_dict
assert "message" in exception_dict
assert "status_code" in exception_dict
assert "details" in exception_dict
# Verify values
assert exception_dict["error_code"] == "STORE_ALREADY_EXISTS"
assert exception_dict["status_code"] == 409
assert isinstance(exception_dict["details"], dict)
def test_authorization_exception_user_details(self, db, test_user, inactive_store):
"""Test authorization exceptions include user context."""
with pytest.raises(UnauthorizedStoreAccessException) as exc_info:
self.service.get_store_by_code(db, inactive_store.store_code, test_user)
exception = exc_info.value
assert exception.details["store_code"] == inactive_store.store_code
assert exception.details["user_id"] == test_user.id
assert "Unauthorized access" in exception.message
def test_not_found_exception_details(self, db, test_user):
"""Test not found exceptions include identifier details."""
with pytest.raises(StoreNotFoundException) as exc_info:
self.service.get_store_by_code(db, "NOTEXIST", test_user)
exception = exc_info.value
assert exception.status_code == 404
assert exception.error_code == "STORE_NOT_FOUND"
@pytest.mark.unit
@pytest.mark.stores
class TestStoreServiceIdentifier:
"""Tests for get_store_by_identifier method."""
def setup_method(self):
self.service = StoreService()
def test_get_store_by_identifier_with_id(self, db, test_store):
"""Test getting store by numeric ID string."""
store = self.service.get_store_by_identifier(db, str(test_store.id))
assert store is not None
assert store.id == test_store.id
def test_get_store_by_identifier_with_code(self, db, test_store):
"""Test getting store by store_code."""
store = self.service.get_store_by_identifier(db, test_store.store_code)
assert store is not None
assert store.store_code == test_store.store_code
def test_get_store_by_identifier_case_insensitive(self, db, test_store):
"""Test getting store by store_code is case insensitive."""
store = self.service.get_store_by_identifier(
db, test_store.store_code.lower()
)
assert store is not None
assert store.id == test_store.id
def test_get_store_by_identifier_not_found(self, db):
"""Test getting non-existent store."""
with pytest.raises(StoreNotFoundException):
self.service.get_store_by_identifier(db, "NONEXISTENT_CODE")
@pytest.mark.unit
@pytest.mark.stores
class TestStoreServicePermissions:
"""Tests for permission checking methods."""
def setup_method(self):
self.service = StoreService()
def test_can_update_store_admin(self, db, test_admin, test_store):
"""Test admin can always update store."""
store = db.query(Store).filter(Store.id == test_store.id).first()
assert self.service.can_update_store(store, test_admin) is True
def test_can_update_store_owner(self, db, test_user, test_store):
"""Test owner can update store."""
store = db.query(Store).filter(Store.id == test_store.id).first()
assert self.service.can_update_store(store, test_user) is True
def test_can_update_store_non_owner(self, db, other_merchant, test_store):
"""Test non-owner cannot update store."""
from app.modules.tenancy.models import User
store = db.query(Store).filter(Store.id == test_store.id).first()
other_user = db.query(User).filter(User.id == other_merchant.owner_user_id).first()
# Clear any StoreUser relationships
assert self.service.can_update_store(store, other_user) is False
def test_is_store_owner_true(self, db, test_user, test_store):
"""Test _is_store_owner returns True for owner."""
store = db.query(Store).filter(Store.id == test_store.id).first()
assert self.service._is_store_owner(store, test_user) is True
def test_is_store_owner_false(self, db, other_merchant, test_store):
"""Test _is_store_owner returns False for non-owner."""
from app.modules.tenancy.models import User
store = db.query(Store).filter(Store.id == test_store.id).first()
other_user = db.query(User).filter(User.id == other_merchant.owner_user_id).first()
assert self.service._is_store_owner(store, other_user) is False
@pytest.mark.unit
@pytest.mark.stores
class TestStoreServiceUpdate:
"""Tests for update methods."""
def setup_method(self):
self.service = StoreService()
def test_update_store_success(self, db, test_user, test_store):
"""Test successfully updating store profile."""
from pydantic import BaseModel
class StoreUpdate(BaseModel):
name: str | None = None
description: str | None = None
class Config:
extra = "forbid"
update_data = StoreUpdate(
name="Updated Store Name",
description="Updated description",
)
store = self.service.update_store(
db, test_store.id, update_data, test_user
)
db.commit()
assert store.name == "Updated Store Name"
assert store.description == "Updated description"
def test_update_store_unauthorized(self, db, other_merchant, test_store):
"""Test update fails for unauthorized user."""
from pydantic import BaseModel
from app.modules.tenancy.exceptions import InsufficientPermissionsException
from app.modules.tenancy.models import User
class StoreUpdate(BaseModel):
name: str | None = None
class Config:
extra = "forbid"
other_user = db.query(User).filter(User.id == other_merchant.owner_user_id).first()
update_data = StoreUpdate(name="Unauthorized Update")
with pytest.raises(InsufficientPermissionsException):
self.service.update_store(
db, test_store.id, update_data, other_user
)
def test_update_store_not_found(self, db, test_admin):
"""Test update fails for non-existent store."""
from pydantic import BaseModel
class StoreUpdate(BaseModel):
name: str | None = None
class Config:
extra = "forbid"
update_data = StoreUpdate(name="Update")
with pytest.raises(StoreNotFoundException):
self.service.update_store(db, 99999, update_data, test_admin)
def test_update_marketplace_settings_success(self, db, test_user, test_store):
"""Test successfully updating marketplace settings."""
marketplace_config = {
"letzshop_csv_url_fr": "https://example.com/fr.csv",
"letzshop_csv_url_en": "https://example.com/en.csv",
}
result = self.service.update_marketplace_settings(
db, test_store.id, marketplace_config, test_user
)
db.commit()
assert result["message"] == "Marketplace settings updated successfully"
assert result["letzshop_csv_url_fr"] == "https://example.com/fr.csv"
assert result["letzshop_csv_url_en"] == "https://example.com/en.csv"
def test_update_marketplace_settings_unauthorized(
self, db, other_merchant, test_store
):
"""Test marketplace settings update fails for unauthorized user."""
from app.modules.tenancy.exceptions import InsufficientPermissionsException
from app.modules.tenancy.models import User
other_user = db.query(User).filter(User.id == other_merchant.owner_user_id).first()
marketplace_config = {"letzshop_csv_url_fr": "https://example.com/fr.csv"}
with pytest.raises(InsufficientPermissionsException):
self.service.update_marketplace_settings(
db, test_store.id, marketplace_config, other_user
)
def test_update_marketplace_settings_not_found(self, db, test_admin):
"""Test marketplace settings update fails for non-existent store."""
marketplace_config = {"letzshop_csv_url_fr": "https://example.com/fr.csv"}
with pytest.raises(StoreNotFoundException):
self.service.update_marketplace_settings(
db, 99999, marketplace_config, test_admin
)
@pytest.mark.unit
@pytest.mark.stores
class TestStoreServiceSingleton:
"""Test singleton instance."""
def test_singleton_exists(self):
"""Test store_service singleton exists."""
from app.modules.tenancy.services.store_service import store_service
assert store_service is not None
assert isinstance(store_service, StoreService)

427
app/modules/tenancy/tests/unit/test_store_team_service.py Normal file
View File

@@ -0,0 +1,427 @@
# tests/unit/services/test_store_team_service.py
"""Unit tests for StoreTeamService."""
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, Store, StoreUser, StoreUserType
from app.modules.tenancy.services.store_team_service import store_team_service
# =============================================================================
# FIXTURES
# =============================================================================
@pytest.fixture
def team_store(db, test_merchant):
"""Create a store for team tests."""
unique_id = str(uuid.uuid4())[:8]
store = Store(
merchant_id=test_merchant.id,
store_code=f"TEAMSTORE_{unique_id.upper()}",
subdomain=f"teamstore{unique_id.lower()}",
name=f"Team Store {unique_id}",
is_active=True,
is_verified=True,
)
db.add(store)
db.commit()
db.refresh(store)
return store
@pytest.fixture
def store_owner(db, team_store, test_user):
"""Create an owner for the team store."""
store_user = StoreUser(
store_id=team_store.id,
user_id=test_user.id,
user_type=StoreUserType.OWNER.value,
is_active=True,
)
db.add(store_user)
db.commit()
db.refresh(store_user)
return store_user
@pytest.fixture
def team_member(db, team_store, other_user, auth_manager):
"""Create a team member for the store."""
# Create a role first
role = Role(
store_id=team_store.id,
name="staff",
permissions=["orders.view", "products.view"],
)
db.add(role)
db.commit()
store_user = StoreUser(
store_id=team_store.id,
user_id=other_user.id,
user_type=StoreUserType.TEAM_MEMBER.value,
role_id=role.id,
is_active=True,
invitation_accepted_at=datetime.utcnow(),
)
db.add(store_user)
db.commit()
db.refresh(store_user)
return store_user
@pytest.fixture
def pending_invitation(db, team_store, 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="store",
is_active=False,
)
db.add(new_user)
db.commit()
# Create role
role = Role(
store_id=team_store.id,
name="support",
permissions=["support.view"],
)
db.add(role)
db.commit()
# Create pending store user
store_user = StoreUser(
store_id=team_store.id,
user_id=new_user.id,
user_type=StoreUserType.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(store_user)
db.commit()
db.refresh(store_user)
return store_user
@pytest.fixture
def expired_invitation(db, team_store, 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="store",
is_active=False,
)
db.add(new_user)
db.commit()
# Create role
role = Role(
store_id=team_store.id,
name="viewer",
permissions=["read_only"],
)
db.add(role)
db.commit()
# Create expired store user (sent 8 days ago, expires in 7)
store_user = StoreUser(
store_id=team_store.id,
user_id=new_user.id,
user_type=StoreUserType.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(store_user)
db.commit()
db.refresh(store_user)
return store_user
# =============================================================================
# INVITE TEAM MEMBER TESTS
# =============================================================================
# TestStoreTeamServiceInvite removed — check_team_limit was refactored in SubscriptionService
# =============================================================================
# ACCEPT INVITATION TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestStoreTeamServiceAccept:
"""Test suite for accepting invitations."""
def test_accept_invitation_success(self, db, pending_invitation):
"""Test accepting a valid invitation."""
result = store_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):
store_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):
store_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:
store_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 TestStoreTeamServiceRemove:
"""Test suite for removing team members."""
def test_remove_team_member_success(self, db, team_store, team_member):
"""Test removing a team member."""
result = store_team_service.remove_team_member(
db=db,
store=team_store,
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_store, store_owner):
"""Test removing owner raises exception."""
with pytest.raises(CannotRemoveOwnerException):
store_team_service.remove_team_member(
db=db,
store=team_store,
user_id=store_owner.user_id,
)
def test_remove_nonexistent_user_raises_error(self, db, team_store):
"""Test removing non-existent user raises exception."""
with pytest.raises(UserNotFoundException):
store_team_service.remove_team_member(
db=db,
store=team_store,
user_id=99999,
)
# =============================================================================
# UPDATE MEMBER ROLE TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestStoreTeamServiceUpdateRole:
"""Test suite for updating member roles."""
def test_update_role_success(self, db, team_store, team_member):
"""Test updating a member's role."""
result = store_team_service.update_member_role(
db=db,
store=team_store,
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_store, store_owner):
"""Test updating owner's role raises exception."""
with pytest.raises(CannotRemoveOwnerException):
store_team_service.update_member_role(
db=db,
store=team_store,
user_id=store_owner.user_id,
new_role_name="staff",
)
def test_update_role_with_custom_permissions(self, db, team_store, team_member):
"""Test updating role with custom permissions."""
custom_perms = ["orders.view", "orders.edit", "reports.view"]
result = store_team_service.update_member_role(
db=db,
store=team_store,
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_store):
"""Test updating non-existent user raises exception."""
with pytest.raises(UserNotFoundException):
store_team_service.update_member_role(
db=db,
store=team_store,
user_id=99999,
new_role_name="staff",
)
# =============================================================================
# GET TEAM MEMBERS TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestStoreTeamServiceGetMembers:
"""Test suite for getting team members."""
def test_get_team_members(self, db, team_store, store_owner, team_member):
"""Test getting all team members."""
members = store_team_service.get_team_members(db, team_store)
assert len(members) >= 2
user_ids = [m["id"] for m in members]
assert store_owner.user_id in user_ids
assert team_member.user_id in user_ids
def test_get_team_members_excludes_inactive(
self, db, team_store, store_owner, team_member
):
"""Test getting only active team members."""
# Deactivate team_member
team_member.is_active = False
db.commit()
members = store_team_service.get_team_members(
db, team_store, include_inactive=False
)
user_ids = [m["id"] for m in members]
assert store_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_store, store_owner, team_member
):
"""Test getting all members including inactive."""
# Deactivate team_member
team_member.is_active = False
db.commit()
members = store_team_service.get_team_members(
db, team_store, include_inactive=True
)
user_ids = [m["id"] for m in members]
assert store_owner.user_id in user_ids
assert team_member.user_id in user_ids
# =============================================================================
# GET STORE ROLES TESTS
# =============================================================================
@pytest.mark.unit
@pytest.mark.tenancy
class TestStoreTeamServiceGetRoles:
"""Test suite for getting store roles."""
def test_get_store_roles_existing(self, db, team_store, team_member):
"""Test getting roles when they exist."""
# team_member fixture creates a role
roles = store_team_service.get_store_roles(db, team_store.id)
assert len(roles) >= 1
role_names = [r["name"] for r in roles]
assert "staff" in role_names
def test_get_store_roles_creates_defaults(self, db, team_store):
"""Test default roles are created if none exist."""
roles = store_team_service.get_store_roles(db, team_store.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_store_roles_returns_permissions(self, db, team_store):
"""Test roles include permissions."""
roles = store_team_service.get_store_roles(db, team_store.id)
for role in roles:
assert "permissions" in role
assert isinstance(role["permissions"], list)

138
app/modules/tenancy/tests/unit/test_team_model.py Normal file
View File

@@ -0,0 +1,138 @@
# tests/unit/models/database/test_team.py
"""Unit tests for StoreUser and Role database models."""
import pytest
from app.modules.tenancy.models import Role, Store, StoreUser
@pytest.mark.unit
@pytest.mark.database
class TestRoleModel:
"""Test Role model."""
def test_role_creation(self, db, test_store):
"""Test Role model creation."""
role = Role(
store_id=test_store.id,
name="Manager",
permissions=["products.create", "orders.view"],
)
db.add(role)
db.commit()
db.refresh(role)
assert role.id is not None
assert role.store_id == test_store.id
assert role.name == "Manager"
assert "products.create" in role.permissions
assert "orders.view" in role.permissions
def test_role_default_permissions(self, db, test_store):
"""Test Role model with default empty permissions."""
role = Role(
store_id=test_store.id,
name="Viewer",
)
db.add(role)
db.commit()
db.refresh(role)
assert role.permissions == [] or role.permissions is None
def test_role_store_relationship(self, db, test_store):
"""Test Role-Store relationship."""
role = Role(
store_id=test_store.id,
name="Admin",
permissions=["*"],
)
db.add(role)
db.commit()
db.refresh(role)
assert role.store is not None
assert role.store.id == test_store.id
@pytest.mark.unit
@pytest.mark.database
class TestStoreUserModel:
"""Test StoreUser model."""
def test_store_user_creation(self, db, test_store, test_user):
"""Test StoreUser model for team management."""
# Create a role
role = Role(
store_id=test_store.id,
name="Manager",
permissions=["products.create", "orders.view"],
)
db.add(role)
db.commit()
# Create store user
store_user = StoreUser(
store_id=test_store.id,
user_id=test_user.id,
role_id=role.id,
is_active=True,
)
db.add(store_user)
db.commit()
db.refresh(store_user)
assert store_user.id is not None
assert store_user.store_id == test_store.id
assert store_user.user_id == test_user.id
assert store_user.role.name == "Manager"
assert "products.create" in store_user.role.permissions
def test_store_user_relationships(self, db, test_store, test_user):
"""Test StoreUser relationships."""
role = Role(
store_id=test_store.id,
name="Staff",
permissions=["orders.view"],
)
db.add(role)
db.commit()
store_user = StoreUser(
store_id=test_store.id,
user_id=test_user.id,
role_id=role.id,
is_active=True,
)
db.add(store_user)
db.commit()
db.refresh(store_user)
assert store_user.store is not None
assert store_user.user is not None
assert store_user.role is not None
assert store_user.store.store_code == test_store.store_code
assert store_user.user.email == test_user.email
def test_store_user_with_active_flag(self, db, test_store, test_user):
"""Test StoreUser is_active field."""
role = Role(
store_id=test_store.id,
name="Default",
permissions=[],
)
db.add(role)
db.commit()
# Create with explicit is_active=True
store_user = StoreUser(
store_id=test_store.id,
user_id=test_user.id,
role_id=role.id,
is_active=True,
)
db.add(store_user)
db.commit()
db.refresh(store_user)
assert store_user.is_active is True

200
app/modules/tenancy/tests/unit/test_team_service.py Normal file
View File

@@ -0,0 +1,200 @@
# tests/unit/services/test_team_service.py
"""
Unit tests for TeamService.
Tests cover:
- Get team members
- Invite team member
- Update team member
- Remove team member
- Get store roles
"""
from datetime import UTC, datetime
from unittest.mock import MagicMock
import pytest
from app.exceptions import ValidationException
from app.modules.tenancy.services.team_service import TeamService, team_service
from app.modules.tenancy.models import Role, StoreUser
@pytest.mark.unit
@pytest.mark.service
class TestTeamServiceGetMembers:
"""Test get_team_members functionality"""
def test_get_team_members_empty(self, db, test_store, test_user):
"""Test get_team_members returns empty list when no members"""
service = TeamService()
result = service.get_team_members(db, test_store.id, test_user)
assert isinstance(result, list)
def test_get_team_members_with_data(self, db, test_store_with_store_user, test_user):
"""Test get_team_members returns member data or raises"""
service = TeamService()
try:
result = service.get_team_members(
db, test_store_with_store_user.id, test_user
)
assert isinstance(result, list)
if len(result) > 0:
member = result[0]
assert "id" in member
assert "email" in member
except ValidationException:
# This is expected if the store user has no role
pass
@pytest.mark.unit
@pytest.mark.service
class TestTeamServiceInvite:
"""Test invite_team_member functionality"""
def test_invite_team_member_placeholder(self, db, test_store, test_user):
"""Test invite_team_member returns placeholder response"""
service = TeamService()
result = service.invite_team_member(
db,
test_store.id,
{"email": "newmember@example.com", "role": "member"},
test_user,
)
assert "message" in result
assert result["email"] == "newmember@example.com"
@pytest.mark.unit
@pytest.mark.service
class TestTeamServiceUpdate:
"""Test update_team_member functionality"""
def test_update_team_member_not_found(self, db, test_store, test_user):
"""Test update_team_member raises for non-existent member"""
service = TeamService()
with pytest.raises(ValidationException) as exc_info:
service.update_team_member(
db,
test_store.id,
99999, # Non-existent user
{"role_id": 1},
test_user,
)
assert "failed" in str(exc_info.value).lower()
def test_update_team_member_success(
self, db, test_store_with_store_user, test_store_user, test_user
):
"""Test update_team_member updates member"""
service = TeamService()
# Get the store_user record
store_user = (
db.query(StoreUser)
.filter(StoreUser.store_id == test_store_with_store_user.id)
.first()
)
if store_user:
result = service.update_team_member(
db,
test_store_with_store_user.id,
store_user.user_id,
{"is_active": True},
test_user,
)
db.commit()
assert result["message"] == "Team member updated successfully"
assert result["user_id"] == store_user.user_id
@pytest.mark.unit
@pytest.mark.service
class TestTeamServiceRemove:
"""Test remove_team_member functionality"""
def test_remove_team_member_not_found(self, db, test_store, test_user):
"""Test remove_team_member raises for non-existent member"""
service = TeamService()
with pytest.raises(ValidationException) as exc_info:
service.remove_team_member(
db,
test_store.id,
99999, # Non-existent user
test_user,
)
assert "failed" in str(exc_info.value).lower()
def test_remove_team_member_success(
self, db, test_store_with_store_user, test_store_user, test_user
):
"""Test remove_team_member soft deletes member"""
service = TeamService()
# Get the store_user record
store_user = (
db.query(StoreUser)
.filter(StoreUser.store_id == test_store_with_store_user.id)
.first()
)
if store_user:
result = service.remove_team_member(
db,
test_store_with_store_user.id,
store_user.user_id,
test_user,
)
db.commit()
assert result is True
# Verify soft delete
db.refresh(store_user)
assert store_user.is_active is False
@pytest.mark.unit
@pytest.mark.service
class TestTeamServiceRoles:
"""Test get_store_roles functionality"""
def test_get_store_roles_empty(self, db, test_store):
"""Test get_store_roles returns empty list when no roles"""
service = TeamService()
result = service.get_store_roles(db, test_store.id)
assert isinstance(result, list)
def test_get_store_roles_with_data(self, db, test_store_with_store_user):
"""Test get_store_roles returns role data"""
# Create a role for the store
role = Role(
store_id=test_store_with_store_user.id,
name="Test Role",
permissions=["view_orders", "edit_products"],
)
db.add(role)
db.commit()
service = TeamService()
result = service.get_store_roles(db, test_store_with_store_user.id)
assert len(result) >= 1
role_data = next((r for r in result if r["name"] == "Test Role"), None)
if role_data:
assert role_data["permissions"] == ["view_orders", "edit_products"]
@pytest.mark.unit
@pytest.mark.service
class TestTeamServiceSingleton:
"""Test singleton instance"""
def test_singleton_exists(self):
"""Test team_service singleton exists"""
assert team_service is not None
assert isinstance(team_service, TeamService)

105
app/modules/tenancy/tests/unit/test_user_model.py Normal file
View File

@@ -0,0 +1,105 @@
# tests/unit/models/database/test_user.py
"""Unit tests for User database model."""
import pytest
from sqlalchemy.exc import IntegrityError
from app.modules.tenancy.models import User
@pytest.mark.unit
@pytest.mark.database
class TestUserModel:
"""Test User model."""
def test_user_creation(self, db):
"""Test User model creation and relationships."""
user = User(
email="db_test@example.com",
username="dbtest",
hashed_password="hashed_password_123",
role="user",
is_active=True,
)
db.add(user)
db.commit()
db.refresh(user)
assert user.id is not None
assert user.email == "db_test@example.com"
assert user.username == "dbtest"
assert user.role == "user"
assert user.is_active is True
assert user.created_at is not None
assert user.updated_at is not None
def test_user_email_uniqueness(self, db):
"""Test email unique constraint."""
user1 = User(
email="unique@example.com",
username="user1",
hashed_password="hash1",
)
db.add(user1)
db.commit()
# Duplicate email should raise error
with pytest.raises(IntegrityError):
user2 = User(
email="unique@example.com",
username="user2",
hashed_password="hash2",
)
db.add(user2)
db.commit()
def test_user_username_uniqueness(self, db):
"""Test username unique constraint."""
user1 = User(
email="user1@example.com",
username="sameusername",
hashed_password="hash1",
)
db.add(user1)
db.commit()
# Duplicate username should raise error
with pytest.raises(IntegrityError):
user2 = User(
email="user2@example.com",
username="sameusername",
hashed_password="hash2",
)
db.add(user2)
db.commit()
def test_user_default_values(self, db):
"""Test User model default values."""
user = User(
email="defaults@example.com",
username="defaultuser",
hashed_password="hash",
)
db.add(user)
db.commit()
db.refresh(user)
assert user.is_active is True # Default
assert user.role == "store" # Default (UserRole.STORE)
def test_user_optional_fields(self, db):
"""Test User model with optional fields."""
user = User(
email="optional@example.com",
username="optionaluser",
hashed_password="hash",
first_name="John",
last_name="Doe",
)
db.add(user)
db.commit()
db.refresh(user)
assert user.first_name == "John"
assert user.last_name == "Doe"