test(0.13.0): code lié, webhook PR, standup, règles — 10 tests, 157 au total

2026-10-11 14:21:11 -04:00
parent 7eef8bb604
commit 370dafd266
1 changed files with 202 additions and 0 deletions
@@ -0,0 +1,202 @@
+"""Palier 10 (0.13.0): code lié (#1), PR fusionnée (#2),
+standup automatique (#6), règles d'automatisation (#8)."""
+
+import json
+from datetime import datetime, timedelta
+
+from app import code_links as cl
+from app import models
+from app import rules as rules_mod
+from tests.conftest import csrf_from
+
+
+# ── #1 Code lié : fonctions pures ─────────────────────────────────
+
+
+def test_extract_refs_and_branch_numbers():
+    assert cl.extract_issue_refs("Fixes #12 et rapport #3, pas #x") == {12, 3}
+    assert cl.extract_issue_refs("a#b #12") == {12}
+    assert cl.extract_issue_refs(None) == set()
+    assert cl.branch_issue_number("feature/12-login") == 12
+    assert cl.branch_issue_number("issue-7") == 7
+    assert cl.branch_issue_number("42") == 42
+    assert cl.branch_issue_number("release-2026") == 2026
+    assert cl.branch_issue_number("main") is None
+    assert cl.branch_issue_number("hotfix/no-number") is None
+
+
+def test_code_endpoint(client, fake_gitea):
+    r = client.get("/api/issues/Projets/demo/1/code")
+    assert r.status_code == 200
+    data = r.json()
+    assert data["branches"] == ["feature/1-tache-un"]
+    assert [p["number"] for p in data["pulls"]] == [16]
+    assert [c["sha"] for c in data["commits"]] == ["abcdef1"]
+    assert data["summary"]["pr"]["label"] == "ouverte"
+    board = client.get("/board/Projets/demo")
+    assert "PR #16 ouverte" in board.text
+    assert "1 commit" in board.text
+
+
+# ── #2 PR fusionnée → carte Terminé ───────────────────────────────
+
+
+def _pr_payload(merged=True, title="Corrige tout", body="Fixes #1",
+                ref="feature/1-tache-un"):
+    return {
+        "action": "closed",
+        "repository": {"full_name": "Projets/demo"},
+        "pull_request": {
+            "number": 16, "title": title, "body": body,
+            "merged": merged, "head": {"ref": ref},
+        },
+    }
+
+
+def _post_webhook(client, payload, event="pull_request"):
+    return client.post(
+        "/webhooks/gitea", json=payload, headers={"X-Gitea-Event": event}
+    )
+
+
+def test_pr_merged_moves_card_and_closes_issue(client, fake_gitea):
+    token = csrf_from(client)
+    client.post("/api/move", data={
+        "repo_full_name": "Projets/demo", "issue_number": 1,
+        "from_column": "backlog", "to_column": "in_progress",
+        "to_order": "1", "csrf_token": token,
+    })
+    r = _post_webhook(client, _pr_payload())
+    assert r.status_code == 200
+    assert r.json()["handled"] == "pr-merged:#1:closed"
+    assert ("Projets", "demo", 1, "closed") in fake_gitea.state_changes
+    board = models.get_board_state("Projets/demo")
+    assert 1 in [c["issue_number"] for c in board["done"]]
+    events = models.get_events(repo_full_name="Projets/demo")
+    assert any(
+        e["event_type"] == "closed" and e["source"] == "webhook"
+        for e in events
+    )
+
+
+def test_pr_merged_by_branch_only_and_idempotent(client, fake_gitea):
+    payload = _pr_payload(title="Sans référence", body="", ref="issue-5-focus")
+    r = _post_webhook(client, payload)
+    assert "#5:closed" in r.json()["handled"]
+    r2 = _post_webhook(client, payload)  # echo: no double journal entry
+    closed_events = [
+        e for e in models.get_events(repo_full_name="Projets/demo")
+        if e["event_type"] == "closed" and e["issue_number"] == 5
+    ]
+    assert len(closed_events) == 1
+    assert r2.json()["handled"] in ("pr-merged:#5:noop", "pr-merged:#5:closed")
+
+
+def test_pr_not_merged_ignored(client, fake_gitea):
+    r = _post_webhook(client, _pr_payload(merged=False))
+    assert r.json()["handled"] == "ignored"
+    assert models.get_events(repo_full_name="Projets/demo") == []
+
+
+# ── #6 Standup ────────────────────────────────────────────────────
+
+
+def test_standup_from_journal(client, fake_gitea):
+    yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S")
+    models.log_event("Projets/demo", 3, "closed", to_column="done")
+    with models.get_db() as conn:
+        conn.execute(
+            "UPDATE board_events SET created_at = ? WHERE issue_number = 3",
+            (yesterday,),
+        )
+    r = client.get("/api/standup")
+    assert r.status_code == 200
+    text = r.json()["text"]
+    assert "Standup" in text
+    assert "1 fermée(s)" in text
+    assert "Ancienne tâche" in text
+    page = client.get("/standup")
+    assert page.status_code == 200
+    assert "Copier le standup" in page.text
+
+
+# ── #8 Règles ─────────────────────────────────────────────────────
+
+
+def test_parse_rules_validation():
+    rules, errors = rules_mod.parse_rules(
+        '[{"name": "ok", "when": {"column": "review"}, "then": {"notify": true}}]'
+    )
+    assert errors == [] and rules[0]["name"] == "ok"
+    _r, errors = rules_mod.parse_rules("{pas json")
+    assert errors and "JSON invalide" in errors[0]
+    _r, errors = rules_mod.parse_rules(
+        '[{"when": {"column": "review"}, "then": {"close_issue": true}}]'
+    )
+    assert errors and "interdit" in errors[0]
+    _r, errors = rules_mod.parse_rules(
+        '[{"when": {"column": "nope"}, "then": {"notify": true}}]'
+    )
+    assert errors and "colonne inconnue" in errors[0]
+
+
+def test_matching_cards_conditions():
+    columns = {
+        "review": [
+            {"number": 1, "labels": [{"name": "bloqué"}]},
+            {"number": 2, "labels": []},
+        ],
+        "in_progress": [{"number": 3, "labels": []}],
+    }
+    ages = {1: {"days": 5}, 2: {"days": 1}, 3: {"days": 9}}
+    rule = {"name": "r", "when": {"column": "review", "min_age_days": 3},
+            "then": {"notify": True}}
+    assert [i["number"] for i, _ in rules_mod.matching_cards(rule, columns, ages)] == [1]
+    rule2 = {"name": "r2", "when": {"column": "review", "label": "bloqué"},
+             "then": {"notify": True}}
+    assert [i["number"] for i, _ in rules_mod.matching_cards(rule2, columns, ages)] == [1]
+    rule3 = {"name": "r3", "when": {"column": "review", "wip_over": 1},
+             "then": {"notify": True}}
+    assert len(rules_mod.matching_cards(rule3, columns, ages)) == 2
+    rule4 = {"name": "r4", "when": {"column": "in_progress", "wip_over": 3},
+             "then": {"notify": True}}
+    assert rules_mod.matching_cards(rule4, columns, ages) == []
+
+
+def test_rules_run_end_to_end(client, fake_gitea, monkeypatch):
+    monkeypatch.setenv(
+        "KANBANTE_RULES",
+        json.dumps([{
+            "name": "Marque le focus",
+            "when": {"column": "in_progress"},
+            "then": {"add_label": "urgent"},
+        }]),
+    )
+    token = csrf_from(client)
+    client.post("/api/move", data={
+        "repo_full_name": "Projets/demo", "issue_number": 5,
+        "from_column": "backlog", "to_column": "in_progress",
+        "to_order": "5", "csrf_token": token,
+    })
+    r = client.post("/api/rules/run", data={"csrf_token": token})
+    assert r.status_code == 200
+    actions = r.json()["summary"]["actions"]
+    assert len(actions) == 1
+    assert actions[0]["number"] == 5
+    assert "urgent" in actions[0]["did"][0]
+    detail = client.get("/api/issues/Projets/demo/5")
+    assert "urgent" in [l["name"] for l in detail.json()["issue"]["labels"]]
+    events = models.get_events(repo_full_name="Projets/demo")
+    assert any(e["event_type"] == "rule" for e in events)
+    # Second run: label already present -> no new action.
+    r2 = client.post("/api/rules/run", data={"csrf_token": token})
+    assert r2.json()["summary"]["actions"] == []
+    page = client.get("/rules")
+    assert page.status_code == 200
+    assert "Marque le focus" in page.text
+
+
+def test_rules_page_unconfigured(client, fake_gitea):
+    page = client.get("/rules")
+    assert page.status_code == 200
+    assert "KANBANTE_RULES" in page.text