fix(billing): use tier_id instead of tier_code for feature limit endpoints
Some checks failed
Some checks failed
Tier codes are not unique across platforms (e.g., "essential" exists for OMS, marketplace, and loyalty). Using tier_code caused feature limits to be saved to the wrong tier. Switched to tier_id (unique PK) in routes, service, and frontend JS. Added comprehensive unit and integration tests including cross-platform isolation regression tests. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,423 @@
|
||||
# app/modules/billing/tests/integration/test_admin_features_routes.py
|
||||
"""
|
||||
Integration tests for admin feature management API routes.
|
||||
|
||||
Tests the feature limit endpoints at:
|
||||
/api/v1/admin/subscriptions/features/*
|
||||
|
||||
Covers:
|
||||
- GET /features/catalog
|
||||
- GET /features/tiers/{tier_id}/limits
|
||||
- PUT /features/tiers/{tier_id}/limits
|
||||
- Regression: tiers with duplicate codes across platforms are isolated by tier_id
|
||||
|
||||
Uses super_admin_headers fixture which bypasses module access checks.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
from app.modules.billing.models import SubscriptionTier
|
||||
from app.modules.billing.models.tier_feature_limit import TierFeatureLimit
|
||||
from app.modules.tenancy.models import Platform
|
||||
|
||||
# ============================================================================
|
||||
# Fixtures
|
||||
# ============================================================================
|
||||
|
||||
BASE = "/api/v1/admin/subscriptions/features"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ft_platform(db):
|
||||
"""Create a platform for feature route tests."""
|
||||
platform = Platform(
|
||||
code=f"feat_{uuid.uuid4().hex[:8]}",
|
||||
name="Feature Test Platform",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(platform)
|
||||
db.commit()
|
||||
db.refresh(platform)
|
||||
return platform
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ft_second_platform(db):
|
||||
"""Second platform for cross-platform isolation tests."""
|
||||
platform = Platform(
|
||||
code=f"feat2_{uuid.uuid4().hex[:8]}",
|
||||
name="Feature Test Platform 2",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(platform)
|
||||
db.commit()
|
||||
db.refresh(platform)
|
||||
return platform
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ft_tier(db, ft_platform):
|
||||
"""Create a tier for feature route tests."""
|
||||
tier = SubscriptionTier(
|
||||
code=f"essential_{uuid.uuid4().hex[:6]}",
|
||||
name="Essential",
|
||||
price_monthly_cents=1000,
|
||||
display_order=0,
|
||||
is_active=True,
|
||||
is_public=True,
|
||||
platform_id=ft_platform.id,
|
||||
)
|
||||
db.add(tier)
|
||||
db.commit()
|
||||
db.refresh(tier)
|
||||
return tier
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ft_duplicate_code_tiers(db, ft_platform, ft_second_platform):
|
||||
"""Create two tiers with the SAME code but different platforms.
|
||||
|
||||
This is the exact scenario that caused the tier_code ambiguity bug.
|
||||
"""
|
||||
tier_a = SubscriptionTier(
|
||||
code="essential",
|
||||
name="Essential (Platform A)",
|
||||
price_monthly_cents=1000,
|
||||
display_order=0,
|
||||
is_active=True,
|
||||
is_public=True,
|
||||
platform_id=ft_platform.id,
|
||||
)
|
||||
tier_b = SubscriptionTier(
|
||||
code="essential",
|
||||
name="Essential (Platform B)",
|
||||
price_monthly_cents=2000,
|
||||
display_order=0,
|
||||
is_active=True,
|
||||
is_public=True,
|
||||
platform_id=ft_second_platform.id,
|
||||
)
|
||||
db.add(tier_a)
|
||||
db.add(tier_b)
|
||||
db.commit()
|
||||
db.refresh(tier_a)
|
||||
db.refresh(tier_b)
|
||||
return tier_a, tier_b
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ft_tier_with_features(db, ft_tier):
|
||||
"""Pre-populate a tier with feature limits."""
|
||||
features = [
|
||||
TierFeatureLimit(tier_id=ft_tier.id, feature_code="basic_shop", limit_value=None),
|
||||
TierFeatureLimit(tier_id=ft_tier.id, feature_code="team_members", limit_value=5),
|
||||
]
|
||||
for f in features:
|
||||
db.add(f)
|
||||
db.commit()
|
||||
# Refresh so the tier's selectin-loaded feature_limits relationship is up to date
|
||||
db.refresh(ft_tier)
|
||||
return features
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Feature Catalog
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.billing
|
||||
class TestFeatureCatalog:
|
||||
"""Tests for GET /features/catalog."""
|
||||
|
||||
def test_get_catalog(self, client, super_admin_headers):
|
||||
"""Returns the feature catalog grouped by category."""
|
||||
response = client.get(f"{BASE}/catalog", headers=super_admin_headers)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "features" in data
|
||||
assert isinstance(data["features"], dict)
|
||||
|
||||
def test_catalog_requires_auth(self, client):
|
||||
"""Catalog endpoint requires authentication."""
|
||||
response = client.get(f"{BASE}/catalog")
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# GET Tier Feature Limits
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.billing
|
||||
class TestGetTierFeatureLimits:
|
||||
"""Tests for GET /features/tiers/{tier_id}/limits."""
|
||||
|
||||
def test_get_limits_empty(self, client, super_admin_headers, ft_tier):
|
||||
"""Returns empty list for tier with no features."""
|
||||
response = client.get(
|
||||
f"{BASE}/tiers/{ft_tier.id}/limits",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json() == []
|
||||
|
||||
def test_get_limits_with_features(
|
||||
self, client, super_admin_headers, ft_tier, ft_tier_with_features
|
||||
):
|
||||
"""Returns feature limit entries for a tier."""
|
||||
response = client.get(
|
||||
f"{BASE}/tiers/{ft_tier.id}/limits",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data) == 2
|
||||
codes = {e["feature_code"] for e in data}
|
||||
assert codes == {"basic_shop", "team_members"}
|
||||
# Check limit values
|
||||
for entry in data:
|
||||
assert entry["enabled"] is True
|
||||
if entry["feature_code"] == "team_members":
|
||||
assert entry["limit_value"] == 5
|
||||
else:
|
||||
assert entry["limit_value"] is None
|
||||
|
||||
def test_get_limits_requires_auth(self, client, ft_tier):
|
||||
"""Endpoint requires authentication."""
|
||||
response = client.get(f"{BASE}/tiers/{ft_tier.id}/limits")
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# PUT Tier Feature Limits
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.billing
|
||||
class TestUpsertTierFeatureLimits:
|
||||
"""Tests for PUT /features/tiers/{tier_id}/limits."""
|
||||
|
||||
def test_save_features(self, client, super_admin_headers, ft_tier):
|
||||
"""Saves feature limits and returns the saved entries."""
|
||||
# Get valid feature codes from catalog
|
||||
catalog = client.get(f"{BASE}/catalog", headers=super_admin_headers).json()
|
||||
all_codes = []
|
||||
for features in catalog["features"].values():
|
||||
for f in features:
|
||||
all_codes.append(f["code"])
|
||||
|
||||
# Use the first two valid codes
|
||||
entries = [
|
||||
{"feature_code": all_codes[0], "limit_value": None, "enabled": True},
|
||||
{"feature_code": all_codes[1], "limit_value": 10, "enabled": True},
|
||||
]
|
||||
|
||||
response = client.put(
|
||||
f"{BASE}/tiers/{ft_tier.id}/limits",
|
||||
json=entries,
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data) == 2
|
||||
|
||||
def test_save_replaces_existing(
|
||||
self, client, super_admin_headers, ft_tier, ft_tier_with_features
|
||||
):
|
||||
"""Saving new features replaces the old ones entirely."""
|
||||
# Get a valid feature code
|
||||
catalog = client.get(f"{BASE}/catalog", headers=super_admin_headers).json()
|
||||
valid_code = next(
|
||||
f["code"]
|
||||
for features in catalog["features"].values()
|
||||
for f in features
|
||||
)
|
||||
|
||||
entries = [
|
||||
{"feature_code": valid_code, "limit_value": None, "enabled": True},
|
||||
]
|
||||
|
||||
response = client.put(
|
||||
f"{BASE}/tiers/{ft_tier.id}/limits",
|
||||
json=entries,
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data) == 1
|
||||
|
||||
# Verify old features are gone
|
||||
get_response = client.get(
|
||||
f"{BASE}/tiers/{ft_tier.id}/limits",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert len(get_response.json()) == 1
|
||||
|
||||
def test_save_empty_clears_features(
|
||||
self, client, super_admin_headers, ft_tier, ft_tier_with_features
|
||||
):
|
||||
"""Saving an empty list removes all features."""
|
||||
response = client.put(
|
||||
f"{BASE}/tiers/{ft_tier.id}/limits",
|
||||
json=[],
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json() == []
|
||||
|
||||
# Verify cleared
|
||||
get_response = client.get(
|
||||
f"{BASE}/tiers/{ft_tier.id}/limits",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert get_response.json() == []
|
||||
|
||||
def test_save_rejects_invalid_feature_codes(self, client, super_admin_headers, ft_tier):
|
||||
"""Returns error for unknown feature codes."""
|
||||
entries = [
|
||||
{"feature_code": "totally_fake_feature_xyz", "limit_value": None, "enabled": True},
|
||||
]
|
||||
|
||||
response = client.put(
|
||||
f"{BASE}/tiers/{ft_tier.id}/limits",
|
||||
json=entries,
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
# Should fail validation
|
||||
assert response.status_code in (400, 422)
|
||||
|
||||
def test_save_requires_auth(self, client, ft_tier):
|
||||
"""Endpoint requires authentication."""
|
||||
response = client.put(
|
||||
f"{BASE}/tiers/{ft_tier.id}/limits",
|
||||
json=[],
|
||||
)
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Cross-Platform Isolation (Regression Test)
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.billing
|
||||
class TestCrossPlatformIsolation:
|
||||
"""
|
||||
Regression tests for the tier_code ambiguity bug.
|
||||
|
||||
When multiple tiers share the same code (e.g., "essential" across platforms),
|
||||
feature operations must use tier_id to avoid saving to the wrong tier.
|
||||
"""
|
||||
|
||||
def test_features_saved_to_correct_tier(
|
||||
self, client, super_admin_headers, ft_duplicate_code_tiers
|
||||
):
|
||||
"""Features saved by tier_id go to the correct tier, not the first match."""
|
||||
tier_a, tier_b = ft_duplicate_code_tiers
|
||||
|
||||
# Get valid feature codes
|
||||
catalog = client.get(f"{BASE}/catalog", headers=super_admin_headers).json()
|
||||
codes = [
|
||||
f["code"]
|
||||
for features in catalog["features"].values()
|
||||
for f in features
|
||||
]
|
||||
|
||||
# Save features to tier B (second platform)
|
||||
entries_b = [
|
||||
{"feature_code": codes[0], "limit_value": None, "enabled": True},
|
||||
{"feature_code": codes[1], "limit_value": 50, "enabled": True},
|
||||
]
|
||||
response = client.put(
|
||||
f"{BASE}/tiers/{tier_b.id}/limits",
|
||||
json=entries_b,
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert len(response.json()) == 2
|
||||
|
||||
# Tier A should still have 0 features
|
||||
response_a = client.get(
|
||||
f"{BASE}/tiers/{tier_a.id}/limits",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response_a.status_code == 200
|
||||
assert len(response_a.json()) == 0
|
||||
|
||||
# Tier B should have 2 features
|
||||
response_b = client.get(
|
||||
f"{BASE}/tiers/{tier_b.id}/limits",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response_b.status_code == 200
|
||||
assert len(response_b.json()) == 2
|
||||
|
||||
def test_features_do_not_leak_between_same_code_tiers(
|
||||
self, client, super_admin_headers, ft_duplicate_code_tiers
|
||||
):
|
||||
"""Saving features to one tier doesn't affect another with the same code."""
|
||||
tier_a, tier_b = ft_duplicate_code_tiers
|
||||
|
||||
# Get valid feature codes
|
||||
catalog = client.get(f"{BASE}/catalog", headers=super_admin_headers).json()
|
||||
codes = [
|
||||
f["code"]
|
||||
for features in catalog["features"].values()
|
||||
for f in features
|
||||
]
|
||||
|
||||
# Save different features to each tier
|
||||
client.put(
|
||||
f"{BASE}/tiers/{tier_a.id}/limits",
|
||||
json=[{"feature_code": codes[0], "limit_value": None, "enabled": True}],
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
client.put(
|
||||
f"{BASE}/tiers/{tier_b.id}/limits",
|
||||
json=[{"feature_code": codes[1], "limit_value": None, "enabled": True}],
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
|
||||
# Each tier should have exactly its own feature
|
||||
resp_a = client.get(f"{BASE}/tiers/{tier_a.id}/limits", headers=super_admin_headers)
|
||||
resp_b = client.get(f"{BASE}/tiers/{tier_b.id}/limits", headers=super_admin_headers)
|
||||
|
||||
assert len(resp_a.json()) == 1
|
||||
assert resp_a.json()[0]["feature_code"] == codes[0]
|
||||
|
||||
assert len(resp_b.json()) == 1
|
||||
assert resp_b.json()[0]["feature_code"] == codes[1]
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Feature Count in Tier List (End-to-End)
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.billing
|
||||
class TestTierListFeatureCount:
|
||||
"""Tests that the tier list endpoint includes correct feature counts."""
|
||||
|
||||
def test_tier_list_includes_feature_codes(
|
||||
self, client, super_admin_headers, ft_tier, ft_tier_with_features
|
||||
):
|
||||
"""GET /tiers returns feature_codes for each tier."""
|
||||
response = client.get(
|
||||
"/api/v1/admin/subscriptions/tiers",
|
||||
headers=super_admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
# Find our test tier in the response
|
||||
tiers = response.json()["tiers"]
|
||||
our_tier = next((t for t in tiers if t["id"] == ft_tier.id), None)
|
||||
assert our_tier is not None
|
||||
assert len(our_tier["feature_codes"]) == 2
|
||||
assert set(our_tier["feature_codes"]) == {"basic_shop", "team_members"}
|
||||
Reference in New Issue
Block a user