156 lines
5.1 KiB
Python
156 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
import cloud.internal_api.api as internal_api
|
|
from cloud.control_config import CloudControlConfig
|
|
from runtime.tool_calling_client import ToolCallDecision
|
|
from cloud_api.app import create_app
|
|
|
|
|
|
class _FakePlannerClient:
|
|
def __init__(self) -> None:
|
|
self.calls = 0
|
|
|
|
def decide(self, **_kwargs) -> ToolCallDecision:
|
|
self.calls += 1
|
|
return ToolCallDecision(tool_name="tap", arguments={"x": 1, "y": 2})
|
|
|
|
|
|
def _login_admin(client: TestClient) -> dict[str, str]:
|
|
client.app.state.cloud_services.user_auth_service.create_user(
|
|
username="admin",
|
|
display_name="Administrator",
|
|
role="admin",
|
|
password="correct-horse-battery-staple",
|
|
must_change_password=False,
|
|
)
|
|
assert (
|
|
client.post(
|
|
"/v1/auth/login",
|
|
json={"username": "admin", "password": "correct-horse-battery-staple"},
|
|
).status_code
|
|
== 200
|
|
)
|
|
csrf = client.cookies.get("amcp_csrf")
|
|
assert csrf is not None
|
|
return {"X-CSRF-Token": csrf}
|
|
|
|
|
|
def _create_profile(
|
|
client: TestClient, headers: dict[str, str], *, name: str, model: str
|
|
) -> dict:
|
|
response = client.post(
|
|
"/v1/planner/providers",
|
|
headers=headers,
|
|
json={
|
|
"name": name,
|
|
"provider_type": "openai-compatible",
|
|
"model": model,
|
|
"base_url": "https://compat.example/v1",
|
|
"timeout_seconds": 30,
|
|
"api_key": f"key-for-{name}",
|
|
},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
return response.json()
|
|
|
|
|
|
def _enroll_host(client: TestClient) -> tuple[str, dict[str, str]]:
|
|
enrollment = client.post(
|
|
"/internal/v1/enrollments",
|
|
json={"agent_instance_id": "agent-a", "host_token": "host-token-" + ("a" * 40)},
|
|
)
|
|
assert enrollment.status_code == 201, enrollment.text
|
|
return enrollment.json()["host_id"], {
|
|
"Authorization": "Bearer host-token-" + ("a" * 40)
|
|
}
|
|
|
|
|
|
def _decision_payload(host_id: str) -> dict:
|
|
return {
|
|
"host_id": host_id,
|
|
"system_prompt": "system",
|
|
"user_prompt": "user",
|
|
"tools": [{"name": "tap", "description": "tap", "parameters": {}}],
|
|
"timeout_seconds": 10,
|
|
}
|
|
|
|
|
|
def test_planner_uses_the_newly_activated_database_profile(monkeypatch) -> None:
|
|
from cryptography.fernet import Fernet
|
|
|
|
monkeypatch.setenv(
|
|
"CLOUD_LLM_PROVIDER_ENCRYPTION_KEY", Fernet.generate_key().decode()
|
|
)
|
|
resolved_profiles = []
|
|
fake = _FakePlannerClient()
|
|
|
|
def build(resolved):
|
|
resolved_profiles.append(resolved)
|
|
return fake
|
|
|
|
monkeypatch.setattr(internal_api, "build_cloud_planner_client", build)
|
|
app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
|
|
with TestClient(app) as client:
|
|
admin_headers = _login_admin(client)
|
|
first = _create_profile(
|
|
client, admin_headers, name="First", model="first-model"
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/v1/planner/providers/{first['id']}/activate",
|
|
headers=admin_headers,
|
|
json={"expected_settings_revision": 0},
|
|
).status_code
|
|
== 200
|
|
)
|
|
host_id, host_headers = _enroll_host(client)
|
|
|
|
first_decision = client.post(
|
|
f"/internal/v1/hosts/{host_id}/planner/decide",
|
|
headers=host_headers,
|
|
json=_decision_payload(host_id),
|
|
)
|
|
assert first_decision.status_code == 200, first_decision.text
|
|
assert resolved_profiles[-1].profile.model == "first-model"
|
|
assert resolved_profiles[-1].api_key == "key-for-First"
|
|
|
|
second = _create_profile(
|
|
client, admin_headers, name="Second", model="second-model"
|
|
)
|
|
settings = client.get("/v1/planner/providers").json()["settings"]
|
|
activated = client.post(
|
|
f"/v1/planner/providers/{second['id']}/activate",
|
|
headers=admin_headers,
|
|
json={"expected_settings_revision": settings["revision"]},
|
|
)
|
|
assert activated.status_code == 200, activated.text
|
|
|
|
second_decision = client.post(
|
|
f"/internal/v1/hosts/{host_id}/planner/decide",
|
|
headers=host_headers,
|
|
json=_decision_payload(host_id),
|
|
)
|
|
assert second_decision.status_code == 200, second_decision.text
|
|
assert resolved_profiles[-1].profile.model == "second-model"
|
|
assert fake.calls == 2
|
|
|
|
|
|
def test_planner_fails_closed_without_an_active_database_profile(monkeypatch) -> None:
|
|
monkeypatch.delenv("CLOUD_LLM_PROVIDER_ENCRYPTION_KEY", raising=False)
|
|
app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
|
|
with TestClient(app) as client:
|
|
host_id, host_headers = _enroll_host(client)
|
|
response = client.post(
|
|
f"/internal/v1/hosts/{host_id}/planner/decide",
|
|
headers=host_headers,
|
|
json=_decision_payload(host_id),
|
|
)
|
|
|
|
assert response.status_code == 502
|
|
assert response.json() == {
|
|
"code": "planner_unavailable",
|
|
"detail": "no active database Provider profile",
|
|
}
|