219 lines
10 KiB
Python
219 lines
10 KiB
Python
"""Betrieb: /live, /ready, /metrics, Worker-Herzschlag und -Probes, Logs, Backup und Prüfung."""
|
|
|
|
import io
|
|
import json
|
|
import logging
|
|
import socket
|
|
import tarfile
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from app import backup
|
|
from app.core import logs
|
|
from app.core.crypto import Vault, generate_key
|
|
from app.data_sources.models import DataSource
|
|
from app.identity import Principal
|
|
from app.main import create_app
|
|
from app.tasks import monitor
|
|
from app.tasks.monitor import WorkerHeartbeat
|
|
from app.tasks.worker import Worker
|
|
from app.tools import service as tools
|
|
from app.uploads.models import Upload
|
|
from helpers import create_workspace
|
|
|
|
CSV = b"region,umsatz\nNord,100\nSued,250\n"
|
|
TOKEN = "metrik-token-1234"
|
|
|
|
|
|
@pytest.fixture
|
|
def ws(browser, people, db):
|
|
anna = browser("anna")
|
|
workspace_id = create_workspace(anna, people["team_demo"], "Betrieb")
|
|
anna.post(f"/workspaces/{workspace_id}/uploads", files={"file": ("umsatz.csv", CSV)})
|
|
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)
|
|
chat_id = int(anna.post(f"/workspaces/{workspace_id}/chats").headers["location"].rsplit("/", 1)[1])
|
|
return {"anna": anna, "id": workspace_id, "chat_id": chat_id}
|
|
|
|
|
|
def test_live_und_ready(app, settings):
|
|
client = TestClient(app)
|
|
for path in ("/live", "/healthz"):
|
|
assert client.get(path).json() == {"status": "ok"}
|
|
body = client.get("/ready").json()
|
|
assert body["status"] == "ok"
|
|
assert set(body["checks"]) == {"database", "migrations", "uploads", "secret_key"}
|
|
assert client.get("/readyz").status_code == 200
|
|
|
|
|
|
def test_ready_meldet_fehlenden_key_und_kaputte_datenbank(settings, db):
|
|
client = TestClient(create_app(settings.model_copy(update={"secret_key": ""}), listen=False))
|
|
response = client.get("/ready")
|
|
assert response.status_code == 503 and response.json()["checks"]["secret_key"] == "AIW_SECRET_KEY fehlt"
|
|
broken = settings.model_copy(update={"database_url": "postgresql+psycopg://aiw:aiw@127.0.0.1:1/nix"})
|
|
client = TestClient(create_app(broken, listen=False))
|
|
response = client.get("/ready")
|
|
assert response.status_code == 503 and "nicht erreichbar" in response.json()["checks"]["database"]
|
|
assert client.get("/live").status_code == 200 # live prüft bewusst keine Datenbank
|
|
|
|
|
|
def test_metrics_nur_mit_token(settings, db, ws, fake_provider):
|
|
client = TestClient(create_app(settings, listen=False))
|
|
assert client.get("/metrics").status_code == 404 # ohne AIW_METRICS_TOKEN aus
|
|
client = TestClient(create_app(settings.model_copy(update={"metrics_token": TOKEN}), listen=False))
|
|
assert client.get("/metrics").status_code == 401
|
|
assert client.get("/metrics", headers={"Authorization": "Bearer falsch"}).status_code == 401
|
|
|
|
fake_provider.script = ["Antwort."]
|
|
ws["anna"].post(f"/chats/{ws['chat_id']}/messages", {"content": "Hallo?"}, headers={"HX-Request": "true"})
|
|
Worker(db, settings).run_once()
|
|
ws["anna"].get(f"/chats/{ws['chat_id']}")
|
|
text = client.get("/metrics", headers={"Authorization": f"Bearer {TOKEN}"}).text
|
|
assert 'aiw_runs_finished_total{kind="chat",status="completed"}' in text
|
|
assert "aiw_model_calls_total{" in text and 'outcome="ok"' in text
|
|
assert 'aiw_http_requests_total{method="GET",route="/chats/{chat_id}",status="200"}' in text
|
|
assert "aiw_workers_alive 1.0" in text and "aiw_state_up 1.0" in text
|
|
assert "aiw_runs_oldest_queued_seconds 0.0" in text
|
|
assert 'route="/live"' not in text and f'/chats/{ws["chat_id"]}' not in text # keine Probes, keine IDs
|
|
|
|
|
|
def test_worker_herzschlag_und_probes(db, settings):
|
|
settings = settings.model_copy(update={"worker_http_port": 0})
|
|
worker = Worker(db, settings)
|
|
worker.run_once()
|
|
with db.session() as session:
|
|
beat = session.scalar(select(WorkerHeartbeat))
|
|
assert beat.worker_id == worker.worker_id
|
|
|
|
probe = monitor.Probe()
|
|
assert monitor.serve(db, settings, probe) is None # Port 0 = aus
|
|
|
|
with socket.socket() as s:
|
|
s.bind(("127.0.0.1", 0))
|
|
port = s.getsockname()[1]
|
|
live = settings.model_copy(update={"worker_http_port": port, "worker_http_host": "127.0.0.1",
|
|
"metrics_token": TOKEN, "worker_stale_seconds": 60})
|
|
server = monitor.serve(db, live, probe)
|
|
try:
|
|
base = f"http://127.0.0.1:{port}"
|
|
assert json.load(urllib.request.urlopen(f"{base}/live"))["status"] == "ok"
|
|
assert json.load(urllib.request.urlopen(f"{base}/ready"))["status"] == "ok"
|
|
request = urllib.request.Request(f"{base}/metrics", headers={"Authorization": f"Bearer {TOKEN}"})
|
|
assert b"aiw_worker_last_loop_timestamp_seconds" in urllib.request.urlopen(request).read()
|
|
probe.last -= 120 # Schleife hängt
|
|
with pytest.raises(urllib.error.HTTPError) as e:
|
|
urllib.request.urlopen(f"{base}/live")
|
|
assert e.value.code == 503
|
|
finally:
|
|
server.shutdown()
|
|
monitor.forget(db, worker.worker_id)
|
|
with db.session() as session:
|
|
assert session.scalar(select(WorkerHeartbeat)) is None
|
|
|
|
|
|
def test_request_id_und_zugriffszeile(app, caplog):
|
|
client = TestClient(app)
|
|
with caplog.at_level(logging.INFO, logger="aiw.access"):
|
|
response = client.get("/login?next=/geheim", headers={"X-Request-ID": "proxy-abc-123"})
|
|
client.get("/live")
|
|
assert response.headers["X-Request-ID"] == "proxy-abc-123"
|
|
lines = [r for r in caplog.records if r.name == "aiw.access"]
|
|
assert len(lines) == 1 # Probes nicht im Log
|
|
assert lines[0].route == "/login" and lines[0].status == 200 and "geheim" not in lines[0].getMessage()
|
|
assert client.get("/live", headers={"X-Request-ID": "<script>"}).headers["X-Request-ID"] != "<script>"
|
|
|
|
|
|
def test_json_logs_mit_kontext_und_maskierung():
|
|
record = logging.LogRecord("aiw.test", logging.WARNING, __file__, 1,
|
|
"Fehler bei postgresql://u:geheim@host/db mit Bearer abcdef123456 und sk-abcdefgh1234",
|
|
None, None)
|
|
record.detail = "password=hunter2"
|
|
before = logs.current()
|
|
with logs.scope(run_id="r-1", workspace_id=7):
|
|
logs.ContextFilter().filter(record)
|
|
entry = json.loads(logs.JsonFormatter().format(record))
|
|
assert entry["run_id"] == "r-1" and entry["workspace_id"] == 7 and entry["level"] == "warning"
|
|
for secret in ("geheim", "abcdef123456", "abcdefgh1234", "hunter2"):
|
|
assert secret not in json.dumps(entry)
|
|
assert logs.current() == before
|
|
|
|
|
|
def test_run_logs_tragen_run_id(ws, db, settings, fake_provider, caplog):
|
|
fake_provider.script = ["Antwort."]
|
|
ws["anna"].post(f"/chats/{ws['chat_id']}/messages", {"content": "Hallo?"}, headers={"HX-Request": "true"})
|
|
with caplog.at_level(logging.INFO):
|
|
Worker(db, settings).run_once()
|
|
events = {getattr(r, "event", None): r for r in caplog.records}
|
|
assert events["run_finished"].outcome == "completed"
|
|
assert events["run_finished"].run_id == events["model_call"].run_id
|
|
assert events["model_call"].workspace_id == ws["id"]
|
|
assert "Hallo" not in caplog.text and "Antwort." not in caplog.text # keine Inhalte
|
|
|
|
|
|
def test_fehlerseite_zeigt_fehler_id(app):
|
|
@app.get("/kaputt")
|
|
def broken():
|
|
raise RuntimeError("Absicht")
|
|
response = TestClient(app, raise_server_exceptions=False).get("/kaputt", headers={"X-Request-ID": "fehler-1234"})
|
|
assert response.status_code == 500 and "fehler-1234" in response.text and "Absicht" not in response.text
|
|
|
|
|
|
def test_backup_restore_und_check(ws, db, settings, tmp_path):
|
|
with db.session() as session:
|
|
upload = session.scalar(select(Upload))
|
|
assert upload.status == "ok"
|
|
result = backup.backup(db, settings, tmp_path / "sicherung")
|
|
assert result.files == 1 and not result.missing and result.archive.name.endswith(".tar.gz")
|
|
with tarfile.open(result.archive) as tar:
|
|
manifest = json.load(tar.extractfile("manifest.json"))
|
|
assert manifest["files"][0]["upload_id"] == upload.id
|
|
|
|
path = Path(settings.upload_dir) / str(upload.workspace_id) / f"{upload.id}.duckdb"
|
|
original = path.read_bytes()
|
|
path.unlink()
|
|
check = backup.check(db, settings)
|
|
assert not check.ok and check.details["missing_uploads"] == [upload.id]
|
|
|
|
restored = backup.restore(db, settings, result.archive)
|
|
assert restored.restored == 1 and path.read_bytes() == original
|
|
assert backup.restore(db, settings, result.archive).unchanged == 1
|
|
path.write_bytes(b"anders")
|
|
assert backup.restore(db, settings, result.archive).conflicts
|
|
assert backup.restore(db, settings, result.archive, force=True).restored == 1
|
|
|
|
(path.parent / "999.duckdb").write_bytes(b"x")
|
|
check = backup.check(db, settings)
|
|
assert check.ok and check.details["orphan_files"] == [f"{upload.workspace_id}/999.duckdb"]
|
|
|
|
|
|
def test_check_erkennt_falschen_master_key(ws, db, settings, people):
|
|
with db.session() as session:
|
|
session.add(DataSource(workspace_id=ws["id"], name="x", kind="git", config={},
|
|
secret=Vault(settings.secret_key).encrypt("pw"), exposure_confirmed_by=people["anna"],
|
|
created_by=people["anna"]))
|
|
assert backup.check(db, settings).checks["secrets"] == "ok"
|
|
other = settings.model_copy(update={"secret_key": generate_key()})
|
|
result = backup.check(db, other)
|
|
assert "nicht entschlüsselbar" in result.checks["secrets"]
|
|
assert any(item.startswith("Datenquelle ") for item in result.details["secrets_failed"])
|
|
|
|
|
|
def test_restore_lehnt_pfade_ausserhalb_ab(ws, db, settings, tmp_path):
|
|
with db.session() as session:
|
|
upload = session.scalar(select(Upload))
|
|
archive = tmp_path / "boese.tar.gz"
|
|
manifest = {"format": 1, "files": [{"path": "uploads/../../etc/x", "upload_id": upload.id, "sha256": "0"}]}
|
|
data = json.dumps(manifest).encode()
|
|
with tarfile.open(archive, "w:gz") as tar:
|
|
info = tarfile.TarInfo("manifest.json")
|
|
info.size = len(data)
|
|
tar.addfile(info, io.BytesIO(data))
|
|
with pytest.raises(ValueError, match="Ungültiger Pfad"):
|
|
backup.restore(db, settings, archive)
|