import json from app.tasks import order_comms as oc from app.tasks.order_comms import ( TASK, TAXONOMY, WORLD_FRIENDLY_IDS, WORLD_UUIDS, execute_tool, generate, lookup_truth, parse_output, score, ) def _answer(truth: dict, entities: list[dict], classification: dict | None = None) -> str: obj = { "classification": classification or truth["classification"], "entities": [{**e, "confidence": 0.9, "reasoning": "because"} for e in entities], "search_summary": "looked it up", } return "```json\n" + json.dumps(obj) + "\n```" def test_generation_deterministic_and_diverse(): texts = set() for seed in range(1, 80): truth, text = generate(seed) assert generate(seed) == (truth, text) # deterministic assert lookup_truth(f"comm-{seed}") == truth # truth re-derivable from id texts.add(text) assert len(texts) > 75 # near-unique renderings def test_gold_is_valid_by_construction(): """Every gold entity must actually exist in the world, and intent labels are valid.""" for seed in range(1, 200): truth, _ = generate(seed) cat = truth["classification"]["category"] assert truth["classification"]["subcategory"] in TAXONOMY[cat] for e in [truth["primary"], *truth["required"], *truth["optional"]]: if e["friendly_id"]: assert e["friendly_id"].upper() in WORLD_FRIENDLY_IDS, (seed, e) assert e["entity_id"] in WORLD_UUIDS or e["entity_id"].upper() in WORLD_FRIENDLY_IDS def test_search_entities_exact_and_fuzzy_and_errors(): truth, _ = generate(4) onum = next(e["friendly_id"] for e in truth["required"] if e["entity_type"] == "Order") r = json.loads(execute_tool("search_entities", {"search_terms": onum, "reason": "x", "entity_types": ["Order"]})) assert r["results"][0]["fields"]["order_number"] == onum assert r["results"][0]["score"] >= 100 # exact-match boost short = json.loads(execute_tool("search_entities", {"search_terms": "ab", "reason": "x"})) assert "INVALID_ARGUMENT" in short["error"] miss = json.loads(execute_tool("search_entities", {"search_terms": "zzqq nonexistent thing", "reason": "x"})) assert "NOT_FOUND" in miss["error"] # phone is deliberately NOT searchable -> must use SQL phone_search = json.loads(execute_tool("search_entities", {"search_terms": "+12029729975", "reason": "x"})) assert "error" in phone_search def test_execute_sql_chain_and_guardrails(): rows = json.loads(execute_tool( "execute_sql", {"sql": "SELECT customer_id, full_name FROM shop.customers LIMIT 1", "reason": "x"})) assert rows["row_count"] == 1 and "customer_id" in rows["columns"] # canonical phone -> orders join runs cid = rows["rows"][0]["customer_id"] joined = json.loads(execute_tool("execute_sql", {"sql": ( "SELECT o.order_number, p.product_name FROM shop.orders o " "JOIN shop.order_items oi ON oi.order_id=o.order_id " "JOIN shop.products p ON p.sku=oi.sku WHERE o.customer_id=?".replace("?", f"'{cid}'")), "reason": "x"})) assert "order_number" in joined["columns"] assert "error" in json.loads(execute_tool("execute_sql", {"sql": "DELETE FROM shop.orders", "reason": "x"})) assert "error" in json.loads(execute_tool("execute_sql", {"sql": "SELECT 1; DROP TABLE shop.orders", "reason": "x"})) assert "error" in json.loads(execute_tool("execute_sql", {"sql": "SELECT missing_col FROM shop.orders", "reason": "x"})) def test_auto_limit_applied(): res = json.loads(execute_tool("execute_sql", {"sql": "SELECT * FROM shop.order_items", "reason": "x"})) assert res["row_count"] <= 50 # auto LIMIT 50 even though the table is larger def test_perfect_answer_scores_full_marks(): for seed in (4, 11, 23, 42): truth, _ = generate(seed) gold = truth["required"] + truth["optional"] s = score(truth, parse_output(_answer(truth, gold))) assert s["intent_exact"] == 1.0 assert s["primary_found"] == 1.0 assert s["entity_id_correct"] == 1.0 assert s["both_correct"] == 1.0 assert s["no_hallucinated_id"] == 1.0 assert s["recall"] == 1.0 def test_wrong_answer_scores_below_correct(): truth, _ = generate(4) gold = truth["required"] + truth["optional"] good = score(truth, parse_output(_answer(truth, gold))) # wrong intent + a real-but-irrelevant order, no uuid other = next(o for o in oc.ORDERS if o["order_number"] != truth["required"][-1]["friendly_id"]) bad = score(truth, parse_output(_answer( truth, [{"entity_type": "Order", "friendly_id": other["order_number"], "entity_id": None}], classification={"category": "Account", "subcategory": "login_issue"}))) assert bad["both_correct"] < good["both_correct"] assert bad["primary_found"] == 0.0 assert bad["intent_exact"] == 0.0 assert bad["f1"] < good["f1"] assert bad["no_hallucinated_id"] == 1.0 # the distractor is a real order, not hallucinated def test_hallucinated_id_penalized(): truth, _ = generate(4) s = score(truth, parse_output(_answer( truth, [{"entity_type": "Order", "friendly_id": "ORD-999999-0000", "entity_id": "not-a-real-uuid"}]))) assert s["no_hallucinated_id"] == 0.0 def test_parse_output_tolerates_garbage(): s = score(*(generate(4)[0],) * 0 or (generate(4)[0], parse_output("no json here, sorry"))) assert s["intent_format_ok"] == 0.0 assert s["tag_format_ok"] == 0.0 assert s["both_correct"] == 0.0 def test_parse_output_picks_last_json_block(): text = "draft:\n```json\n{\"classification\": {\"category\": \"X\"}}\n```\n" \ "final:\n```json\n{\"classification\": {\"category\": \"Shipping\", " \ "\"subcategory\": \"late_delivery\"}, \"entities\": [], \"search_summary\": \"s\"}\n```" parsed = parse_output(text) assert parsed["category"] == "Shipping" and parsed["subcategory"] == "late_delivery" def test_task_registered_with_tools(): from app.tasks import REGISTRY assert REGISTRY.get("order-comms") is TASK assert TASK.tools and TASK.execute_tool is not None assert TASK.ui["output"] == "comms"