""" 15-round focused test: IPM treatment limits always stated when recommending products. Every round triggers a product recommendation and checks that: 1. An AllTech product is named 2. A treatment count limit (max N / anno / stagione / season) is explicitly stated 3. No generic chemicals are recommended by name """ import httpx import time import sys import io sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") BASE_URL = "http://localhost:8000" SESSION_ID = f"test_ipm_limits_{int(time.time())}" TIMEOUT = 60.0 SEP = "-" * 80 ALLTECH_NAMES = [ "procrop", "sol-plex", "chitosano", "lecitina", "equisetum", "algae", "alltech", "chitosan", ] LIMIT_WORDS = [ # Italian "max", "massimo", "massima", "al massimo", "trattamenti", "trattamento", "anno", "stagione", "ciclo", "applicazioni", "applicazione", "ipm", "2026", "limite", "limitaz", # English "maximum", "season", "per year", "applications", "limit", "times", ] GENERIC_CHEMICALS = [ "folpet", "mancozeb", "cymoxanil", "fluazinam", "dithianon", "azoxystrobin", "tebuconazole", "fosetil", "fosfonato", "metalaxyl", "iprodione", "chlorothalonil", ] def has_alltech(r: str) -> bool: rl = r.lower() return any(name in rl for name in ALLTECH_NAMES) def has_limit(r: str) -> bool: rl = r.lower() return any(w in rl for w in LIMIT_WORDS) def no_generics(r: str) -> bool: rl = r.lower() return not any(chem in rl for chem in GENERIC_CHEMICALS) # ── 15 rounds across crops, diseases, languages, farming types ───────────────── ROUNDS = [ # 1 — Vite / oidio / conventional / Italian { "round": 1, "label": "Vite oidio — convenzionale (IT)", "message": "ho l'oidio sulle mie viti al BBCH 57, cosa uso?", "crop_type": "vite", "farming_type": "conventional", "bbch_stage": "57", "expected": "AllTech product + IPM max N treatments stated", }, # 2 — Vite / peronospora / conventional / Italian { "round": 2, "label": "Vite peronospora — convenzionale (IT)", "message": "peronospora sulle viti, quali prodotti posso usare?", "crop_type": "vite", "farming_type": "conventional", "bbch_stage": "71", "expected": "AllTech product + IPM limit stated", }, # 3 — Pomodoro / alternaria / conventional / English { "round": 3, "label": "Pomodoro alternaria — conventional (EN)", "message": "I have early blight on my tomatoes, what should I spray?", "crop_type": "pomodoro", "farming_type": "conventional", "bbch_stage": "71", "expected": "English response, AllTech product + IPM limit stated", }, # 4 — Vite / botrytis / organic / Italian { "round": 4, "label": "Vite botrytis — biologico (IT)", "message": "ho la muffa grigia sui grappoli, sono in biologico, cosa faccio?", "crop_type": "vite", "farming_type": "organic", "bbch_stage": "79", "expected": "Organic AllTech product + IPM limit stated", }, # 5 — Melo / ticchiolatura / conventional / Italian { "round": 5, "label": "Melo ticchiolatura — convenzionale (IT)", "message": "ticchiolatura sul melo, dimmi cosa usare e quante volte posso trattare", "crop_type": "melo", "farming_type": "conventional", "bbch_stage": "65", "expected": "AllTech product + explicit IPM max applications", }, # 6 — Olivo / occhio di pavone / conventional / Italian { "round": 6, "label": "Olivo occhio di pavone — convenzionale (IT)", "message": "vedo macchie circolari giallastre sulle foglie degli olivi, trattamento?", "crop_type": "olivo", "farming_type": "conventional", "bbch_stage": None, "expected": "AllTech product + IPM limit", }, # 7 — Pomodoro peronospora / organic / English { "round": 7, "label": "Pomodoro peronospora — organic (EN)", "message": "downy mildew on my tomatoes, I farm organically, what products?", "crop_type": "pomodoro", "farming_type": "organic", "bbch_stage": "71", "expected": "Organic AllTech product + limit stated in English", }, # 8 — Pero / ticchiolatura / conventional / Italian { "round": 8, "label": "Pero ticchiolatura — convenzionale (IT)", "message": "ticchiolatura sul pero BBCH 65, quale prodotto e quanti trattamenti?", "crop_type": "pero", "farming_type": "conventional", "bbch_stage": "65", "expected": "AllTech product + explicit treatment count limit", }, # 9 — Fragola / botrytis / organic / Italian { "round": 9, "label": "Fragola botrytis — biologico (IT)", "message": "botrytis sulle fragole in biologico, come la combatto?", "crop_type": "fragola", "farming_type": "organic", "bbch_stage": "71", "expected": "Organic AllTech product + IPM limit", }, # 10 — Vite / peronospora / organic / English (language switch) { "round": 10, "label": "Vite peronospora — organic (EN)", "message": "downy mildew on my organic vineyard, what can I apply and how many times?", "crop_type": "vite", "farming_type": "organic", "bbch_stage": "57", "expected": "English, organic AllTech product + limit stated", }, # 11 — Pomodoro / botrytis / conventional / Italian { "round": 11, "label": "Pomodoro botrytis — convenzionale (IT)", "message": "botrytis sul pomodoro in fase di fruttificazione, prodotto e limite trattamenti?", "crop_type": "pomodoro", "farming_type": "conventional", "bbch_stage": "79", "expected": "AllTech product + IPM limit clearly stated", }, # 12 — Melo / oidio / organic / Italian { "round": 12, "label": "Melo oidio — biologico (IT)", "message": "ho l'oidio sul melo, pratico biologico, cosa posso usare?", "crop_type": "melo", "farming_type": "organic", "bbch_stage": "65", "expected": "Organic AllTech product + limit stated", }, # 13 — Pomodoro / alternaria follow-up / conventional / Italian { "round": 13, "label": "Follow-up: quanti trattamenti rimangono?", "message": "ho già fatto 2 trattamenti, quanti me ne rimangono secondo il disciplinare?", "crop_type": "pomodoro", "farming_type": "conventional", "bbch_stage": "71", "expected": "References IPM limit, calculates remaining treatments", }, # 14 — Vite / esca/mal dell'esca / conventional / Italian { "round": 14, "label": "Vite esca — convenzionale (IT)", "message": "le mie viti hanno i sintomi dell'esca, cosa posso fare e con che prodotto?", "crop_type": "vite", "farming_type": "conventional", "bbch_stage": None, "expected": "AllTech product or agronomic advice + any applicable limit", }, # 15 — Pomodoro / peronospora / conventional / English (final regression) { "round": 15, "label": "Pomodoro peronospora — conventional (EN) regression", "message": "my tomatoes have downy mildew, recommend a product with the IPM treatment limit", "crop_type": "pomodoro", "farming_type": "conventional", "bbch_stage": "71", "expected": "English, AllTech product, explicit IPM limit", }, ] def run_round(client: httpx.Client, r: dict, session_id: str) -> dict: payload = { "session_id": session_id, "message": r["message"], "farming_type": r.get("farming_type", "conventional"), "weather_enabled": False, } if r.get("crop_type"): payload["crop_type"] = r["crop_type"] if r.get("bbch_stage"): payload["bbch_stage"] = r["bbch_stage"] t0 = time.time() resp = client.post(f"{BASE_URL}/api/chat", json=payload, timeout=TIMEOUT) elapsed = time.time() - t0 if resp.status_code != 200: return {"error": f"HTTP {resp.status_code}: {resp.text[:300]}", "elapsed": elapsed} data = resp.json() return {"response": data.get("response", ""), "elapsed": elapsed} def main(): print("\n" + "=" * 80) print(" 15-ROUND TEST: IPM Limits Always Stated") print(f" Session: {SESSION_ID}") print("=" * 80 + "\n") passed_checks = 0 total_checks = 0 round_results = [] # Fresh session per round — each is independent with httpx.Client() as client: for r in ROUNDS: # New session each round so history doesn't interfere round_session = f"{SESSION_ID}_r{r['round']}" print(SEP) print(f" ROUND {r['round']:02d} - {r['label']}") print(SEP) print(f" USER : {r['message']}") print(f" CROP : {r.get('crop_type','-')} FARMING: {r.get('farming_type','-')} BBCH: {r.get('bbch_stage','-')}") result = run_round(client, {**r}, round_session) if "error" in result: print(f" [ERROR] {result['error']}") round_results.append({"round": r["round"], "label": r["label"], "error": True}) total_checks += 3 continue resp_text = result["response"] print(f"\n MARCO : {resp_text[:500]}{'...' if len(resp_text) > 500 else ''}") print(f" TIME : {result['elapsed']:.1f}s") # Three checks per round c_alltech = has_alltech(resp_text) c_limit = has_limit(resp_text) c_no_generics = no_generics(resp_text) checks = { "alltech_product_named": c_alltech, "ipm_limit_stated": c_limit, "no_generic_chemicals": c_no_generics, } round_pass = all(checks.values()) round_results.append({ "round": r["round"], "label": r["label"], "pass": round_pass, "checks": checks, }) print(f"\n EXPECTED : {r['expected']}") print(" CHECKS:") for name, ok in checks.items(): icon = "[OK] " if ok else "[FAIL]" print(f" {icon} {name}") total_checks += 1 if ok: passed_checks += 1 print(f" ROUND: {'PASS' if round_pass else 'FAIL'}") print() # ── Summary ──────────────────────────────────────────────────────────────── print("\n" + "=" * 80) print(" RESULTS SUMMARY") print("=" * 80) passed_rounds = sum(1 for r in round_results if r.get("pass")) total_rounds = len([r for r in round_results if "error" not in r]) accuracy = (passed_checks / total_checks * 100) if total_checks else 0 for r in round_results: if "error" in r: print(f" Round {r['round']:02d}: [ERROR] {r['label']}") else: icon = "[PASS]" if r["pass"] else "[FAIL]" cp = sum(1 for v in r["checks"].values() if v) ct = len(r["checks"]) print(f" Round {r['round']:02d}: {icon} [{cp}/{ct}] - {r['label']}") if not r["pass"]: for cname, ok in r["checks"].items(): if not ok: print(f" FAIL: {cname}") print(f"\n Rounds passed : {passed_rounds}/{total_rounds}") print(f" Checks passed : {passed_checks}/{total_checks}") print(f" Accuracy : {accuracy:.1f}%") print("=" * 80 + "\n") return accuracy if __name__ == "__main__": acc = main() sys.exit(0 if acc >= 85.0 else 1)