diff --git a/backend/tests/test_storefronts_invariants.py b/backend/tests/test_storefronts_invariants.py new file mode 100644 index 0000000..6ce0e60 --- /dev/null +++ b/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