"""Shopify mandatory privacy webhooks (GDPR): shop/redact and the two customers/* topics.""" from __future__ import annotations from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock from uuid import uuid4 import pytest from src.core.exceptions import NotFoundError from src.domain.external_admin.models import ( INSTALLATION_STATUS_CONNECTED, INSTALLATION_STATUS_DISCONNECTED, INSTALLATION_STATUS_REDACTED, ExternalAdminInstallation, ) from src.domain.external_admin.repository import ExternalAdminInstallationRepository from src.domain.external_admin.schemas import ShopifyWebhookEnvelope from src.domain.external_admin.service import ExternalAdminService SHOP = "acme.myshopify.com" def _installation(**over) -> ExternalAdminInstallation: inst = ExternalAdminInstallation( id=uuid4(), project_id=uuid4(), workspace_id=uuid4(), installation_type="shopify", status=INSTALLATION_STATUS_CONNECTED, installation_token_enc="encrypted-token", shop_domain=SHOP, site_url="https://acme.example", plugin_version="1.2.3", companion_version="0.9", last_error="boom", metadata_json={"shopify_webhooks": [{"topic": "products/update"}]}, ) for key, value in over.items(): setattr(inst, key, value) return inst def _service(installation=None, removed=None): repo = MagicMock() repo.get_by_shop_domain = AsyncMock(return_value=installation) repo.save = AsyncMock() repo.redact_installation_data = AsyncMock(return_value=removed or {"external_credentials": 1}) others = [MagicMock() for _ in range(10)] svc = ExternalAdminService(repo, *others) return svc, repo def _envelope(topic: str, shop: str = SHOP, payload=None) -> ShopifyWebhookEnvelope: return ShopifyWebhookEnvelope( topic=topic, shopDomain=shop, payload=payload if payload is not None else {}, receivedAt="2026-09-27T10:00:00Z", ) # ── shop/redact ────────────────────────────────────────────────────────────── async def test_shop_redact_erases_the_shop_and_leaves_only_a_tombstone(): inst = _installation() removed = {"external_admin_chat_messages": 4, "external_credentials": 1} svc, repo = _service(inst, removed) out = await svc.handle_shopify_webhook(_envelope("shop/redact")) assert out == {"ok": True, "topic": "shop/redact", "status": "shop_data_erased"} repo.redact_installation_data.assert_awaited_once_with(inst.id) repo.save.assert_awaited_once_with(inst) assert inst.status == INSTALLATION_STATUS_REDACTED assert inst.installation_token_enc is None assert inst.shop_domain is None assert inst.site_url is None assert inst.last_error is None assert inst.plugin_version is None and inst.companion_version is None assert inst.disconnected_at is not None # Webhook history is gone; the only metadata left is the audit of the erasure. assert list(inst.metadata_json) == ["privacy_requests"] (record,) = inst.metadata_json["privacy_requests"] assert record["topic"] == "shop/redact" assert record["outcome"] == "shop_data_erased" assert record["removed"] == removed async def test_shop_redact_is_idempotent_because_shopify_retries(): inst = _installation() svc, repo = _service(inst) await svc.handle_shopify_webhook(_envelope("shop/redact")) # The tombstone no longer matches the shop domain, so a retry finds nothing. repo.get_by_shop_domain.return_value = None repo.redact_installation_data.reset_mock() repo.save.reset_mock() out = await svc.handle_shopify_webhook(_envelope("shop/redact")) assert out["status"] == "no_data_held" repo.redact_installation_data.assert_not_awaited() repo.save.assert_not_awaited() async def test_shop_redact_erases_even_when_we_still_think_it_is_connected(): """A missed app/uninstalled webhook must not block a legal erasure request.""" inst = _installation(status=INSTALLATION_STATUS_CONNECTED) svc, repo = _service(inst) await svc.handle_shopify_webhook(_envelope("shop/redact")) repo.redact_installation_data.assert_awaited_once() async def test_shop_redact_normalizes_the_shop_domain_before_lookup(): svc, repo = _service(_installation()) await svc.handle_shopify_webhook( _envelope("shop/redact", shop="HTTPS://Acme.MyShopify.com/admin") ) repo.get_by_shop_domain.assert_awaited_once_with(SHOP) # ── unknown shops answer 200, never 404 ────────────────────────────────────── @pytest.mark.parametrize("topic", ["shop/redact", "customers/redact", "customers/data_request"]) async def test_privacy_topics_for_an_unknown_shop_succeed(topic): svc, repo = _service(None) out = await svc.handle_shopify_webhook(_envelope(topic)) assert out == {"ok": True, "topic": topic, "status": "no_data_held"} repo.save.assert_not_awaited() # ── customers/* ────────────────────────────────────────────────────────────── @pytest.mark.parametrize("topic", ["customers/redact", "customers/data_request"]) async def test_customer_requests_are_recorded_without_storing_the_customer(topic): inst = _installation() svc, repo = _service(inst) payload = { "customer": {"id": 191167, "email": "john@example.com", "phone": "+1555"}, "orders_to_redact": [299938], } out = await svc.handle_shopify_webhook(_envelope(topic, payload=payload)) assert out == {"ok": True, "topic": topic, "status": "no_customer_data_held"} repo.redact_installation_data.assert_not_awaited() assert inst.status == INSTALLATION_STATUS_CONNECTED assert inst.shop_domain == SHOP # Existing metadata is kept; the request is appended. assert inst.metadata_json["shopify_webhooks"] == [{"topic": "products/update"}] (record,) = inst.metadata_json["privacy_requests"] assert record["topic"] == topic and record["outcome"] == "no_customer_data_held" # The payload (personal data) is never persisted. blob = repr(inst.metadata_json) for secret in ("john@example.com", "+1555", "191167", "299938"): assert secret not in blob async def test_customer_request_history_is_bounded(): inst = _installation( metadata_json={ "privacy_requests": [{"topic": "customers/redact", "n": i} for i in range(80)] } ) svc, _ = _service(inst) await svc.handle_shopify_webhook(_envelope("customers/data_request")) assert len(inst.metadata_json["privacy_requests"]) == 50 # ── other topics are unchanged ─────────────────────────────────────────────── async def test_uninstall_still_disconnects(): inst = _installation() svc, repo = _service(inst) out = await svc.handle_shopify_webhook(_envelope("app/uninstalled")) assert out["status"] == INSTALLATION_STATUS_DISCONNECTED assert inst.status == INSTALLATION_STATUS_DISCONNECTED repo.redact_installation_data.assert_not_awaited() async def test_non_privacy_topics_for_an_unknown_shop_still_404(): svc, _ = _service(None) with pytest.raises(NotFoundError): await svc.handle_shopify_webhook(_envelope("products/update")) # ── repository: what "erase" actually deletes ──────────────────────────────── class _RecordingSession: def __init__(self): self.statements = [] async def execute(self, stmt): self.statements.append(stmt) return SimpleNamespace(rowcount=2) async def test_redact_deletes_every_installation_table_in_dependency_order(): db = _RecordingSession() installation_id = uuid4() removed = await ExternalAdminInstallationRepository(db).redact_installation_data( installation_id ) assert list(removed) == [ "external_admin_chat_jobs", "external_admin_chat_messages", "external_admin_chat_sessions", "assistant_action_sessions", "site_understanding_runs", "wp_mutation_actions", "external_credentials", ] assert all(count == 2 for count in removed.values()) for stmt in db.statements: sql = str(stmt.compile()) assert sql.startswith("DELETE FROM") # Never an unscoped delete. assert "installation_id = :installation_id" in sql assert stmt.compile().params["installation_id_1"] == installation_id