ewoooc/tests/test_pixelrag_vlm_route_readiness_service.py

import json


class FakeResponse:
    def __init__(self, status_code=200, payload=None, text=""):
        self.status_code = status_code
        self._payload = payload or {}
        self.text = text

    def json(self):
        return self._payload


def test_pixelrag_vlm_route_readiness_selects_installed_candidate(monkeypatch):
    from services import pixelrag_vlm_route_readiness_service as service

    hosts = [
        "http://primary:11434",
        "http://secondary:11434",
    ]
    monkeypatch.setattr(service, "_approved_hosts", lambda: hosts)

    def fake_get(url, timeout):
        if url.startswith("http://primary"):
            raise TimeoutError("primary timeout")
        return FakeResponse(
            payload={
                "models": [
                    {"name": "gemma3:4b"},
                    {"name": "bge-m3:latest"},
                ]
            }
        )

    monkeypatch.setattr(service.requests, "get", fake_get)
    payload = service.build_pixelrag_vlm_route_readiness(
        model="minicpm-v:latest",
        timeout_seconds=1,
    )

    assert payload["policy"] == "read_only_pixelrag_vlm_route_readiness_v1"
    assert payload["status"] == "warning"
    assert payload["summary"]["reachable_host_count"] == 1
    assert payload["summary"]["configured_model_available_count"] == 0
    assert payload["summary"]["candidate_model"] == "gemma3:4b"
    assert payload["summary"]["model_route_ready"] is True
    assert payload["controlled_apply"]["model_call"] is False
    assert payload["controlled_apply"]["writes_database"] is False
    assert payload["next_machine_action"] == "run_pixelrag_vlm_replay_worker_execute_with_selected_model"


def test_pixelrag_vlm_route_readiness_critical_when_no_candidate(monkeypatch):
    from services import pixelrag_vlm_route_readiness_service as service

    monkeypatch.setattr(service, "_approved_hosts", lambda: ["http://secondary:11434"])
    monkeypatch.setattr(
        service.requests,
        "get",
        lambda url, timeout: FakeResponse(payload={"models": [{"name": "bge-m3:latest"}]}),
    )

    payload = service.build_pixelrag_vlm_route_readiness(model="minicpm-v:latest")

    assert payload["status"] == "critical"
    assert payload["success"] is False
    assert payload["summary"]["model_route_ready"] is False
    assert payload["next_machine_action"] == "install_or_configure_pixelrag_vlm_model_on_approved_ollama_host"


def test_pixelrag_vlm_route_readiness_generate_probe_success(monkeypatch):
    from services import pixelrag_vlm_route_readiness_service as service

    monkeypatch.setattr(service, "_approved_hosts", lambda: ["http://secondary:11434"])
    monkeypatch.setattr(
        service.requests,
        "get",
        lambda url, timeout: FakeResponse(payload={"models": [{"name": "gemma3:4b"}]}),
    )
    monkeypatch.setattr(
        service.requests,
        "post",
        lambda url, json, timeout: FakeResponse(payload={"response": "OK", "done": True}),
    )

    payload = service.build_pixelrag_vlm_route_readiness(
        model="minicpm-v:latest",
        probe_generate=True,
        probe_timeout_seconds=1,
    )

    assert payload["status"] == "warning"
    assert payload["summary"]["candidate_model"] == "gemma3:4b"
    assert payload["summary"]["tag_model_route_ready"] is True
    assert payload["summary"]["model_route_ready"] is True
    assert payload["summary"]["generate_probe_performed"] is True
    assert payload["summary"]["generate_probe_ok_count"] == 1
    assert payload["summary"]["generate_route_ready"] is True
    assert payload["summary"]["generate_ready_host"] == "http://secondary:11434"
    assert payload["controlled_apply"]["model_call"] is True


def test_pixelrag_vlm_route_readiness_generate_probe_failure_blocks_route(monkeypatch):
    from services import pixelrag_vlm_route_readiness_service as service

    monkeypatch.setattr(service, "_approved_hosts", lambda: ["http://secondary:11434"])
    monkeypatch.setattr(
        service.requests,
        "get",
        lambda url, timeout: FakeResponse(payload={"models": [{"name": "gemma3:4b"}]}),
    )

    def fake_post(url, json, timeout):
        raise TimeoutError("generate timeout")

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

    payload = service.build_pixelrag_vlm_route_readiness(
        model="minicpm-v:latest",
        probe_generate=True,
        probe_timeout_seconds=1,
    )

    assert payload["status"] == "critical"
    assert payload["success"] is False
    assert payload["summary"]["candidate_model"] == "gemma3:4b"
    assert payload["summary"]["tag_model_route_ready"] is True
    assert payload["summary"]["model_route_ready"] is False
    assert payload["summary"]["generate_probe_performed"] is True
    assert payload["summary"]["generate_probe_ok_count"] == 0
    assert payload["summary"]["generate_route_ready"] is False
    assert payload["next_machine_action"] == "repair_ollama_vlm_generate_runtime_or_proxy_timeout"


def test_pixelrag_vlm_route_readiness_route_returns_readback(monkeypatch):
    from flask import Flask
    from routes import system_public_routes as routes
    from services import pixelrag_vlm_route_readiness_service as service

    monkeypatch.setattr(
        service,
        "build_pixelrag_vlm_route_readiness",
        lambda **kwargs: {
            "success": True,
            "policy": "read_only_pixelrag_vlm_route_readiness_v1",
            "status": "ok",
            "summary": {"model_route_ready": True, "candidate_model": "gemma3:4b"},
            "next_machine_action": "run_pixelrag_vlm_replay_worker_execute",
        },
    )

    app = Flask(__name__)
    with app.test_request_context(
        "/api/ai-automation/pixelrag-vlm-route-readiness?model=minicpm-v:latest"
        "&probe_generate=true&probe_timeout=4"
    ):
        response = routes.ai_automation_pixelrag_vlm_route_readiness_api.__wrapped__()
    payload = response.get_json()

    assert payload["policy"] == "read_only_pixelrag_vlm_route_readiness_v1"
    assert payload["summary"]["candidate_model"] == "gemma3:4b"


def test_pixelrag_vlm_route_readiness_cli_outputs_json(monkeypatch, tmp_path):
    from services import pixelrag_vlm_route_readiness_service as service

    monkeypatch.setattr(service, "_approved_hosts", lambda: ["http://secondary:11434"])
    monkeypatch.setattr(
        service.requests,
        "get",
        lambda url, timeout: FakeResponse(payload={"models": [{"name": "gemma3:4b"}]}),
    )

    payload = service.build_pixelrag_vlm_route_readiness(model="minicpm-v:latest")
    assert json.dumps(payload, ensure_ascii=False)
    assert payload["summary"]["candidate_model"] == "gemma3:4b"