EvoScientist/tests/test_provider_test_api.py

"""Tests for POST /api/model-registry/test (design doc 9.4, 4.3, 5.1, 6.2).

A Task 2 style fake DNS plus a local threaded HTTP server (registered as a
``development_endpoints`` entry) simulates an OpenAI-compatible provider:
normal chat completions, 401, 404, connection refusal, and tool-call
requests. Covers the full verification flow — frozen resolution, credential
resolution, the minimal real call, capability probes, the guarded
``model_verifications`` upsert, response redaction, and the
``model_config:test`` scope — plus every section 9.4 error path.
"""

from __future__ import annotations

import json
import socket
import threading
import time
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import jwt
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import ec
from starlette.applications import Starlette
from starlette.testclient import TestClient

from EvoScientist.model_registry.auth import BffAuthenticator
from EvoScientist.model_registry.endpoint_policy import EndpointPolicy
from EvoScientist.model_registry.hashing import configuration_hash
from EvoScientist.model_registry.http_api import (
    ApiServices,
    model_registry_routes,
)
from EvoScientist.model_registry.platform import DelegationPublicKey
from EvoScientist.model_registry.provider_test import ProviderTester
from EvoScientist.model_registry.resolver import ModelRegistryResolver
from EvoScientist.model_registry.safe_transport import (
    build_safe_async_http_client,
    build_safe_http_client,
)
from EvoScientist.model_registry.schemas import (
    CredentialWrite,
    DevelopmentEndpoint,
    RegistryV4,
)
from EvoScientist.model_registry.snapshots import SnapshotService
from EvoScientist.model_registry.store import ModelRuntimeStore

SERVICE_TOKEN = "bff-service-token"
SECRET = "sk-live-9876abcd"
MODEL_REF = {"provider_id": "test-provider", "model_key": "test-model"}

_PRIVATE_KEY = ec.generate_private_key(ec.SECP256R1())
_PRIVATE_PEM = _PRIVATE_KEY.private_bytes(
    serialization.Encoding.PEM,
    serialization.PrivateFormat.PKCS8,
    serialization.NoEncryption(),
)
_PUBLIC_PEM = _PRIVATE_KEY.public_key().public_bytes(
    serialization.Encoding.PEM,
    serialization.PublicFormat.SubjectPublicKeyInfo,
)


# --- fake OpenAI-compatible provider -----------------------------------------


def _completion(content):
    return {
        "id": "chatcmpl-test",
        "object": "chat.completion",
        "created": 1,
        "model": "test-model",
        "choices": [
            {
                "index": 0,
                "message": {"role": "assistant", "content": content},
                "finish_reason": "stop",
            }
        ],
        "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
    }


class _OpenAIHandler(BaseHTTPRequestHandler):
    """OpenAI-compatible chat completions with a switchable failure mode."""

    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        body = json.loads(self.rfile.read(length))
        self.server.requests.append(
            {
                "path": self.path,
                "authorization": self.headers.get("Authorization"),
                "body": body,
            }
        )
        mode = self.server.mode
        if mode == "reject" or (mode == "fail_tools" and body.get("tools")):
            status = 401 if mode == "reject" else 400
            self._respond(
                status, {"error": {"message": "rejected", "type": "auth_error"}}
            )
            return
        if mode == "missing_model":
            self._respond(
                404,
                {"error": {"message": "model not found", "type": "not_found"}},
            )
            return
        # The structured-output probe sends response_format; answer with JSON.
        content = json.dumps({"answer": "ok"}) if body.get("response_format") else "ok"
        self._respond(200, _completion(content))

    def _respond(self, status, payload):
        body = json.dumps(payload).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


@pytest.fixture
def openai_server():
    server = ThreadingHTTPServer(("127.0.0.1", 0), _OpenAIHandler)
    server.requests = []
    server.mode = "ok"
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    yield server
    server.shutdown()
    server.server_close()
    thread.join()


@pytest.fixture
def closed_port():
    """A loopback port with no listener (connection refused)."""
    probe = socket.socket()
    probe.bind(("127.0.0.1", 0))
    port = probe.getsockname()[1]
    probe.close()
    return port


def _fake_getaddrinfo(counter):
    def fake(host, port, *args, **kwargs):
        counter["calls"] += 1
        assert host == "localhost"
        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", port))]

    return fake


# --- registry fixtures ---------------------------------------------------------


def _model_runtime(**overrides):
    payload = {
        "limit_mode": "combined",
        "context_window_tokens": 1048576,
        "max_input_tokens": None,
        "max_output_tokens": 32768,
        "min_effective_input_tokens": 8192,
        "fixed_system_reserve_tokens": 4096,
        "fixed_tools_reserve_tokens": 8192,
        "fixed_attachments_reserve_tokens": 4096,
        "limits_status": "confirmed",
        "limits_source": "provider",
        "temperature": None,
        "top_p": None,
        "reasoning_effort": "auto",
        "declared_capabilities": {
            "tools": True,
            "vision": True,
            "structured_output": True,
        },
    }
    payload.update(overrides)
    return payload


def _provider_payload(port, **overrides):
    provider = {
        "id": "test-provider",
        "name": "Test Provider",
        "adapter": "openai-compatible",
        "base_url": f"http://localhost:{port}/v1",
        "auth": {"mode": "api_key", "credential_id": "test-key"},
        "enabled": True,
        "runtime": {
            "timeout_seconds": 120,
            "max_retries": 2,
            "default_temperature": 0.7,
            "default_top_p": 0.95,
            "default_reasoning_effort": "auto",
        },
        "models": [
            {
                "key": "test-model",
                "name": "Test Model",
                "upstream_model_id": "test-model",
                "enabled": False,
                "runtime": _model_runtime(),
            }
        ],
    }
    provider.update(overrides)
    return provider


def _registry_payload(port, **provider_overrides):
    return {
        "version": 4,
        "revision": 1,
        "state": "bootstrap",
        "defaults": {"primary": None},
        "providers": [_provider_payload(port, **provider_overrides)],
    }


def _save(store, port, *, credential=True, **provider_overrides):
    writes = (
        [CredentialWrite(credential_id="test-key", secret_value=SECRET)]
        if credential
        else []
    )
    return store.save_registry(
        expected_revision=1,
        registry=RegistryV4.model_validate(
            _registry_payload(port, **provider_overrides)
        ),
        credential_writes=writes,
    )


# --- app fixtures --------------------------------------------------------------


@pytest.fixture
def store(tmp_path):
    return ModelRuntimeStore(config_dir=tmp_path)


def _policy_for(port):
    return EndpointPolicy(
        [
            DevelopmentEndpoint(
                id="test-provider",
                url=f"http://localhost:{port}/v1",
                label="Test provider",
            )
        ]
    )


def _services_for(store, port):
    resolver = ModelRegistryResolver(store)
    policy = _policy_for(port)
    fake_dns = _fake_getaddrinfo({"calls": 0})
    tester = ProviderTester(
        store,
        resolver,
        policy,
        sync_client_builder=lambda timeout, retries: build_safe_http_client(
            policy, timeout=timeout, retries=retries, getaddrinfo=fake_dns
        ),
        async_client_builder=lambda timeout, retries: build_safe_async_http_client(
            policy, timeout=timeout, retries=retries, getaddrinfo=fake_dns
        ),
    )
    return ApiServices(
        store=store,
        resolver=resolver,
        snapshot_service=SnapshotService(store, resolver),
        endpoint_policy=policy,
        authenticator=BffAuthenticator(
            service_token=SERVICE_TOKEN,
            service_token_hash=None,
            delegation_keys=(
                DelegationPublicKey(deployment_id="webui-1", public_key=_PUBLIC_PEM),
            ),
            jti_store=store,
        ),
        provider_tester=tester,
    )


@pytest.fixture
def services(store, openai_server):
    return _services_for(store, openai_server.server_address[1])


def _client_for(services):
    app = Starlette(routes=model_registry_routes(lambda: services))
    return TestClient(app)


@pytest.fixture
def client(services):
    return _client_for(services)


@pytest.fixture
def saved_store(store, openai_server):
    _save(store, openai_server.server_address[1])
    return store


@pytest.fixture
def test_client(client, saved_store):
    return client


# --- auth helpers ---------------------------------------------------------------


def _headers(scopes):
    now = int(time.time())
    claims = {
        "iss": "WebUI",
        "aud": "EvoScientist",
        "sub": "user-1",
        "scopes": list(scopes),
        "deployment_id": "webui-1",
        "iat": now,
        "exp": now + 30,
        "jti": uuid.uuid4().hex,
    }
    return {
        "Authorization": f"Bearer {SERVICE_TOKEN}",
        "X-Evo-Actor": jwt.encode(claims, _PRIVATE_PEM, algorithm="ES256"),
    }


def _test_headers():
    return _headers(["model_config:test"])


def _post(client, *, model_ref=None, expected_revision=2, headers=None):
    return client.post(
        "/api/model-registry/test",
        headers=headers or _test_headers(),
        json={
            "expected_registry_revision": expected_revision,
            "model_ref": model_ref or MODEL_REF,
        },
    )


# --- success paths ---------------------------------------------------------------


def test_success_full_flow(test_client, saved_store, openai_server):
    response = _post(test_client)
    assert response.status_code == 200
    body = response.json()
    assert set(body) == {
        "ok",
        "registry_revision",
        "model_ref",
        "adapter_spec_revision",
        "effective_request_options",
        "model_status",
        "latency_ms",
    }
    assert body["ok"] is True
    assert body["registry_revision"] == 2
    assert body["model_ref"] == MODEL_REF
    assert body["adapter_spec_revision"] == 1
    # The redacted build_request output matches the section 9.4 example shape.
    assert body["effective_request_options"] == {
        "timeout": 120,
        "max_retries": 2,
        "max_tokens": 32768,
        "temperature": 0.7,
        "top_p": 0.95,
    }
    assert isinstance(body["latency_ms"], int)
    assert body["latency_ms"] >= 0
    # The model was saved disabled: verified, not yet selectable.
    assert body["model_status"]["state"] == "verified"
    assert body["model_status"]["selectable"] is False
    assert body["model_status"]["verification"]["status"] == "passed"
    assert body["model_status"]["effective_capabilities"] == {
        "tools": True,
        "vision": True,
        "structured_output": True,
    }
    # The response never carries the secret (section 5.1).
    assert SECRET not in response.text

    # The five-tuple record was upserted (section 4.3).
    registry = saved_store.load_registry()
    provider = registry.find_provider("test-provider")
    model = provider.find_model("test-model")
    records = saved_store.list_model_verifications()
    assert len(records) == 1
    record = records[0]
    assert record["provider_id"] == "test-provider"
    assert record["model_key"] == "test-model"
    assert record["configuration_hash"] == configuration_hash(provider, model)
    assert record["credential_revision"] == 1
    assert record["adapter_spec_revision"] == 1
    assert record["result"] == "passed"
    assert record["verified_capabilities"] == {
        "tools": True,
        "vision": True,
        "structured_output": True,
    }
    assert record["error_code"] is None

    # The provider saw the minimal chat call plus one request per declared
    # capability probe; every request carried the resolved credential.
    requests = openai_server.requests
    assert len(requests) == 4
    assert all(request["path"] == "/v1/chat/completions" for request in requests)
    assert all(request["authorization"] == f"Bearer {SECRET}" for request in requests)
    assert any("tools" in request["body"] for request in requests)
    assert any("response_format" in request["body"] for request in requests)
    assert any("image_url" in json.dumps(request["body"]) for request in requests)


def test_enabled_model_reports_enabled_status(client, store, openai_server):
    port = openai_server.server_address[1]
    provider = _provider_payload(port)
    provider["models"][0]["enabled"] = True
    _save(store, port, models=provider["models"])
    response = _post(client)
    assert response.status_code == 200
    body = response.json()
    assert body["ok"] is True
    assert body["model_status"]["state"] == "enabled"
    assert body["model_status"]["selectable"] is True


def test_tools_probe_failure_marks_capability_false(
    test_client, saved_store, openai_server
):
    openai_server.mode = "fail_tools"
    response = _post(test_client)
    assert response.status_code == 200
    body = response.json()
    assert body["ok"] is True
    assert body["model_status"]["effective_capabilities"]["tools"] is False
    assert body["model_status"]["effective_capabilities"]["vision"] is True

    record = saved_store.list_model_verifications()[0]
    assert record["result"] == "passed"
    assert record["verified_capabilities"] == {
        "tools": False,
        "vision": True,
        "structured_output": True,
    }


# --- precondition and provider errors -------------------------------------------


def test_unsaved_model_ref_is_404(test_client, saved_store):
    response = _post(
        test_client,
        model_ref={"provider_id": "test-provider", "model_key": "free-draft"},
    )
    assert response.status_code == 404
    assert response.json()["code"] == "MODEL_NOT_FOUND"

    response = _post(
        test_client, model_ref={"provider_id": "draft", "model_key": "test-model"}
    )
    assert response.status_code == 404
    assert response.json()["code"] == "MODEL_NOT_FOUND"
    assert saved_store.list_model_verifications() == []


def test_missing_credential_is_422(client, store, openai_server):
    _save(store, openai_server.server_address[1], credential=False)
    response = _post(client)
    assert response.status_code == 422
    assert response.json()["code"] == "CREDENTIAL_NOT_CONFIGURED"
    assert store.list_model_verifications() == []


def test_credential_rejected_records_failed_result(
    test_client, saved_store, openai_server
):
    openai_server.mode = "reject"
    response = _post(test_client)
    assert response.status_code == 422
    body = response.json()
    assert body["code"] == "CREDENTIAL_REJECTED"
    assert SECRET not in response.text

    record = saved_store.list_model_verifications()[0]
    assert record["result"] == "failed"
    assert record["error_code"] == "CREDENTIAL_REJECTED"
    assert record["verified_capabilities"] == {
        "tools": False,
        "vision": False,
        "structured_output": False,
    }


def test_provider_404_is_model_not_found(test_client, saved_store, openai_server):
    openai_server.mode = "missing_model"
    response = _post(test_client)
    # The section 9.5 table pins MODEL_NOT_FOUND to 404 for every source.
    assert response.status_code == 404
    assert response.json()["code"] == "MODEL_NOT_FOUND"
    record = saved_store.list_model_verifications()[0]
    assert record["result"] == "failed"
    assert record["error_code"] == "MODEL_NOT_FOUND"


def test_connection_refused_is_provider_unreachable(store, closed_port):
    # The policy must register the unreachable port; the failure must come
    # from the connection attempt, not the endpoint allowlist.
    _save(store, closed_port)
    client = _client_for(_services_for(store, closed_port))
    response = _post(client)
    assert response.status_code == 422
    assert response.json()["code"] == "PROVIDER_UNREACHABLE"
    record = store.list_model_verifications()[0]
    assert record["result"] == "failed"
    assert record["error_code"] == "PROVIDER_UNREACHABLE"


def test_unregistered_local_endpoint_is_rejected(client, store):
    # Saved directly through the store (no save-time policy hook), so the
    # test endpoint must re-validate the base URL before any I/O.
    port = 9  # unregistered loopback endpoint
    store.save_registry(
        expected_revision=1,
        registry=RegistryV4.model_validate(_registry_payload(port)),
        credential_writes=[
            CredentialWrite(credential_id="test-key", secret_value=SECRET)
        ],
    )
    response = _post(client)
    assert response.status_code == 422
    assert response.json()["code"] == "ENDPOINT_NOT_ALLOWED"
    assert store.list_model_verifications() == []


# --- verification record transaction --------------------------------------------


def test_concurrent_registry_change_is_409_and_writes_nothing(
    test_client, saved_store, services, monkeypatch
):
    tester = services.provider_tester
    original_execute = tester._execute

    def mutate_then_execute(*args, **kwargs):
        # A concurrent admin save between resolution and the record write.
        registry = saved_store.load_registry()
        payload = registry.model_dump(mode="json")
        payload["providers"][0]["models"][0]["name"] = "Renamed Mid-Test"
        saved_store.save_registry(
            expected_revision=registry.revision,
            registry=RegistryV4.model_validate(payload),
        )
        return original_execute(*args, **kwargs)

    monkeypatch.setattr(tester, "_execute", mutate_then_execute)

    response = _post(test_client)
    assert response.status_code == 409
    assert response.json()["code"] == "MODEL_CONFIGURATION_CHANGED"
    # The stale test result must not be recorded for the new configuration.
    assert saved_store.list_model_verifications() == []


def test_stale_expected_revision_is_409(test_client, saved_store):
    response = _post(test_client, expected_revision=99)
    assert response.status_code == 409
    assert response.json()["code"] == "MODEL_CONFIGURATION_CHANGED"
    assert saved_store.list_model_verifications() == []


# --- authentication ---------------------------------------------------------------


def test_requires_test_scope(test_client, saved_store):
    response = _post(test_client, headers=_headers(["model_config:write"]))
    assert response.status_code == 403
    assert response.json()["code"] == "FORBIDDEN"


def test_requires_authentication(test_client, saved_store):
    response = test_client.post(
        "/api/model-registry/test",
        json={"expected_registry_revision": 2, "model_ref": MODEL_REF},
    )
    assert response.status_code == 401