Onyscribe-Admin/tests/test_agent_preview.py

"""Änderungen des KI-Agenten müssen in Vorschau (live, auch PDF) und nach der Freigabe sichtbar sein."""
import io
from pathlib import Path

import pytest
from fastapi.testclient import TestClient
from pypdf import PdfReader

from app.agent import loop
from app.agent.backends import Turn
from app.db import session
from app.main import app
from app.models import ChatMessage, Tenant
from app.services import preview
from app.services.importer import import_all
from app.settings import LEGACY_PLATFORM


@pytest.fixture(scope="module")
def client():
    with TestClient(app) as c:
        with session() as db:
            import_all(db, Path(LEGACY_PLATFORM))
        yield c


def pdf_text(b: bytes) -> str:
    return "\n".join(p.extract_text() or "" for p in PdfReader(io.BytesIO(b)).pages)


class Scripted:
    name, model = "scripted", "x"

    def __init__(self, *calls):
        self.calls, self.i = calls, 0

    def step(self, system, messages, tools):
        self.i += 1
        if self.i == 1:
            return Turn([{"type": "tool_use", "id": str(n), "name": nm, "input": a} for n, (nm, a) in enumerate(self.calls)])
        return Turn([{"type": "text", "text": "Fertig."}])


CSS = "body { background: #ffee99; } h2 { color: #cc0000; }"
SAMPLE = "===TERMINART===\nWorkshop\n===/TERMINART===\n===PROJEKT===\nAgentenprojekt\n===/PROJEKT===\nAnlass & Kontext\nEin Text nur aus dem Agenten-Beispiel.\n"


def test_live_preview_sees_files_written_by_the_running_agent(client):
    """Regression: Dateien lagen in der offenen Transaktion des Agenten, die Vorschau (andere Sitzung) bekam KeyError."""
    with session() as db:
        t = db.get(Tenant, "sprinz")
        base = preview.run(db, t, "draft", "pdf")
        events = []
        for ev in loop.run(db, t, "x", backend=Scripted(("write_file", {"path": "protocol.css", "content": CSS}), ("set_config", {"key": "pdf_projekt_label", "value": "AGENTENTHEMA"}))):
            events.append(ev)
            if ev["type"] == "changed":
                with session() as other:                              # wie der Preview-Request: eigene Sitzung, Agent-Transaktion läuft noch
                    live = preview.run(other, other.get(Tenant, "sprinz"), f"ws:{ev['run']}", "pdf")
                assert live != base                                  # kein KeyError, und die Änderung ist im PDF
    changed = [e for e in events if e["type"] == "changed"]
    assert "protocol.css" in changed[0]["paths"] and any("pdf_projekt_label" in p for p in changed[-1]["paths"])


def test_sample_set_by_agent_reaches_live_preview_proposal_and_draft(client):
    with session() as db:
        t = db.get(Tenant, "revier")
        events = list(loop.run(db, t, "x", backend=Scripted(("set_sample", {"text": SAMPLE}))))
        db.commit()
        run = next(e["run"] for e in events if e["type"] == "mode")
        prop = next(e for e in events if e["type"] == "proposal")["proposal"]
        assert prop["sample"] == SAMPLE and any(i["path"] == "Vorschau-Beispielprotokoll" for i in prop["items"])
        assert "Agentenprojekt" in pdf_text(preview.run(db, t, f"ws:{run}", "pdf"))                  # live
        assert "Agentenprojekt" not in pdf_text(preview.run(db, t, "draft", "pdf"))                   # Entwurf unberührt
        m = ChatMessage(tenant_id="revier", role="assistant", content="ok", proposal=prop, proposal_status="open")
        db.add(m)
        db.commit()
        assert "Agentenprojekt" in pdf_text(preview.run(db, t, f"proposal:{m.id}", "pdf"))            # Vorschlag
        loop.apply_proposal(db, t, prop)
        db.commit()
        assert "Agentenprojekt" in pdf_text(preview.run(db, t, "draft", "pdf"))                       # nach Freigabe im Entwurf


def test_prompt_test_result_becomes_the_pdf_sample(client, monkeypatch):
    from app.services import llm
    llm_text = SAMPLE.replace("Agentenprojekt", "LLMERGEBNIS")
    monkeypatch.setattr(llm, "summarize", lambda db, system, user, model=None, tenant_id=None: {"text": llm_text, "model": "m", "input_tokens": 1, "output_tokens": 1, "truncated": False})
    with session() as db:
        t = db.get(Tenant, "demo")
        events = list(loop.run(db, t, "x", backend=Scripted(("run_prompt_test", {}))))
        run = next(e["run"] for e in events if e["type"] == "mode")
        assert "LLMERGEBNIS" in pdf_text(preview.run(db, t, f"ws:{run}", "pdf"))
        assert next(e for e in events if e["type"] == "proposal")["proposal"]["sample"] == llm_text


def test_api_flow_through_sse_and_apply(client, monkeypatch):
    """Über die echte API: Agent schreibt CSS + Beispiel → Vorschlag → Freigabe → Entwurf-PDF zeigt es."""
    monkeypatch.setattr(loop, "make_backend", lambda db, tid=None: Scripted(("write_file", {"path": "protocol.css", "content": CSS}), ("set_sample", {"text": SAMPLE})))
    r = client.post("/api/tenants/onyva/chat", json={"message": "mach was"})
    import json
    ev = [json.loads(l[6:]) for l in r.text.split("\n\n") if l.startswith("data: ")]
    assert any(e["type"] == "changed" and "protocol.css" in e["paths"] for e in ev)
    msg = ev[-1]["message"]
    live = client.get(f"/preview/onyva/pdf?source=proposal:{msg['id']}")
    assert live.status_code == 200 and "Agentenprojekt" in pdf_text(live.content)
    assert client.post(f"/api/tenants/onyva/chat/{msg['id']}/apply").status_code == 200
    assert "Agentenprojekt" in pdf_text(client.get("/preview/onyva/pdf?source=draft").content)
    assert client.get("/api/tenants/onyva/state").json()["files"]["protocol.css"]