"""Shopify installation lifecycle: per-installation token, reinstall, and webhook addressing.""" from __future__ import annotations from datetime import UTC, datetime from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock from uuid import uuid4 import httpx import pytest from src.core.exceptions import ConflictError, ForbiddenError from src.domain.external_admin import router as router_module from src.domain.external_admin.models import ( INSTALLATION_STATUS_CONNECTED, INSTALLATION_STATUS_DISCONNECTED, INSTALLATION_STATUS_ERROR, INSTALLATION_STATUS_PENDING, ExternalAdminInstallation, ) from src.domain.external_admin.repository import ExternalAdminInstallationRepository from src.domain.external_admin.schemas import ( ShopifyConnectProjectRequest, ShopifyOAuthCallbackRequest, ) from src.domain.external_admin.service import ExternalAdminService from src.integrations import shopify_admin_client from src.integrations.shopify_admin_client import ShopifyAdminClient SHOP = "acme.myshopify.com" WORKSPACE = uuid4() def _installation(**over) -> ExternalAdminInstallation: inst = ExternalAdminInstallation( id=uuid4(), project_id=uuid4(), workspace_id=WORKSPACE, installation_type="shopify", status=INSTALLATION_STATUS_PENDING, shop_domain=SHOP, metadata_json={}, ) for key, value in over.items(): setattr(inst, key, value) return inst def _service(installs=None, credentials=None): installs = installs or MagicMock() for name in ("get_by_id", "get_by_project", "save", "create", "get_by_shop_domain"): if not isinstance(getattr(installs, name, None), AsyncMock): setattr(installs, name, AsyncMock()) if not isinstance(getattr(installs, "list_by_shop_domain", None), AsyncMock): installs.list_by_shop_domain = AsyncMock(return_value=[]) if credentials is None: credentials = MagicMock() credentials.get_active = AsyncMock(return_value=None) credentials.get_max_version = AsyncMock(return_value=0) credentials.create = AsyncMock() credentials.save = AsyncMock() projects = MagicMock() projects.get_by_id = AsyncMock( return_value=MagicMock(workspace_id=WORKSPACE, type="ai_assist_shopify") ) others = [MagicMock() for _ in range(8)] svc = ExternalAdminService( installs, credentials, others[0], others[1], others[2], projects, *others[3:] ) svc._encrypt = lambda value: f"enc({value})" return svc, installs, credentials # ── per-installation token ─────────────────────────────────────────────────── def _oauth_service(inst): svc, installs, credentials = _service() installs.get_by_id = AsyncMock(return_value=inst) svc._exchange_shopify_oauth_code = AsyncMock(return_value="shpat_shopify_access_token") svc._register_shopify_webhooks = AsyncMock() return svc, installs def _callback(inst) -> ShopifyOAuthCallbackRequest: return ShopifyOAuthCallbackRequest(code="auth-code", shop=SHOP, state=str(inst.id)) async def test_oauth_callback_mints_a_token_and_keeps_only_its_hash_and_an_encrypted_copy(): inst = _installation() svc, installs = _oauth_service(inst) out = await svc.handle_shopify_oauth_callback(_callback(inst)) token = out["installation_token"] assert len(token) >= 40 assert token != "shpat_shopify_access_token" # never Shopify's own access token assert inst.installation_token_enc == f"enc({token})" assert inst.metadata_json["installation_token_hash"] == svc._token_hash(token) assert token not in repr(inst.metadata_json) # the clear value is not stored assert inst.status == INSTALLATION_STATUS_CONNECTED installs.save.assert_awaited() svc._register_shopify_webhooks.assert_awaited_once_with(SHOP, "shpat_shopify_access_token") async def test_every_install_replaces_the_token_and_retires_the_old_one(): inst = _installation() svc, _ = _oauth_service(inst) first = (await svc.handle_shopify_oauth_callback(_callback(inst)))["installation_token"] first_hash = inst.metadata_json["installation_token_hash"] second = (await svc.handle_shopify_oauth_callback(_callback(inst)))["installation_token"] assert first != second assert inst.metadata_json["installation_token_hash"] == svc._token_hash(second) assert inst.metadata_json["installation_token_hash"] != first_hash # only the latest resolves async def test_reinstalling_reconnects_a_disconnected_installation(): inst = _installation(status=INSTALLATION_STATUS_DISCONNECTED, disconnected_at=datetime.now(UTC)) svc, _ = _oauth_service(inst) await svc.handle_shopify_oauth_callback(_callback(inst)) assert inst.status == INSTALLATION_STATUS_CONNECTED assert inst.disconnected_at is None async def test_the_issued_token_authenticates_that_installation_and_nothing_else(): inst = _installation() svc, installs = _oauth_service(inst) token = (await svc.handle_shopify_oauth_callback(_callback(inst)))["installation_token"] good_hash = svc._token_hash(token) installs.get_by_installation_token_hash = AsyncMock( side_effect=lambda h: inst if h == good_hash else None ) installs.get_by_installation_token = AsyncMock(return_value=None) resolved = await svc._resolve_shopify_installation_for_chat(token, str(inst.id), SHOP) assert resolved is inst with pytest.raises(ForbiddenError, match="invalid_installation_token"): await svc._resolve_shopify_installation_for_chat("not-the-token", str(inst.id), SHOP) with pytest.raises(ForbiddenError, match="installation_id_mismatch"): await svc._resolve_shopify_installation_for_chat(token, str(uuid4()), SHOP) with pytest.raises(ForbiddenError, match="shop_domain_mismatch"): await svc._resolve_shopify_installation_for_chat( token, str(inst.id), "victim.myshopify.com" ) async def test_an_unknown_installation_still_fails_the_callback(): svc, installs, _ = _service() installs.get_by_id = AsyncMock(return_value=None) installs.get_by_shop_domain = AsyncMock(return_value=None) with pytest.raises(Exception, match="external_admin_installation_not_found"): await svc.handle_shopify_oauth_callback( ShopifyOAuthCallbackRequest(code="c", shop=SHOP, state=str(uuid4())) ) # ── the routes ─────────────────────────────────────────────────────────────── def _app(svc): from fastapi import FastAPI app = FastAPI() app.include_router(router_module.plugin_router) app.dependency_overrides[router_module._svc] = lambda: svc app.dependency_overrides[router_module._require_shopify_edge] = lambda: None return app async def _post(app, path, body): transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://core") as client: return await client.post(path, json=body) # note: no Authorization header async def test_the_oauth_callback_needs_no_bearer_and_returns_the_token(): svc = MagicMock() svc.handle_shopify_oauth_callback = AsyncMock( return_value={ "ok": True, "installation_id": str(uuid4()), "shop_domain": SHOP, "redirect_url": f"https://{SHOP}/admin/apps/p2hs-ai", "installation_token": "tok-123", } ) res = await _post( _app(svc), "/v1/external-admin/shopify/oauth/callback", {"code": "c", "shop": SHOP, "state": "s"}, ) assert res.status_code == 200 assert res.json()["installation_token"] == "tok-123" async def test_the_shop_webhook_needs_no_bearer(): svc = MagicMock() svc.handle_shopify_webhook = AsyncMock(return_value={"ok": True, "status": "disconnected"}) body = { "topic": "app/uninstalled", "shopDomain": SHOP, "payload": {}, "receivedAt": "2026-09-27T00:00:00Z", } res = await _post(_app(svc), "/v1/external-admin/shopify/webhooks", body) assert res.status_code == 200 svc.handle_shopify_webhook.assert_awaited_once() async def test_the_authenticated_shopify_routes_still_require_a_bearer(): svc = MagicMock() res = await _post( _app(svc), "/v1/external-admin/shopify/health", {"installation_id": str(uuid4())} ) assert res.status_code == 422 # Authorization header is required and was not sent # ── reinstall: one installation per shop, reused ───────────────────────────── async def test_reconnecting_reuses_the_workspaces_existing_project(): existing = _installation(status=INSTALLATION_STATUS_DISCONNECTED) svc, installs, _ = _service() installs.list_by_shop_domain = AsyncMock(return_value=[existing]) assert ( await svc.resolve_shopify_connect_target(WORKSPACE, "ACME.MyShopify.com") == existing.project_id ) installs.list_by_shop_domain.assert_awaited_once_with(SHOP) # normalized async def test_another_workspaces_live_installation_blocks_the_connect(): other = _installation(workspace_id=uuid4(), status=INSTALLATION_STATUS_CONNECTED) svc, installs, _ = _service() installs.list_by_shop_domain = AsyncMock(return_value=[other]) with pytest.raises(ConflictError, match="shop_already_connected"): await svc.resolve_shopify_connect_target(WORKSPACE, SHOP) @pytest.mark.parametrize( "status", [INSTALLATION_STATUS_DISCONNECTED, INSTALLATION_STATUS_PENDING, INSTALLATION_STATUS_ERROR], ) async def test_another_workspaces_dead_installation_does_not_block(status): other = _installation(workspace_id=uuid4(), status=status) svc, installs, _ = _service() installs.list_by_shop_domain = AsyncMock(return_value=[other]) assert await svc.resolve_shopify_connect_target(WORKSPACE, SHOP) is None async def test_a_shop_nobody_has_connected_gets_a_new_project(): svc, _, _ = _service() assert await svc.resolve_shopify_connect_target(WORKSPACE, SHOP) is None async def test_pending_installation_restarts_a_disconnected_one_instead_of_adding_another(): existing = _installation( status=INSTALLATION_STATUS_DISCONNECTED, disconnected_at=datetime.now(UTC) ) svc, installs, _ = _service() installs.get_by_project = AsyncMock(return_value=existing) out = await svc.create_shopify_pending_installation(existing.project_id, WORKSPACE, SHOP) assert out.installation_id == str(existing.id) # the same installation assert existing.status == INSTALLATION_STATUS_PENDING assert existing.disconnected_at is None installs.create.assert_not_awaited() async def test_pending_installation_leaves_a_connected_one_connected(): existing = _installation(status=INSTALLATION_STATUS_CONNECTED) svc, installs, _ = _service() installs.get_by_project = AsyncMock(return_value=existing) await svc.create_shopify_pending_installation(existing.project_id, WORKSPACE, SHOP) assert existing.status == INSTALLATION_STATUS_CONNECTED # re-authorising must not drop it async def test_pending_installation_is_refused_while_another_project_holds_the_shop(): other = _installation(status=INSTALLATION_STATUS_CONNECTED) # a different project svc, installs, _ = _service() installs.list_by_shop_domain = AsyncMock(return_value=[other]) installs.get_by_project = AsyncMock(return_value=None) with pytest.raises(ConflictError, match="shop_already_connected"): await svc.create_shopify_pending_installation(uuid4(), WORKSPACE, SHOP) installs.create.assert_not_awaited() async def test_pending_installation_creates_one_when_none_exists(): svc, installs, _ = _service() installs.get_by_project = AsyncMock(return_value=None) project_id = uuid4() await svc.create_shopify_pending_installation(project_id, WORKSPACE, SHOP) installs.create.assert_awaited_once() created = installs.create.await_args.args[0] assert (created.project_id, created.shop_domain, created.status) == ( project_id, SHOP, INSTALLATION_STATUS_PENDING, ) def _connect_route_args(svc, project_service): return dict( body=ShopifyConnectProjectRequest( shop_domain=SHOP, name="Shopify AI - acme", description=None ), workspace_id=WORKSPACE, user=SimpleNamespace(id=uuid4()), project_service=project_service, svc=svc, ) async def test_connect_route_reuses_the_existing_project_and_creates_none(): reused = uuid4() svc = MagicMock() svc.resolve_shopify_connect_target = AsyncMock(return_value=reused) svc.create_shopify_pending_installation = AsyncMock(return_value="result") project_service = MagicMock(create_project=AsyncMock()) out = await router_module.create_shopify_connect_project( **_connect_route_args(svc, project_service) ) assert out == "result" project_service.create_project.assert_not_awaited() svc.create_shopify_pending_installation.assert_awaited_once_with(reused, WORKSPACE, SHOP) async def test_connect_route_creates_a_project_for_a_new_shop(): new_project = SimpleNamespace(id=uuid4()) svc = MagicMock() svc.resolve_shopify_connect_target = AsyncMock(return_value=None) svc.create_shopify_pending_installation = AsyncMock(return_value="result") project_service = MagicMock(create_project=AsyncMock(return_value=new_project)) await router_module.create_shopify_connect_project(**_connect_route_args(svc, project_service)) project_service.create_project.assert_awaited_once() svc.create_shopify_pending_installation.assert_awaited_once_with( new_project.id, WORKSPACE, SHOP ) async def test_connect_route_refuses_before_creating_any_project_when_the_shop_is_taken(): svc = MagicMock() svc.resolve_shopify_connect_target = AsyncMock( side_effect=ConflictError("shop_already_connected") ) project_service = MagicMock(create_project=AsyncMock()) with pytest.raises(ConflictError): await router_module.create_shopify_connect_project( **_connect_route_args(svc, project_service) ) project_service.create_project.assert_not_awaited() # no orphan project # ── the shop lookup can no longer throw on duplicates ──────────────────────── class _Session: def __init__(self): self.statements = [] async def execute(self, stmt): self.statements.append(stmt) result = MagicMock() result.scalars.return_value.first.return_value = None result.scalars.return_value.all.return_value = [] return result async def test_shop_lookup_prefers_the_live_installation_and_takes_exactly_one(): db = _Session() await ExternalAdminInstallationRepository(db).get_by_shop_domain(SHOP) sql = str(db.statements[0].compile()).upper() assert "ORDER BY CASE" in sql assert sql.index("CASE") < sql.index("UPDATED_AT DESC") # live status first, then most recent assert "LIMIT" in sql # a single row: scalar_one_or_none() is what used to raise async def test_listing_by_shop_is_ordered_the_same_way(): db = _Session() assert await ExternalAdminInstallationRepository(db).list_by_shop_domain(SHOP) == [] assert "ORDER BY CASE" in str(db.statements[0].compile()).upper() # ── webhook addresses ──────────────────────────────────────────────────────── class _FakeHttpClient: posts: list[dict] = [] async def __aenter__(self): return self async def __aexit__(self, *exc): return False async def post(self, url, headers=None, json=None): self.posts.append(json["webhook"]) return SimpleNamespace( status_code=201, json=lambda: {"webhook": {"id": len(self.posts)}}, text="" ) async def test_webhooks_are_registered_at_hyphenated_paths_the_bff_serves(monkeypatch): _FakeHttpClient.posts = [] monkeypatch.setattr(shopify_admin_client.httpx, "AsyncClient", lambda **kw: _FakeHttpClient()) monkeypatch.setattr(ShopifyAdminClient, "_get", AsyncMock(return_value={"webhooks": []})) topics = [ "app/uninstalled", "customers/data_request", "customers/redact", "shop/redact", "products/update", "themes/delete", "shop/update", ] await ShopifyAdminClient(SHOP, "token").register_webhooks( "https://shopifyapp.p2hs.com/", topics ) addresses = {w["topic"]: w["address"] for w in _FakeHttpClient.posts} assert ( addresses["customers/data_request"] == "https://shopifyapp.p2hs.com/webhooks/shopify/customers-data-request" ) assert ( addresses["app/uninstalled"] == "https://shopifyapp.p2hs.com/webhooks/shopify/app-uninstalled" ) assert addresses["shop/redact"] == "https://shopifyapp.p2hs.com/webhooks/shopify/shop-redact" assert ( addresses["products/update"] == "https://shopifyapp.p2hs.com/webhooks/shopify/products-update" ) for address in addresses.values(): assert "_" not in address.rsplit("/", 1)[1] # the BFF's slugs are hyphen-only async def test_registration_is_idempotent_for_an_existing_correct_subscription(monkeypatch): _FakeHttpClient.posts = [] monkeypatch.setattr(shopify_admin_client.httpx, "AsyncClient", lambda **kw: _FakeHttpClient()) existing = { "webhooks": [ { "id": 7, "topic": "customers/data_request", "address": "https://shopifyapp.p2hs.com/webhooks/shopify/customers-data-request", } ] } monkeypatch.setattr(ShopifyAdminClient, "_get", AsyncMock(return_value=existing)) results = await ShopifyAdminClient(SHOP, "token").register_webhooks( "https://shopifyapp.p2hs.com", ["customers/data_request"] ) assert results[0]["action"] == "exists" assert _FakeHttpClient.posts == []