ontorefactor-governance/tests/test_paas.py

import json

from ontorefactor_governance.paas import paas_status, record_agent_run


class FakeResponse:
    def __enter__(self):
        return self

    def __exit__(self, *_):
        return False

    def read(self):
        return b'{"run_id":"paas-run-1"}'


def _clear(monkeypatch):
    for name in (
        "AGENTPAAS_BASE_URL", "AGENTPAAS_API_KEY", "ONTOREFACTOR_PAAS_ASSET_AGENT_ID",
        "ONTOREFACTOR_PAAS_SEMANTIC_AGENT_ID", "ONTOREFACTOR_PAAS_VALIDATION_AGENT_ID",
        "ONTOREFACTOR_PAAS_DECISION_AGENT_ID",
    ):
        monkeypatch.delenv(name, raising=False)


def test_paas_is_optional(monkeypatch):
    _clear(monkeypatch)

    assert paas_status()["configured"] is False
    assert record_agent_run(
        "asset-construction-agent", input_summary={}, output_summary={}
    ) == {"role": "asset-construction-agent", "status": "not_configured"}


def test_paas_records_only_role_summary_when_configured(monkeypatch):
    _clear(monkeypatch)
    monkeypatch.setenv("AGENTPAAS_BASE_URL", "https://paas.example.cn")
    monkeypatch.setenv("AGENTPAAS_API_KEY", "paas-secret")
    monkeypatch.setenv("ONTOREFACTOR_PAAS_ASSET_AGENT_ID", "agent-asset")
    captured = {}

    def fake_urlopen(request, timeout):
        captured["url"] = request.full_url
        captured["authorization"] = request.get_header("Authorization")
        captured["payload"] = json.loads(request.data)
        captured["timeout"] = timeout
        return FakeResponse()

    monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
    result = record_agent_run(
        "asset-construction-agent",
        input_summary={"build_id": "ob_1", "source_count": 4},
        output_summary={"elements": 42},
        duration_ms=12,
    )

    assert result == {"role": "asset-construction-agent", "status": "recorded", "run_id": "paas-run-1"}
    assert captured["url"] == "https://paas.example.cn/api/v1/agents/agent-asset/runs/record"
    assert captured["authorization"] == "Bearer paas-secret"
    assert captured["payload"]["source"] == "ontorefactor-governance"
    assert "ob_1" in captured["payload"]["input"]
    assert captured["timeout"] == 5