backend/tests/test_storefronts_invariants.py

@@ -0,0 +1,64 @@
+"""INV-4's sharp edges (SD-0001 §6.8): concurrent second-storefront refusal, and the
+membership schema staying many-capable (the rule is a deletable guard, not a schema law)."""
+import threading
+
+import psycopg
+import pytest
+
+from app.domains import storefronts
+from app.platform import db
+
+
+@pytest.fixture()
+def migrated_url(fresh_db_url):
+    with psycopg.connect(fresh_db_url) as conn:
+        db.migrate(conn)
+    return fresh_db_url
+
+
+def test_inv_4_concurrent_creates_exactly_one_wins(migrated_url):
+    with psycopg.connect(migrated_url) as setup:
+        acct = setup.execute(
+            "INSERT INTO account (email) VALUES ('racer@example.com') RETURNING id"
+        ).fetchone()[0]
+        setup.commit()
+
+    barrier = threading.Barrier(2)
+    results: list[str] = []
+
+    def racer():
+        with psycopg.connect(migrated_url) as conn:
+            barrier.wait()
+            try:
+                storefronts.create_storefront(conn, acct, "racer@example.com", None)
+                results.append("created")
+            except storefronts.AlreadyOwnsStorefront:
+                results.append("refused")
+
+    threads = [threading.Thread(target=racer) for _ in range(2)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert sorted(results) == ["created", "refused"]
+    with psycopg.connect(migrated_url) as check:
+        count = check.execute(
+            "SELECT count(*) FROM storefront_membership WHERE account_id = %s", (acct,)
+        ).fetchone()[0]
+    assert count == 1
+
+
+def test_inv_4_schema_stays_many_capable(migrated_url):
+    # No UNIQUE(account_id) anywhere in the storefront path: the only unique constraint on
+    # storefront_membership is its composite PK, so many-per-account stays a deletable
+    # service rule (R-5 mitigation).
+    with psycopg.connect(migrated_url) as conn:
+        uniques = conn.execute(
+            "SELECT i.indkey::text FROM pg_index i"
+            " JOIN pg_class c ON c.oid = i.indrelid"
+            " WHERE c.relname = 'storefront_membership' AND (i.indisunique OR i.indisprimary)"
+        ).fetchall()
+    # exactly one unique index (the composite PK over two columns)
+    assert len(uniques) == 1
+    assert len(uniques[0][0].split()) == 2