ewoooc/tests/test_nemotron_decision_canary_service.py

import inspect
import json
import subprocess
import sys


class _Response:
    def __init__(self, payload, status_code=200):
        self._payload = payload
        self.status_code = status_code

    def raise_for_status(self):
        if self.status_code >= 400:
            import requests

            raise requests.HTTPError(f"HTTP {self.status_code}")

    def json(self):
        return self._payload


def _tags(service, *, digest=None):
    return {
        "models": [{
            "name": service.NEMOTRON_OLLAMA_MODEL,
            "model": service.NEMOTRON_OLLAMA_MODEL,
            "digest": digest or service.NEMOTRON_OLLAMA_EXPECTED_DIGEST,
            "details": {
                "parameter_size": "14.8B",
                "quantization_level": "Q4_K_M",
            },
        }]
    }


def test_nemotron_canary_readback_never_calls_network(tmp_path, monkeypatch):
    from services import nemotron_decision_canary_service as service

    monkeypatch.setattr(
        service.requests,
        "get",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(
            AssertionError("network call forbidden")
        ),
    )
    payload = service.run_nemotron_decision_canary(output_root=tmp_path)

    assert payload["status"] == "warning"
    assert payload["controlled_apply"]["network_call"] is False
    assert payload["controlled_apply"]["database_write"] is False


def test_nemotron_canary_executes_decision_only_on_verified_secondary(
    tmp_path, monkeypatch
):
    import requests
    from services import nemotron_decision_canary_service as service

    def fake_get(url, timeout):
        assert timeout == service.DEFAULT_MODEL_IDENTITY_TIMEOUT_SEC
        if service.OLLAMA_HOST_PRIMARY in url:
            raise requests.ConnectionError("primary unavailable")
        return _Response(_tags(service))

    posted = []

    def fake_post(url, json, timeout):
        posted.append((url, json, timeout))
        return _Response({
            "message": {
                "role": "assistant",
                "content": "",
                "tool_calls": [{
                    "function": {
                        "name": "trigger_price_alert",
                        "arguments": {
                            "sku": "CANARY-NEMO-001",
                            "name": "NemoTron 受控決策驗證品",
                            "gap_pct": 22.4,
                            "sales_delta": -35.0,
                            "action": "建立價格告警候選",
                            "confidence": 0.91,
                        },
                    }
                }],
            },
            "prompt_eval_count": 100,
            "eval_count": 20,
        })

    monkeypatch.setattr(service.requests, "get", fake_get)
    monkeypatch.setattr(service.requests, "post", fake_post)

    payload = service.run_nemotron_decision_canary(
        output_root=tmp_path,
        execute=True,
        write_receipt=True,
        timeout_sec=45,
        run_id="nemotron-canary-test-001",
    )

    assert payload["status"] == "canary_passed"
    execution = payload["execution"]
    assert execution["selected_host"] == service.OLLAMA_HOST_SECONDARY
    assert execution["decision_tool"] == "trigger_price_alert"
    assert execution["decision_sku"] == "CANARY-NEMO-001"
    assert execution["tool_execution_count"] == 0
    assert execution["database_call_performed"] is False
    assert execution["writes_price_tables"] is False
    assert execution["writes_ai_insights"] is False
    assert execution["telegram_sent"] is False
    assert execution["rollback_terminal"] == "decision_only_no_tool_or_data_write"
    assert posted[0][0].endswith("/api/chat")
    assert posted[0][1]["keep_alive"] == 0
    assert posted[0][1]["think"] is False
    assert posted[0][2] == 45
    assert payload["latest_execution"]["canary_passed"] is True

    acknowledged = service.acknowledge_nemotron_decision_canary(
        execution["receipt_path"],
        telegram_status="acknowledged",
        telegram_sent=1,
    )
    assert acknowledged["terminal_status"] == "verified_decision_only_no_write"
    assert acknowledged["closure_receipt"]["telegram_acknowledgement"] == (
        "acknowledged"
    )
    readback = service.run_nemotron_decision_canary(output_root=tmp_path)
    assert readback["status"] == "verified"
    assert readback["latest_execution"]["terminal_status"] == (
        "verified_decision_only_no_write"
    )


def test_nemotron_canary_blocks_model_digest_drift_before_model_call(
    tmp_path, monkeypatch
):
    from services import nemotron_decision_canary_service as service

    monkeypatch.setattr(
        service.requests,
        "get",
        lambda *_args, **_kwargs: _Response(_tags(service, digest="deadbeef")),
    )
    model_calls = []
    monkeypatch.setattr(
        service.requests,
        "post",
        lambda *_args, **_kwargs: model_calls.append(True),
    )

    payload = service.run_nemotron_decision_canary(
        output_root=tmp_path,
        execute=True,
        write_receipt=True,
    )

    assert payload["status"] == "canary_failed"
    assert payload["execution"]["model_call_performed"] is False
    assert payload["execution"]["error"] == (
        "no_approved_gcp_host_with_expected_nemotron_digest"
    )
    assert model_calls == []


def test_nemotron_canary_route_is_read_only():
    from routes import system_public_routes as routes

    source = inspect.getsource(routes.ai_automation_nemotron_decision_canary_api)
    assert "execute=False" in source
    assert "write_receipt=False" in source


def test_nemotron_canary_cli_outputs_machine_readable_readback(tmp_path):
    completed = subprocess.run(
        [
            sys.executable,
            "scripts/ops/run_nemotron_decision_canary.py",
            "--output-root",
            str(tmp_path),
        ],
        capture_output=True,
        check=False,
        text=True,
    )

    assert completed.returncode == 0
    payload = json.loads(completed.stdout)
    assert payload["policy"] == "controlled_nemotron_decision_only_canary_v1"
    assert payload["controlled_apply"]["tool_execution_count"] == 0