From 370dafd2662fb325199f3b2123c3b11f61c3842f Mon Sep 17 00:00:00 2001 From: Spark Date: Sun, 11 Oct 2026 14:21:11 -0400 Subject: [PATCH] =?UTF-8?q?test(0.13.0):=20code=20li=C3=A9,=20webhook=20PR?= =?UTF-8?q?,=20standup,=20r=C3=A8gles=20=E2=80=94=2010=20tests,=20157=20au?= =?UTF-8?q?=20total?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_palier10.py | 202 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 202 insertions(+) create mode 100644 tests/test_palier10.py diff --git a/tests/test_palier10.py b/tests/test_palier10.py new file mode 100644 index 0000000..eb90bb1 --- /dev/null +++ b/tests/test_palier10.py @@ -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