150 lines
7.7 KiB
Python
150 lines
7.7 KiB
Python
"""Nutzung, Kosten, Runs und Audit-Log im UI (→ Konzept 13, Phase 5 Administration)."""
|
|
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from sqlalchemy import select, update
|
|
|
|
from app.ai.models import Model
|
|
from app.audit.models import AuditEvent
|
|
from app.tasks.worker import Worker
|
|
from app.usage import service as usage
|
|
from app.usage.models import UsageRecord
|
|
from helpers import connect_repo, create_workspace
|
|
|
|
USAGE = {"input_tokens": 1_000_000, "output_tokens": 200_000, "total_tokens": 1_200_000,
|
|
"input_tokens_details": {"cached_tokens": 400_000}, "output_tokens_details": {"reasoning_tokens": 50_000}}
|
|
|
|
|
|
def make_admin(db, user_id):
|
|
from app.identity.models import User
|
|
with db.session() as session:
|
|
session.get(User, user_id).is_admin = True
|
|
|
|
|
|
@pytest.fixture
|
|
def setup(browser, people, git_repo, db):
|
|
make_admin(db, people["anna"])
|
|
anna = browser("anna")
|
|
workspace_id = create_workspace(anna, people["team_demo"])
|
|
connect_repo(anna, workspace_id, git_repo)
|
|
anna.post(f"/workspaces/{workspace_id}/members", {"username": "ben", "role": "member"})
|
|
chat_id = int(anna.post(f"/workspaces/{workspace_id}/chats").headers["location"].rsplit("/", 1)[1])
|
|
return {"anna": anna, "ben": browser("ben"), "workspace_id": workspace_id, "chat_id": chat_id}
|
|
|
|
|
|
def ask(b, chat_id, text="Hallo"):
|
|
b.post(f"/chats/{chat_id}/messages", {"content": text}, headers={"HX-Request": "true"})
|
|
|
|
|
|
def test_tokens_und_preis():
|
|
counts = usage.tokens(USAGE)
|
|
assert counts == {"input_tokens": 1_000_000, "cached_tokens": 400_000, "output_tokens": 200_000,
|
|
"reasoning_tokens": 50_000}
|
|
model = Model(input_price=Decimal("2"), cached_price=Decimal("0.5"), output_price=Decimal("8"))
|
|
# 600k * 2 + 400k * 0,5 + 200k * 8 = 1,2 + 0,2 + 1,6
|
|
assert usage.price(model, counts) == Decimal("3")
|
|
model.cached_price = None
|
|
assert usage.price(model, counts) == Decimal("3.6")
|
|
assert usage.price(Model(), counts) is None
|
|
assert usage.tokens({"input_tokens": -5, "output_tokens": "x"})["input_tokens"] == 0
|
|
|
|
|
|
def test_preise_im_modell(setup, db, people):
|
|
anna = setup["anna"]
|
|
with db.session() as session:
|
|
model_id = session.scalar(select(Model.id))
|
|
base = {"provider_id": people["provider"], "name": "gpt-6-luna", "label": "GPT-6 Luna", "model_class": "standard",
|
|
"context_window": "128000", "supports_tools": "1", "active": "1", "model_id": str(model_id)}
|
|
assert "Modell gespeichert" in anna.post("/admin/models", {**base, "input_price": "1,25", "output_price": "10"}).text
|
|
with db.session() as session:
|
|
model = session.get(Model, model_id)
|
|
assert model.input_price == Decimal("1.25") and model.output_price == Decimal("10")
|
|
assert model.cached_price is None
|
|
assert "1,25 / 10 USD" in anna.get("/admin/models").text
|
|
assert anna.post("/admin/models", {**base, "input_price": "1"}).status_code == 400
|
|
assert anna.post("/admin/models", {**base, "input_price": "abc", "output_price": "1"}).status_code == 400
|
|
assert anna.post("/admin/models", {**base, "input_price": "-1", "output_price": "1"}).status_code == 400
|
|
|
|
|
|
def test_verbrauch_wird_je_aufruf_erfasst(setup, db, settings, fake_provider, people):
|
|
from app.identity import Principal
|
|
from app.tools import service as tools
|
|
admin = Principal(user_id=people["anna"], username="anna", display_name="Anna", is_admin=True,
|
|
team_ids=frozenset({people["team_demo"]}))
|
|
with db.session() as session:
|
|
tools.classify_suggested(session, admin)
|
|
with db.session() as session:
|
|
session.execute(update(Model).values(input_price=Decimal("2"), cached_price=Decimal("0.5"),
|
|
output_price=Decimal("8")))
|
|
fake_provider.usage = USAGE
|
|
fake_provider.script = [[("git_list_dir", {})], "Fertig."]
|
|
ask(setup["anna"], setup["chat_id"])
|
|
Worker(db, settings).run_once()
|
|
with db.session() as session:
|
|
records = list(session.scalars(select(UsageRecord)))
|
|
# Zwei Modellaufrufe (Tool-Runde und Antwort), je mit Kosten
|
|
assert len(records) == 2 and all(r.cost == Decimal("3") for r in records)
|
|
assert records[0].workspace_id == setup["workspace_id"] and records[0].team_id is not None
|
|
# Verbrauch bleibt nach dem Löschen des Chats
|
|
setup["anna"].post(f"/chats/{setup['chat_id']}/delete")
|
|
with db.session() as session:
|
|
assert len(list(session.scalars(select(UsageRecord)))) == 2
|
|
|
|
|
|
def test_uebersicht_und_runs(setup, db, settings, fake_provider):
|
|
with db.session() as session:
|
|
session.execute(update(Model).values(input_price=Decimal("2"), output_price=Decimal("8")))
|
|
fake_provider.usage = USAGE
|
|
ask(setup["anna"], setup["chat_id"])
|
|
Worker(db, settings).run_once()
|
|
page = setup["anna"].get("/admin/usage").text
|
|
assert "Demo" in page and "1.200.000" in page and "3,60 USD" in page and "fertig: 1" in page
|
|
page = setup["anna"].get("/admin/usage?group=model").text
|
|
assert "gpt-6-luna" in page
|
|
page = setup["anna"].get("/admin/usage?group=day&period=7d").text
|
|
assert "3,60" in page
|
|
assert setup["anna"].get("/admin/usage?period=custom&start=2020-01-01&end=2020-01-31").status_code == 200
|
|
assert "keine Modellaufrufe" in setup["anna"].get("/admin/usage?period=custom&start=2020-01-01&end=2020-01-31").text
|
|
assert setup["anna"].get("/admin/usage?period=custom&start=x").status_code == 400
|
|
runs = setup["anna"].get("/admin/runs").text
|
|
assert "fertig" in runs and f"/chats/{setup['chat_id']}" in runs and "1.200.000" in runs
|
|
assert "Keine Runs im Zeitraum" in setup["anna"].get("/admin/runs?status=failed").text
|
|
# Workspace-Admin sieht seinen Workspace, Mitglieder nicht
|
|
assert "3,60" in setup["anna"].get(f"/workspaces/{setup['workspace_id']}/usage").text
|
|
assert setup["anna"].get(f"/workspaces/{setup['workspace_id']}/runs").status_code == 200
|
|
assert setup["ben"].get(f"/workspaces/{setup['workspace_id']}/usage").status_code == 403
|
|
assert setup["ben"].get("/admin/usage").status_code == 403
|
|
assert setup["ben"].get("/admin/runs").status_code == 403
|
|
|
|
|
|
def test_audit_log(setup, db, browser):
|
|
anna, ben = setup["anna"], setup["ben"]
|
|
page = anna.get("/admin/audit").text
|
|
assert "workspace.created" in page and "source.tested" in page and "System" not in page.split("<tbody>")[0]
|
|
page = anna.get("/admin/audit?action=source.").text
|
|
assert "source.tested" in page and "workspace.created" not in page.split("<tbody>")[-1]
|
|
assert "Keine Ereignisse" in anna.get("/admin/audit?start=2020-01-01&end=2020-01-02").text
|
|
assert anna.get("/admin/audit?start=kaputt").status_code == 400
|
|
csv = anna.get("/admin/audit.csv?action=source.").text
|
|
assert csv.startswith("id;zeit;aktion") and "source.tested" in csv and "workspace.created" not in csv
|
|
# Workspace-Admin: nur Ereignisse des Workspace
|
|
with db.session() as session:
|
|
session.add(AuditEvent(action="fremd.ereignis", workspace_id=None))
|
|
page = anna.get(f"/workspaces/{setup['workspace_id']}/audit").text
|
|
assert "source.tested" in page and "fremd.ereignis" not in page
|
|
assert "fremd.ereignis" in anna.get("/admin/audit").text
|
|
assert ben.get(f"/workspaces/{setup['workspace_id']}/audit").status_code == 403
|
|
assert ben.get("/admin/audit").status_code == 403
|
|
assert browser("dora").get(f"/workspaces/{setup['workspace_id']}/audit.csv").status_code == 404
|
|
|
|
|
|
def test_audit_blaettern(setup, db, monkeypatch):
|
|
from app.audit import views
|
|
monkeypatch.setattr(views, "PAGE", 2)
|
|
page = setup["anna"].get("/admin/audit").text
|
|
assert "Ältere Ereignisse" in page
|
|
import re
|
|
link = re.search(r'href="(/admin/audit\?[^"]*before=\d+[^"]*)"', page).group(1).replace("&", "&")
|
|
older = setup["anna"].get(link).text
|
|
assert older.count("<tr>") >= 2
|