type-two
d75f27a691
- FastAPI twin of DealGod (async SQLAlchemy/asyncpg). wowplatter/v1: ping, inventory/lookup, inventory/intake (enriches from discogs_full by release_id). - schema v2: unified inventory (vinyl+general), slim crate, sales/sale_items. Discogs catalog NOT copied — joined from discogs_full at read time. - migrate.py: 34,543 shop rows moved (vs ~3.2M in the WowPlatter clone). - Smoke test green; plan/readme carry the architecture + UI boundary. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
74 lines
2.4 KiB
Python
74 lines
2.4 KiB
Python
import os
|
|
os.environ["RECORDGOD_TOKEN"] = "test-tok"
|
|
|
|
from fastapi.testclient import TestClient
|
|
from app.main import app
|
|
from app.db import get_db
|
|
|
|
AUTH = {"Authorization": "Bearer test-tok"}
|
|
|
|
|
|
class _FakeResult:
|
|
def __init__(self, row): self._row = row
|
|
def mappings(self): return self
|
|
def first(self): return self._row
|
|
|
|
|
|
def _override(row):
|
|
async def _dep():
|
|
class _DB:
|
|
async def execute(self, *a, **k): return _FakeResult(row)
|
|
async def commit(self): pass
|
|
yield _DB()
|
|
return _dep
|
|
|
|
|
|
def test_ping_ok():
|
|
r = TestClient(app).get("/wowplatter/v1/ping", headers=AUTH)
|
|
assert r.status_code == 200
|
|
b = r.json()
|
|
assert b["ok"] and b["store"] and b["plan"] and b["version"]
|
|
|
|
|
|
def test_auth_required():
|
|
c = TestClient(app)
|
|
assert c.get("/wowplatter/v1/ping").status_code == 401
|
|
assert c.get("/wowplatter/v1/ping", headers={"Authorization": "Bearer wrong"}).status_code == 401
|
|
|
|
|
|
def test_lookup_in_stock():
|
|
app.dependency_overrides[get_db] = _override({
|
|
"sku": "MRP-AB12CD", "price": 39.95, "qty": 1, "condition": "VG+",
|
|
"status": "publish", "product_url": "https://store/x", "release_id": 12345})
|
|
r = TestClient(app).get("/wowplatter/v1/inventory/lookup?release_id=12345", headers=AUTH)
|
|
app.dependency_overrides.clear()
|
|
assert r.status_code == 200
|
|
b = r.json()
|
|
assert b["in_stock"] is True and b["sku"] == "MRP-AB12CD" and b["price"] == 39.95
|
|
|
|
|
|
def test_lookup_not_stocked():
|
|
app.dependency_overrides[get_db] = _override(None)
|
|
r = TestClient(app).get("/wowplatter/v1/inventory/lookup?release_id=999", headers=AUTH)
|
|
app.dependency_overrides.clear()
|
|
assert r.json() == {"ok": True, "in_stock": False, "release_id": 999}
|
|
|
|
|
|
def test_intake_stages_without_mirror():
|
|
# no release_id → no mirror call; just generate a sku and stage.
|
|
app.dependency_overrides[get_db] = _override(None)
|
|
r = TestClient(app).post("/wowplatter/v1/inventory/intake",
|
|
headers=AUTH, json={"kind": "vinyl", "price": 25.0, "condition": "VG+"})
|
|
app.dependency_overrides.clear()
|
|
b = r.json()
|
|
assert b["ok"] and b["staged"] is True and b["sku"].startswith("MRP-")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_ping_ok()
|
|
test_auth_required()
|
|
test_lookup_in_stock()
|
|
test_lookup_not_stocked()
|
|
test_intake_stages_without_mirror()
|
|
print("ok — ping, auth, lookup(in/out), intake all pass")
|