"""Integration test for the /run-eval endpoint against a real deployment.""" import time import requests from tests.integration import config _POLL_INTERVAL_SECONDS = 1 _POLL_TIMEOUT_SECONDS = 60 _REQUEST_TIMEOUT_SECONDS = 10 _PAYLOAD = { "runs": 1, "pipeline": {"pipeline_name": "integration-test", "pipeline_id": "1"}, "prompts": { "capital_of_uk": { "prompt": "What is the capital of the UK, answer in one word.", "judges": [ { "name": "mentions_london", "instructions": "Answer contains London", } ], } }, } def _poll_job(job_id: str) -> dict: deadline = time.monotonic() + _POLL_TIMEOUT_SECONDS while time.monotonic() < deadline: response = requests.get( f"{config.QA_BASE_URL}/run-eval/{job_id}", timeout=_REQUEST_TIMEOUT_SECONDS ) assert response.status_code == 200, "\nReason: {}\nURL: {}".format( response.reason, response.url ) body = response.json() if body["status"] not in ("pending", "running"): return body time.sleep(_POLL_INTERVAL_SECONDS) raise TimeoutError(f"job {job_id} did not finish within {_POLL_TIMEOUT_SECONDS}s") def test_run_eval_with_no_mcp_runs_prompt_and_judge() -> None: """No `mcp` configured — the model should answer with no tools available, and the judge should confirm the answer names the correct city.""" response = requests.post( f"{config.QA_BASE_URL}/run-eval", json=_PAYLOAD, timeout=_REQUEST_TIMEOUT_SECONDS, ) assert response.status_code == 202, "\nReason: {}\nURL: {}".format( response.reason, response.url ) job_id = response.json()["job_id"] job = _poll_job(job_id) assert job["status"] == "succeeded", job.get("error") run = job["result"]["prompts"]["capital_of_uk"]["runs"][0] assert run["tools_called"] == [] assert run["tool_correct"] is True judge_results = {jr["name"]: jr for jr in run["judge_results"]} assert judge_results["mentions_london"]["passed"] is True