Ty
cec03c5bd3
- inference: _taxonomy_lists hands the model the synced species leaves + wood leaves + case-insensitive attribute terms (was furniture roots / empty) - inference: CALL1_SEED pins exact flat schema + *_confidence keys, drops grade - inference: _gate maps field->confidence key correctly (was species_id_confidence), reads nested shape as safety net, omits grade (METADATA-40) - review.svelte: figure gets AI assist, grade dropped from AI (owner judgment) - MultiSelect: grouped mode (parents non-selectable headers, selectable leaves) - METADATA-40 updated: grade not suggested by Call 1
259 lines
16 KiB
Python
259 lines
16 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_create_empty_sku_stored_as_null(client, db):
|
|
# Business rule (middle tier): empty SKU at create = "not set yet" -> stored as NULL,
|
|
# not rejected by the DB CHECK. This is the UAT "Continue to Review" path.
|
|
slab_id = str(uuid.uuid4())
|
|
payload = {"id": slab_id, "sku": "", "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}
|
|
resp = client.post("/api/v1/slabs", json=payload)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["sku"] is None
|
|
# A second empty-SKU slab is allowed (NULLs are not unique-conflicts).
|
|
resp2 = client.post("/api/v1/slabs", json={**payload, "id": str(uuid.uuid4())})
|
|
assert resp2.status_code == 201
|
|
|
|
|
|
def test_create_null_sku_allowed(client, db):
|
|
slab_id = str(uuid.uuid4())
|
|
payload = {"id": slab_id, "sku": None, "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}
|
|
resp = client.post("/api/v1/slabs", json=payload)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["sku"] is None
|
|
|
|
|
|
def test_gate_reads_flat_confidence_keys(client, db):
|
|
# Regression: the gate's confidence keys are 'species_confidence' / 'edge_confidence'
|
|
# etc., NOT f'{field}_confidence' (field 'species_id' would give the wrong key). A
|
|
# high-confidence response must surface its picks; low-confidence ones must not.
|
|
from backend.inference import _gate, _parse_json
|
|
good = '```json\n{"species_id": 284, "species_confidence": 0.85, "wood_category_ids": [251], "wood_category_confidence": 1.0, "edge_type_term_id": 323, "edge_confidence": 0.95, "figure_term_ids": [332], "figure_confidence": 0.9, "character": [], "inclusions": ["bark"], "voids": [], "checks": []}\n```'
|
|
out = _gate(_parse_json(good))
|
|
assert out["species_id"] == 284
|
|
assert out["wood_category_ids"] == [251]
|
|
assert out["edge_type_term_id"] == 323
|
|
assert out["figure_term_ids"] == [332]
|
|
# grade must never be gated in (METADATA-40)
|
|
assert "grade_term_id" not in out
|
|
# low-confidence fields are dropped (METADATA-45)
|
|
low = '```json\n{"species_id": 284, "species_confidence": 0.4, "wood_category_ids": [251], "wood_category_confidence": 0.8, "edge_type_term_id": 323, "edge_confidence": 0.2, "figure_term_ids": [332], "figure_confidence": 0.9}\n```'
|
|
out2 = _gate(_parse_json(low))
|
|
assert out2["wood_category_ids"] == [251]
|
|
assert out2["figure_term_ids"] == [332]
|
|
assert "species_id" not in out2
|
|
assert "edge_type_term_id" not in out2
|
|
|
|
|
|
def test_ready_gate_rejects_malformed_sku(client, db):
|
|
# REVIEW-45 enforced in the middle tier: a present-but-out-of-range/char-set SKU is not ready.
|
|
from backend.review import compute_ready, sku_valid
|
|
base = {"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,"price":100,"price_source":"recommendation","title":"Test","short_description":"short","description":"desc","content_source":"manual"}
|
|
# valid
|
|
assert sku_valid("AB-1") is True
|
|
ready, missing = compute_ready({**base, "sku": "AB-1"})
|
|
assert ready is True and "sku" not in missing
|
|
# too short
|
|
assert sku_valid("AB") is False
|
|
ready, missing = compute_ready({**base, "sku": "AB"})
|
|
assert ready is False and "sku" in missing
|
|
# bad char
|
|
ready, missing = compute_ready({**base, "sku": "AB!"})
|
|
assert ready is False and "sku" in missing
|
|
# too long
|
|
ready, missing = compute_ready({**base, "sku": "A" * 25})
|
|
assert ready is False and "sku" in missing
|
|
# missing (None)
|
|
ready, missing = compute_ready({**base, "sku": None})
|
|
assert ready is False and "sku" in missing
|
|
|
|
|
|
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
|