333 lines
16 KiB
Python
333 lines
16 KiB
Python
"""Modelle und Zugänge: Verwaltung, Auflösung je Team, Antworten über den Adapter (→ Konzept 8.1)."""
|
|
|
|
import json
|
|
import re
|
|
import uuid
|
|
|
|
import httpx
|
|
import pytest
|
|
from sqlalchemy import select, update
|
|
|
|
from app.ai import service as ai
|
|
from app.ai.models import Model, ProviderKey
|
|
from app.ai.provider import GuardrailBlocked, ProviderError, ResponsesProvider
|
|
from app.chat.models import Message
|
|
from app.core.crypto import Vault
|
|
from app.tasks import states
|
|
from app.tasks.models import TaskRun
|
|
from app.tasks.worker import Worker
|
|
from helpers import connect_repo, create_workspace
|
|
|
|
|
|
def make_admin(db, user_id: int) -> None:
|
|
from app.identity.models import User
|
|
with db.session() as session:
|
|
session.get(User, user_id).is_admin = True
|
|
|
|
|
|
@pytest.fixture
|
|
def admin(browser, people, db):
|
|
make_admin(db, people["anna"])
|
|
return browser("anna")
|
|
|
|
|
|
@pytest.fixture
|
|
def chat(browser, people, git_repo):
|
|
anna = browser("anna")
|
|
workspace_id = create_workspace(anna, people["team_demo"])
|
|
connect_repo(anna, workspace_id, git_repo)
|
|
location = anna.post(f"/workspaces/{workspace_id}/chats").headers["location"]
|
|
return anna, workspace_id, int(location.rsplit("/", 1)[1])
|
|
|
|
|
|
def ask(b, chat_id: int, text: str = "Hallo") -> uuid.UUID:
|
|
response = b.post(f"/chats/{chat_id}/messages", {"content": text}, headers={"HX-Request": "true"})
|
|
assert response.status_code == 200, response.text
|
|
return uuid.UUID(re.search(r'/runs/([0-9a-f-]+)/events', response.text).group(1))
|
|
|
|
|
|
def run_of(db, run_id) -> TaskRun:
|
|
with db.session() as session:
|
|
return session.get(TaskRun, run_id)
|
|
|
|
|
|
# --- Verwaltung ---
|
|
|
|
def test_nur_plattform_admin(browser, people):
|
|
ben = browser("ben")
|
|
assert ben.get("/admin/models").status_code == 403
|
|
assert ben.post("/admin/providers", {"name": "X", "base_url": "https://x.example"}).status_code == 403
|
|
assert ben.post("/admin/keys", {"provider_id": people["provider"], "value": "sk-1234567890"}).status_code == 403
|
|
assert "Modelle und Zugänge" not in ben.get("/workspaces").text
|
|
|
|
|
|
def test_key_wird_verschluesselt_und_nie_angezeigt(admin, people, db, settings):
|
|
response = admin.post("/admin/keys", {"provider_id": people["provider"], "team_id": people["team_demo"],
|
|
"value": "sk-geheim-team-demo-9876"})
|
|
assert response.status_code == 200 and "nicht wieder angezeigt" in response.text
|
|
page = admin.get("/admin/models").text
|
|
assert "sk-geheim" not in page and "…9876" in page
|
|
with db.session() as session:
|
|
key = session.scalar(select(ProviderKey).where(ProviderKey.team_id == people["team_demo"]))
|
|
assert "sk-geheim" not in key.secret
|
|
assert Vault(settings.secret_key).decrypt(key.secret) == "sk-geheim-team-demo-9876"
|
|
|
|
|
|
def test_key_ersetzen_statt_doppelt(admin, people, db):
|
|
for value in ("sk-erster-key-1111", "sk-zweiter-key-2222"):
|
|
admin.post("/admin/keys", {"provider_id": people["provider"], "value": value})
|
|
with db.session() as session:
|
|
hints = list(session.scalars(select(ProviderKey.hint).where(ProviderKey.team_id.is_(None))))
|
|
assert hints == ["2222"]
|
|
|
|
|
|
def test_ohne_master_key_kein_speichern(settings, db, people, browser):
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import create_app
|
|
from helpers import Browser
|
|
make_admin(db, people["anna"])
|
|
with TestClient(create_app(settings.model_copy(update={"secret_key": ""}))) as client:
|
|
b = Browser(client).login("anna")
|
|
assert "AIW_SECRET_KEY" in b.get("/admin/models").text
|
|
response = b.post("/admin/keys", {"provider_id": people["provider"], "value": "sk-1234567890"})
|
|
assert response.status_code == 400
|
|
|
|
|
|
@pytest.mark.parametrize("url", ["http://api.example.com/v1", "https://user:pw@api.example.com", "ftp://x", ""])
|
|
def test_unzulaessige_basis_url(admin, url):
|
|
response = admin.post("/admin/providers", {"name": "X", "base_url": url})
|
|
assert response.status_code in (400, 422)
|
|
|
|
|
|
def test_localhost_per_http_erlaubt():
|
|
assert ai.check_base_url("http://127.0.0.1:4000/v1/") == "http://127.0.0.1:4000/v1"
|
|
|
|
|
|
def test_modell_anlegen_und_standard_wechseln(admin, people, db):
|
|
response = admin.post("/admin/models", {"provider_id": people["provider"], "name": "gpt-6-luna-mini",
|
|
"label": "Luna Mini", "active": "1", "supports_tools": "1"})
|
|
assert response.status_code == 200, response.text
|
|
with db.session() as session:
|
|
mini = session.scalar(select(Model).where(Model.name == "gpt-6-luna-mini"))
|
|
assert not mini.is_default
|
|
admin.post(f"/admin/models/{mini.id}/default")
|
|
with db.session() as session:
|
|
assert session.scalar(select(Model.name).where(Model.is_default)) == "gpt-6-luna-mini"
|
|
|
|
|
|
def test_standardmodell_nicht_deaktivierbar(admin, people, db):
|
|
with db.session() as session:
|
|
luna = session.scalar(select(Model).where(Model.is_default))
|
|
response = admin.post("/admin/models", {"model_id": luna.id, "provider_id": people["provider"],
|
|
"name": "gpt-6-luna"})
|
|
assert response.status_code == 400 and "Standardmodell" in response.text
|
|
|
|
|
|
def test_verbindung_testen(admin, people, monkeypatch):
|
|
monkeypatch.setattr(ResponsesProvider, "list_models", lambda self, target: ["gpt-6-luna", "andere"])
|
|
response = admin.post(f"/admin/providers/{people['provider']}/test", headers={"HX-Request": "true"})
|
|
assert "Verbindung ok, 2 Modelle" in response.text and "gpt-6-luna" in response.text
|
|
|
|
|
|
# --- Modellwahl im Workspace ---
|
|
|
|
def test_workspace_modell_waehlen(chat, people, db, fake_provider, settings, browser):
|
|
anna, workspace_id, chat_id = chat
|
|
with db.session() as session:
|
|
mini = Model(provider_id=people["provider"], name="gpt-6-luna-mini", label="Luna Mini")
|
|
session.add(mini)
|
|
assert anna.post(f"/workspaces/{workspace_id}/model", {"model_id": mini.id}).status_code == 303
|
|
assert "Luna Mini" in anna.get(f"/workspaces/{workspace_id}").text
|
|
ask(anna, chat_id)
|
|
Worker(db, settings).run_once()
|
|
assert fake_provider.calls[-1]["target"].model == "gpt-6-luna-mini"
|
|
# Deaktiviert der Plattform-Admin das Modell, gilt wieder der Standard
|
|
with db.session() as session:
|
|
session.execute(update(Model).where(Model.id == mini.id).values(active=False))
|
|
ask(anna, chat_id, "Nochmal")
|
|
Worker(db, settings).run_once()
|
|
assert fake_provider.calls[-1]["target"].model == "gpt-6-luna"
|
|
|
|
|
|
def test_workspace_modell_nur_admin_und_nur_aktiv(chat, people, db, browser):
|
|
anna, workspace_id, _ = chat
|
|
with db.session() as session:
|
|
off = Model(provider_id=people["provider"], name="alt", label="Alt", active=False)
|
|
session.add(off)
|
|
assert anna.post(f"/workspaces/{workspace_id}/model", {"model_id": off.id}).status_code == 400
|
|
anna.post(f"/workspaces/{workspace_id}/members", {"username": "ben", "role": "member"})
|
|
assert browser("ben").post(f"/workspaces/{workspace_id}/model", {"model_id": ""}).status_code == 403
|
|
|
|
|
|
# --- Auflösung und Antworten ---
|
|
|
|
def test_team_key_hat_vorrang(chat, people, db, settings, fake_provider):
|
|
anna, _, chat_id = chat
|
|
with db.session() as session:
|
|
session.add(ProviderKey(provider_id=people["provider"], team_id=people["team_demo"],
|
|
secret=Vault(settings.secret_key).encrypt("sk-team-demo"), hint="demo",
|
|
updated_by=people["anna"]))
|
|
ask(anna, chat_id)
|
|
Worker(db, settings).run_once()
|
|
assert fake_provider.calls[-1]["target"].api_key == "sk-team-demo"
|
|
|
|
|
|
def test_antwort_mit_verlauf_und_reasoning(chat, db, settings, fake_provider):
|
|
anna, _, chat_id = chat
|
|
first = ask(anna, chat_id, "Erste Frage")
|
|
Worker(db, settings).run_once()
|
|
assert run_of(db, first).status == states.COMPLETED
|
|
with db.session() as session:
|
|
reply = session.scalar(select(Message).where(Message.role == "assistant"))
|
|
assert reply.model == "gpt-6-luna" and reply.usage["total_tokens"] == 7 and reply.usage["context_window"] == 128000
|
|
ask(anna, chat_id, "Zweite Frage")
|
|
Worker(db, settings).run_once()
|
|
sent = fake_provider.calls[-1]["input"]
|
|
assert sent[0] == {"role": "user", "content": "Erste Frage"}
|
|
# Gespeicherte Elemente gehen unverändert zurück, inklusive verschlüsseltem Reasoning
|
|
assert sent[1]["type"] == "reasoning" and sent[1]["encrypted_content"] == "verschluesselt"
|
|
assert sent[-1] == {"role": "user", "content": "Zweite Frage"}
|
|
assert "Projekt X" in fake_provider.calls[-1]["instructions"]
|
|
|
|
|
|
def test_kein_key_endet_sofort_mit_meldung(chat, db, settings):
|
|
anna, _, chat_id = chat
|
|
with db.session() as session:
|
|
session.execute(ProviderKey.__table__.delete())
|
|
run_id = ask(anna, chat_id)
|
|
Worker(db, settings).run_once()
|
|
run = run_of(db, run_id)
|
|
assert run.status == states.FAILED and run.attempts == 1
|
|
assert "kein API-Key" in anna.get(f"/runs/{run_id}/answer").text
|
|
|
|
|
|
def test_guardrail_blockiert(chat, db, settings, fake_provider):
|
|
anna, _, chat_id = chat
|
|
fake_provider.error = GuardrailBlocked("guardrail_prompt_injection", "Prompt Injection erkannt")
|
|
run_id = ask(anna, chat_id)
|
|
Worker(db, settings).run_once()
|
|
assert run_of(db, run_id).status == states.BLOCKED_BY_GUARDRAIL
|
|
assert "guardrail_prompt_injection" in anna.get(f"/runs/{run_id}/answer").text
|
|
|
|
|
|
def test_voruebergehender_fehler_wird_wiederholt(chat, db, settings, fake_provider):
|
|
anna, _, chat_id = chat
|
|
fake_provider.error = ProviderError("503", retryable=True, user_message="ausgelastet")
|
|
run_id = ask(anna, chat_id)
|
|
Worker(db, settings).run_once()
|
|
assert run_of(db, run_id).status == states.QUEUED
|
|
|
|
|
|
def test_abgelehnter_key_ohne_wiederholung(chat, db, settings, fake_provider):
|
|
anna, _, chat_id = chat
|
|
fake_provider.error = ProviderError("401", retryable=False, user_message="Der API-Key wurde abgelehnt.")
|
|
run_id = ask(anna, chat_id)
|
|
Worker(db, settings).run_once()
|
|
run = run_of(db, run_id)
|
|
assert run.status == states.FAILED and run.attempts == 1 and "401" in run.error
|
|
|
|
|
|
# --- Adapter gegen die echte SDK-Schicht, Netz durch MockTransport ersetzt ---
|
|
|
|
TARGET = ai.Target(base_url="https://api.openai.com/v1", api_key="sk-test", model="gpt-6-luna",
|
|
model_label="Luna", supports_reasoning=True)
|
|
|
|
|
|
def sse(*events: dict) -> bytes:
|
|
return "".join(f"event: {e['type']}\ndata: {json.dumps(e)}\n\n" for e in events).encode()
|
|
|
|
|
|
def response_object(status: str = "completed", text: str = "Hallo Welt", **extra) -> dict:
|
|
return {"id": "resp_1", "object": "response", "created_at": 0, "model": "gpt-6-luna", "status": status,
|
|
"parallel_tool_calls": True, "tool_choice": "auto", "tools": [], "error": None,
|
|
"incomplete_details": None, "instructions": None, "metadata": {},
|
|
"output": [{"type": "reasoning", "id": "rs_1", "summary": [], "encrypted_content": "enc"},
|
|
{"type": "message", "id": "msg_1", "role": "assistant", "status": "completed",
|
|
"content": [{"type": "output_text", "text": text, "annotations": []}]}],
|
|
"usage": {"input_tokens": 5, "output_tokens": 2, "total_tokens": 7,
|
|
"input_tokens_details": {"cached_tokens": 0},
|
|
"output_tokens_details": {"reasoning_tokens": 0}}, **extra}
|
|
|
|
|
|
def provider_with(handler) -> tuple[ResponsesProvider, list[httpx.Request]]:
|
|
seen: list[httpx.Request] = []
|
|
|
|
def record(request: httpx.Request) -> httpx.Response:
|
|
seen.append(request)
|
|
return handler(request)
|
|
|
|
return ResponsesProvider(http_client=httpx.Client(transport=httpx.MockTransport(record))), seen
|
|
|
|
|
|
def test_adapter_streamt_und_sendet_zustandslos():
|
|
body = sse({"type": "response.created", "sequence_number": 0, "response": response_object("in_progress")},
|
|
{"type": "response.output_text.delta", "sequence_number": 1, "item_id": "msg_1", "output_index": 1,
|
|
"content_index": 0, "delta": "Hallo ", "logprobs": []},
|
|
{"type": "response.output_text.delta", "sequence_number": 2, "item_id": "msg_1", "output_index": 1,
|
|
"content_index": 0, "delta": "Welt", "logprobs": []},
|
|
{"type": "response.completed", "sequence_number": 3, "response": response_object()})
|
|
provider, seen = provider_with(lambda r: httpx.Response(200, content=body,
|
|
headers={"content-type": "text/event-stream"}))
|
|
deltas: list[str] = []
|
|
result = provider.respond(TARGET, instructions="Sei knapp.", input=[{"role": "user", "content": "Hi"}],
|
|
on_text=deltas.append)
|
|
assert deltas == ["Hallo ", "Welt"] and result.text == "Hallo Welt"
|
|
assert result.items[0]["encrypted_content"] == "enc" and result.usage["total_tokens"] == 7
|
|
sent = json.loads(seen[0].content)
|
|
assert seen[0].url.path == "/v1/responses" and seen[0].headers["authorization"] == "Bearer sk-test"
|
|
assert sent["store"] is False and sent["stream"] is True
|
|
assert sent["include"] == ["reasoning.encrypted_content"] and "tools" not in sent
|
|
|
|
|
|
def test_adapter_guardrail_als_http_fehler():
|
|
error = {"error": {"code": "guardrail_violation", "type": "guardrail", "message": "Blockiert"}}
|
|
provider, _ = provider_with(lambda r: httpx.Response(400, json=error))
|
|
with pytest.raises(GuardrailBlocked) as caught:
|
|
provider.respond(TARGET, instructions="", input=[], on_text=lambda _: None)
|
|
assert caught.value.code == "guardrail_violation"
|
|
|
|
|
|
@pytest.mark.parametrize("status, retryable", [(401, False), (404, False), (429, True), (503, True), (400, False)])
|
|
def test_adapter_http_fehler(status, retryable):
|
|
provider, _ = provider_with(lambda r: httpx.Response(status, json={"error": {"message": "x", "code": "c"}}))
|
|
with pytest.raises(ProviderError) as caught:
|
|
provider.respond(TARGET, instructions="", input=[], on_text=lambda _: None)
|
|
assert caught.value.retryable is retryable
|
|
|
|
|
|
def test_adapter_failed_im_stream():
|
|
failed = response_object("failed", error={"code": "server_error", "message": "kaputt"})
|
|
body = sse({"type": "response.failed", "sequence_number": 0, "response": failed})
|
|
provider, _ = provider_with(lambda r: httpx.Response(200, content=body,
|
|
headers={"content-type": "text/event-stream"}))
|
|
with pytest.raises(ProviderError) as caught:
|
|
provider.respond(TARGET, instructions="", input=[], on_text=lambda _: None)
|
|
assert caught.value.retryable
|
|
|
|
|
|
def test_adapter_netzfehler_wird_wiederholt():
|
|
def broken(request):
|
|
raise httpx.ConnectError("weg")
|
|
provider, _ = provider_with(broken)
|
|
with pytest.raises(ProviderError) as caught:
|
|
provider.respond(TARGET, instructions="", input=[], on_text=lambda _: None)
|
|
assert caught.value.retryable
|
|
|
|
|
|
def test_adapter_tool_aufruf_und_tools_im_request():
|
|
call = {"type": "function_call", "id": "fc_1", "call_id": "call_abc", "name": "git_search",
|
|
"arguments": "{\"pattern\": \"TIMEOUT\"}", "status": "completed"}
|
|
done = response_object(output=[call])
|
|
body = sse({"type": "response.output_item.done", "sequence_number": 0, "output_index": 0, "item": call},
|
|
{"type": "response.completed", "sequence_number": 1, "response": done})
|
|
provider, seen = provider_with(lambda r: httpx.Response(200, content=body,
|
|
headers={"content-type": "text/event-stream"}))
|
|
tools = [{"type": "function", "name": "git_search", "description": "Suche", "strict": False,
|
|
"parameters": {"type": "object", "properties": {"pattern": {"type": "string"}}}}]
|
|
history = [{"role": "user", "content": "Hi"},
|
|
{"type": "function_call_output", "call_id": "call_0", "output": "alt"}]
|
|
result = provider.respond(TARGET, instructions="", input=history, on_text=lambda _: None, tools=tools)
|
|
assert result.calls == [call] and result.text == ""
|
|
sent = json.loads(seen[0].content)
|
|
assert sent["tools"] == tools and sent["tool_choice"] == "auto" and sent["input"] == history
|