121 lines
3.4 KiB
Python
121 lines
3.4 KiB
Python
"""Completion-gate tests for Hermes external Kanban workers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
SCRIPT = (
|
|
Path(__file__).parents[2]
|
|
/ "services/hermes/scripts/cli_lane_goal.py"
|
|
)
|
|
SPEC = importlib.util.spec_from_file_location("cli_lane_goal_test", SCRIPT)
|
|
assert SPEC and SPEC.loader
|
|
goal = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = goal
|
|
SPEC.loader.exec_module(goal)
|
|
|
|
|
|
def _result(**overrides):
|
|
value = {
|
|
"status": "completed",
|
|
"summary": "All acceptance criteria passed and the branch was pushed.",
|
|
"changed_files": ["src/example.py"],
|
|
"tests_run": ["pytest -q: 12 passed"],
|
|
"artifacts": [],
|
|
"blockers": [],
|
|
}
|
|
value.update(overrides)
|
|
return value
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"result",
|
|
[
|
|
_result(status="incomplete", summary="The full suite is still running."),
|
|
_result(summary="The broad rerun remains active before the final push."),
|
|
_result(tests_run=["Full suite — in progress (11m)"]),
|
|
_result(blockers=["remote head was not verified"]),
|
|
],
|
|
)
|
|
def test_unfinished_completion_evidence_is_rejected(result):
|
|
assert goal.unfinished_result_reason(result)
|
|
|
|
|
|
def test_completed_pending_state_test_name_is_not_a_false_positive():
|
|
result = _result(tests_run=["pytest tests/test_pending_build_state.py: 8 passed"])
|
|
|
|
assert goal.unfinished_result_reason(result) is None
|
|
|
|
|
|
def test_completed_review_findings_are_not_task_blockers():
|
|
result = _result(
|
|
summary="Review complete; the pull request is not merge-ready.",
|
|
findings=["A casting-table variant fails open."],
|
|
)
|
|
|
|
assert goal.unfinished_result_reason(result) is None
|
|
|
|
|
|
class JudgeResponse:
|
|
def __init__(self, verdict: str, reason: str):
|
|
self.body = json.dumps(
|
|
{
|
|
"choices": [
|
|
{
|
|
"message": {
|
|
"content": json.dumps(
|
|
{"verdict": verdict, "reason": reason}
|
|
)
|
|
}
|
|
}
|
|
]
|
|
}
|
|
).encode()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def read(self):
|
|
return self.body
|
|
|
|
|
|
def test_goal_judge_uses_local_structured_verdict():
|
|
observed = {}
|
|
|
|
def request(req, timeout):
|
|
observed["payload"] = json.loads(req.data)
|
|
observed["timeout"] = timeout
|
|
return JudgeResponse("continue", "remote head evidence is missing")
|
|
|
|
accepted, reason = goal.judge_goal_completion(
|
|
"Run tests, push, and verify the remote head.",
|
|
_result(summary="Tests passed."),
|
|
open_request=request,
|
|
)
|
|
|
|
assert accepted is False
|
|
assert reason == "remote head evidence is missing"
|
|
assert observed["payload"]["model"] == "qwen2.5:14b-instruct-q4_0"
|
|
assert observed["payload"]["response_format"]["json_schema"]["strict"] is True
|
|
assert observed["timeout"] == 120
|
|
|
|
|
|
def test_goal_judge_fails_closed_on_invalid_response():
|
|
accepted, reason = goal.judge_goal_completion(
|
|
"Finish the task.",
|
|
_result(),
|
|
open_request=lambda *_args, **_kwargs: JudgeResponse("unknown", "bad"),
|
|
)
|
|
|
|
assert accepted is False
|
|
assert "local completion judge unavailable" in reason
|