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, "feature_tag_ids": [1], "feature_tag_confidence": 0.8, "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] assert out["feature_tag_ids"] == [1] # grade must never be gated in (METADATA-40) assert "grade_term_id" not in out # free-text bags dropped (#86) assert "character" not in out assert "inclusions" 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 def test_stale_field_returns_live_options_not_submitted_value(client, db): """#82 / PUBLISH-20: current_options is the live taxonomy list, not the bad id.""" slab_id = str(uuid.uuid4()) payload = { "id": slab_id, "sku": "SKU-STALE", "length_in": 10, "thickness_in": 2, "sqft": 20, "bdft": 5, "width_min_in": 1, "width_max_in": 2, "width_avg_in": 1.5, } assert client.post("/api/v1/slabs", json=payload).status_code == 201 resp = client.put(f"/api/v1/slabs/{slab_id}", json={"client_rev": 1, "edge_type_term_id": 999}) assert resp.status_code == 422 body = resp.json() assert body["error"]["code"] == "stale_field" opts = body["error"]["detail"]["current_options"] assert isinstance(opts, list) and opts assert all("id" in o and "name" in o for o in opts) assert 999 not in [o["id"] for o in opts] assert "Edge type" in body["error"]["message"] or "edge" in body["error"]["message"].lower() def test_create_extra_key_and_bad_taxonomy_are_4xx(client, db): """#83: extra keys and bad taxonomy ids must not 500.""" slab_id = str(uuid.uuid4()) base = { "id": slab_id, "sku": "SKU-XTRA", "length_in": 10, "thickness_in": 2, "sqft": 20, "bdft": 5, "width_min_in": 1, "width_max_in": 2, "width_avg_in": 1.5, "status": "published", # server-managed; must be dropped, not bind-break } r = client.post("/api/v1/slabs", json=base) assert r.status_code == 201, r.text assert r.json()["status"] == "calibrated" # Bad taxonomy on create r2 = client.post("/api/v1/slabs", json={ **base, "id": str(uuid.uuid4()), "sku": "SKU-BADTAX", "species_id": 999, }) assert r2.status_code == 422 assert r2.json()["error"]["code"] == "stale_field" # Bad enum r3 = client.post("/api/v1/slabs", json={ **base, "id": str(uuid.uuid4()), "sku": "SKU-BADENUM", "price_source": "not-a-source", }) assert r3.status_code == 400 assert r3.json()["error"]["code"] == "validation"