Ty
7287a55707
Ship locked expert Call 1 seed: order cats before edge, species=best confidence, mill-count edge rules, Wild for cookies/burl, multi figure, feature_tag_ids only (no free-text bags, no grade). LISTS keys match the seed; feat- prefix stripped for the model. Gate and review UI apply feature_tag_ids. Seed auto-activates on deploy when still seed-authored.
312 lines
14 KiB
Python
312 lines
14 KiB
Python
"""S7 inference tests (METADATA-105/106, 45, 55, 70, SETTINGS-45)."""
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import backend.db as db_mod
|
|
|
|
|
|
@pytest.fixture
|
|
def env(monkeypatch):
|
|
tmp = tempfile.mkdtemp(prefix="inf_")
|
|
os.environ["SLAB_DB_PATH"] = os.path.join(tmp, "i.db")
|
|
os.environ["SLAB_IMAGES_DIR"] = tmp
|
|
os.environ["WOO_BASE_URL"] = "https://x.com"
|
|
os.environ["SLAB_AES_KEY"] = base64.b64encode(b"0123456789abcdef0123456789abcdef").decode()
|
|
db_mod._reset_for_tests()
|
|
from backend.crypto import encrypt
|
|
|
|
conn = db_mod.init_db()
|
|
conn.execute("INSERT OR REPLACE INTO settings(key,value,value_enc) VALUES ('inference_base_url',NULL,?)", (encrypt("https://infer.example.com"),))
|
|
conn.execute("INSERT OR REPLACE INTO settings(key,value,value_enc) VALUES ('inference_api_key',NULL,?)", (encrypt("sk-test"),))
|
|
conn.execute("INSERT OR REPLACE INTO settings(key,value,value_enc) VALUES ('inference_model','gpt-test',NULL)")
|
|
conn.commit()
|
|
from backend.app import app
|
|
|
|
yield TestClient(app), conn
|
|
db_mod._reset_for_tests()
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
def test_prompt_seed_and_revision(env):
|
|
client, conn = env
|
|
r = client.get("/api/v1/prompts/call1")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["active_content"] is not None
|
|
assert len(body["revisions"]) == 1 # seed only
|
|
|
|
r = client.post("/api/v1/prompts/call1/revisions", json={"content": "revised prompt v2"})
|
|
assert r.status_code == 201
|
|
new_rev = r.json()["id"]
|
|
# new revision becomes active
|
|
r = client.get("/api/v1/prompts/call1")
|
|
assert r.json()["active_revision_id"] == new_rev
|
|
assert r.json()["active_content"] == "revised prompt v2"
|
|
assert len(r.json()["revisions"]) == 2
|
|
|
|
# activate back to seed
|
|
seed_rev = 1
|
|
r = client.post("/api/v1/prompts/call1/activate", json={"revision_id": seed_rev})
|
|
assert r.status_code == 200
|
|
assert client.get("/api/v1/prompts/call1").json()["active_content"] is not None
|
|
|
|
|
|
def test_call1_seed_is_locked_expert_prompt(env):
|
|
"""#86: shipped Call 1 seed is the locked expert prompt (no free-text bags, no grade)."""
|
|
client, _conn = env
|
|
from backend.inference import CALL1_SEED
|
|
|
|
body = client.get("/api/v1/prompts/call1").json()
|
|
assert body["active_content"] == CALL1_SEED
|
|
assert "wood identification expert" in CALL1_SEED
|
|
assert "feature_tag_ids" in CALL1_SEED
|
|
assert "Decide wood categories before edge type" in CALL1_SEED
|
|
assert "highest-confidence" in CALL1_SEED
|
|
assert "Wild edge" in CALL1_SEED
|
|
assert "cookies and burl" in CALL1_SEED.lower() or "cookies and burl" in CALL1_SEED
|
|
assert "grade_term_id" not in CALL1_SEED
|
|
assert '"character"' not in CALL1_SEED
|
|
assert '"inclusions"' not in CALL1_SEED
|
|
assert "character\":" not in CALL1_SEED.replace(" ", "")
|
|
|
|
|
|
def test_taxonomy_lists_call1_closed_shape():
|
|
"""#86: Call 1 lists are closed Woo picks only — no grade/thickness/moisture; feat names stripped."""
|
|
import tempfile, os, shutil
|
|
tmp = tempfile.mkdtemp(prefix="tax_")
|
|
os.environ["SLAB_DB_PATH"] = os.path.join(tmp, "t.db")
|
|
os.environ["SLAB_IMAGES_DIR"] = tmp
|
|
os.environ.setdefault("WOO_BASE_URL", "https://x.com")
|
|
os.environ.setdefault(
|
|
"SLAB_AES_KEY",
|
|
base64.b64encode(b"0123456789abcdef0123456789abcdef").decode(),
|
|
)
|
|
db_mod._reset_for_tests()
|
|
conn = db_mod.init_db()
|
|
conn.execute("INSERT INTO woo_categories(id,name,slug,parent_id,woo_id) VALUES (10,'Species','species',NULL,10)")
|
|
conn.execute("INSERT INTO woo_categories(id,name,slug,parent_id,woo_id) VALUES (11,'Walnut','walnut',10,11)")
|
|
conn.execute("INSERT INTO woo_categories(id,name,slug,parent_id,woo_id) VALUES (20,'Wcat','wcat',NULL,20)")
|
|
conn.execute("INSERT INTO woo_categories(id,name,slug,parent_id,woo_id) VALUES (21,'Live edge wood','liveedge',20,21)")
|
|
conn.execute("INSERT INTO woo_attributes(id,name,slug,woo_id) VALUES (6,'Edge type','pa_etype',6)")
|
|
conn.execute("INSERT INTO woo_attributes(id,name,slug,woo_id) VALUES (8,'Figure','pa_fig',8)")
|
|
conn.execute("INSERT INTO woo_attributes(id,name,slug,woo_id) VALUES (9,'Grade','pa_gr',9)")
|
|
conn.execute("INSERT INTO woo_attributes(id,name,slug,woo_id) VALUES (10,'Thickness','pa_thk',10)")
|
|
conn.execute("INSERT INTO woo_attributes(id,name,slug,woo_id) VALUES (7,'Moisture','pa_mst',7)")
|
|
conn.execute("INSERT INTO woo_attribute_terms(id,attribute_id,name,slug,woo_id) VALUES (321,6,'Square edge','sqe',321)")
|
|
conn.execute("INSERT INTO woo_attribute_terms(id,attribute_id,name,slug,woo_id) VALUES (328,8,'Plain','pl',328)")
|
|
conn.execute("INSERT INTO woo_attribute_terms(id,attribute_id,name,slug,woo_id) VALUES (342,9,'Fancy','fnc',342)")
|
|
conn.execute("INSERT INTO woo_attribute_terms(id,attribute_id,name,slug,woo_id) VALUES (350,10,'1\"- 1 1/2\"','4-6qtr',350)")
|
|
conn.execute("INSERT INTO woo_attribute_terms(id,attribute_id,name,slug,woo_id) VALUES (326,7,'Kiln dried','kd',326)")
|
|
conn.execute("INSERT INTO woo_tags(id,name,slug,woo_id) VALUES (1,'feat-knot','feat-knot',1)")
|
|
conn.execute("INSERT INTO woo_tags(id,name,slug,woo_id) VALUES (2,'fig-curl','fig-curl',2)")
|
|
conn.commit()
|
|
from backend.inference import _taxonomy_lists
|
|
|
|
tax = _taxonomy_lists(conn)
|
|
assert set(tax.keys()) == {"species", "wood_categories", "edge_terms", "figure_terms", "feature_tags"}
|
|
assert tax["species"] == [{"id": 11, "name": "Walnut"}]
|
|
assert tax["wood_categories"] == [{"id": 21, "name": "Live edge wood"}]
|
|
assert tax["edge_terms"] == [{"id": 321, "name": "Square edge"}]
|
|
assert tax["figure_terms"] == [{"id": 328, "name": "Plain"}]
|
|
# feat- prefix stripped for the model; fig- tags are not feature_tags
|
|
assert tax["feature_tags"] == [{"id": 1, "name": "knot"}]
|
|
assert "grade_terms" not in tax
|
|
assert "thickness_terms" not in tax
|
|
assert "moisture_terms" not in tax
|
|
db_mod._reset_for_tests()
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
def test_gate_feature_tag_ids_not_free_text():
|
|
"""#86: gate keeps feature_tag_ids at threshold; drops free-text bags and grade."""
|
|
from backend.inference import _gate
|
|
|
|
parsed = {
|
|
"species_id": 11, "species_confidence": 0.9,
|
|
"wood_category_ids": [21], "wood_category_confidence": 0.8,
|
|
"edge_type_term_id": 321, "edge_confidence": 0.85,
|
|
"figure_term_ids": [328, 335], "figure_confidence": 0.75,
|
|
"feature_tag_ids": [1], "feature_tag_confidence": 0.8,
|
|
"grade_term_id": 342, "grade_confidence": 0.99,
|
|
"character": ["rustic"], "inclusions": ["bark"], "voids": [], "checks": ["end"],
|
|
}
|
|
out = _gate(parsed)
|
|
assert out["species_id"] == 11
|
|
assert out["wood_category_ids"] == [21]
|
|
assert out["edge_type_term_id"] == 321
|
|
assert out["figure_term_ids"] == [328, 335]
|
|
assert out["feature_tag_ids"] == [1]
|
|
assert "grade_term_id" not in out
|
|
assert "character" not in out
|
|
assert "inclusions" not in out
|
|
assert "voids" not in out
|
|
assert "checks" not in out
|
|
# below threshold feature tags dropped
|
|
low = {**parsed, "feature_tag_confidence": 0.2}
|
|
assert "feature_tag_ids" not in _gate(low)
|
|
|
|
|
|
def test_call1_confidence_gate(env, monkeypatch):
|
|
client, conn = env
|
|
# enable inference
|
|
conn.execute("INSERT OR REPLACE INTO settings(key,value,value_enc) VALUES ('inference_enabled','true',NULL)")
|
|
conn.execute("INSERT INTO woo_categories(id,name,slug,parent_id,woo_id) VALUES (1,'Black Walnut','black-walnut',NULL,100)")
|
|
conn.execute("INSERT INTO slabs(id,sku,length_in,thickness_in,status,server_rev,client_rev,created_at,updated_at) VALUES ('s1','SLAB-01',100,2,'calibrated',1,1,'t','t')")
|
|
conn.execute("INSERT INTO slab_photos(id,slab_id,seq,processed_blob,created_at) VALUES ('p1','s1',1,X'89504E47','t')")
|
|
conn.commit()
|
|
from backend.inference import InferenceClient
|
|
|
|
def fake_chat(self, messages):
|
|
# Capture list header for #86 shape
|
|
text = messages[0]["content"][-1]["text"]
|
|
assert "LISTS:" in text
|
|
assert "grade_terms" not in text
|
|
assert "species_leaves" not in text
|
|
return json.dumps({
|
|
"species_id": 1, "species_confidence": 0.9,
|
|
"wood_category_ids": [1], "wood_category_confidence": 0.5, # below threshold
|
|
"grade_term_id": 7, "grade_confidence": 0.99, # never suggested
|
|
"figure_term_ids": [2], "figure_confidence": 0.8, # at/above
|
|
"feature_tag_ids": [9], "feature_tag_confidence": 0.9,
|
|
"character": ["x"], "inclusions": ["y"],
|
|
})
|
|
monkeypatch.setattr(InferenceClient, "chat", fake_chat)
|
|
r = client.post("/api/v1/slabs/s1/inference/call1")
|
|
assert r.status_code == 200, r.text
|
|
s = r.json()["suggestions"]
|
|
assert s.get("species_id") == 1
|
|
assert s.get("figure_term_ids") == [2]
|
|
assert s.get("feature_tag_ids") == [9]
|
|
assert "wood_category_ids" not in s # gated out
|
|
assert "grade_term_id" not in s # never
|
|
assert "character" not in s
|
|
# inference_log written with call=1
|
|
assert conn.execute("SELECT call FROM inference_log").fetchone()["call"] == 1
|
|
|
|
|
|
def test_call2_stores_description(env, monkeypatch):
|
|
client, conn = env
|
|
conn.execute("INSERT OR REPLACE INTO settings(key,value,value_enc) VALUES ('content_llm_enabled','true',NULL)")
|
|
conn.execute("INSERT INTO slabs(id,sku,length_in,thickness_in,width_avg_in,bdft,title,status,server_rev,client_rev,created_at,updated_at) VALUES ('s1','SLAB-01',100,2,20,18,'Walnut Slab','ready',1,1,'t','t')")
|
|
conn.execute("INSERT INTO slab_photos(id,slab_id,seq,processed_blob,created_at) VALUES ('p1','s1',1,X'89504E47','t')")
|
|
conn.commit()
|
|
from backend.inference import InferenceClient
|
|
|
|
monkeypatch.setattr(InferenceClient, "chat", lambda self, messages: "<p>A fine walnut slab.</p>")
|
|
r = client.post("/api/v1/slabs/s1/inference/call2")
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["description"] == "<p>A fine walnut slab.</p>"
|
|
row = conn.execute("SELECT description, content_source FROM slabs WHERE id='s1'").fetchone()
|
|
assert row["description"] == "<p>A fine walnut slab.</p>"
|
|
assert row["content_source"] == "llm"
|
|
|
|
|
|
def test_inference_off_call1_409(env):
|
|
client, conn = env
|
|
conn.execute("INSERT INTO slabs(id,sku,status,server_rev,client_rev,created_at,updated_at) VALUES ('s1','SLAB-01','calibrated',1,1,'t','t')")
|
|
conn.commit()
|
|
r = client.post("/api/v1/slabs/s1/inference/call1")
|
|
assert r.status_code == 409 # inference_enabled default off
|
|
|
|
|
|
def test_server_description_inference_off(env):
|
|
from backend.inference import server_description
|
|
|
|
d = server_description({"title": "Figured Walnut Slab", "length_in": 114.0, "width_avg_in": 24.0, "thickness_in": 1.75, "bdft": 18.4})
|
|
assert "114.0in" in d and "18.4" in d
|
|
|
|
|
|
def test_test_inference_unconfigured(env):
|
|
client, conn = env
|
|
conn.execute("UPDATE settings SET value_enc=NULL WHERE key='inference_api_key'")
|
|
conn.commit()
|
|
r = client.post("/api/v1/admin/test-inference")
|
|
assert r.status_code == 200
|
|
assert r.json()["ok"] is False and r.json()["mode"] == "fail"
|
|
|
|
|
|
def _rgba_png(w: int, h: int) -> bytes:
|
|
"""RGBA PNG: opaque body + transparent top-left block (capture-like alpha)."""
|
|
from PIL import Image
|
|
import io
|
|
im = Image.new("RGBA", (w, h), (10, 20, 30, 255))
|
|
for y in range(min(80, h)):
|
|
for x in range(min(80, w)):
|
|
im.putpixel((x, y), (0, 0, 0, 0))
|
|
buf = io.BytesIO()
|
|
im.save(buf, format="PNG")
|
|
return buf.getvalue()
|
|
|
|
|
|
def test_fit_long_edge_png_downscales_over_1024():
|
|
"""METADATA-35: Call 1 images long edge <= 1024; never upscale; keep alpha."""
|
|
from backend.inference import fit_long_edge_png
|
|
from PIL import Image
|
|
import io
|
|
|
|
src = _rgba_png(800, 1067) # inventory 3:4, long edge 1067
|
|
out = fit_long_edge_png(src, max_long_edge=1024)
|
|
im = Image.open(io.BytesIO(out)).convert("RGBA")
|
|
assert max(im.size) == 1024
|
|
assert min(im.size) < 1024
|
|
w, h = im.size
|
|
assert abs((w / h) - (800 / 1067)) < 0.02
|
|
# Transparent block still transparent near origin after downscale
|
|
assert im.getpixel((0, 0))[3] == 0
|
|
assert im.getpixel((5, 5))[3] == 0
|
|
|
|
|
|
def test_fit_long_edge_png_leaves_small_unchanged_dims():
|
|
from backend.inference import fit_long_edge_png
|
|
from PIL import Image
|
|
import io
|
|
|
|
src = _rgba_png(400, 300)
|
|
out = fit_long_edge_png(src, max_long_edge=1024)
|
|
im = Image.open(io.BytesIO(out))
|
|
assert im.size == (400, 300)
|
|
|
|
|
|
def test_call1_sends_long_edge_at_most_1024(env, monkeypatch):
|
|
"""METADATA-35 end-to-end: messages to the model use <=1024 long edge."""
|
|
client, conn = env
|
|
conn.execute("INSERT OR REPLACE INTO settings(key,value,value_enc) VALUES ('inference_enabled','true',NULL)")
|
|
conn.execute(
|
|
"INSERT INTO slabs(id,sku,length_in,thickness_in,status,server_rev,client_rev,created_at,updated_at) "
|
|
"VALUES ('s1','SLAB-01',100,2,'calibrated',1,1,'t','t')"
|
|
)
|
|
big = _rgba_png(800, 1067)
|
|
conn.execute(
|
|
"INSERT INTO slab_photos(id,slab_id,seq,processed_blob,created_at) VALUES ('p1','s1',1,?, 't')",
|
|
(big,),
|
|
)
|
|
conn.commit()
|
|
from backend.inference import InferenceClient
|
|
from PIL import Image
|
|
import io
|
|
|
|
captured = {}
|
|
|
|
def fake_chat(self, messages):
|
|
captured["messages"] = messages
|
|
return json.dumps({"species_id": 1, "species_confidence": 0.9})
|
|
|
|
monkeypatch.setattr(InferenceClient, "chat", fake_chat)
|
|
r = client.post("/api/v1/slabs/s1/inference/call1")
|
|
assert r.status_code == 200, r.text
|
|
content = captured["messages"][0]["content"]
|
|
imgs = [p for p in content if isinstance(p, dict) and p.get("type") == "image_url"]
|
|
assert len(imgs) == 1
|
|
url = imgs[0]["image_url"]["url"]
|
|
assert url.startswith("data:image/png;base64,")
|
|
raw = base64.b64decode(url.split(",", 1)[1])
|
|
im = Image.open(io.BytesIO(raw))
|
|
assert max(im.size) <= 1024
|