forked from Gitlink/gitlink-cli
399 lines
16 KiB
Python
399 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from scripts.gitlink_workflow import (
|
|
apply_triage_plan,
|
|
build_triage_plan,
|
|
build_issue_comment_command,
|
|
classify_issue_with_skill_rules,
|
|
load_skill_triage_rules,
|
|
normalize_issues,
|
|
normalize_prs,
|
|
normalize_releases,
|
|
render_markdown_report,
|
|
render_release_notes,
|
|
run_triage_workflow,
|
|
summarize_workflow,
|
|
)
|
|
|
|
|
|
class WorkflowTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.now = datetime(2026, 5, 15, 12, 0, tzinfo=timezone.utc)
|
|
self.repo_info = {
|
|
"name": "forgeplus",
|
|
"description": "demo repo",
|
|
"default_branch": "master",
|
|
}
|
|
|
|
def write_skill(self, directory: str) -> Path:
|
|
ruleset = {
|
|
"version": 1,
|
|
"mode": "rule",
|
|
"skill": {"name": "gitlink-issue-triage-rules", "version": "9.9.9"},
|
|
"defaults": {"dry_run": True, "priority": "P3"},
|
|
"priority_ids": {"P0": 4, "P1": 3, "P2": 2, "P3": 1, "critical": 4, "high": 3, "normal": 2, "low": 1},
|
|
"labels_by_type": {
|
|
"bug": ["缺陷", "bug"],
|
|
"feature": ["功能", "enhancement"],
|
|
"question": ["疑问", "question"],
|
|
"docs": ["文档", "documentation"],
|
|
"security": ["缺陷", "security"],
|
|
},
|
|
"assigners_by_type": {
|
|
"bug": [153579],
|
|
"feature": [153579],
|
|
"question": [153579],
|
|
"docs": [153579],
|
|
"security": [153579],
|
|
},
|
|
"rules": [
|
|
{
|
|
"id": "security-sensitive",
|
|
"type": "security",
|
|
"label": ["缺陷", "security"],
|
|
"priority": "critical",
|
|
"match": {"any_keyword": ["token", "漏洞"]},
|
|
},
|
|
{
|
|
"id": "bug-crash",
|
|
"type": "bug",
|
|
"label": ["缺陷", "bug"],
|
|
"priority": "high",
|
|
"match": {"any_keyword": ["crash", "崩溃"]},
|
|
},
|
|
{
|
|
"id": "feature-request",
|
|
"type": "feature",
|
|
"label": ["功能", "enhancement"],
|
|
"priority": "normal",
|
|
"match": {"any_keyword": ["建议", "feature"]},
|
|
},
|
|
{
|
|
"id": "question-default",
|
|
"type": "question",
|
|
"label": ["疑问", "question"],
|
|
"priority": "low",
|
|
"match": {"any_keyword": ["请问", "how to"]},
|
|
},
|
|
{
|
|
"id": "docs-default",
|
|
"type": "docs",
|
|
"label": ["文档", "documentation"],
|
|
"priority": "low",
|
|
"match": {"any_keyword": ["README", "typo"]},
|
|
},
|
|
],
|
|
}
|
|
path = Path(directory) / "SKILL.md"
|
|
path.write_text(
|
|
"---\n"
|
|
"name: gitlink-issue-triage-rules\n"
|
|
"version: 9.9.9\n"
|
|
"---\n\n"
|
|
"<!-- TRIAGE_RULES_JSON_START -->\n"
|
|
"```json\n"
|
|
f"{json.dumps(ruleset, ensure_ascii=False, indent=2)}\n"
|
|
"```\n"
|
|
"<!-- TRIAGE_RULES_JSON_END -->\n",
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
def test_normalize_issue_payload(self) -> None:
|
|
payload = {
|
|
"data": {
|
|
"issues": [
|
|
{
|
|
"project_issues_index": 1,
|
|
"subject": "feat: add report",
|
|
"status_id": 1,
|
|
"status_name": "新增",
|
|
"updated_at": "2026-05-10T10:00:00Z",
|
|
"labels": [{"name": "enhancement"}],
|
|
}
|
|
]
|
|
}
|
|
}
|
|
issues = normalize_issues(payload)
|
|
self.assertEqual(len(issues), 1)
|
|
self.assertEqual(issues[0]["title"], "feat: add report")
|
|
self.assertEqual(issues[0]["labels"], ["enhancement"])
|
|
self.assertEqual(issues[0]["state"], "open")
|
|
|
|
def test_normalize_pr_payload(self) -> None:
|
|
payload = {
|
|
"data": {
|
|
"merge_requests": [
|
|
{
|
|
"pull_request_number": 10,
|
|
"title": "fix: bug",
|
|
"pull_request_status": 1,
|
|
"merged_at": "2026-05-14T10:00:00Z",
|
|
}
|
|
]
|
|
}
|
|
}
|
|
prs = normalize_prs(payload)
|
|
self.assertEqual(len(prs), 1)
|
|
self.assertTrue(prs[0]["merged"])
|
|
self.assertEqual(prs[0]["state"], "merged")
|
|
|
|
def test_normalize_release_payload(self) -> None:
|
|
payload = {"data": {"releases": [{"id": 5, "name": "v1.0.0"}]}}
|
|
releases = normalize_releases(payload)
|
|
self.assertEqual(len(releases), 1)
|
|
self.assertEqual(releases[0]["title"], "v1.0.0")
|
|
|
|
def test_summary_and_report(self) -> None:
|
|
issues = [
|
|
{
|
|
"id": "1",
|
|
"title": "feat: add report",
|
|
"state": "open",
|
|
"created_at": datetime(2026, 5, 5, 12, 0, tzinfo=timezone.utc),
|
|
"updated_at": datetime(2026, 5, 10, 12, 0, tzinfo=timezone.utc),
|
|
"labels": ["enhancement"],
|
|
},
|
|
{
|
|
"id": "2",
|
|
"title": "fix: stale issue",
|
|
"state": "open",
|
|
"created_at": datetime(2026, 4, 20, 12, 0, tzinfo=timezone.utc),
|
|
"updated_at": datetime(2026, 5, 1, 12, 0, tzinfo=timezone.utc),
|
|
"labels": ["bug"],
|
|
},
|
|
]
|
|
prs = [
|
|
{
|
|
"id": "10",
|
|
"title": "feat: workflow",
|
|
"state": "merged",
|
|
"created_at": datetime(2026, 5, 12, 12, 0, tzinfo=timezone.utc),
|
|
"updated_at": datetime(2026, 5, 14, 12, 0, tzinfo=timezone.utc),
|
|
"merged_at": datetime(2026, 5, 14, 12, 0, tzinfo=timezone.utc),
|
|
"merged": True,
|
|
"labels": [],
|
|
},
|
|
{
|
|
"id": "11",
|
|
"title": "chore: cleanup",
|
|
"state": "open",
|
|
"created_at": datetime(2026, 5, 1, 12, 0, tzinfo=timezone.utc),
|
|
"updated_at": datetime(2026, 5, 2, 12, 0, tzinfo=timezone.utc),
|
|
"merged_at": None,
|
|
"merged": False,
|
|
"labels": [],
|
|
},
|
|
]
|
|
releases = [{"id": "1", "title": "v1.0.0", "created_at": datetime(2026, 5, 14, 12, 0, tzinfo=timezone.utc)}]
|
|
summary = summarize_workflow(self.repo_info, issues, prs, releases, self.now, 7)
|
|
report = render_markdown_report(summary)
|
|
self.assertIn("# forgeplus 自动化周报", report)
|
|
self.assertIn("Issues 总数", report)
|
|
self.assertIn("超窗 Issue", report)
|
|
self.assertIn("feature", report)
|
|
release_notes = render_release_notes(summary)
|
|
self.assertIn("Release Notes", release_notes)
|
|
self.assertIn("变更分类", release_notes)
|
|
self.assertEqual(summary["counts"]["issues_stale"], 1)
|
|
self.assertEqual(summary["counts"]["prs_merged"], 1)
|
|
self.assertIn("feature", summary["pr_buckets"])
|
|
|
|
def test_issue_comment_command_uses_number_flag(self) -> None:
|
|
command = build_issue_comment_command(2, "demo")
|
|
self.assertEqual(command, ["issue", "+comment", "--number", "2", "--body", "demo"])
|
|
self.assertNotIn("-i", command)
|
|
|
|
def test_load_skill_triage_rules_extracts_json_block(self) -> None:
|
|
with TemporaryDirectory() as tmp:
|
|
skill_path = self.write_skill(tmp)
|
|
ruleset = load_skill_triage_rules(skill_path)
|
|
|
|
self.assertEqual(ruleset["skill"]["name"], "gitlink-issue-triage-rules")
|
|
self.assertEqual(ruleset["skill"]["version"], "9.9.9")
|
|
self.assertEqual(len(ruleset["rules"]), 5)
|
|
|
|
def test_classify_issue_with_skill_rules_matches_common_types(self) -> None:
|
|
with TemporaryDirectory() as tmp:
|
|
ruleset = load_skill_triage_rules(self.write_skill(tmp))
|
|
|
|
cases = [
|
|
("token 泄露漏洞", "security", "security-sensitive"),
|
|
("CLI crash when upload", "bug", "bug-crash"),
|
|
("建议新增报表导出", "feature", "feature-request"),
|
|
("请问 how to 配置项目", "question", "question-default"),
|
|
("README typo", "docs", "docs-default"),
|
|
]
|
|
for title, expected_type, expected_rule in cases:
|
|
result = classify_issue_with_skill_rules(
|
|
{"id": "1", "title": title, "labels": [], "raw": {"description": title}},
|
|
ruleset,
|
|
)
|
|
self.assertEqual(result["detected_type"], expected_type)
|
|
self.assertEqual(result["rule_id"], expected_rule)
|
|
|
|
def test_classify_issue_with_skill_rules_marks_unknown(self) -> None:
|
|
with TemporaryDirectory() as tmp:
|
|
ruleset = load_skill_triage_rules(self.write_skill(tmp))
|
|
|
|
result = classify_issue_with_skill_rules(
|
|
{"id": "9", "title": "General note", "labels": [], "raw": {"description": "nothing special"}},
|
|
ruleset,
|
|
)
|
|
|
|
self.assertEqual(result["detected_type"], "unknown")
|
|
self.assertEqual(result["source"], "skill-json")
|
|
|
|
def test_build_triage_plan_resolves_label_and_assigner(self) -> None:
|
|
plan = build_triage_plan(
|
|
{
|
|
"results": [
|
|
{
|
|
"issue": {"number": 1, "title": "CLI crash"},
|
|
"detected_type": "bug",
|
|
"priority": "P1",
|
|
"confidence": 88,
|
|
"matched_rules": ["matched keyword: crash"],
|
|
}
|
|
]
|
|
},
|
|
{"data": {"issue_tags": [{"id": 10, "name": "缺陷"}]}},
|
|
{"data": {"assigners": [{"id": 153579, "login": "Angel123456", "name": "Angel"}]}},
|
|
{
|
|
"labels_by_type": {"bug": "缺陷"},
|
|
"assigners_by_type": {"bug": ["Angel123456"]},
|
|
"priority_ids": {"P1": 3},
|
|
},
|
|
owner="owner",
|
|
repo="repo",
|
|
)
|
|
|
|
self.assertEqual(plan["analyzed"], 1)
|
|
item = plan["items"][0]
|
|
self.assertEqual(item["status"], "planned")
|
|
self.assertEqual(item["label_id"], 10)
|
|
self.assertEqual(item["assigner_ids"], [153579])
|
|
self.assertEqual(item["priority_id"], 3)
|
|
|
|
def test_build_triage_plan_warns_without_interrupting(self) -> None:
|
|
plan = build_triage_plan(
|
|
{
|
|
"results": [
|
|
{
|
|
"issue": {"number": 2, "title": "General note"},
|
|
"detected_type": "unknown",
|
|
"priority": "P3",
|
|
},
|
|
{
|
|
"issue": {"number": 3, "title": "Slow API"},
|
|
"detected_type": "performance",
|
|
"priority": "P2",
|
|
},
|
|
]
|
|
},
|
|
{"data": {"issue_tags": []}},
|
|
{"data": {"assigners": []}},
|
|
{
|
|
"labels_by_type": {"performance": "缺陷"},
|
|
"assigners_by_type": {"performance": ["missing-user"]},
|
|
"priority_ids": {"P2": 2, "P3": 1},
|
|
},
|
|
)
|
|
|
|
self.assertEqual(plan["items"][0]["status"], "skipped")
|
|
self.assertEqual(plan["items"][1]["status"], "planned")
|
|
self.assertGreaterEqual(len(plan["warnings"]), 3)
|
|
|
|
def test_apply_triage_plan_merges_existing_metadata_and_comments(self) -> None:
|
|
plan = build_triage_plan(
|
|
{
|
|
"results": [
|
|
{
|
|
"issue": {"number": 4, "title": "Bug: upload crash"},
|
|
"detected_type": "bug",
|
|
"priority": "P1",
|
|
"confidence": 91,
|
|
}
|
|
]
|
|
},
|
|
{"data": {"issue_tags": [{"id": 10, "name": "缺陷"}]}},
|
|
{"data": {"assigners": [{"id": 153579, "login": "Angel123456"}]}},
|
|
{
|
|
"labels_by_type": {"bug": "缺陷"},
|
|
"assigners_by_type": {"bug": [153579]},
|
|
"priority_ids": {"P1": 3},
|
|
},
|
|
)
|
|
calls: list[list[str]] = []
|
|
|
|
def fake_runner(command: list[str], owner: str, repo: str, cli_bin: str | None = None):
|
|
calls.append(command)
|
|
if command[:2] == ["issue", "+view"]:
|
|
return {
|
|
"data": {
|
|
"subject": "Bug: upload crash",
|
|
"description": "existing body",
|
|
"tags": [{"id": 99, "name": "已有"}],
|
|
"assigners": [{"id": 88, "login": "maintainer"}],
|
|
}
|
|
}
|
|
return {"ok": True}
|
|
|
|
applied = apply_triage_plan(plan, "owner", "repo", fake_runner, comment=True)
|
|
self.assertEqual(applied["items"][0]["status"], "applied")
|
|
self.assertEqual(calls[0][:2], ["issue", "+view"])
|
|
self.assertEqual(calls[1][:2], ["issue", "+update"])
|
|
self.assertEqual(calls[2][:2], ["issue", "+comment"])
|
|
self.assertIn("--tag-ids", calls[1])
|
|
self.assertIn("99,10", calls[1])
|
|
self.assertIn("--assigner-ids", calls[1])
|
|
self.assertIn("88,153579", calls[1])
|
|
self.assertIn("--priority-id", calls[1])
|
|
self.assertIn("3", calls[1])
|
|
|
|
def test_run_triage_workflow_dry_run_does_not_update_issue(self) -> None:
|
|
calls: list[list[str]] = []
|
|
|
|
def fake_runner(command: list[str], owner: str, repo: str, cli_bin: str | None = None):
|
|
calls.append(command)
|
|
if command[:2] == ["issue", "+list"]:
|
|
return {"data": {"issues": [{"project_issues_index": 1, "subject": "CLI crash", "description": "crash"}]}}
|
|
if command[:2] == ["label", "+list"]:
|
|
return {"data": {"issue_tags": [{"id": 10, "name": "缺陷"}]}}
|
|
if command[:2] == ["issue", "+assigners"]:
|
|
return {"data": {"assigners": [{"id": 153579, "login": "Angel123456"}]}}
|
|
raise AssertionError(f"unexpected command in dry-run: {command}")
|
|
|
|
with TemporaryDirectory() as tmp:
|
|
plan, markdown_path, json_path = run_triage_workflow(
|
|
"owner",
|
|
"repo",
|
|
{
|
|
"labels_by_type": {"bug": "缺陷"},
|
|
"assigners_by_type": {"bug": [153579]},
|
|
"priority_ids": {"P1": 3},
|
|
},
|
|
apply_triage=False,
|
|
output_dir=Path(tmp),
|
|
base_name="demo",
|
|
triage_skill=self.write_skill(tmp),
|
|
runner=fake_runner,
|
|
)
|
|
|
|
self.assertTrue(plan["dry_run"])
|
|
self.assertEqual(plan["source"], "skill-json")
|
|
self.assertEqual(plan["items"][0]["rule_id"], "bug-crash")
|
|
self.assertTrue(markdown_path.name.endswith("_triage_plan.md"))
|
|
self.assertTrue(json_path.name.endswith("_triage_plan.json"))
|
|
self.assertNotIn(["issue", "+update"], [call[:2] for call in calls])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|