From 11357ffdb41778b707d59979d6fdefb2b31ae884 Mon Sep 17 00:00:00 2001 From: Kimi Agent Date: Wed, 18 Mar 2026 20:54:02 -0400 Subject: [PATCH] test: add comprehensive unit tests for agentic_loop.py (#345) Co-authored-by: Kimi Agent Co-committed-by: Kimi Agent --- tests/test_agentic_loop.py | 236 ++++++++++++++++++++++++++++++++++++- 1 file changed, 233 insertions(+), 3 deletions(-) diff --git a/tests/test_agentic_loop.py b/tests/test_agentic_loop.py index 929252bf..5600c1d9 100644 --- a/tests/test_agentic_loop.py +++ b/tests/test_agentic_loop.py @@ -1,14 +1,22 @@ """Unit tests for the agentic loop module. -Tests cover planning, execution, max_steps enforcement, failure -adaptation, progress callbacks, and response cleaning. +Tests cover data structures, plan parsing, planning, execution, +max_steps enforcement, failure adaptation, double-failure, +progress callbacks, broadcast helper, summary logic, and +response cleaning. """ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from timmy.agentic_loop import _parse_steps, run_agentic_loop +from timmy.agentic_loop import ( + AgenticResult, + AgenticStep, + _broadcast_progress, + _parse_steps, + run_agentic_loop, +) # --------------------------------------------------------------------------- # Helpers @@ -27,6 +35,27 @@ def _mock_run(content: str): # --------------------------------------------------------------------------- +class TestDataStructures: + def test_agentic_step_fields(self): + step = AgenticStep( + step_num=1, description="Do X", result="Done", status="completed", duration_ms=42 + ) + assert step.step_num == 1 + assert step.status == "completed" + assert step.duration_ms == 42 + + def test_agentic_result_defaults(self): + r = AgenticResult(task_id="abc", task="test", summary="ok") + assert r.steps == [] + assert r.status == "completed" + assert r.total_duration_ms == 0 + + +# --------------------------------------------------------------------------- +# _parse_steps +# --------------------------------------------------------------------------- + + class TestParseSteps: def test_numbered_with_dot(self): text = "1. Search for data\n2. Write to file\n3. Verify" @@ -43,6 +72,19 @@ class TestParseSteps: def test_empty_returns_empty(self): assert _parse_steps("") == [] + def test_whitespace_only_returns_empty(self): + assert _parse_steps(" \n \n ") == [] + + def test_leading_whitespace_in_numbered(self): + text = " 1. First\n 2. Second" + assert _parse_steps(text) == ["First", "Second"] + + def test_mixed_numbered_and_plain(self): + """When numbered lines are present, only those are returned.""" + text = "Here is the plan:\n1. Step one\n2. Step two\nGood luck!" + result = _parse_steps(text) + assert result == ["Step one", "Step two"] + # --------------------------------------------------------------------------- # run_agentic_loop @@ -231,3 +273,191 @@ async def test_planning_failure_returns_failed(): assert result.status == "failed" assert "Planning failed" in result.summary + + +@pytest.mark.asyncio +async def test_empty_plan_returns_failed(): + """Planning that produces no steps results in 'failed'.""" + mock_agent = MagicMock() + mock_agent.run = MagicMock(return_value=_mock_run("")) + + with ( + patch("timmy.agentic_loop._get_loop_agent", return_value=mock_agent), + patch("timmy.agentic_loop._broadcast_progress", new_callable=AsyncMock), + ): + result = await run_agentic_loop("Do nothing") + + assert result.status == "failed" + assert "no steps" in result.summary.lower() + + +@pytest.mark.asyncio +async def test_double_failure_marks_step_failed(): + """When both execution and adaptation fail, step status is 'failed'.""" + mock_agent = MagicMock() + mock_agent.run = MagicMock( + side_effect=[ + _mock_run("1. Do something"), + Exception("Step failed"), + Exception("Adaptation also failed"), + ] + ) + + with ( + patch("timmy.agentic_loop._get_loop_agent", return_value=mock_agent), + patch("timmy.agentic_loop._broadcast_progress", new_callable=AsyncMock), + ): + result = await run_agentic_loop("Try and fail", max_steps=1) + + assert len(result.steps) == 1 + assert result.steps[0].status == "failed" + assert "Failed" in result.steps[0].result + assert result.status == "partial" + + +@pytest.mark.asyncio +async def test_broadcast_progress_ignores_ws_errors(): + """_broadcast_progress swallows import/connection errors.""" + with patch( + "timmy.agentic_loop.ws_manager", + create=True, + side_effect=ImportError("no ws"), + ): + # Should not raise + await _broadcast_progress("test.event", {"key": "value"}) + + +@pytest.mark.asyncio +async def test_broadcast_progress_sends_to_ws(): + """_broadcast_progress calls ws_manager.broadcast.""" + mock_ws = AsyncMock() + with patch("infrastructure.ws_manager.handler.ws_manager", mock_ws): + await _broadcast_progress("agentic.plan_ready", {"task_id": "abc"}) + mock_ws.broadcast.assert_awaited_once_with("agentic.plan_ready", {"task_id": "abc"}) + + +@pytest.mark.asyncio +async def test_summary_counts_step_statuses(): + """Summary string includes completed, adapted, and failed counts.""" + mock_agent = MagicMock() + mock_agent.run = MagicMock( + side_effect=[ + _mock_run("1. A\n2. B\n3. C"), + _mock_run("A done"), + Exception("B broke"), + _mock_run("B adapted"), + Exception("C broke"), + Exception("C adapt broke too"), + ] + ) + + with ( + patch("timmy.agentic_loop._get_loop_agent", return_value=mock_agent), + patch("timmy.agentic_loop._broadcast_progress", new_callable=AsyncMock), + ): + result = await run_agentic_loop("A B C", max_steps=3) + + assert "1 adapted" in result.summary + assert "1 failed" in result.summary + assert result.status == "partial" + + +@pytest.mark.asyncio +async def test_task_id_is_set(): + """Result has a non-empty task_id.""" + mock_agent = MagicMock() + mock_agent.run = MagicMock(side_effect=[_mock_run("1. X"), _mock_run("done")]) + + with ( + patch("timmy.agentic_loop._get_loop_agent", return_value=mock_agent), + patch("timmy.agentic_loop._broadcast_progress", new_callable=AsyncMock), + ): + result = await run_agentic_loop("One step") + + assert result.task_id + assert len(result.task_id) == 8 + + +@pytest.mark.asyncio +async def test_total_duration_is_set(): + """Result.total_duration_ms is a positive integer.""" + mock_agent = MagicMock() + mock_agent.run = MagicMock(side_effect=[_mock_run("1. X"), _mock_run("done")]) + + with ( + patch("timmy.agentic_loop._get_loop_agent", return_value=mock_agent), + patch("timmy.agentic_loop._broadcast_progress", new_callable=AsyncMock), + ): + result = await run_agentic_loop("Quick task") + + assert result.total_duration_ms >= 0 + + +@pytest.mark.asyncio +async def test_agent_run_without_content_attr(): + """When agent.run() returns an object without .content, str() is used.""" + + class PlanResult: + def __str__(self): + return "1. Only step" + + class StepResult: + def __str__(self): + return "Step result" + + mock_agent = MagicMock() + mock_agent.run = MagicMock(side_effect=[PlanResult(), StepResult()]) + + with ( + patch("timmy.agentic_loop._get_loop_agent", return_value=mock_agent), + patch("timmy.agentic_loop._broadcast_progress", new_callable=AsyncMock), + ): + result = await run_agentic_loop("Fallback test", max_steps=1) + + assert len(result.steps) == 1 + + +@pytest.mark.asyncio +async def test_adapted_step_calls_on_progress(): + """on_progress is called even for adapted steps.""" + events = [] + + async def on_progress(desc, step, total): + events.append((desc, step)) + + mock_agent = MagicMock() + mock_agent.run = MagicMock( + side_effect=[ + _mock_run("1. Risky step"), + Exception("boom"), + _mock_run("Adapted result"), + ] + ) + + with ( + patch("timmy.agentic_loop._get_loop_agent", return_value=mock_agent), + patch("timmy.agentic_loop._broadcast_progress", new_callable=AsyncMock), + ): + await run_agentic_loop("Adapt test", max_steps=1, on_progress=on_progress) + + assert len(events) == 1 + assert "[Adapted]" in events[0][0] + + +@pytest.mark.asyncio +async def test_broadcast_called_for_each_phase(): + """_broadcast_progress is called for plan_ready, step_complete, and task_complete.""" + mock_agent = MagicMock() + mock_agent.run = MagicMock(side_effect=[_mock_run("1. Do it"), _mock_run("Done")]) + + broadcast = AsyncMock() + with ( + patch("timmy.agentic_loop._get_loop_agent", return_value=mock_agent), + patch("timmy.agentic_loop._broadcast_progress", broadcast), + ): + await run_agentic_loop("One step task", max_steps=1) + + event_names = [call.args[0] for call in broadcast.call_args_list] + assert "agentic.plan_ready" in event_names + assert "agentic.step_complete" in event_names + assert "agentic.task_complete" in event_names