203 lines
7.8 KiB
Python
203 lines
7.8 KiB
Python
"""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
|