Ty
cc5241f7da
- backend/slabs.py: create (calibrated, client-derived stored as sent), list (excl soft-deleted, most recent), get, update (single compare-and-set on server_rev, 409 revision_conflict / 409 duplicate_sku / 422 stale_field, status server-derived), delete (soft, published rejected). Explicit column allowlists, parameterized SQL, JSON array fields, commits. - backend/app.py: wire slabs router. - frontend: src/api/slabs.ts client + create/slabs/SlabList.svelte + dev toggle. - tests: 9 slabs tests (create/dup/list/get/update/rev-conflict/dup-sku/soft-delete/ published-delete) with full taxonomy seeding. Verified: pytest 23 passed; check_requirements + check_coverage exit 0; npm build ok.
193 lines
11 KiB
Python
193 lines
11 KiB
Python
import os
|
|
import json
|
|
import uuid
|
|
import datetime
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
import backend.db as db_mod
|
|
|
|
|
|
# Helper to set environment for tests
|
|
|
|
def _setup_env(monkeypatch, tmp):
|
|
db_path = os.path.join(tmp, "test.db")
|
|
img_dir = os.path.join(tmp, "images")
|
|
os.mkdir(img_dir)
|
|
monkeypatch.setenv("SLAB_DB_PATH", db_path)
|
|
monkeypatch.setenv("SLAB_IMAGES_DIR", img_dir)
|
|
monkeypatch.setenv("WOO_BASE_URL", "https://example.com")
|
|
monkeypatch.setenv("SLAB_AES_KEY", "aGVsbGxvb2xvY2g=") # dummy base64
|
|
monkeypatch.setenv("LOG_LEVEL", "INFO")
|
|
db_mod._reset_for_tests()
|
|
|
|
# Taxonomy seeding helper
|
|
|
|
def _seed_taxonomy(conn):
|
|
# create some categories
|
|
conn.execute("INSERT OR REPLACE INTO woo_categories(id, name, slug, parent_id, woo_id) VALUES (1, 'Species A', 'species-a', NULL, 10)")
|
|
conn.execute("INSERT OR REPLACE INTO woo_categories(id, name, slug, parent_id, woo_id) VALUES (2, 'Wood Cat A', 'wood-cat-a', NULL, 20)")
|
|
# attribute terms
|
|
conn.execute("INSERT OR REPLACE INTO woo_attributes(id, name, slug, woo_id) VALUES (1, 'Edge Type', 'edge-type', 30)")
|
|
conn.execute("INSERT OR REPLACE INTO woo_attribute_terms(id, attribute_id, name, slug, woo_id) VALUES (1, 1, 'Edge A', 'edge-a', 40)")
|
|
# tags
|
|
conn.execute("INSERT OR REPLACE INTO woo_tags(id, name, slug, woo_id) VALUES (1, 'Fig Tag A', 'fig-tag-a', 50)")
|
|
conn.execute("INSERT OR REPLACE INTO woo_tags(id, name, slug, woo_id) VALUES (2, 'Feat Tag A', 'feat-tag-a', 60)")
|
|
conn.commit()
|
|
|
|
@pytest.fixture
|
|
def db(tmp_path, monkeypatch):
|
|
_setup_env(monkeypatch, tmp_path)
|
|
conn = db_mod.get_connection()
|
|
db_mod.init_db()
|
|
_seed_taxonomy(conn)
|
|
return conn
|
|
_setup_env(monkeypatch, tmp_path)
|
|
conn = db_mod.get_connection()
|
|
db_mod.init_db()
|
|
_seed_taxonomy(conn)
|
|
return conn
|
|
|
|
# Helper to get client
|
|
@pytest.fixture
|
|
def client(db):
|
|
from backend.app import app
|
|
return TestClient(app)
|
|
|
|
# Utility to format now
|
|
_now_iso = lambda: datetime.datetime.utcnow().replace(microsecond=0).isoformat()
|
|
|
|
# --- Tests ---
|
|
|
|
|
|
def test_create_and_fetch(client, db):
|
|
slab_id = str(uuid.uuid4())
|
|
payload = {
|
|
"id": slab_id,
|
|
"sku": "SKU001",
|
|
"length_in": 10.0,
|
|
"thickness_in": 2.0,
|
|
"sqft": 20.0,
|
|
"bdft": 5.0,
|
|
"width_min_in": 1.0,
|
|
"width_max_in": 2.0,
|
|
"width_avg_in": 1.5,
|
|
"species_id": 1,
|
|
"wood_category_ids": json.dumps([2]),
|
|
"edge_type_term_id": 1,
|
|
"figure_term_ids": json.dumps([1]),
|
|
"grade_term_id": 1,
|
|
"thickness_term_id": 1,
|
|
"moisture_term_id": 1,
|
|
"fig_tag_ids": json.dumps([1]),
|
|
"feat_tag_ids": json.dumps([2]),
|
|
"price": 100.0,
|
|
"price_source": "recommendation",
|
|
"title": "Test Slab",
|
|
"short_description": "short",
|
|
"description": "desc",
|
|
"content_source": "manual",
|
|
"species_confidence": 0.9,
|
|
"figure_confidence": 0.8,
|
|
"grade_confidence": 0.7,
|
|
"inference_status": "inferring",
|
|
"calibration_photo_id": None
|
|
}
|
|
resp = client.post("/api/v1/slabs", json=payload)
|
|
assert resp.status_code == 201
|
|
body = resp.json()
|
|
assert body["status"] == "calibrated"
|
|
assert body["client_rev"] == 1
|
|
assert body["server_rev"] == 1
|
|
for key in ["length_in","thickness_in","sqft","bdft","width_min_in","width_max_in","width_avg_in"]:
|
|
assert body[key] == payload[key]
|
|
|
|
|
|
def test_duplicate_sku(client, db):
|
|
slab_id = str(uuid.uuid4())
|
|
payload = {"id": slab_id, "sku": "SKU001", "length_in":10,"thickness_in":2,"sqft":20,"bdft":5,"width_min_in":1,"width_max_in":2,"width_avg_in":1.5,"species_id":1,"wood_category_ids":json.dumps([2]),"edge_type_term_id":1,"figure_term_ids":json.dumps([1]),"grade_term_id":1,"thickness_term_id":1,"moisture_term_id":1,"fig_tag_ids":json.dumps([1]),"feat_tag_ids":json.dumps([2]),"price":100,"price_source":"recommendation","title":"Test","short_description":"short","description":"desc","content_source":"manual","species_confidence":0.9,"figure_confidence":0.8,"grade_confidence":0.7,"inference_status":"inferring","calibration_photo_id":None}
|
|
resp1 = client.post("/api/v1/slabs", json=payload)
|
|
assert resp1.status_code == 201
|
|
resp2 = client.post("/api/v1/slabs", json=payload)
|
|
assert resp2.status_code == 409
|
|
|
|
|
|
def test_list_and_exclude_deleted(client, db):
|
|
# create two slabs
|
|
for i in [1,2]:
|
|
slab_id = str(uuid.uuid4())
|
|
payload = {"id": slab_id, "sku": f"SKU{i}", "length_in":10,"thickness_in":2,"sqft":20,"bdft":5,"width_min_in":1,"width_max_in":2,"width_avg_in":1.5,"species_id":1,"wood_category_ids":json.dumps([2]),"edge_type_term_id":1,"figure_term_ids":json.dumps([1]),"grade_term_id":1,"thickness_term_id":1,"moisture_term_id":1,"fig_tag_ids":json.dumps([1]),"feat_tag_ids":json.dumps([2]),"price":100,"price_source":"recommendation","title":"Test","short_description":"short","description":"desc","content_source":"manual","species_confidence":0.9,"figure_confidence":0.8,"grade_confidence":0.7,"inference_status":"inferring","calibration_photo_id":None}
|
|
client.post("/api/v1/slabs", json=payload)
|
|
# delete first created
|
|
resp_list = client.get("/api/v1/slabs")
|
|
ids = [r["id"] for r in resp_list.json()]
|
|
client.delete(f"/api/v1/slabs/{ids[0]}")
|
|
# list again
|
|
resp2 = client.get("/api/v1/slabs")
|
|
body = resp2.json()
|
|
assert len(body) == 1
|
|
assert body[0]["id"] == ids[1]
|
|
|
|
|
|
def test_get_not_found(client, db):
|
|
resp = client.get("/api/v1/slabs/nonexistent")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_update_success(client, db):
|
|
slab_id = str(uuid.uuid4())
|
|
payload = {"id": slab_id, "sku": "SKU001", "length_in":10,"thickness_in":2,"sqft":20,"bdft":5,"width_min_in":1,"width_max_in":2,"width_avg_in":1.5,"species_id":1,"wood_category_ids":json.dumps([2]),"edge_type_term_id":1,"figure_term_ids":json.dumps([1]),"grade_term_id":1,"thickness_term_id":1,"moisture_term_id":1,"fig_tag_ids":json.dumps([1]),"feat_tag_ids":json.dumps([2]),"price":100,"price_source":"recommendation","title":"Test","short_description":"short","description":"desc","content_source":"manual","species_confidence":0.9,"figure_confidence":0.8,"grade_confidence":0.7,"inference_status":"inferring","calibration_photo_id":None}
|
|
client.post("/api/v1/slabs", json=payload)
|
|
# update
|
|
update_payload = {"client_rev":1, "sku":"SKU002"}
|
|
resp = client.put(f"/api/v1/slabs/{slab_id}", json=update_payload)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["sku"] == "SKU002"
|
|
assert body["server_rev"] == 2
|
|
|
|
|
|
def test_update_revision_conflict(client, db):
|
|
slab_id = str(uuid.uuid4())
|
|
payload = {"id": slab_id, "sku": "SKU001", "length_in":10,"thickness_in":2,"sqft":20,"bdft":5,"width_min_in":1,"width_max_in":2,"width_avg_in":1.5,"species_id":1,"wood_category_ids":json.dumps([2]),"edge_type_term_id":1,"figure_term_ids":json.dumps([1]),"grade_term_id":1,"thickness_term_id":1,"moisture_term_id":1,"fig_tag_ids":json.dumps([1]),"feat_tag_ids":json.dumps([2]),"price":100,"price_source":"recommendation","title":"Test","short_description":"short","description":"desc","content_source":"manual","species_confidence":0.9,"figure_confidence":0.8,"grade_confidence":0.7,"inference_status":"inferring","calibration_photo_id":None}
|
|
client.post("/api/v1/slabs", json=payload)
|
|
# wrong client_rev
|
|
resp = client.put(f"/api/v1/slabs/{slab_id}", json={"client_rev":99, "sku":"SKU002"})
|
|
assert resp.status_code == 409
|
|
|
|
|
|
def test_update_duplicate_sku(client, db):
|
|
# create two slabs
|
|
id1, id2 = str(uuid.uuid4()), str(uuid.uuid4())
|
|
payload1 = {"id": id1, "sku": "SKU001", "length_in":10,"thickness_in":2,"sqft":20,"bdft":5,"width_min_in":1,"width_max_in":2,"width_avg_in":1.5,"species_id":1,"wood_category_ids":json.dumps([2]),"edge_type_term_id":1,"figure_term_ids":json.dumps([1]),"grade_term_id":1,"thickness_term_id":1,"moisture_term_id":1,"fig_tag_ids":json.dumps([1]),"feat_tag_ids":json.dumps([2]),"price":100,"price_source":"recommendation","title":"Test","short_description":"short","description":"desc","content_source":"manual","species_confidence":0.9,"figure_confidence":0.8,"grade_confidence":0.7,"inference_status":"inferring","calibration_photo_id":None}
|
|
payload2 = {"id": id2, "sku": "SKU002", "length_in":10,"thickness_in":2,"sqft":20,"bdft":5,"width_min_in":1,"width_max_in":2,"width_avg_in":1.5,"species_id":1,"wood_category_ids":json.dumps([2]),"edge_type_term_id":1,"figure_term_ids":json.dumps([1]),"grade_term_id":1,"thickness_term_id":1,"moisture_term_id":1,"fig_tag_ids":json.dumps([1]),"feat_tag_ids":json.dumps([2]),"price":100,"price_source":"recommendation","title":"Test","short_description":"short","description":"desc","content_source":"manual","species_confidence":0.9,"figure_confidence":0.8,"grade_confidence":0.7,"inference_status":"inferring","calibration_photo_id":None}
|
|
client.post("/api/v1/slabs", json=payload1)
|
|
client.post("/api/v1/slabs", json=payload2)
|
|
# update first to sku of second
|
|
resp = client.put(f"/api/v1/slabs/{id1}", json={"client_rev":1, "sku":"SKU002"})
|
|
assert resp.status_code == 409
|
|
|
|
|
|
def test_soft_delete(client, db):
|
|
slab_id = str(uuid.uuid4())
|
|
payload = {"id": slab_id, "sku": "SKU001", "length_in":10,"thickness_in":2,"sqft":20,"bdft":5,"width_min_in":1,"width_max_in":2,"width_avg_in":1.5,"species_id":1,"wood_category_ids":json.dumps([2]),"edge_type_term_id":1,"figure_term_ids":json.dumps([1]),"grade_term_id":1,"thickness_term_id":1,"moisture_term_id":1,"fig_tag_ids":json.dumps([1]),"feat_tag_ids":json.dumps([2]),"price":100,"price_source":"recommendation","title":"Test","short_description":"short","description":"desc","content_source":"manual","species_confidence":0.9,"figure_confidence":0.8,"grade_confidence":0.7,"inference_status":"inferring","calibration_photo_id":None}
|
|
client.post("/api/v1/slabs", json=payload)
|
|
client.delete(f"/api/v1/slabs/{slab_id}")
|
|
resp = client.get("/api/v1/slabs")
|
|
body = resp.json()
|
|
assert all(r["id"] != slab_id for r in body)
|
|
resp_get = client.get(f"/api/v1/slabs/{slab_id}")
|
|
assert resp_get.status_code == 404
|
|
|
|
|
|
def test_delete_published_conflict(client, db):
|
|
slab_id = str(uuid.uuid4())
|
|
payload = {"id": slab_id, "sku": "SKU001", "length_in":10,"thickness_in":2,"sqft":20,"bdft":5,"width_min_in":1,"width_max_in":2,"width_avg_in":1.5,"species_id":1,"wood_category_ids":json.dumps([2]),"edge_type_term_id":1,"figure_term_ids":json.dumps([1]),"grade_term_id":1,"thickness_term_id":1,"moisture_term_id":1,"fig_tag_ids":json.dumps([1]),"feat_tag_ids":json.dumps([2]),"price":100,"price_source":"recommendation","title":"Test","short_description":"short","description":"desc","content_source":"manual","species_confidence":0.9,"figure_confidence":0.8,"grade_confidence":0.7,"inference_status":"inferring","calibration_photo_id":None}
|
|
client.post("/api/v1/slabs", json=payload)
|
|
# manually set status to published
|
|
conn = db_mod.get_connection()
|
|
conn.execute("UPDATE slabs SET status = 'published' WHERE id = ?", (slab_id,))
|
|
conn.commit()
|
|
resp = client.delete(f"/api/v1/slabs/{slab_id}")
|
|
assert resp.status_code == 409
|