136 lines
5.8 KiB
Python
136 lines
5.8 KiB
Python
"""§6.4 /api/products/* endpoint scenarios (PUC-2/3/3a/4/5/5a/8 + gates)."""
|
|
import io
|
|
import re
|
|
from contextlib import contextmanager
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import create_app
|
|
|
|
GOOD_CSV = b"Handle,Title,Vendor,Variant Price\nmoon-mug,Moon Mug,Acme,18.00\n"
|
|
|
|
|
|
@contextmanager
|
|
def _merchant_client(fresh_db_url, email="m@example.com"):
|
|
with TestClient(create_app(database_url=fresh_db_url)) as client:
|
|
client.post("/api/auth/request-code", json={"email": email})
|
|
code = re.search(r"\b(\d{6})\b", client.app.state.mailer.outbox[-1].body).group(1)
|
|
client.post("/api/auth/verify", json={"email": email, "code": code})
|
|
client.post("/api/storefronts", json={})
|
|
yield client
|
|
|
|
|
|
def _upload(client, data=GOOD_CSV, name="cat.csv"):
|
|
return client.post("/api/products/imports", files={"file": (name, io.BytesIO(data), "text/csv")})
|
|
|
|
|
|
def test_upload_returns_201_draft(fresh_db_url):
|
|
with _merchant_client(fresh_db_url) as client:
|
|
resp = _upload(client)
|
|
assert resp.status_code == 201
|
|
body = resp.json()
|
|
assert body["summary"]["adds"] == 1 and body["dialect"] == "canonical"
|
|
|
|
|
|
def test_upload_rejections_carry_codes(fresh_db_url):
|
|
with _merchant_client(fresh_db_url) as client:
|
|
resp = _upload(client, b"Vendor\nAcme\n")
|
|
assert resp.status_code == 400
|
|
assert resp.json()["error"]["code"] == "missing_required_column"
|
|
resp = _upload(client, b"Handle,Title\n" + b"x" * (10 * 1024 * 1024 + 1))
|
|
assert resp.status_code == 413
|
|
|
|
|
|
def test_unauthenticated_401_and_no_storefront_404(fresh_db_url):
|
|
with TestClient(create_app(database_url=fresh_db_url)) as client:
|
|
assert _upload(client).status_code == 401
|
|
client.post("/api/auth/request-code", json={"email": "x@example.com"})
|
|
code = re.search(r"\b(\d{6})\b", client.app.state.mailer.outbox[-1].body).group(1)
|
|
client.post("/api/auth/verify", json={"email": "x@example.com", "code": code})
|
|
assert _upload(client).status_code == 404
|
|
|
|
|
|
def test_preview_confirm_run_flow(fresh_db_url):
|
|
with _merchant_client(fresh_db_url) as client:
|
|
draft = _upload(client).json()
|
|
recs = client.get(f"/api/products/imports/drafts/{draft['id']}/records").json()["records"]
|
|
assert recs[0]["kind"] == "add"
|
|
run_id = client.post(f"/api/products/imports/drafts/{draft['id']}/confirm").json()["run_id"]
|
|
run = client.get(f"/api/products/imports/runs/{run_id}").json()
|
|
assert run["products_added"] == 1 and run["by"] == "m@example.com"
|
|
assert client.get("/api/products/summary").json()["product_count"] == 1
|
|
assert client.get("/api/products/imports/runs").json()["runs"][0]["id"] == run_id
|
|
|
|
|
|
def test_cancel_no_trace_puc3a(fresh_db_url):
|
|
with _merchant_client(fresh_db_url) as client:
|
|
draft = _upload(client).json()
|
|
assert client.delete(f"/api/products/imports/drafts/{draft['id']}").status_code == 204
|
|
assert client.get(f"/api/products/imports/drafts/{draft['id']}").status_code == 404
|
|
assert client.get("/api/products/imports/runs").json()["runs"] == []
|
|
|
|
|
|
def test_confirm_conflicts(fresh_db_url):
|
|
with _merchant_client(fresh_db_url) as client:
|
|
d1 = _upload(client).json()
|
|
client.post(f"/api/products/imports/drafts/{d1['id']}/confirm")
|
|
d2 = _upload(client).json()
|
|
resp = client.post(f"/api/products/imports/drafts/{d2['id']}/confirm")
|
|
assert resp.status_code == 409 and resp.json()["error"]["code"] == "nothing_to_apply"
|
|
|
|
|
|
def test_sample_csv_served(fresh_db_url):
|
|
with TestClient(create_app(database_url=fresh_db_url)) as client:
|
|
resp = client.get("/api/products/sample.csv")
|
|
assert resp.status_code == 200
|
|
assert resp.headers["content-type"].startswith("text/csv")
|
|
assert resp.text.startswith("Handle,Title,")
|
|
|
|
|
|
def test_sample_csv_imports_clean(fresh_db_url):
|
|
"""DOC-3 honesty: our own sample must validate with zero errors."""
|
|
with _merchant_client(fresh_db_url) as client:
|
|
sample = client.get("/api/products/sample.csv").content
|
|
body = _upload(client, sample, "sample.csv").json()
|
|
assert body["summary"]["errors"] == 0 and body["summary"]["adds"] == 2
|
|
|
|
|
|
def test_export_returns_canonical_csv(fresh_db_url):
|
|
with _merchant_client(fresh_db_url) as client:
|
|
draft = _upload(client).json()
|
|
client.post(f"/api/products/imports/drafts/{draft['id']}/confirm")
|
|
resp = client.get("/api/products/export")
|
|
assert resp.status_code == 200
|
|
assert resp.headers["content-type"].startswith("text/csv")
|
|
assert "attachment" in resp.headers["content-disposition"]
|
|
body = resp.text
|
|
assert body.splitlines()[0].startswith("Handle,")
|
|
assert "moon-mug" in body
|
|
|
|
|
|
def test_export_status_filter_respected(fresh_db_url):
|
|
with _merchant_client(fresh_db_url) as client:
|
|
# GOOD_CSV's moon-mug has no Status column → defaults to active.
|
|
draft = _upload(client).json()
|
|
client.post(f"/api/products/imports/drafts/{draft['id']}/confirm")
|
|
assert "moon-mug" in client.get("/api/products/export?status=active").text
|
|
# No archived products → 409.
|
|
assert client.get("/api/products/export?status=archived").status_code == 409
|
|
|
|
|
|
def test_export_empty_catalog_409(fresh_db_url):
|
|
with _merchant_client(fresh_db_url) as client:
|
|
resp = client.get("/api/products/export")
|
|
assert resp.status_code == 409
|
|
assert resp.json()["error"]["code"] == "empty_catalog"
|
|
|
|
|
|
def test_export_requires_merchant(fresh_db_url):
|
|
with TestClient(create_app(database_url=fresh_db_url)) as client:
|
|
assert client.get("/api/products/export").status_code == 401
|
|
|
|
|
|
def test_export_bad_status_422(fresh_db_url):
|
|
with _merchant_client(fresh_db_url) as client:
|
|
assert client.get("/api/products/export?status=bogus").status_code == 422
|