mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-05-19 07:54:10 +00:00
77 lines
3.4 KiB
Python
77 lines
3.4 KiB
Python
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import patch
|
|
|
|
import agent_preflight
|
|
|
|
|
|
class AgentPreflightTest(unittest.TestCase):
|
|
def test_audit_preflight_passes_when_branch_and_active_claim_match(self) -> None:
|
|
control_plane = {
|
|
"prerelease_branch": "pulse/v6-release",
|
|
"active_profile_id": "v6",
|
|
"active_target_id": "v6-rc-stabilization",
|
|
}
|
|
# Use a window around `now` so the claim stays active regardless of
|
|
# when the test is run. The fixture must remain valid as wall-clock
|
|
# time advances; a fixed historical window expires and silently
|
|
# breaks the active-claim split.
|
|
now = datetime.now(timezone.utc)
|
|
claimed_at = (now - timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
expires_at = (now + timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
status_payload = {
|
|
"work_claims": [
|
|
{
|
|
"id": "codex-lane-l13",
|
|
"agent_id": "codex",
|
|
"work_item": {"kind": "lane", "id": "L13"},
|
|
"claimed_at": claimed_at,
|
|
"expires_at": expires_at,
|
|
}
|
|
]
|
|
}
|
|
|
|
with patch.object(agent_preflight, "active_control_plane", return_value=control_plane), patch.object(
|
|
agent_preflight, "load_status_payload", return_value=status_payload
|
|
), patch.object(agent_preflight, "current_branch", return_value="pulse/v6-release"):
|
|
report = agent_preflight.audit_preflight(agent_id="codex", require_active_claim=True)
|
|
|
|
self.assertEqual(report["errors"], [])
|
|
self.assertEqual(report["active_claim_count"], 1)
|
|
self.assertEqual(report["agent_active_claim_count"], 1)
|
|
|
|
def test_audit_preflight_requires_active_claim_for_agent(self) -> None:
|
|
control_plane = {
|
|
"prerelease_branch": "pulse/v6-release",
|
|
"active_profile_id": "v6",
|
|
"active_target_id": "v6-rc-stabilization",
|
|
}
|
|
status_payload = {"work_claims": []}
|
|
|
|
with patch.object(agent_preflight, "active_control_plane", return_value=control_plane), patch.object(
|
|
agent_preflight, "load_status_payload", return_value=status_payload
|
|
), patch.object(agent_preflight, "current_branch", return_value="pulse/v6-release"):
|
|
report = agent_preflight.audit_preflight(agent_id="codex", require_active_claim=True)
|
|
|
|
self.assertTrue(report["errors"])
|
|
self.assertIn("expected exactly one active work claim", report["errors"][0])
|
|
|
|
def test_audit_preflight_flags_branch_mismatch(self) -> None:
|
|
control_plane = {
|
|
"prerelease_branch": "pulse/v6-release",
|
|
"active_profile_id": "v6",
|
|
"active_target_id": "v6-rc-stabilization",
|
|
}
|
|
status_payload = {"work_claims": []}
|
|
|
|
with patch.object(agent_preflight, "active_control_plane", return_value=control_plane), patch.object(
|
|
agent_preflight, "load_status_payload", return_value=status_payload
|
|
), patch.object(agent_preflight, "current_branch", return_value="main"):
|
|
report = agent_preflight.audit_preflight(agent_id="codex", require_active_claim=False)
|
|
|
|
self.assertTrue(report["errors"])
|
|
self.assertIn("does not match active prerelease branch", report["errors"][0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|