"""Tests for MR-triggered plugin action runs and progress fan-out."""

from __future__ import annotations

import asyncio
from collections.abc import Generator
from pathlib import Path
from unittest.mock import patch

import pygit2
import pytest
from fastapi import Request
from fastapi.testclient import TestClient
from pygittools.merge import MR_REF_PREFIX, MergeRequest
from starlette.responses import PlainTextResponse, Response

from pygitweb.action_runs import ACTION_RUNS, ActionRunRegistry, start_mr_action_runs
from pygitweb.api.action import Action, ActionProgress, MrEvent, MrTrigger, ProgressSink
from pygitweb.config import settings
from pygitweb.main import app


class _ProgressAction(Action):
    def __init__(self) -> None:
        self.events: list[MrEvent] = []

    def action_name(self) -> str:
        return "progress_demo"

    def action(self, project: str, req: Request | None = None) -> Response | None:
        return PlainTextResponse(project)

    def mr_triggers(self) -> frozenset[MrTrigger]:
        return frozenset({MrTrigger.OPENED, MrTrigger.UPDATED})

    async def on_mr(self, event: MrEvent, progress: ProgressSink) -> None:
        self.events.append(event)
        await progress.report(ActionProgress(message="step-1", fraction=0.25))
        await asyncio.sleep(0.01)
        await progress.report(ActionProgress(message="step-2", fraction=0.75))
        await progress.report(ActionProgress(message="finished", fraction=1.0, done=True, ok=True))


def _commit_chain(repo: pygit2.Repository) -> tuple[pygit2.Oid, pygit2.Oid]:
    sig = pygit2.Signature("tester", "tester@example.com")
    tb = repo.TreeBuilder()
    tree = tb.write()
    a = repo.create_commit(None, sig, sig, "a", tree, [])
    repo.create_reference("refs/heads/main", a)
    b = repo.create_commit(None, sig, sig, "b", tree, [a])
    repo.create_reference("refs/heads/feature", b)
    return a, b


@pytest.fixture
def mr_project(tmp_path: Path) -> Generator[dict[str, object], None, None]:
    root = tmp_path / "action-runs"
    root.mkdir()
    repo_dir = root / "demo"
    repo_dir.mkdir()
    repo = pygit2.init_repository(str(repo_dir), bare=False)
    _a, b = _commit_chain(repo)
    projects_list = root / "projects.list"
    projects_list.write_text("demo tester\n", encoding="utf-8")
    plugin = _ProgressAction()
    with (
        patch.object(settings, "PROJECTROOT", str(root)),
        patch.object(settings, "PROJECTS_LIST", str(projects_list)),
        patch.object(settings, "STRICT_EXPORT", False),
        patch.object(settings, "EXPORT_OK", ""),
        patch.object(settings, "AUTH", False),
        patch("pygitweb.plugin_loader.PLUGIN_ACTIONS", {"progress_demo": plugin}),
    ):
        client = TestClient(app)
        yield {
            "client": client,
            "project": "demo",
            "repo": repo,
            "plugin": plugin,
            "tip_b": str(b),
        }


def test_start_mr_triggers_reports_progress(mr_project: dict[str, object]) -> None:
    repo = mr_project["repo"]
    assert isinstance(repo, pygit2.Repository)
    plugin = mr_project["plugin"]
    assert isinstance(plugin, _ProgressAction)
    mr = MergeRequest(
        pygit2.Oid(hex=str(mr_project["tip_b"])),
        pygit2.Signature("tester", "tester@example.com"),
        ours="refs/heads/main",
        theirs="refs/heads/feature",
        title="Feature",
        name=f"{MR_REF_PREFIX}abcdabcdabcdabcd",
    )
    mr.write(repo)
    registry = ActionRunRegistry()
    run_ids = registry.start_mr_triggers(
        "demo",
        MrTrigger.OPENED,
        mr,
        actions={"progress_demo": plugin},
    )
    assert len(run_ids) == 1
    run = registry.get(run_ids[0])
    assert run is not None

    async def wait_done() -> ActionProgress:
        for _ in range(100):
            if run.current.done:
                return run.current
            await asyncio.sleep(0.02)
        return run.current

    final = asyncio.run(wait_done())
    assert final.done is True
    assert final.ok is True
    assert final.message == "finished"
    assert plugin.events and plugin.events[0].trigger is MrTrigger.OPENED
    assert any(p.message == "step-1" for p in run.history)


def test_create_mr_starts_opened_runs(mr_project: dict[str, object]) -> None:
    client = mr_project["client"]
    assert isinstance(client, TestClient)
    plugin = mr_project["plugin"]
    assert isinstance(plugin, _ProgressAction)
    post = client.post(
        "/mr/create",
        data={
            "project": "demo",
            "theirs": "refs/heads/feature",
            "ours": "refs/heads/main",
            "title": "From form",
        },
        follow_redirects=False,
    )
    assert post.status_code == 303
    location = post.headers["location"]
    assert "runs=" in location
    assert "a=tag" in location

    async def wait_plugin() -> None:
        for _ in range(100):
            if plugin.events:
                return
            await asyncio.sleep(0.02)

    asyncio.run(wait_plugin())
    assert plugin.events
    assert plugin.events[0].trigger is MrTrigger.OPENED
    assert plugin.events[0].title == "From form"

    run_ids = location.split("runs=")[1].split("&")[0]
    resp = client.get("/mr/runs", params={"project": "demo", "ids": run_ids})
    assert resp.status_code == 200
    body = resp.json()
    assert body["project"] == "demo"
    assert body["runs"]
    assert body["runs"][0]["action"] == "progress_demo"


def test_runs_sse_streams_until_done(mr_project: dict[str, object]) -> None:
    repo = mr_project["repo"]
    assert isinstance(repo, pygit2.Repository)
    plugin = mr_project["plugin"]
    assert isinstance(plugin, _ProgressAction)
    client = mr_project["client"]
    assert isinstance(client, TestClient)
    mr = MergeRequest(
        pygit2.Oid(hex=str(mr_project["tip_b"])),
        pygit2.Signature("tester", "tester@example.com"),
        ours="refs/heads/main",
        title="SSE",
        name=f"{MR_REF_PREFIX}sse0sse0sse0sse0",
    )
    mr.write(repo)
    run_ids = start_mr_action_runs("demo", MrTrigger.OPENED, mr, actions={"progress_demo": plugin})
    assert run_ids
    with client.stream(
        "GET",
        f"/mr/runs/{run_ids[0]}/events",
        params={"project": "demo"},
    ) as resp:
        assert resp.status_code == 200
        chunks: list[str] = []
        for line in resp.iter_lines():
            if line.startswith("data: "):
                chunks.append(line.removeprefix("data: "))
                if '"done": true' in line or '"done":true' in line:
                    break
    assert chunks
    assert any("finished" in chunk or "completed" in chunk or "step" in chunk for chunk in chunks)


def test_action_defaults_skip_triggers() -> None:
    class _Plain(Action):
        def action_name(self) -> str:
            return "plain"

        def action(self, project: str, req: Request | None = None) -> Response | None:
            return None

    plain = _Plain()
    assert plain.mr_triggers() == frozenset()
    mr = MergeRequest(
        "0" * 40,
        "x <x@y>",
        ours="refs/heads/main",
        title="t",
        name=f"{MR_REF_PREFIX}ffffffffffffffff",
    )
    # Invalid head oid — start_mr_triggers should still skip because no triggers.
    ids = ActionRunRegistry().start_mr_triggers(
        "demo",
        MrTrigger.OPENED,
        mr,
        actions={"plain": plain},
    )
    assert ids == []


def test_parse_and_registry_helpers() -> None:
    from pygitweb.action_runs import parse_run_ids

    assert parse_run_ids(None) == []
    assert parse_run_ids(" a, b , ") == ["a", "b"]
    assert ACTION_RUNS.get("missing") is None


def test_refresh_from_theirs_starts_updated_runs(mr_project: dict[str, object]) -> None:
    repo = mr_project["repo"]
    assert isinstance(repo, pygit2.Repository)
    plugin = mr_project["plugin"]
    assert isinstance(plugin, _ProgressAction)
    client = mr_project["client"]
    assert isinstance(client, TestClient)
    sig = pygit2.Signature("tester", "tester@example.com")
    main = repo.references["refs/heads/main"].target
    feature = repo.references["refs/heads/feature"].target
    mr = MergeRequest(
        main,
        sig,
        ours="refs/heads/main",
        theirs="refs/heads/feature",
        title="Track feature",
        name=f"{MR_REF_PREFIX}upd0upd0upd0upd0",
    )
    mr_oid = mr.write(repo)
    assert str(mr.target) == str(main)
    assert str(feature) != str(main)

    resp = client.get(
        f"/project/{mr_project['project']}",
        params={"a": "tag", "h": str(mr_oid)},
        follow_redirects=False,
    )
    assert resp.status_code == 302
    location = resp.headers["location"]
    assert "runs=" in location

    async def wait_updated() -> None:
        for _ in range(100):
            if any(e.trigger is MrTrigger.UPDATED for e in plugin.events):
                return
            await asyncio.sleep(0.02)

    asyncio.run(wait_updated())
    assert any(e.trigger is MrTrigger.UPDATED for e in plugin.events)