205 lines
7.4 KiB
Python
205 lines
7.4 KiB
Python
"""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
|