query/validate.py

"""Gold-query validation gate for the brain.

Each case asserts that the grounded answer (a) resolves the expected entity and/or
(b) contains expected substrings. Run: python -m query.validate
"""
from __future__ import annotations

from query.answer import answer

# (question, expected_matched_name_or_None, [substrings that must appear in answer])
GOLD = [
    ("which SPs write PO_POMAS_PUR_ORDER_HDR", "PO_POMAS_PUR_ORDER_HDR", ["write", "/wiki/sp/"]),
    ("what tables does pocrmn_sp_podbyr read", "pocrmn_sp_podbyr", ["read"]),
    ("what is the API for creating a PO", "CreatePO", ["CreatePO", "PoCrt"]),
    ("how do I amend a purchase order", "PoAmnd", ["Amend", "AmendPO"]),
    ("which API lets me view PO details", "GetPODetails", ["GetPODetails", "PoViewDtls"]),
    ("what screens are in PoCrt", "PoCrt", ["PoCrtMain", "/wiki/screen/"]),
    ("tell me about CreatePO", "CreatePO", ["REST endpoint", "fields"]),
    ("which errors does poemn_sp_aprgrd raise", "poemn_sp_aprgrd", ["/wiki/error/"]),
    ("what stored procedures handle approval", None, ["/wiki/sp/"]),
    ("which stored procedures does PoCrt run", "PoCrt", ["/wiki/sp/"]),
    ("tell me about error 2130164", "2130164", ["Dropship", "Raised by"]),
    ("what does poaprmn_sp_apr_hdrsav write", "poaprmn_sp_apr_hdrsav", ["write"]),
    ("which activity shows PoCrtMain", "PoCrtMain", ["PoCrt"]),
    ("columns of PO_POSHD_SCHEDULE_DTL", "PO_POSHD_SCHEDULE_DTL", ["column"]),
    ("schedule distribution api", None, ["ScheduleDistribution", "/wiki/api/"]),
]


def main():
    passed = 0
    for q, exp_match, subs in GOLD:
        res = answer(q)
        ans = res.get("answer", "")
        matched = (res.get("matched") or {}).get("name")
        ok_match = (exp_match is None) or (matched == exp_match)
        ok_subs = all(s.lower() in ans.lower() for s in subs)
        ok = ok_match and ok_subs
        passed += ok
        flag = "PASS" if ok else "FAIL"
        detail = "" if ok else f"  (matched={matched!r} exp={exp_match!r} subs_ok={ok_subs})"
        print(f"[{flag}] {q}{detail}")
    print(f"\n{passed}/{len(GOLD)} gold queries passed.")
    return passed == len(GOLD)


if __name__ == "__main__":
    import sys
    sys.exit(0 if main() else 1)