sysadminandClaude Opus 4.8
275a362a46
Structured claim/progress heartbeats post on issue claim and via gitea_post_heartbeat. Read-only gitea_reconcile_issue_claims inventories active, stale, phantom, and PR-backed claims; gitea_cleanup_stale_claims supports dry-run cleanup of reclaimable labels. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
"""Tests for issue claim heartbeat leases (#268)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import issue_claim_heartbeat as ich # noqa: E402
|
|
|
|
|
|
class TestHeartbeatFormatting(unittest.TestCase):
|
|
def test_format_and_parse_roundtrip(self):
|
|
body = ich.format_heartbeat_body(
|
|
kind="claim",
|
|
issue_number=268,
|
|
branch="feat/issue-268-heartbeat-leases",
|
|
phase="claimed",
|
|
profile="prgs-author",
|
|
next_action="implement module",
|
|
)
|
|
parsed = ich.parse_heartbeat_comment(body)
|
|
self.assertIsNotNone(parsed)
|
|
assert parsed is not None
|
|
self.assertEqual(parsed["issue_number"], 268)
|
|
self.assertEqual(parsed["branch"], "feat/issue-268-heartbeat-leases")
|
|
self.assertEqual(parsed["phase"], "claimed")
|
|
|
|
|
|
class TestClaimClassification(unittest.TestCase):
|
|
NOW = datetime(2026, 7, 7, 12, 0, tzinfo=timezone.utc)
|
|
|
|
def _comment(self, *, minutes_ago: int, kind: str = "progress") -> dict:
|
|
ts = (self.NOW - timedelta(minutes=minutes_ago)).isoformat()
|
|
body = ich.format_heartbeat_body(
|
|
kind=kind,
|
|
issue_number=268,
|
|
branch="feat/issue-268-heartbeat-leases",
|
|
phase="implementation",
|
|
profile="prgs-author",
|
|
)
|
|
return {"id": 1, "body": body, "created_at": ts, "updated_at": ts}
|
|
|
|
def test_active_claim_within_lease(self):
|
|
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
|
|
result = ich.classify_issue_claim(
|
|
issue=issue,
|
|
comments=[self._comment(minutes_ago=5)],
|
|
now=self.NOW,
|
|
)
|
|
self.assertEqual(result["status"], "active")
|
|
|
|
def test_stale_claim_without_branch(self):
|
|
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
|
|
result = ich.classify_issue_claim(
|
|
issue=issue,
|
|
comments=[self._comment(minutes_ago=45)],
|
|
now=self.NOW,
|
|
)
|
|
self.assertEqual(result["status"], "stale")
|
|
|
|
def test_reclaimable_after_sixty_minutes_without_branch(self):
|
|
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
|
|
result = ich.classify_issue_claim(
|
|
issue=issue,
|
|
comments=[self._comment(minutes_ago=90)],
|
|
now=self.NOW,
|
|
)
|
|
self.assertEqual(result["status"], "reclaimable")
|
|
|
|
def test_awaiting_review_when_open_pr_exists(self):
|
|
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
|
|
open_prs = [
|
|
{
|
|
"number": 400,
|
|
"title": "feat",
|
|
"body": "Closes #268",
|
|
"head": {"ref": "feat/issue-268-heartbeat-leases"},
|
|
}
|
|
]
|
|
result = ich.classify_issue_claim(
|
|
issue=issue,
|
|
comments=[self._comment(minutes_ago=120)],
|
|
open_prs=open_prs,
|
|
now=self.NOW,
|
|
)
|
|
self.assertEqual(result["status"], "awaiting_review")
|
|
|
|
def test_phantom_claim_without_heartbeat(self):
|
|
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
|
|
result = ich.classify_issue_claim(
|
|
issue=issue,
|
|
comments=[{"id": 1, "body": "working on it"}],
|
|
now=self.NOW,
|
|
)
|
|
self.assertEqual(result["status"], "phantom")
|
|
|
|
|
|
class TestInventory(unittest.TestCase):
|
|
def test_build_cleanup_plan_flags_phantom(self):
|
|
inventory = {
|
|
"entries": [
|
|
{"issue_number": 1, "status": "phantom", "reasons": ["no heartbeat"]},
|
|
{"issue_number": 2, "status": "active", "reasons": []},
|
|
]
|
|
}
|
|
plan = ich.build_cleanup_plan(inventory)
|
|
self.assertEqual(len(plan), 1)
|
|
self.assertEqual(plan[0]["issue_number"], 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |