operator
122 lines
5.1 KiB
Python
122 lines
5.1 KiB
Python
"""test_box_readback_loopback.py — Unit tests for Readback Gate and True Loopback Engine."""
|
|
|
|
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
from datetime import datetime, timezone, timedelta
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "bin"))
|
|
import box_readback_loopback as brl
|
|
|
|
|
|
class TestReadbackValidation(unittest.TestCase):
|
|
|
|
def test_strict_tag_match(self):
|
|
text = "[READBACK] Ticket #219 | Branch: dev/pip/219-work | Plan: run pytest"
|
|
valid, excerpt = brl.validate_readback(text, 219, "pip")
|
|
self.assertTrue(valid)
|
|
self.assertIn("Ticket #219", excerpt)
|
|
|
|
def test_semantic_fallback_match(self):
|
|
text = "Understood. I am working on ticket 219 on dev/pip/219-work."
|
|
valid, excerpt = brl.validate_readback(text, 219, "pip")
|
|
self.assertTrue(valid)
|
|
self.assertIn("219", excerpt)
|
|
|
|
def test_irrelevant_message_rejected(self):
|
|
text = "Heartbeat check-in completed, all systems green."
|
|
valid, excerpt = brl.validate_readback(text, 219, "pip")
|
|
self.assertFalse(valid)
|
|
self.assertEqual(excerpt, "")
|
|
|
|
def test_wrong_ticket_rejected(self):
|
|
text = "[READBACK] Ticket #218 | Branch: dev/pip/218-work"
|
|
valid, excerpt = brl.validate_readback(text, 219, "pip")
|
|
self.assertFalse(valid)
|
|
|
|
|
|
class TestLoopbackEscalation(unittest.TestCase):
|
|
|
|
@patch("box_readback_loopback.acp.get_passive_cognitive_state")
|
|
@patch("box_readback_loopback.gitea_api")
|
|
def test_loopback_silence_when_active(self, mock_gitea, mock_cog):
|
|
# 5 minutes inactive -> silence
|
|
now = datetime.now(timezone.utc)
|
|
recent = (now - timedelta(minutes=5)).isoformat()
|
|
mock_gitea.return_value = [
|
|
{"number": 219, "title": "Test", "created_at": recent, "assignee": {"username": "dev"}}
|
|
]
|
|
mock_cog.return_value = {"status": "IDLE", "cognitive_lock": False}
|
|
|
|
actions = brl.run_loopback_sweep(dry_run=True, verbose=False)
|
|
self.assertEqual(len(actions), 0)
|
|
|
|
@patch("box_readback_loopback.acp.get_passive_cognitive_state")
|
|
@patch("box_readback_loopback.gitea_api")
|
|
def test_loopback_tier1_after_15m_when_idle(self, mock_gitea, mock_cog):
|
|
# 20 minutes inactive -> Tier 1
|
|
now = datetime.now(timezone.utc)
|
|
old = (now - timedelta(minutes=20)).isoformat()
|
|
mock_gitea.side_effect = lambda ep, **kwargs: (
|
|
[{"number": 999, "title": "Stalled", "created_at": old, "assignee": {"username": "dev"}}]
|
|
if ep == "/repos/super/box/issues?state=open"
|
|
else []
|
|
)
|
|
mock_cog.return_value = {"status": "IDLE", "cognitive_lock": False}
|
|
|
|
actions = brl.run_loopback_sweep(dry_run=True, verbose=False)
|
|
self.assertTrue(any("Tier 1" in a for a in actions))
|
|
|
|
@patch("box_readback_loopback.acp.get_passive_cognitive_state")
|
|
@patch("box_readback_loopback.gitea_api")
|
|
def test_loopback_defers_when_thinking(self, mock_gitea, mock_cog):
|
|
# 25 minutes inactive, but agent is THINKING -> defer
|
|
now = datetime.now(timezone.utc)
|
|
old = (now - timedelta(minutes=25)).isoformat()
|
|
mock_gitea.side_effect = lambda ep, **kwargs: (
|
|
[{"number": 999, "title": "Stalled", "created_at": old, "assignee": {"username": "dev"}}]
|
|
if ep == "/repos/super/box/issues?state=open"
|
|
else []
|
|
)
|
|
mock_cog.return_value = {"status": "THINKING", "cognitive_lock": True}
|
|
|
|
actions = brl.run_loopback_sweep(dry_run=True, verbose=False)
|
|
self.assertEqual(len(actions), 0)
|
|
|
|
@patch("box_readback_loopback.acp.get_passive_cognitive_state")
|
|
@patch("box_readback_loopback.gitea_api")
|
|
def test_loopback_tier2_after_45m(self, mock_gitea, mock_cog):
|
|
# 50 minutes inactive -> Tier 2
|
|
now = datetime.now(timezone.utc)
|
|
old = (now - timedelta(minutes=50)).isoformat()
|
|
mock_gitea.side_effect = lambda ep, **kwargs: (
|
|
[{"number": 999, "title": "Stalled", "created_at": old, "assignee": {"username": "dev"}}]
|
|
if ep == "/repos/super/box/issues?state=open"
|
|
else []
|
|
)
|
|
mock_cog.return_value = {"status": "IDLE", "cognitive_lock": False}
|
|
|
|
actions = brl.run_loopback_sweep(dry_run=True, verbose=False)
|
|
self.assertTrue(any("Tier 2" in a for a in actions))
|
|
|
|
@patch("box_readback_loopback.acp.get_passive_cognitive_state")
|
|
@patch("box_readback_loopback.gitea_api")
|
|
def test_loopback_tier3_after_90m(self, mock_gitea, mock_cog):
|
|
# 95 minutes inactive -> Tier 3
|
|
now = datetime.now(timezone.utc)
|
|
old = (now - timedelta(minutes=95)).isoformat()
|
|
mock_gitea.side_effect = lambda ep, **kwargs: (
|
|
[{"number": 999, "title": "Stalled", "created_at": old, "assignee": {"username": "dev"}}]
|
|
if ep == "/repos/super/box/issues?state=open"
|
|
else []
|
|
)
|
|
mock_cog.return_value = {"status": "IDLE", "cognitive_lock": False}
|
|
|
|
actions = brl.run_loopback_sweep(dry_run=True, verbose=False)
|
|
self.assertTrue(any("Tier 3" in a for a in actions))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|