aios/tests/test_action_contract_regressions.py

from __future__ import annotations

import inspect
import json
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "runtime/agent-os"))

from aios_chatgpt_bridge_v2 import (  # noqa: E402
    OPERATION_REQUIRED_FIELDS,
    operation_contracts,
    render_openapi_contract_fragment,
)
from aios_chatgpt_bridge_v2_server import GovernedRuntime  # noqa: E402
from aios_gitops_job_controller import JobState  # noqa: E402
from gitea_tag_publication import validate_gitea_api_url  # noqa: E402
from silent_release_hardening import ReleaseSagaStore, SilentReleaseActionService  # noqa: E402
from release_version import validate_release_version  # noqa: E402
import aios_model_api_staging as model_api  # noqa: E402


def test_canonical_release_version_is_shared_and_accepts_v2r2c1():
    assert validate_release_version("V2R2C1") == "V2R2C1"
    assert validate_release_version("V2R2C1-repair") == "V2R2C1-repair"
    with pytest.raises(ValueError):
        validate_release_version("v2r2c1")


def test_gitea_url_validation_is_startup_safe():
    assert validate_gitea_api_url("https://g.cnnetmkt.top/git/api/v1") == (
        "https://g.cnnetmkt.top/git/api/v1"
    )
    with pytest.raises(ValueError, match="HTTP"):
        validate_gitea_api_url("g.cnnetmkt.top/git/api/v1")


def test_public_release_reads_use_the_bridge_canonical_evaluator():
    query = {
        "version": ["V2R2C1"],
        "target_oid": ["998d30304787e31e5ae607c0bd86b5edb2b0e5ad"],
        "evidence_closure_oid": ["a8b906d1bc24088c19647e98de6a679268f6d0a2"],
        "approval_plan_id": ["sha256:plan"],
    }
    assert model_api.governed_release_query_payload(query) == {
        "version": "V2R2C1",
        "target_oid": "998d30304787e31e5ae607c0bd86b5edb2b0e5ad",
        "evidence_closure_oid": "a8b906d1bc24088c19647e98de6a679268f6d0a2",
        "approval_plan_id": "sha256:plan",
    }
    source = inspect.getsource(model_api.H.do_GET)
    assert "publication_publisher()" in source
    assert "get_release_consistency" in source
    assert "run_release_acceptance" in source


def test_governed_read_routes_are_bound_to_authenticated_facade():
    route_script = (
        ROOT / "ops/gitops-migration/bind_external_action_surface_final_staging.sh"
    ).read_text()
    for path in (
        "/api/governed-write-capabilities",
        "/api/control-plane-snapshot",
        "/api/online-agent-task",
        "/api/online-agent-task-events",
    ):
        assert f"location = {path}" in route_script
        assert f"proxy_pass http://127.0.0.1:18890{path};" in route_script
    assert "location /api/ {" in route_script
    assert "Authorization $http_authorization" in route_script


def test_commit_files_runtime_and_published_schema_have_same_required_fields():
    runtime = operation_contracts()["commitFilesGoverned"]["required_fields"]
    assert runtime == list(OPERATION_REQUIRED_FIELDS["commitFilesGoverned"])
    fragment = render_openapi_contract_fragment()
    published = (ROOT / "docs/gitops/openapi-chatgpt-actions.yaml").read_text()
    assert fragment in published
    for field in runtime:
        assert f"                - {field}" in fragment
    assert "required: [path, content_base64]" in fragment


class _Store:
    def __init__(self, row):
        self.row = row
        self.receipt_calls = 0

    def read(self, _job_id):
        return self.row

    def _receipt(self, row, duplicate=False):
        self.receipt_calls += 1
        return type("Receipt", (), {"as_dict": lambda _self: {
            "job_id": row["job_id"],
            "state": row["state"],
            "duplicate": duplicate,
        }})()


class _Controller:
    def __init__(self, row):
        self.store = _Store(row)


def _runtime_for(row):
    runtime = object.__new__(GovernedRuntime)
    controller = _Controller(row)
    runtime._controller_for_job = lambda _job_id: controller
    runtime._executor_from_row = lambda *_args: (_ for _ in ()).throw(
        AssertionError("terminal reconcile must not reconstruct an executor")
    )
    return runtime, controller


@pytest.mark.parametrize(
    "state",
    [
        JobState.COMPLETED.value,
        JobState.FAILED.value,
        JobState.CANCELLED.value,
        JobState.MANUAL_OWNER_REVIEW_REQUIRED.value,
    ],
)
def test_terminal_reconcile_is_idempotent_without_durable_binding(state):
    row = {
        "job_id": "job-terminal",
        "state": state,
        "request_json": json.dumps({}),
    }
    runtime, controller = _runtime_for(row)
    result = GovernedRuntime.job(runtime, {"job_id": row["job_id"]}, "reconcile")
    assert result["duplicate"] is True
    assert controller.store.receipt_calls == 1


def test_active_reconcile_still_uses_executor():
    row = {
        "job_id": "job-active",
        "state": JobState.RECONCILIATION_REQUIRED.value,
        "request_json": json.dumps({}),
    }
    runtime, controller = _runtime_for(row)
    class Executor:
        def reconcile(self, job_id):
            return {"job_id": job_id, "state": "COMPLETED"}
    runtime._executor_from_row = lambda *_args: Executor()
    assert GovernedRuntime.job(runtime, {"job_id": row["job_id"]}, "reconcile") == {
        "job_id": "job-active",
        "state": "COMPLETED",
    }


def test_missing_release_saga_is_a_not_found_empty_state(tmp_path):
    service = object.__new__(SilentReleaseActionService)
    service.sagas = ReleaseSagaStore(tmp_path)
    service.execution_enabled = False
    result = service.get_release_saga("never-created-action-contract")
    assert result["status"] == "NOT_FOUND"
    assert result["state"] == "NOT_FOUND"
    assert not list(tmp_path.rglob("*"))