Spaces:
Sleeping
Sleeping
| 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" | |