80 lines
3.2 KiB
Python
80 lines
3.2 KiB
Python
"""SD-0002 invariants: INV-10 (never deletes), INV-14 (two-storefront zero bleed),
|
|
apply transactionality (§6.8), TEL-6."""
|
|
import json
|
|
import logging
|
|
|
|
import psycopg
|
|
import pytest
|
|
|
|
from app.domains import products
|
|
from app.domains.products import repo, service
|
|
from app.platform import db
|
|
|
|
CSV_A = b"Handle,Title,Variant Price\nmug,Mug,10.00\ntee,Tee,20.00\n"
|
|
CSV_PARTIAL = b"Handle,Title,Variant Price\nmug,Mug,12.00\n"
|
|
|
|
|
|
@pytest.fixture()
|
|
def migrated_conn(fresh_db_url):
|
|
with psycopg.connect(fresh_db_url) as conn:
|
|
db.migrate(conn)
|
|
yield conn
|
|
|
|
|
|
def _merchant(conn, email="m@example.com", shop="Shop"):
|
|
acct = conn.execute("INSERT INTO account (email) VALUES (%s) RETURNING id", (email,)).fetchone()[0]
|
|
sf = conn.execute("INSERT INTO storefront (name) VALUES (%s) RETURNING id", (shop,)).fetchone()[0]
|
|
conn.execute("INSERT INTO storefront_membership (account_id, storefront_id) VALUES (%s,%s)", (acct, sf))
|
|
conn.commit()
|
|
return acct, sf
|
|
|
|
|
|
def _import(conn, acct, sf, data):
|
|
d = products.import_validate(conn, sf, acct, "f.csv", data)
|
|
return products.confirm_draft(conn, sf, acct, d["id"])
|
|
|
|
|
|
def test_inv10_partial_file_never_deletes(migrated_conn):
|
|
acct, sf = _merchant(migrated_conn)
|
|
_import(migrated_conn, acct, sf, CSV_A)
|
|
before = migrated_conn.execute("SELECT count(*) FROM product").fetchone()[0]
|
|
_import(migrated_conn, acct, sf, CSV_PARTIAL)
|
|
after = migrated_conn.execute("SELECT count(*) FROM product").fetchone()[0]
|
|
assert after >= before == 2
|
|
|
|
|
|
def test_inv14_two_storefronts_zero_bleed(migrated_conn):
|
|
acct1, sf1 = _merchant(migrated_conn)
|
|
acct2, sf2 = _merchant(migrated_conn, "n@example.com", "Other")
|
|
_import(migrated_conn, acct1, sf1, CSV_A)
|
|
assert products.summary(migrated_conn, sf2)["product_count"] == 0
|
|
assert products.list_runs(migrated_conn, sf2) == []
|
|
_import(migrated_conn, acct2, sf2, CSV_A)
|
|
assert products.summary(migrated_conn, sf2)["product_count"] == 2
|
|
run1 = products.list_runs(migrated_conn, sf1)[0]
|
|
with pytest.raises(products.RunNotFound):
|
|
products.get_run(migrated_conn, sf2, run1["id"])
|
|
|
|
|
|
def test_apply_failure_rolls_back_whole_transaction_tel6(migrated_conn, monkeypatch, caplog):
|
|
acct, sf = _merchant(migrated_conn)
|
|
d = products.import_validate(migrated_conn, sf, acct, "f.csv", CSV_A)
|
|
|
|
def boom(*a, **k):
|
|
raise RuntimeError("mid-apply crash")
|
|
monkeypatch.setattr(service.repo, "insert_run_errors", boom)
|
|
lg = logging.getLogger("ecomm")
|
|
prior = lg.propagate
|
|
lg.propagate = True
|
|
try:
|
|
with caplog.at_level(logging.INFO, logger="ecomm.telemetry"):
|
|
with pytest.raises(RuntimeError):
|
|
products.confirm_draft(migrated_conn, sf, acct, d["id"])
|
|
finally:
|
|
lg.propagate = prior
|
|
assert migrated_conn.execute("SELECT count(*) FROM product").fetchone()[0] == 0
|
|
assert migrated_conn.execute("SELECT count(*) FROM import_run").fetchone()[0] == 0
|
|
assert migrated_conn.execute("SELECT count(*) FROM import_draft").fetchone()[0] == 1
|
|
events = [json.loads(r.message) for r in caplog.records if r.name == "ecomm.telemetry"]
|
|
assert any(e["event"] == "import_apply_failed" and e["error_class"] == "RuntimeError" for e in events)
|