From 1fd70cc0fb1c55c23e9a7a2905253e4317c68666 Mon Sep 17 00:00:00 2001 From: rcholic Date: Fri, 2 Jan 2026 15:51:24 -0800 Subject: [PATCH 1/2] testing --- tests/integration/test_agent_workflows.py | 424 ++++++++++++++++++++++ tests/test_utils_browser.py | 152 ++++++++ 2 files changed, 576 insertions(+) create mode 100644 tests/integration/test_agent_workflows.py create mode 100644 tests/test_utils_browser.py diff --git a/tests/integration/test_agent_workflows.py b/tests/integration/test_agent_workflows.py new file mode 100644 index 0000000..bebd7d2 --- /dev/null +++ b/tests/integration/test_agent_workflows.py @@ -0,0 +1,424 @@ +""" +Integration tests for SentienceAgent workflows. + +Tests multi-step agent scenarios and error recovery without requiring real browser. +Uses mocks to simulate realistic browser behavior. +""" + +from unittest.mock import Mock, patch + +import pytest + +from sentience.agent import SentienceAgent +from sentience.llm_provider import LLMProvider, LLMResponse +from sentience.models import BBox, Element, Snapshot, Viewport, VisualCues +from sentience.protocols import BrowserProtocol, PageProtocol + + +class MockLLMProvider(LLMProvider): + """Mock LLM provider for integration testing""" + + def __init__(self, responses=None): + self.responses = responses or [] + self.call_count = 0 + self.calls = [] + + def generate(self, system_prompt: str, user_prompt: str, **kwargs): + self.calls.append({"system": system_prompt, "user": user_prompt, "kwargs": kwargs}) + + if self.responses: + response = self.responses[self.call_count % len(self.responses)] + else: + response = "CLICK(1)" + + self.call_count += 1 + + return LLMResponse( + content=response, + prompt_tokens=100, + completion_tokens=20, + total_tokens=120, + model_name="mock-model", + ) + + def supports_json_mode(self) -> bool: + return True + + @property + def model_name(self) -> str: + return "mock-model" + + +class MockPage(PageProtocol): + """Mock page that implements PageProtocol""" + + def __init__(self, url: str = "https://example.com"): + self._url = url + + @property + def url(self) -> str: + return self._url + + def evaluate(self, script: str, *args, **kwargs): + return {} + + def goto(self, url: str, **kwargs): + self._url = url + + def wait_for_timeout(self, timeout: int): + pass + + def wait_for_load_state(self, state: str = "load", timeout: int | None = None): + pass + + def wait_for_function(self, expression: str, timeout: int | None = None): + pass + + +class MockBrowser(BrowserProtocol): + """Mock browser for integration testing""" + + def __init__(self): + self._page = MockPage() + self._started = False + self.api_key = None # Required by snapshot function + self.api_url = None # Required by snapshot function + self._context = Mock() # Mock context for storage state + + def start(self): + self._started = True + + @property + def page(self) -> PageProtocol | None: + return self._page if self._started else None + + def goto(self, url: str): + if self._page: + self._page.goto(url) + + def close(self, output_path=None): + self._started = False + return output_path + + @property + def context(self): + return self._context + + +def create_mock_snapshot(elements=None): + """Create a mock snapshot for testing""" + if elements is None: + elements = [ + Element( + id=1, + role="button", + text="Click Me", + importance=900, + bbox=BBox(x=100, y=200, width=80, height=30), + visual_cues=VisualCues(is_primary=True, is_clickable=True), + ), + Element( + id=2, + role="input", + text="Search", + importance=800, + bbox=BBox(x=100, y=250, width=200, height=30), + visual_cues=VisualCues(is_primary=False, is_clickable=True), + ), + ] + return Snapshot( + status="success", + timestamp="2024-12-24T10:00:00Z", + url="https://example.com", + viewport=Viewport(width=1920, height=1080), + elements=elements, + ) + + +class TestAgentMultiStepWorkflows: + """Test multi-step agent workflows""" + + def test_agent_multi_step_click_then_type(self): + """Test agent performing multiple actions in sequence.""" + browser = MockBrowser() + browser.start() + llm = MockLLMProvider(responses=["CLICK(2)", 'TYPE(2, "search query")']) + agent = SentienceAgent(browser, llm, verbose=False) + + with ( + patch("sentience.agent.snapshot") as mock_snapshot, + patch("sentience.action_executor.click") as mock_click, + patch("sentience.action_executor.type_text") as mock_type, + ): + from sentience.models import ActionResult + + mock_snapshot.return_value = create_mock_snapshot() + mock_click.return_value = ActionResult( + success=True, duration_ms=150, outcome="dom_updated" + ) + mock_type.return_value = ActionResult( + success=True, duration_ms=200, outcome="dom_updated" + ) + + # First action: click input + result1 = agent.act("Click the search input", max_retries=0) + assert result1.success is True + assert result1.action == "click" + assert mock_click.call_count == 1 + + # Second action: type into input + result2 = agent.act("Type search query into the input", max_retries=0) + assert result2.success is True + assert result2.action == "type" + assert mock_type.call_count == 1 + + # Verify history tracks both actions + assert len(agent.history) == 2 + + def test_agent_workflow_with_retry(self): + """Test agent workflow with retry on failure.""" + browser = MockBrowser() + browser.start() + llm = MockLLMProvider(responses=["CLICK(1)"]) + agent = SentienceAgent(browser, llm, verbose=False) + + with ( + patch("sentience.agent.snapshot") as mock_snapshot, + patch("sentience.action_executor.click") as mock_click, + ): + from sentience.models import ActionResult + + mock_snapshot.return_value = create_mock_snapshot() + # First call fails, second succeeds + mock_click.side_effect = [ + ActionResult(success=False, duration_ms=100, outcome="error"), + ActionResult(success=True, duration_ms=150, outcome="dom_updated"), + ] + + result = agent.act("Click the button", max_retries=1) + + assert result.success is True + assert mock_click.call_count == 2 + assert len(agent.history) == 1 # Only successful attempt recorded + + def test_agent_workflow_url_change(self): + """Test agent workflow that causes URL change.""" + browser = MockBrowser() + browser.start() + llm = MockLLMProvider(responses=["CLICK(1)"]) + agent = SentienceAgent(browser, llm, verbose=False) + + with ( + patch("sentience.agent.snapshot") as mock_snapshot, + patch("sentience.action_executor.click") as mock_click, + ): + from sentience.models import ActionResult + + mock_snapshot.return_value = create_mock_snapshot() + mock_click.return_value = ActionResult( + success=True, duration_ms=150, outcome="navigated", url_changed=True + ) + + result = agent.act("Click the link", max_retries=0) + + assert result.success is True + assert result.url_changed is True + assert result.action == "click" + + def test_agent_workflow_finish_action(self): + """Test agent workflow that finishes successfully.""" + browser = MockBrowser() + browser.start() + llm = MockLLMProvider(responses=["FINISH()"]) + agent = SentienceAgent(browser, llm, verbose=False) + + with patch("sentience.snapshot.snapshot") as mock_snapshot: + mock_snapshot.return_value = create_mock_snapshot() + + result = agent.act("Task is complete", max_retries=0) + + assert result.success is True + assert result.action == "finish" + assert len(agent.history) == 1 + + def test_agent_workflow_token_tracking(self): + """Test that token usage is tracked across workflow.""" + browser = MockBrowser() + browser.start() + llm = MockLLMProvider(responses=["CLICK(1)", "CLICK(2)"]) + agent = SentienceAgent(browser, llm, verbose=False) + + with ( + patch("sentience.agent.snapshot") as mock_snapshot, + patch("sentience.action_executor.click") as mock_click, + ): + from sentience.models import ActionResult + + mock_snapshot.return_value = create_mock_snapshot() + mock_click.return_value = ActionResult( + success=True, duration_ms=150, outcome="dom_updated" + ) + + # Perform two actions + agent.act("Click first button", max_retries=0) + agent.act("Click second button", max_retries=0) + + # Check token stats + stats = agent.get_token_stats() + assert stats.total_tokens > 0 + assert stats.total_prompt_tokens > 0 + assert stats.total_completion_tokens > 0 + assert len(stats.by_action) == 2 # Two actions tracked + + +class TestAgentErrorRecovery: + """Test agent error recovery scenarios""" + + def test_agent_recovery_after_snapshot_failure(self): + """Test agent recovers after snapshot failure.""" + browser = MockBrowser() + browser.start() + llm = MockLLMProvider(responses=["CLICK(1)"]) + agent = SentienceAgent(browser, llm, verbose=False) + + with ( + patch("sentience.agent.snapshot") as mock_snapshot, + patch("sentience.action_executor.click") as mock_click, + ): + from sentience.models import ActionResult, Snapshot + + # First snapshot fails, second succeeds + failed_snapshot = Snapshot( + status="error", + error="Network timeout", + url="https://example.com", + viewport=Viewport(width=1920, height=1080), + elements=[], + ) + mock_snapshot.side_effect = [ + failed_snapshot, + create_mock_snapshot(), + ] + mock_click.return_value = ActionResult( + success=True, duration_ms=150, outcome="dom_updated" + ) + + # Should raise on first attempt, succeed on retry + with pytest.raises(RuntimeError, match="Snapshot failed"): + agent.act("Click button", max_retries=0) + + # With retry, should succeed + result = agent.act("Click button", max_retries=1) + assert result.success is True + + def test_agent_recovery_after_action_failure(self): + """Test agent recovers after action failure.""" + browser = MockBrowser() + browser.start() + llm = MockLLMProvider(responses=["CLICK(1)", "CLICK(1)"]) + agent = SentienceAgent(browser, llm, verbose=False) + + with ( + patch("sentience.agent.snapshot") as mock_snapshot, + patch("sentience.action_executor.click") as mock_click, + ): + from sentience.models import ActionResult + + mock_snapshot.return_value = create_mock_snapshot() + # First action fails, second succeeds + mock_click.side_effect = [ + RuntimeError("Element not found"), + ActionResult(success=True, duration_ms=150, outcome="dom_updated"), + ] + + result = agent.act("Click button", max_retries=1) + + assert result.success is True + assert mock_click.call_count == 2 + + def test_agent_handles_max_retries_exceeded(self): + """Test agent handles max retries exceeded.""" + browser = MockBrowser() + browser.start() + # Need multiple responses for multiple retries + llm = MockLLMProvider(responses=["CLICK(1)", "CLICK(1)", "CLICK(1)"]) + agent = SentienceAgent(browser, llm, verbose=False) + + with ( + patch("sentience.agent.snapshot") as mock_snapshot, + patch("sentience.action_executor.click") as mock_click, + ): + from sentience.models import ActionResult + + mock_snapshot.return_value = create_mock_snapshot() + # Raise exception to trigger retry logic (agent only retries on exceptions, not failed results) + mock_click.side_effect = RuntimeError("Action failed") + + with pytest.raises(RuntimeError, match="Failed after"): + agent.act("Click button", max_retries=2) + + # Should have attempted 3 times (initial + 2 retries) + # Each attempt calls snapshot, LLM, and click + assert mock_click.call_count == 3 + assert mock_snapshot.call_count >= 3 + assert llm.call_count >= 3 + + +class TestAgentStateManagement: + """Test agent state management across actions""" + + def test_agent_history_preservation(self): + """Test that agent history is preserved across actions.""" + browser = MockBrowser() + browser.start() + llm = MockLLMProvider(responses=["CLICK(1)", "CLICK(2)", "FINISH()"]) + agent = SentienceAgent(browser, llm, verbose=False) + + with ( + patch("sentience.agent.snapshot") as mock_snapshot, + patch("sentience.action_executor.click") as mock_click, + ): + from sentience.models import ActionResult + + mock_snapshot.return_value = create_mock_snapshot() + mock_click.return_value = ActionResult( + success=True, duration_ms=150, outcome="dom_updated" + ) + + # Perform multiple actions + agent.act("Click first", max_retries=0) + agent.act("Click second", max_retries=0) + agent.act("Finish", max_retries=0) + + # Verify history contains all actions + assert len(agent.history) == 3 + assert agent.history[0]["goal"] == "Click first" + assert agent.history[1]["goal"] == "Click second" + assert agent.history[2]["goal"] == "Finish" + + def test_agent_step_count_increments(self): + """Test that step count increments across actions.""" + browser = MockBrowser() + browser.start() + llm = MockLLMProvider(responses=["CLICK(1)", "CLICK(2)"]) + agent = SentienceAgent(browser, llm, verbose=False) + + with ( + patch("sentience.agent.snapshot") as mock_snapshot, + patch("sentience.action_executor.click") as mock_click, + ): + from sentience.models import ActionResult + + mock_snapshot.return_value = create_mock_snapshot() + mock_click.return_value = ActionResult( + success=True, duration_ms=150, outcome="dom_updated" + ) + + initial_count = agent._step_count + + agent.act("First action", max_retries=0) + assert agent._step_count == initial_count + 1 + + agent.act("Second action", max_retries=0) + assert agent._step_count == initial_count + 2 + diff --git a/tests/test_utils_browser.py b/tests/test_utils_browser.py new file mode 100644 index 0000000..f5f8c56 --- /dev/null +++ b/tests/test_utils_browser.py @@ -0,0 +1,152 @@ +""" +Unit tests for sentience.utils.browser module. + +Tests browser storage state saving functionality. +""" + +import json +import tempfile +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest + +from sentience.utils.browser import save_storage_state + + +class TestSaveStorageState: + """Tests for save_storage_state function.""" + + def test_save_storage_state_creates_file(self): + """Test that save_storage_state creates a file with storage state.""" + # Create a mock BrowserContext + mock_context = Mock() + mock_context.storage_state.return_value = { + "cookies": [ + { + "name": "session_id", + "value": "abc123", + "domain": "example.com", + "path": "/", + } + ], + "origins": [ + { + "origin": "https://example.com", + "localStorage": [{"name": "user_pref", "value": "dark_mode"}], + } + ], + } + + # Use temporary file + with tempfile.TemporaryDirectory() as tmpdir: + file_path = Path(tmpdir) / "storage.json" + + # Call function + save_storage_state(mock_context, file_path) + + # Verify file was created + assert file_path.exists() + + # Verify content + with open(file_path) as f: + data = json.load(f) + + assert "cookies" in data + assert "origins" in data + assert len(data["cookies"]) == 1 + assert data["cookies"][0]["name"] == "session_id" + + def test_save_storage_state_creates_parent_directories(self): + """Test that save_storage_state creates parent directories if needed.""" + mock_context = Mock() + mock_context.storage_state.return_value = {"cookies": [], "origins": []} + + with tempfile.TemporaryDirectory() as tmpdir: + # Create nested path + file_path = Path(tmpdir) / "nested" / "deep" / "storage.json" + + # Should not raise error + save_storage_state(mock_context, file_path) + + # Verify file was created + assert file_path.exists() + assert file_path.parent.exists() + + def test_save_storage_state_with_string_path(self): + """Test that save_storage_state accepts string paths.""" + mock_context = Mock() + mock_context.storage_state.return_value = {"cookies": [], "origins": []} + + with tempfile.TemporaryDirectory() as tmpdir: + file_path = str(Path(tmpdir) / "storage.json") + + save_storage_state(mock_context, file_path) + + assert Path(file_path).exists() + + def test_save_storage_state_calls_context_storage_state(self): + """Test that save_storage_state calls context.storage_state().""" + mock_context = Mock() + mock_context.storage_state.return_value = {"cookies": [], "origins": []} + + with tempfile.TemporaryDirectory() as tmpdir: + file_path = Path(tmpdir) / "storage.json" + + save_storage_state(mock_context, file_path) + + # Verify storage_state was called + mock_context.storage_state.assert_called_once() + + def test_save_storage_state_json_format(self): + """Test that saved file is valid JSON with indentation.""" + mock_context = Mock() + mock_context.storage_state.return_value = { + "cookies": [{"name": "test", "value": "value"}], + "origins": [], + } + + with tempfile.TemporaryDirectory() as tmpdir: + file_path = Path(tmpdir) / "storage.json" + + save_storage_state(mock_context, file_path) + + # Verify JSON is valid and formatted + with open(file_path) as f: + content = f.read() + # Should have indentation (contains newlines) + assert "\n" in content + # Should be valid JSON + data = json.loads(content) + assert isinstance(data, dict) + + def test_save_storage_state_handles_empty_state(self): + """Test that save_storage_state handles empty storage state.""" + mock_context = Mock() + mock_context.storage_state.return_value = {"cookies": [], "origins": []} + + with tempfile.TemporaryDirectory() as tmpdir: + file_path = Path(tmpdir) / "storage.json" + + save_storage_state(mock_context, file_path) + + with open(file_path) as f: + data = json.load(f) + + assert data == {"cookies": [], "origins": []} + + def test_save_storage_state_prints_success_message(self, capsys): + """Test that save_storage_state prints success message.""" + mock_context = Mock() + mock_context.storage_state.return_value = {"cookies": [], "origins": []} + + with tempfile.TemporaryDirectory() as tmpdir: + file_path = Path(tmpdir) / "storage.json" + + save_storage_state(mock_context, file_path) + + captured = capsys.readouterr() + assert "✅" in captured.out + assert "Saved storage state" in captured.out + assert str(file_path) in captured.out + From 360ceeb1712378a293642927603d9c408af7270b Mon Sep 17 00:00:00 2001 From: rcholic Date: Fri, 2 Jan 2026 16:16:22 -0800 Subject: [PATCH 2/2] add tests --- tests/integration/test_agent_workflows.py | 7 +++---- tests/test_utils_browser.py | 1 - 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/tests/integration/test_agent_workflows.py b/tests/integration/test_agent_workflows.py index bebd7d2..f3bd0fc 100644 --- a/tests/integration/test_agent_workflows.py +++ b/tests/integration/test_agent_workflows.py @@ -189,9 +189,9 @@ def test_agent_workflow_with_retry(self): from sentience.models import ActionResult mock_snapshot.return_value = create_mock_snapshot() - # First call fails, second succeeds + # First call raises exception (triggers retry), second succeeds mock_click.side_effect = [ - ActionResult(success=False, duration_ms=100, outcome="error"), + RuntimeError("Element not found"), ActionResult(success=True, duration_ms=150, outcome="dom_updated"), ] @@ -232,7 +232,7 @@ def test_agent_workflow_finish_action(self): llm = MockLLMProvider(responses=["FINISH()"]) agent = SentienceAgent(browser, llm, verbose=False) - with patch("sentience.snapshot.snapshot") as mock_snapshot: + with patch("sentience.agent.snapshot") as mock_snapshot: mock_snapshot.return_value = create_mock_snapshot() result = agent.act("Task is complete", max_retries=0) @@ -421,4 +421,3 @@ def test_agent_step_count_increments(self): agent.act("Second action", max_retries=0) assert agent._step_count == initial_count + 2 - diff --git a/tests/test_utils_browser.py b/tests/test_utils_browser.py index f5f8c56..145c888 100644 --- a/tests/test_utils_browser.py +++ b/tests/test_utils_browser.py @@ -149,4 +149,3 @@ def test_save_storage_state_prints_success_message(self, capsys): assert "✅" in captured.out assert "Saved storage state" in captured.out assert str(file_path) in captured.out -