65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
"""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
|