237 lines
13 KiB
Python
237 lines
13 KiB
Python
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlmodel import Session, SQLModel, create_engine
|
|
|
|
|
|
@pytest.fixture
|
|
def ai_db(tmp_path, monkeypatch):
|
|
from backend.app.db import session as database
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'managed-ai.db'}", connect_args={"check_same_thread": False})
|
|
monkeypatch.setattr(database, "engine", engine)
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as db:
|
|
yield db
|
|
|
|
|
|
def _configure(db):
|
|
from backend.app.services import managed_ai
|
|
managed_ai.save_config(db, {"enabled": True, "gatewayUrl": "http://198.18.10.150:8000", "capability": "assistant.ops@1", "credential": "mf_" + "a" * 48})
|
|
|
|
|
|
def _response():
|
|
return {"content": "Controleer de beschikbare evidence [1].", "request_id": "request-1", "deployment_id": "deployment-1", "model_key": "gpt-oss-20b", "model_revision": "immutable-revision", "artifact_digest": "sha256:" + "b" * 64, "latency_ms": 100, "usage": {"input_tokens": 50, "output_tokens": 9}}
|
|
|
|
|
|
def test_invocation_uses_exact_capability_and_redacts_known_secrets(ai_db, monkeypatch):
|
|
from backend.app.services import managed_ai
|
|
from backend.app.services.secret_store import set_secret
|
|
_configure(ai_db)
|
|
set_secret(ai_db, "UNRELATED_API_KEY", "private-value-to-omit")
|
|
requests = []
|
|
def handle(request):
|
|
requests.append(request)
|
|
assert request.url.path == "/api/v1/capabilities/assistant.ops@1/invoke"
|
|
body = json.loads(request.content)
|
|
assert "private-value-to-omit" not in request.content.decode()
|
|
assert body["max_output_tokens"] <= 1024
|
|
assert "model" not in body
|
|
assert request.headers["Authorization"] == "Bearer mf_" + "a" * 48
|
|
return httpx.Response(200, json=_response())
|
|
monkeypatch.setattr(managed_ai, "_transport", httpx.MockTransport(handle))
|
|
result = managed_ai.analyze(ai_db, "Wat betekent private-value-to-omit?", [{"rank": 1, "excerpt": "token=private-value-to-omit"}])
|
|
assert len(requests) == 1
|
|
assert result["model"] == "gpt-oss-20b"
|
|
assert result["content"].startswith("Controleer")
|
|
assert managed_ai.get_config(ai_db)["credentialConfigured"] is True
|
|
assert "credential" not in managed_ai.get_config(ai_db)
|
|
|
|
|
|
def test_no_retry_or_fallback_on_gateway_error(ai_db, monkeypatch):
|
|
from backend.app.services import managed_ai
|
|
_configure(ai_db)
|
|
requests = []
|
|
def handle(request):
|
|
requests.append(request)
|
|
return httpx.Response(503, json={"code": "CAPACITY_CONSTRAINED", "message": "Bearer secret-error-content", "correlation_id": "corr-1"})
|
|
monkeypatch.setattr(managed_ai, "_transport", httpx.MockTransport(handle))
|
|
with pytest.raises(managed_ai.ManagedAiError) as error:
|
|
managed_ai.analyze(ai_db, "Vraag", [])
|
|
assert len(requests) == 1
|
|
assert error.value.code == "CAPACITY_CONSTRAINED"
|
|
assert "secret-error-content" not in str(error.value)
|
|
|
|
|
|
@pytest.mark.parametrize("body", [{}, {"content": "", "model_key": "x"}, {"content": "x", "deployment_id": "d"}])
|
|
def test_invalid_response_never_claims_inference_success(ai_db, monkeypatch, body):
|
|
from backend.app.services import managed_ai
|
|
_configure(ai_db)
|
|
monkeypatch.setattr(managed_ai, "_transport", httpx.MockTransport(lambda _: httpx.Response(200, json=body)))
|
|
with pytest.raises(managed_ai.ManagedAiError, match="INVALID_RESPONSE"):
|
|
managed_ai.analyze(ai_db, "Vraag", [])
|
|
|
|
|
|
def test_missing_config_is_explicit_and_does_not_call_transport(ai_db, monkeypatch):
|
|
from backend.app.services import managed_ai
|
|
monkeypatch.setattr(managed_ai, "_transport", httpx.MockTransport(lambda _: pytest.fail("network request without config")))
|
|
assert managed_ai.status(ai_db)["state"] == "not_configured"
|
|
with pytest.raises(managed_ai.ManagedAiError, match="NOT_CONFIGURED"):
|
|
managed_ai.analyze(ai_db, "Vraag", [])
|
|
|
|
|
|
@pytest.mark.parametrize("url", ["http://169.254.169.254", "https://user:pass@example.com", "https://example.com/api", "https://example.com?token=a", "http://127.0.0.1"])
|
|
def test_admin_config_rejects_unsafe_gateway_origins(ai_db, url):
|
|
from backend.app.services import managed_ai
|
|
with pytest.raises(ValueError):
|
|
managed_ai.save_config(ai_db, {"gatewayUrl": url})
|
|
|
|
|
|
def test_capability_check_does_not_imply_model_ready(ai_db, monkeypatch):
|
|
from backend.app.services import managed_ai
|
|
_configure(ai_db)
|
|
monkeypatch.setattr(managed_ai, "_transport", httpx.MockTransport(lambda _: httpx.Response(200, json={"capability": "assistant.ops@1", "project_key": "appops"})))
|
|
assert managed_ai.status(ai_db)["state"] == "unverified"
|
|
|
|
|
|
def test_read_only_can_chat_but_cannot_change_configuration(ai_db, monkeypatch):
|
|
from backend.app.db.models import GatewayUser
|
|
from backend.app.main import app
|
|
from backend.app.services import managed_ai, user_access
|
|
readonly = GatewayUser(id="readonly-ai", provider="authentik", issuer="test", subject="readonly-ai", role="read_only")
|
|
monkeypatch.setattr(user_access, "current_user", lambda _: readonly)
|
|
monkeypatch.setattr(managed_ai, "analyze", lambda *_: {"content": "Beschikbare context [1].", "model": "gpt-oss-20b", "revision": "r1", "deploymentId": "d1", "requestId": "q1", "artifactDigest": "sha256:" + "b" * 64})
|
|
from backend.app.services import knowledge_service
|
|
monkeypatch.setattr(knowledge_service, "query", lambda *_args, **_kwargs: {"citations": [{"rank": 1, "title": "Context", "uri": "appops://state", "excerpt": "safe context"}], "confidence": "medium"})
|
|
client = TestClient(app)
|
|
result = client.post("/api/ai/chat", json={"content": "Wat is de status?"})
|
|
assert result.status_code == 200
|
|
assert result.json()["metadata"]["provider"] == "modelforge"
|
|
assert client.patch("/api/settings/ai", json={"enabled": True}).status_code == 403
|
|
history = client.get("/api/ai/history").json()
|
|
assert len(history) == 2
|
|
other = GatewayUser(id="other", provider="authentik", issuer="test", subject="other", role="read_only")
|
|
monkeypatch.setattr(user_access, "current_user", lambda _: other)
|
|
assert client.get("/api/ai/history").json() == []
|
|
|
|
|
|
def test_chat_failures_return_named_unavailable_not_fixed_answer(ai_db, monkeypatch):
|
|
from backend.app.main import app
|
|
from backend.app.services import knowledge_service, managed_ai
|
|
monkeypatch.setattr(knowledge_service, "query", lambda *_args, **_kwargs: {"citations": [], "confidence": "none"})
|
|
monkeypatch.setattr(managed_ai, "analyze", lambda *_: (_ for _ in ()).throw(managed_ai.ManagedAiError("NOT_CONFIGURED")))
|
|
response = TestClient(app).post("/api/ai/chat", json={"content": "Is alles gezond?"})
|
|
assert response.status_code == 503
|
|
assert response.json()["detail"]["code"] == "NOT_CONFIGURED"
|
|
|
|
|
|
def test_log_analysis_uses_redacted_input_and_real_result(ai_db, monkeypatch):
|
|
from backend.app.main import app
|
|
from backend.app.services import managed_ai
|
|
_configure(ai_db)
|
|
bodies = []
|
|
def handle(request):
|
|
bodies.append(json.loads(request.content))
|
|
assert "unsafe-password" not in request.content.decode()
|
|
result = _response()
|
|
result["content"] = "Er staat een ERROR in de meegegeven log."
|
|
return httpx.Response(200, json=result)
|
|
monkeypatch.setattr(managed_ai, "_transport", httpx.MockTransport(handle))
|
|
response = TestClient(app).post("/api/ai/summarize/logs", json={"logs": "ERROR connection failed password=unsafe-password"})
|
|
assert response.status_code == 200
|
|
assert response.json()["summary"] == "Er staat een ERROR in de meegegeven log."
|
|
assert response.json()["readOnly"] is True
|
|
assert len(bodies) == 1
|
|
|
|
|
|
def test_memory_crud_persists_and_redacts(ai_db):
|
|
from backend.app.main import app
|
|
client = TestClient(app)
|
|
response = client.post("/api/ai/memory", json={"key": "server-note", "value": "password=unsafe-password", "pinned": True})
|
|
assert response.status_code == 200
|
|
memory = response.json()["memory"]
|
|
assert memory["value"] == "password=[REDACTED]"
|
|
assert client.get("/api/ai/memory").json()[0]["id"] == memory["id"]
|
|
assert client.patch(f"/api/ai/memory/{memory['id']}", json={"key": "server-note", "value": "New note"}).status_code == 200
|
|
assert client.get("/api/ai/memory").json()[0]["value"] == "New note"
|
|
assert client.delete(f"/api/ai/memory/{memory['id']}").status_code == 200
|
|
assert client.get("/api/ai/memory").json() == []
|
|
|
|
|
|
def test_unknown_incident_does_not_claim_analysis(ai_db):
|
|
from backend.app.main import app
|
|
response = TestClient(app).post("/api/ai/analyze/incident/not-existing")
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_citations_evidence_and_history_do_not_expose_known_raw_secret(ai_db, monkeypatch):
|
|
from backend.app.main import app
|
|
from backend.app.services import knowledge_service, managed_ai
|
|
from backend.app.services.secret_store import set_secret
|
|
set_secret(ai_db, "UNRELATED_API_KEY", "raw-private-citation-value")
|
|
monkeypatch.setattr(knowledge_service, "query", lambda *_args, **_kwargs: {"citations": [{"rank": 1, "title": "raw-private-citation-value", "uri": "appops://context", "excerpt": "raw-private-citation-value"}], "confidence": "medium"})
|
|
monkeypatch.setattr(managed_ai, "analyze", lambda *_: {"content": "Beschikbare context [1].", "model": "gpt-oss-20b", "revision": "r1", "deploymentId": "d1", "requestId": "q1", "artifactDigest": "sha256:" + "b" * 64})
|
|
client = TestClient(app)
|
|
response = client.post("/api/ai/chat", json={"content": "Wat is de status?"})
|
|
assert response.status_code == 200
|
|
assert "raw-private-citation-value" not in response.text
|
|
assert "raw-private-citation-value" not in client.get("/api/ai/history").text
|
|
|
|
|
|
def test_validation_errors_do_not_echo_sensitive_inputs(ai_db):
|
|
from backend.app.main import app
|
|
value = "mf_" + "a" * 48 + " " * 14
|
|
response = TestClient(app).patch("/api/settings/ai", json={"credential": value})
|
|
assert response.status_code == 422
|
|
assert "mf_" not in response.text
|
|
response = TestClient(app).post("/api/ai/summarize/logs", json={"logs": "password=too-sensitive" * 1000})
|
|
assert response.status_code == 422
|
|
assert "too-sensitive" not in response.text
|
|
|
|
|
|
def test_json_escaping_cannot_bypass_known_secret_redaction(ai_db, monkeypatch):
|
|
from backend.app.services import managed_ai
|
|
from backend.app.services.secret_store import set_secret
|
|
_configure(ai_db)
|
|
private_value = 'complex"private\\value'
|
|
set_secret(ai_db, "UNRELATED_API_KEY", private_value)
|
|
def handle(request):
|
|
body = json.loads(request.content)
|
|
assert json.dumps(private_value, ensure_ascii=False)[1:-1] not in body['messages'][1]['content']
|
|
assert private_value not in body['messages'][1]['content']
|
|
return httpx.Response(200, json=_response())
|
|
monkeypatch.setattr(managed_ai, "_transport", httpx.MockTransport(handle))
|
|
managed_ai.analyze(ai_db, "Vraag", {'logs': 'Unlabeled: ' + private_value})
|
|
|
|
|
|
def test_advisory_activity_does_not_invalidate_security_sources(ai_db):
|
|
from backend.app.db.models import AppSetting
|
|
from backend.app.services import ai_security_reviewer, safety_center_service
|
|
before_review = ai_security_reviewer._source_signature(ai_db, 'all')
|
|
before_safety = safety_center_service._source_signature(ai_db)
|
|
ai_db.add(AppSetting(key='ai:chat:test-user', value_json={'messages': []}))
|
|
ai_db.add(AppSetting(key='ai:modelforge:last_inference', value_json={'model': 'gpt-oss-20b'}))
|
|
ai_db.commit()
|
|
assert ai_security_reviewer._source_signature(ai_db, 'all') == before_review
|
|
assert safety_center_service._source_signature(ai_db) == before_safety
|
|
ai_db.add(AppSetting(key='ai:modelforge', value_json={'enabled': True}))
|
|
ai_db.commit()
|
|
assert ai_security_reviewer._source_signature(ai_db, 'all') != before_review
|
|
assert safety_center_service._source_signature(ai_db) != before_safety
|
|
|
|
|
|
@pytest.mark.parametrize("text", ["evidence " * 2000, "😀漢字" * 3000], ids=["ascii", "unicode"])
|
|
def test_prompt_fits_qualified_runtime_in_utf8_bytes(ai_db, monkeypatch, text):
|
|
from backend.app.services import managed_ai
|
|
_configure(ai_db)
|
|
def handle(request):
|
|
payload = json.loads(request.content)
|
|
input_bytes = sum(len(message["content"].encode("utf-8")) for message in payload["messages"])
|
|
assert input_bytes + payload["max_output_tokens"] + 512 <= 8192
|
|
assert "afgekapt" in payload["messages"][1]["content"]
|
|
return httpx.Response(200, json=_response())
|
|
monkeypatch.setattr(managed_ai, "_transport", httpx.MockTransport(handle))
|
|
result = managed_ai.analyze(ai_db, "Wat betekent dit? " + "😀" * 2048, {"logs": text})
|
|
assert result["contextTruncated"] is True
|