Spaces:
Sleeping
Sleeping
| import asyncio | |
| import unittest | |
| from unittest.mock import AsyncMock, MagicMock, patch | |
| from bot import autonomy_runtime | |
| class TestAutonomyRuntime(unittest.IsolatedAsyncioTestCase): | |
| async def test_bind_unbind(self): | |
| runtime = autonomy_runtime.AutonomyRuntime() | |
| loop = asyncio.get_running_loop() | |
| with patch.object(runtime, "_ensure_loop_task") as mock_ensure: | |
| runtime.bind(loop=loop, mode="irc") | |
| self.assertEqual(runtime._mode, "irc") | |
| runtime.unbind() | |
| self.assertIsNone(runtime._loop) | |
| self.assertIsNone(runtime._loop_task) | |
| async def test_run_tick(self, mock_process, mock_cp): | |
| runtime = autonomy_runtime.AutonomyRuntime() | |
| mock_cp.consume_due_goals.return_value = ["goal1"] | |
| mock_cp.get_config.return_value = {"twitch_channel_login": "default"} | |
| mock_process.return_value = {"status": "success"} | |
| result = await runtime._run_tick(force=True, reason="test") | |
| self.assertTrue(result["ok"]) | |
| self.assertEqual(result["due_goals"], 1) | |
| mock_cp.register_tick.assert_called_with(reason="test") | |
| # Deve agora incluir o channel_id | |
| mock_process.assert_called_once_with("goal1", None, channel_id="default") | |
| async def test_run_tick_registers_goal_session_kpi(self, mock_process, mock_cp, mock_obs): | |
| runtime = autonomy_runtime.AutonomyRuntime() | |
| mock_cp.consume_due_goals.return_value = [{"id": "g1", "risk": "suggest_streamer"}] | |
| mock_cp.get_config.return_value = {"twitch_channel_login": "default"} | |
| mock_cp.register_goal_session_result.return_value = {"met": True} | |
| mock_process.return_value = { | |
| "goal_id": "g1", | |
| "risk": "suggest_streamer", | |
| "outcome": "queued", | |
| } | |
| result = await runtime._run_tick(force=True, reason="test") | |
| mock_cp.register_goal_session_result.assert_called_once_with( | |
| goal_id="g1", | |
| outcome="queued", | |
| ) | |
| self.assertTrue(result["processed"][0]["session_kpi"]["met"]) | |
| mock_obs.record_autonomy_goal.assert_called_once() | |
| async def test_heartbeat_loop(self, mock_sleep, mock_cp): | |
| runtime = autonomy_runtime.AutonomyRuntime() | |
| loop = asyncio.get_running_loop() | |
| runtime.bind(loop=loop, mode="test") | |
| # Mock _run_tick to stop the loop after one call | |
| with patch.object(runtime, "_run_tick", new_callable=AsyncMock) as mock_tick: | |
| def side_effect(*args, **kwargs): | |
| runtime.unbind() | |
| return {"ok": True} | |
| mock_tick.side_effect = side_effect | |
| await runtime._heartbeat_loop() | |
| mock_tick.assert_called_once() | |
| mock_cp.set_loop_running.assert_called() | |
| async def test_run_tick_with_sentiment_triggers(self, mock_process, mock_sent, mock_cp): | |
| runtime = autonomy_runtime.AutonomyRuntime() | |
| mock_cp.consume_due_goals.return_value = [] | |
| mock_cp.get_config.return_value = {"twitch_channel_login": "default"} | |
| mock_sent.should_trigger_anti_boredom.return_value = True | |
| mock_sent.should_trigger_anti_confusion.return_value = True | |
| mock_process.return_value = {"status": "success"} | |
| result = await runtime._run_tick(force=False, reason="test") | |
| self.assertTrue(result["ok"]) | |
| self.assertEqual(len(result["processed"]), 2) | |
| # Verifica chamadas com channel_id | |
| mock_sent.should_trigger_anti_boredom.assert_called_with("default") | |
| mock_sent.should_trigger_anti_confusion.assert_called_with("default") | |
| # Verify call arguments | |
| calls = [call[0][0] for call in mock_process.call_args_list] | |
| self.assertTrue(any(g["id"] == "dynamic-anti-boredom" for g in calls)) | |
| self.assertTrue(any(g["id"] == "dynamic-anti-confusion" for g in calls)) | |
| async def test_run_tick_skips_sentiment_on_force(self, mock_process, mock_sent, mock_cp): | |
| runtime = autonomy_runtime.AutonomyRuntime() | |
| mock_cp.consume_due_goals.return_value = [] | |
| mock_cp.get_config.return_value = {"twitch_channel_login": "default"} | |
| mock_sent.should_trigger_anti_boredom.return_value = True | |
| await runtime._run_tick(force=True, reason="test") | |
| # Should NOT have added dynamic goals because force=True | |
| mock_process.assert_not_called() | |
| def test_run_manual_tick_no_loop(self, mock_run): | |
| runtime = autonomy_runtime.AutonomyRuntime() | |
| runtime._loop = None # Ensure no loop | |
| def run_side_effect(coroutine): | |
| coroutine.close() | |
| return {"ok": True} | |
| mock_run.side_effect = run_side_effect | |
| result = runtime.run_manual_tick(force=True) | |
| self.assertTrue(result["ok"]) | |
| mock_run.assert_called_once() | |
| def test_run_manual_tick_threadsafe(self, mock_run): | |
| runtime = autonomy_runtime.AutonomyRuntime() | |
| loop = MagicMock() | |
| loop.is_closed.return_value = False | |
| loop.is_running.return_value = True | |
| runtime._loop = loop | |
| mock_future = MagicMock() | |
| mock_future.result.return_value = {"ok": True} | |
| def submit_side_effect(coroutine, _loop): | |
| coroutine.close() | |
| return mock_future | |
| mock_run.side_effect = submit_side_effect | |
| result = runtime.run_manual_tick(force=True) | |
| self.assertTrue(result["ok"]) | |
| mock_run.assert_called_once() | |