"""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