atlas-iac/testing/tests/test_hermes_cli_lane_goal.py

"""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, role=goal.IMPLEMENTATION_ROLE)


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, role=goal.IMPLEMENTATION_ROLE) 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, role=goal.IMPLEMENTATION_ROLE) 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


def test_completed_report_with_corrupt_test_evidence_is_still_inspected():
    """A non-list tests_run field neither crashes nor blocks the summary gate."""
    assert goal.unfinished_result_reason(_result(tests_run=None)) is None
    assert goal.unfinished_result_reason(
        _result(
            tests_run={"suite": "still running"},
            summary="The remaining work is still running.",
        )
    )