Compare commits

...

2 Commits

Author SHA1 Message Date
a1cc05cd3d fix(tests): remove stale onboarding redirect tests and mock billing limits
Some checks failed
CI / ruff (push) Successful in 12s
CI / docs (push) Has been cancelled
CI / deploy (push) Has been cancelled
CI / validate (push) Has been cancelled
CI / dependency-scanning (push) Has been cancelled
CI / pytest (push) Has been cancelled
Remove marketplace page redirect-to-onboarding tests that no longer
match the route behavior. Add can_create_store mock to tenancy store
creation tests to bypass billing limit checks.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-12 22:32:35 +01:00
19d267587b fix(loyalty): use private key PEM for JWT signing instead of RSASigner.key
RSASigner doesn't expose a .key attribute. Load the private key string
directly from the service account JSON file for PyJWT encoding. Also
adds fat JWT fallback for demo mode where DRAFT classes reject object
creation via REST API.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-12 22:31:02 +01:00
4 changed files with 52 additions and 126 deletions

View File

@@ -472,34 +472,60 @@ class GoogleWalletService:
if not self.is_configured:
raise GoogleWalletNotConfiguredException()
# Try to create the object via API. If it fails (e.g. demo mode
# where DRAFT classes reject object creation), fall back to
# embedding the full object data in the JWT ("fat JWT").
if not card.google_object_id:
self.create_object(db, card)
try:
self.create_object(db, card)
except WalletIntegrationException:
logger.info(
"Object creation failed for card %s, using fat JWT",
card.id,
)
try:
import jwt
credentials = self._get_credentials()
signer = self._get_signer()
now = datetime.now(tz=UTC)
origins = settings.loyalty_google_wallet_origins or []
issuer_id = settings.loyalty_google_issuer_id
object_id = card.google_object_id or f"{issuer_id}.loyalty_card_{card.id}"
if card.google_object_id:
# Object exists in Google — reference by ID only
payload = {
"loyaltyObjects": [{"id": card.google_object_id}],
}
else:
# Object not created — embed full object data in JWT
object_data = self._build_object_data(card, object_id)
payload = {
"loyaltyObjects": [object_data],
}
claims = {
"iss": credentials.service_account_email,
"aud": "google",
"origins": origins,
"typ": "savetowallet",
"payload": {
"loyaltyObjects": [{"id": card.google_object_id}],
},
"payload": payload,
"iat": now,
"exp": now + timedelta(hours=1),
}
# Sign using the RSASigner's key_id and key bytes (public API)
# Load the private key directly from the service account file
# (RSASigner doesn't expose .key; PyJWT needs the PEM string)
with open(settings.loyalty_google_service_account_json) as f:
sa_data = json.load(f)
private_key = sa_data["private_key"]
token = jwt.encode(
claims,
signer.key,
private_key,
algorithm="RS256",
)
@@ -507,7 +533,7 @@ class GoogleWalletService:
db.commit()
return f"https://pay.google.com/gp/v/save/{token}"
except (AttributeError, ValueError, KeyError) as exc:
except (AttributeError, ValueError, KeyError, OSError) as exc:
logger.error(
"Failed to generate Google Wallet save URL: %s", exc
)

View File

@@ -302,19 +302,6 @@ class TestOnboardingPageRoutes:
class TestMarketplacePageRoutes:
"""Tests for GET /store/{store_code}/marketplace."""
def test_redirects_to_onboarding_when_not_completed(
self, client, db, mp_auth, mp_store, mp_store_with_marketplace,
mp_onboarding_not_completed,
):
"""Marketplace page redirects to onboarding when not completed."""
response = client.get(
f"/store/{mp_store.subdomain}/marketplace",
headers=mp_auth,
follow_redirects=False,
)
assert response.status_code == 302
assert f"/store/{mp_store.subdomain}/onboarding" in response.headers["location"]
def test_renders_when_onboarding_completed(
self, client, db, mp_auth, mp_store, mp_store_with_marketplace,
mp_onboarding_completed,
@@ -327,18 +314,6 @@ class TestMarketplacePageRoutes:
)
assert response.status_code == 200
def test_redirects_when_no_onboarding_record(
self, client, db, mp_auth, mp_store, mp_store_with_marketplace,
):
"""Marketplace page redirects when no onboarding record (not completed)."""
response = client.get(
f"/store/{mp_store.subdomain}/marketplace",
headers=mp_auth,
follow_redirects=False,
)
assert response.status_code == 302
assert f"/store/{mp_store.subdomain}/onboarding" in response.headers["location"]
# ============================================================================
# Letzshop page tests
@@ -350,19 +325,6 @@ class TestMarketplacePageRoutes:
class TestLetzshopPageRoutes:
"""Tests for GET /store/{store_code}/letzshop."""
def test_redirects_to_onboarding_when_not_completed(
self, client, db, mp_auth, mp_store, mp_store_with_marketplace,
mp_onboarding_not_completed,
):
"""Letzshop page redirects to onboarding when not completed."""
response = client.get(
f"/store/{mp_store.subdomain}/letzshop",
headers=mp_auth,
follow_redirects=False,
)
assert response.status_code == 302
assert f"/store/{mp_store.subdomain}/onboarding" in response.headers["location"]
def test_renders_when_onboarding_completed(
self, client, db, mp_auth, mp_store, mp_store_with_marketplace,
mp_onboarding_completed,
@@ -374,15 +336,3 @@ class TestLetzshopPageRoutes:
follow_redirects=False,
)
assert response.status_code == 200
def test_redirects_when_no_onboarding_record(
self, client, db, mp_auth, mp_store, mp_store_with_marketplace,
):
"""Letzshop page redirects when no onboarding record (not completed)."""
response = client.get(
f"/store/{mp_store.subdomain}/letzshop",
headers=mp_auth,
follow_redirects=False,
)
assert response.status_code == 302
assert f"/store/{mp_store.subdomain}/onboarding" in response.headers["location"]

View File

@@ -186,42 +186,15 @@ class TestOnboardingPageGuard:
@pytest.mark.unit
@pytest.mark.marketplace
@pytest.mark.asyncio
class TestMarketplacePageOnboardingGate:
"""Test that marketplace page redirects to onboarding when not completed."""
class TestMarketplacePageRendering:
"""Test that marketplace page renders correctly."""
@patch("app.modules.marketplace.routes.pages.store.OnboardingService")
@patch("app.modules.marketplace.routes.pages.store.get_store_context")
@patch("app.modules.marketplace.routes.pages.store.templates")
async def test_redirects_to_onboarding_when_not_completed(
self, mock_templates, mock_ctx, mock_onboarding_cls, db
async def test_renders_marketplace_page(
self, mock_templates, mock_ctx, db
):
"""Redirect to onboarding if onboarding not completed."""
mock_svc = MagicMock()
mock_svc.is_completed.return_value = False
mock_onboarding_cls.return_value = mock_svc
user = _make_user_context()
response = await store_marketplace_page(
request=_make_request(),
store_code="teststore",
current_user=user,
db=db,
)
assert isinstance(response, RedirectResponse)
assert response.status_code == 302
assert "/store/teststore/onboarding" in response.headers["location"]
@patch("app.modules.marketplace.routes.pages.store.OnboardingService")
@patch("app.modules.marketplace.routes.pages.store.get_store_context")
@patch("app.modules.marketplace.routes.pages.store.templates")
async def test_renders_marketplace_when_onboarding_completed(
self, mock_templates, mock_ctx, mock_onboarding_cls, db
):
"""Render marketplace page if onboarding is completed."""
mock_svc = MagicMock()
mock_svc.is_completed.return_value = True
mock_onboarding_cls.return_value = mock_svc
"""Render marketplace page."""
mock_ctx.return_value = {"request": _make_request()}
mock_templates.TemplateResponse.return_value = "rendered"
@@ -246,42 +219,15 @@ class TestMarketplacePageOnboardingGate:
@pytest.mark.unit
@pytest.mark.marketplace
@pytest.mark.asyncio
class TestLetzshopPageOnboardingGate:
"""Test that letzshop page redirects to onboarding when not completed."""
class TestLetzshopPageRendering:
"""Test that letzshop page renders correctly."""
@patch("app.modules.marketplace.routes.pages.store.OnboardingService")
@patch("app.modules.marketplace.routes.pages.store.get_store_context")
@patch("app.modules.marketplace.routes.pages.store.templates")
async def test_redirects_to_onboarding_when_not_completed(
self, mock_templates, mock_ctx, mock_onboarding_cls, db
async def test_renders_letzshop_page(
self, mock_templates, mock_ctx, db
):
"""Redirect to onboarding if onboarding not completed."""
mock_svc = MagicMock()
mock_svc.is_completed.return_value = False
mock_onboarding_cls.return_value = mock_svc
user = _make_user_context()
response = await store_letzshop_page(
request=_make_request(),
store_code="teststore",
current_user=user,
db=db,
)
assert isinstance(response, RedirectResponse)
assert response.status_code == 302
assert "/store/teststore/onboarding" in response.headers["location"]
@patch("app.modules.marketplace.routes.pages.store.OnboardingService")
@patch("app.modules.marketplace.routes.pages.store.get_store_context")
@patch("app.modules.marketplace.routes.pages.store.templates")
async def test_renders_letzshop_when_onboarding_completed(
self, mock_templates, mock_ctx, mock_onboarding_cls, db
):
"""Render letzshop page if onboarding is completed."""
mock_svc = MagicMock()
mock_svc.is_completed.return_value = True
mock_onboarding_cls.return_value = mock_svc
"""Render letzshop page."""
mock_ctx.return_value = {"request": _make_request()}
mock_templates.TemplateResponse.return_value = "rendered"

View File

@@ -193,7 +193,8 @@ class TestMerchantStoreServiceCreate:
assert result["is_active"] is True
assert result["is_verified"] is False
def test_create_store_duplicate_code(self, db, merchant_owner, merchant_store):
@patch.object(MerchantStoreService, "can_create_store", return_value=(True, None))
def test_create_store_duplicate_code(self, _mock_limit, db, merchant_owner, merchant_store):
"""Test creating store with duplicate store code."""
store_data = {
"name": "Another Store",
@@ -206,7 +207,8 @@ class TestMerchantStoreServiceCreate:
with pytest.raises(StoreAlreadyExistsException):
self.service.create_store(db, merchant_owner.id, store_data)
def test_create_store_duplicate_subdomain(self, db, merchant_owner, merchant_store):
@patch.object(MerchantStoreService, "can_create_store", return_value=(True, None))
def test_create_store_duplicate_subdomain(self, _mock_limit, db, merchant_owner, merchant_store):
"""Test creating store with duplicate subdomain."""
store_data = {
"name": "Another Store",
@@ -219,7 +221,8 @@ class TestMerchantStoreServiceCreate:
with pytest.raises(StoreValidationException):
self.service.create_store(db, merchant_owner.id, store_data)
def test_create_store_nonexistent_merchant(self, db):
@patch.object(MerchantStoreService, "can_create_store", return_value=(True, None))
def test_create_store_nonexistent_merchant(self, _mock_limit, db):
"""Test creating store for non-existent merchant."""
store_data = {
"name": "No Merchant Store",
@@ -232,7 +235,8 @@ class TestMerchantStoreServiceCreate:
with pytest.raises(MerchantNotFoundException):
self.service.create_store(db, 99999, store_data)
def test_create_store_creates_default_roles(self, db, merchant_owner):
@patch.object(MerchantStoreService, "can_create_store", return_value=(True, None))
def test_create_store_creates_default_roles(self, _mock_limit, db, merchant_owner):
"""Test that default roles are created for new store."""
from app.modules.tenancy.models.store import Role