"""Palier 2 (0.4.0): webhooks + SSE live, label sync, milestone boards."""

import asyncio
import hashlib
import hmac
import json

from app import models
from app.main import hub
from tests.conftest import csrf_from


def _move(client, number, to_column, from_column):
    token = csrf_from(client)
    return client.post(
        "/api/move",
        data={
            "repo_full_name": "Projets/demo", "issue_number": number,
            "to_column": to_column, "from_column": from_column,
            "csrf_token": token,
        },
    )


def _webhook(client, payload, event="issues", secret=None, raw=None):
    body = raw if raw is not None else json.dumps(payload).encode()
    headers = {"X-Gitea-Event": event, "Content-Type": "application/json"}
    if secret is not None:
        headers["X-Gitea-Signature"] = hmac.new(
            secret.encode(), body, hashlib.sha256
        ).hexdigest()
    return client.post("/webhooks/gitea", content=body, headers=headers)


# ── Hub SSE ────────────────────────────────────────────────

def test_hub_broadcast_reaches_subscribers():
    async def scenario():
        queue = hub.subscribe()
        try:
            await hub.broadcast({"repo": "a/b", "action": "closed"})
            return await asyncio.wait_for(queue.get(), timeout=1)
        finally:
            hub.unsubscribe(queue)

    payload = asyncio.run(scenario())
    assert payload["action"] == "closed"
    assert hub.subscriber_count == 0


def test_sse_event_stream_frames_and_cleanup():
    # The generator is unit-tested directly: HTTP test clients cannot
    # reliably tear down an infinite SSE stream (verified empirically).
    from app.main import sse_event_stream

    class FakeRequest:
        async def is_disconnected(self):
            return False

    async def scenario():
        gen = sse_event_stream(FakeRequest())
        first = await gen.__anext__()
        await hub.broadcast({"repo": "a/b", "action": "closed"})
        second = await asyncio.wait_for(gen.__anext__(), timeout=5)
        await gen.aclose()
        return first, second

    first, second = asyncio.run(scenario())
    assert first == "retry: 5000\n\n"
    assert second.startswith("event: gitea\ndata: ")
    assert '"action": "closed"' in second
    assert hub.subscriber_count == 0  # finally unsubscribed


# ── Webhooks ───────────────────────────────────────────────

def test_webhook_closed_applies_transition_and_journals(client, fake_gitea):
    models.move_issue("Projets/demo", 1, "todo")
    payload = {
        "action": "closed",
        "repository": {"full_name": "Projets/demo"},
        "issue": {"number": 1},
    }
    r = _webhook(client, payload)
    assert r.status_code == 200
    assert r.json()["handled"] == "closed"
    state = models.get_board_state("Projets/demo")
    assert [c["issue_number"] for c in state["done"]] == [1]
    events = models.get_events("Projets/demo")
    assert events[0]["event_type"] == "closed"
    assert events[0]["source"] == "webhook"
    assert fake_gitea.cache_invalidations >= 1

    # The echo of our own API moves arrives in the same state: no-op,
    # no duplicate journal entry.
    r = _webhook(client, payload)
    assert r.json()["handled"] == "noop"
    assert len(models.get_events("Projets/demo")) == 1


def test_webhook_reopened_moves_back_to_backlog(client):
    models.move_issue("Projets/demo", 1, "done")
    r = _webhook(client, {
        "action": "reopened",
        "repository": {"full_name": "Projets/demo"},
        "issue": {"number": 1},
    })
    assert r.json()["handled"] == "reopened"
    state = models.get_board_state("Projets/demo")
    assert [c["issue_number"] for c in state["backlog"]] == [1]


def test_webhook_signature_enforced_when_secret_set(client, monkeypatch):
    monkeypatch.setenv("KANBANTE_WEBHOOK_SECRET", "s3cret")
    payload = {
        "action": "closed",
        "repository": {"full_name": "Projets/demo"},
        "issue": {"number": 1},
    }
    assert _webhook(client, payload).status_code == 401
    assert _webhook(client, payload, secret="wrong").status_code == 401
    assert _webhook(client, payload, secret="s3cret").status_code == 200


def test_webhook_invalid_json_is_400(client):
    r = _webhook(client, None, raw=b"{not json")
    assert r.status_code == 400


def test_webhook_ignores_unrelated_events(client):
    r = _webhook(client, {
        "action": "created",
        "repository": {"full_name": "Projets/demo"},
        "issue": {"number": 1},
    }, event="issue_comment")
    assert r.status_code == 200
    assert r.json()["handled"] == "ignored"


# ── Synchro par étiquettes ─────────────────────────────────

def test_label_sync_follows_column_moves(client, fake_gitea):
    r = _move(client, 1, "in_progress", "backlog")
    assert r.json()["label_sync"] == "ok"
    assert ("create", "Projets/demo", "kanban:in_progress") in fake_gitea.label_ops
    assert ("add", 1, "kanban:in_progress") in fake_gitea.label_ops

    r = _move(client, 1, "review", "in_progress")
    assert r.json()["label_sync"] == "ok"
    assert ("remove", 1, "kanban:in_progress") in fake_gitea.label_ops
    assert ("add", 1, "kanban:review") in fake_gitea.label_ops

    r = _move(client, 1, "done", "review")
    assert r.json()["label_sync"] == "ok"
    assert ("remove", 1, "kanban:review") in fake_gitea.label_ops
    issue = fake_gitea.issues["Projets/demo"][0]
    assert not any(l["name"].startswith("kanban:") for l in issue["labels"])
    assert any(l["name"] == "bug" for l in issue["labels"])  # user's label kept


def test_label_sync_can_be_disabled(client, fake_gitea, monkeypatch):
    monkeypatch.setenv("KANBANTE_SYNC_LABELS", "0")
    r = _move(client, 1, "in_progress", "backlog")
    assert r.json()["label_sync"] == "skipped"
    assert fake_gitea.label_ops == []


# ── Boards par milestone ───────────────────────────────────

def test_board_filters_by_milestone_with_progress(client):
    models.move_issue("Projets/demo", 3, "done")
    r = client.get("/board/Projets/demo?milestone=Sprint%201")
    assert r.status_code == 200
    assert "Tâche un" in r.text          # open, in Sprint 1
    assert "Tâche focus" not in r.text    # open, no milestone
    assert "milestone-banner" in r.text
    assert "1/2 issues terminées" in r.text
    assert "50%" in r.text
    assert "2026-10-20" in r.text         # due_on from the milestone


def test_board_lists_milestone_chips(client):
    r = client.get("/board/Projets/demo")
    assert "milestone-chip" in r.text
    assert "Sprint 1" in r.text
    assert "milestone=Sprint" in r.text


def test_board_filter_by_milestone_id(client):
    r = client.get("/board/Projets/demo?milestone=1")
    assert "Tâche un" in r.text
    assert "Tâche focus" not in r.text


# ── Refonte UI : marqueurs structurels ─────────────────────

def test_dashboard_uses_new_design_system(client):
    r = client.get("/")
    assert "app-header" in r.text
    assert "nav-pill" in r.text
    # En-tête refondu en 0.20.0 : bande KPI unifiée (les stat-tile
    # ne servent plus qu'au briefing).
    assert "kpi-strip" in r.text
    assert "live.js" in r.text