diff --git a/tests/contract/test_claude_api_client.py b/tests/contract/test_claude_api_client.py index 563b27d..6ef7397 100644 --- a/tests/contract/test_claude_api_client.py +++ b/tests/contract/test_claude_api_client.py @@ -1,288 +1,217 @@ -""" -Contract tests for IClaudeAPIClient interface. +"""Executable contract tests for IClaudeAPIClient using AnthropicAPIClient.""" -Verifies that all implementations of IClaudeAPIClient adhere to the contract. -These tests are run against each concrete implementation to ensure consistency. -""" +from __future__ import annotations -import pytest +import json +from datetime import UTC, datetime from pathlib import Path +from types import SimpleNamespace from uuid import uuid4 -from datetime import datetime + +import pytest +from PIL import Image from src.interfaces.screenshot_service import IClaudeAPIClient +from src.lib.exceptions import APIError, AuthenticationError, OAuthConfigNotFoundError, PayloadTooLargeError from src.models.entities import Screenshot -from src.lib.exceptions import ( - AuthenticationError, - APIError, - PayloadTooLargeError, - OAuthConfigNotFoundError -) - - -class TestIClaudeAPIClientContract: - """ - Contract test suite for IClaudeAPIClient interface. - - All implementations of IClaudeAPIClient MUST pass these tests. - """ - - @pytest.fixture - def client_implementation(self): - """ - Override this fixture in concrete test classes to provide the implementation. - - Example: - @pytest.fixture - def client_implementation(self): - return AnthropicAPIClient() - """ - pytest.skip("Must be implemented by concrete test class") - - @pytest.fixture - def sample_screenshot(self, tmp_path): - """ - Create a sample Screenshot object for testing. - - This fixture should be overridden in concrete tests to provide a real image file. - """ - pytest.skip("Must be implemented by concrete test class to provide real image") - - def test_interface_inheritance(self, client_implementation): - """Test that implementation inherits from IClaudeAPIClient.""" - assert isinstance(client_implementation, IClaudeAPIClient) - - def test_send_multimodal_prompt_returns_string(self, client_implementation, sample_screenshot): - """Test that send_multimodal_prompt() returns a string response.""" - text = "What do you see in this image?" - - response = client_implementation.send_multimodal_prompt(text, sample_screenshot) - - assert isinstance(response, str) - assert len(response) > 0 - - def test_send_multimodal_prompt_with_empty_text(self, client_implementation, sample_screenshot): - """Test that send_multimodal_prompt() handles empty text prompt.""" - # Should still work - image-only analysis - response = client_implementation.send_multimodal_prompt("", sample_screenshot) - - assert isinstance(response, str) - # Response might be empty or contain a default message - - def test_send_multimodal_prompt_with_long_text(self, client_implementation, sample_screenshot): - """Test send_multimodal_prompt() with long text prompt.""" - long_text = "Analyze this screenshot in detail. " * 100 # Long prompt - - response = client_implementation.send_multimodal_prompt(long_text, sample_screenshot) - - assert isinstance(response, str) - - def test_send_multimodal_prompt_validates_screenshot_exists(self, client_implementation, tmp_path): - """Test that send_multimodal_prompt() validates screenshot file exists.""" - # Create screenshot with non-existent file - invalid_screenshot = Screenshot( - id=uuid4(), - timestamp=datetime.now(), - file_path=tmp_path / "nonexistent.jpg", - format="jpeg", - original_size_bytes=0, - optimized_size_bytes=0, - resolution=(800, 600), - source_monitor=0, - capture_method="test", - privacy_zones_applied=False - ) - - with pytest.raises((APIError, FileNotFoundError)): - client_implementation.send_multimodal_prompt("Test", invalid_screenshot) - - def test_validate_oauth_token_returns_bool(self, client_implementation): - """Test that validate_oauth_token() returns boolean.""" - result = client_implementation.validate_oauth_token() - - assert isinstance(result, bool) - - def test_validate_oauth_token_with_valid_token(self, client_implementation): - """Test validate_oauth_token() with valid token.""" - # This test assumes a valid token is configured - # May need to be mocked or skipped in CI - try: - result = client_implementation.validate_oauth_token() - # If we get here without error, result should be True - assert isinstance(result, bool) - except OAuthConfigNotFoundError: - pytest.skip("OAuth config not found - expected in test environment") - - def test_validate_oauth_token_missing_config_raises_error(self, client_implementation, monkeypatch): - """Test that validate_oauth_token() raises error when config missing.""" - # Mock missing config file - monkeypatch.setenv('HOME', '/nonexistent') - - with pytest.raises(OAuthConfigNotFoundError): - client_implementation.validate_oauth_token() - - def test_refresh_oauth_token_executes_without_error(self, client_implementation): - """Test that refresh_oauth_token() can be called.""" - # This test may need to be mocked as it requires valid credentials - try: - client_implementation.refresh_oauth_token() - # If we get here, refresh succeeded - except AuthenticationError: - # Expected if no valid refresh token available - pytest.skip("No valid refresh token - expected in test environment") - except OAuthConfigNotFoundError: - # Expected if config file missing - pytest.skip("OAuth config not found - expected in test environment") - - -class TestIClaudeAPIClientErrorHandling: - """ - Contract tests for error handling in IClaudeAPIClient implementations. - """ - - @pytest.fixture - def client_implementation(self): - """Override in concrete test classes.""" - pytest.skip("Must be implemented by concrete test class") - - @pytest.fixture - def sample_screenshot(self, tmp_path): - """Override in concrete test classes.""" - pytest.skip("Must be implemented by concrete test class") - - def test_send_multimodal_prompt_invalid_token_raises_auth_error(self, client_implementation, sample_screenshot, monkeypatch): - """Test that invalid OAuth token raises AuthenticationError.""" - # Mock invalid token scenario - # Implementation-specific: may need to mock token reading - # with pytest.raises(AuthenticationError): - # client_implementation.send_multimodal_prompt("Test", sample_screenshot) - pytest.skip("Implementation-specific test - requires mocking") - - def test_send_multimodal_prompt_network_error_raises_api_error(self, client_implementation, sample_screenshot): - """Test that network errors raise APIError.""" - # Mock network failure scenario - # Implementation-specific: may need to mock requests - pytest.skip("Implementation-specific test - requires mocking") - - def test_send_multimodal_prompt_too_large_raises_payload_error(self, client_implementation, tmp_path): - """Test that oversized payload raises PayloadTooLargeError.""" - # Create a very large screenshot - large_screenshot = Screenshot( - id=uuid4(), - timestamp=datetime.now(), - file_path=tmp_path / "large.jpg", - format="jpeg", - original_size_bytes=50 * 1024 * 1024, # 50 MB - optimized_size_bytes=50 * 1024 * 1024, - resolution=(10000, 10000), - source_monitor=0, - capture_method="test", - privacy_zones_applied=False - ) - - # Note: This test requires actual large file creation - # May need to be implemented in concrete test class - pytest.skip("Requires actual large file - implement in concrete class") - - def test_refresh_token_invalid_credentials_raises_auth_error(self, client_implementation): - """Test that refresh with invalid credentials raises AuthenticationError.""" - # Mock invalid refresh token scenario - pytest.skip("Implementation-specific test - requires mocking") - - -class TestIClaudeAPIClientIntegration: - """ - Integration-style contract tests that verify end-to-end functionality. - - These tests may be slow and should be marked as such. - """ - - @pytest.fixture - def client_implementation(self): - """Override in concrete test classes.""" - pytest.skip("Must be implemented by concrete test class") - - @pytest.fixture - def sample_screenshot(self, tmp_path): - """Override in concrete test classes.""" - pytest.skip("Must be implemented by concrete test class") - - @pytest.mark.slow - @pytest.mark.integration - def test_full_multimodal_workflow(self, client_implementation, sample_screenshot): - """ - Test complete workflow: validate token, send prompt, receive response. - - This is an integration test that requires: - - Valid OAuth configuration - - Network connectivity - - Valid API key - """ - try: - # Step 1: Validate token - is_valid = client_implementation.validate_oauth_token() - if not is_valid: - pytest.skip("Token not valid - skipping integration test") - - # Step 2: Send multimodal prompt - response = client_implementation.send_multimodal_prompt( - "Describe this image briefly.", - sample_screenshot - ) - - # Step 3: Verify response - assert isinstance(response, str) - assert len(response) > 0 - - except (OAuthConfigNotFoundError, AuthenticationError): - pytest.skip("OAuth not configured - expected in test environment") - except APIError as e: - pytest.fail(f"API error during integration test: {e}") - - @pytest.mark.slow - @pytest.mark.integration - def test_token_refresh_workflow(self, client_implementation): - """ - Test token refresh workflow. - - This is an integration test that requires valid refresh token. - """ - try: - # Attempt to refresh token - client_implementation.refresh_oauth_token() - - # Verify token is now valid - is_valid = client_implementation.validate_oauth_token() - assert is_valid - - except (OAuthConfigNotFoundError, AuthenticationError): - pytest.skip("OAuth not configured - expected in test environment") - - -# NOTE: Concrete test classes will inherit from these and provide actual implementations -# Example: -# class TestAnthropicAPIClient(TestIClaudeAPIClientContract): -# @pytest.fixture -# def client_implementation(self): -# return AnthropicAPIClient() -# -# @pytest.fixture -# def sample_screenshot(self, tmp_path): -# # Create a real test image -# from PIL import Image -# img = Image.new('RGB', (800, 600), color='blue') -# img_path = tmp_path / "test.jpg" -# img.save(img_path, quality=85) -# -# return Screenshot( -# id=uuid4(), -# timestamp=datetime.now(), -# file_path=img_path, -# format="jpeg", -# original_size_bytes=img_path.stat().st_size, -# optimized_size_bytes=img_path.stat().st_size, -# resolution=(800, 600), -# source_monitor=0, -# capture_method="test", -# privacy_zones_applied=False -# ) +from src.services.claude_api_client import AnthropicAPIClient + + +@pytest.fixture() +def sample_screenshot(tmp_path: Path) -> Screenshot: + img_path = tmp_path / "sample.png" + Image.new("RGB", (120, 80), color="white").save(img_path) + size = img_path.stat().st_size + return Screenshot( + id=uuid4(), + timestamp=datetime.now(tz=UTC), + file_path=img_path, + format="png", + original_size_bytes=size, + optimized_size_bytes=size, + resolution=(120, 80), + source_monitor=0, + capture_method="test", + privacy_zones_applied=False, + ) + + +@pytest.fixture() +def client_implementation() -> AnthropicAPIClient: + return AnthropicAPIClient(api_key="sk-test") + + +def test_interface_inheritance(client_implementation: AnthropicAPIClient) -> None: + assert isinstance(client_implementation, IClaudeAPIClient) + + +def test_send_multimodal_prompt_returns_string( + client_implementation: AnthropicAPIClient, + sample_screenshot: Screenshot, + monkeypatch: pytest.MonkeyPatch, +) -> None: + def _post_ok(*_args, **_kwargs): + return SimpleNamespace(status_code=200, text="ok", json=lambda: {"content": [{"text": "ok"}]}) + + monkeypatch.setattr( + "src.services.claude_api_client.requests.post", + _post_ok, + ) + + response = client_implementation.send_multimodal_prompt("hello", sample_screenshot) + + assert response == "ok" + + +def test_send_multimodal_prompt_with_empty_text( + client_implementation: AnthropicAPIClient, + sample_screenshot: Screenshot, + monkeypatch: pytest.MonkeyPatch, +) -> None: + def _post_empty(*_args, **_kwargs): + return SimpleNamespace(status_code=200, text="ok", json=lambda: {"content": [{"text": "empty-ok"}]}) + + monkeypatch.setattr( + "src.services.claude_api_client.requests.post", + _post_empty, + ) + + assert client_implementation.send_multimodal_prompt("", sample_screenshot) == "empty-ok" + + +def test_send_multimodal_prompt_with_long_text( + client_implementation: AnthropicAPIClient, + sample_screenshot: Screenshot, + monkeypatch: pytest.MonkeyPatch, +) -> None: + def _post_long(*_args, **_kwargs): + return SimpleNamespace(status_code=200, text="ok", json=lambda: {"content": [{"text": "long-ok"}]}) + + monkeypatch.setattr( + "src.services.claude_api_client.requests.post", + _post_long, + ) + + response = client_implementation.send_multimodal_prompt("Analyze " * 150, sample_screenshot) + assert isinstance(response, str) + + +def test_send_multimodal_prompt_validates_screenshot_exists(client_implementation: AnthropicAPIClient) -> None: + missing = Screenshot( + id=uuid4(), + timestamp=datetime.now(tz=UTC), + file_path=Path("/definitely/missing.png"), + format="png", + original_size_bytes=0, + optimized_size_bytes=0, + resolution=(10, 10), + source_monitor=0, + capture_method="test", + privacy_zones_applied=False, + ) + + with pytest.raises(APIError, match="Screenshot file not found"): + client_implementation.send_multimodal_prompt("x", missing) + + +def test_validate_oauth_token_returns_bool(client_implementation: AnthropicAPIClient) -> None: + assert client_implementation.validate_oauth_token() is True + + +def test_validate_oauth_token_missing_config_raises_error(tmp_path: Path) -> None: + client = AnthropicAPIClient(oauth_token_path=str(tmp_path / "missing.json")) + + with pytest.raises(OAuthConfigNotFoundError): + client.validate_oauth_token() + + +def test_refresh_oauth_token_executes_without_error(client_implementation: AnthropicAPIClient) -> None: + client_implementation.refresh_oauth_token() + + +def test_send_multimodal_prompt_invalid_token_raises_auth_error( + client_implementation: AnthropicAPIClient, + sample_screenshot: Screenshot, + monkeypatch: pytest.MonkeyPatch, +) -> None: + def _post_unauthorized(*_args, **_kwargs): + return SimpleNamespace(status_code=401, text="bad", json=lambda: {}) + + monkeypatch.setattr( + "src.services.claude_api_client.requests.post", + _post_unauthorized, + ) + + with pytest.raises(AuthenticationError): + client_implementation.send_multimodal_prompt("x", sample_screenshot) + + +def test_send_multimodal_prompt_network_error_raises_api_error( + client_implementation: AnthropicAPIClient, + sample_screenshot: Screenshot, + monkeypatch: pytest.MonkeyPatch, +) -> None: + import requests + + def _raise(*_a, **_k): + raise requests.exceptions.ConnectionError() + + monkeypatch.setattr("src.services.claude_api_client.requests.post", _raise) + + with pytest.raises(APIError, match="Failed to connect"): + client_implementation.send_multimodal_prompt("x", sample_screenshot) + + +def test_send_multimodal_prompt_too_large_raises_payload_error( + client_implementation: AnthropicAPIClient, + sample_screenshot: Screenshot, +) -> None: + oversized = Screenshot( + id=sample_screenshot.id, + timestamp=sample_screenshot.timestamp, + file_path=sample_screenshot.file_path, + format=sample_screenshot.format, + original_size_bytes=sample_screenshot.original_size_bytes, + optimized_size_bytes=10 * 1024 * 1024, + resolution=sample_screenshot.resolution, + source_monitor=sample_screenshot.source_monitor, + capture_method=sample_screenshot.capture_method, + privacy_zones_applied=False, + ) + + with pytest.raises(PayloadTooLargeError): + client_implementation.send_multimodal_prompt("x", oversized) + + +def test_refresh_token_invalid_credentials_raises_auth_error(tmp_path: Path) -> None: + cfg = tmp_path / "oauth.json" + cfg.write_text(json.dumps({}), encoding="utf-8") + client = AnthropicAPIClient(oauth_token_path=str(cfg)) + + with pytest.raises(AuthenticationError): + client.refresh_oauth_token() + + +def test_full_multimodal_workflow( + client_implementation: AnthropicAPIClient, + sample_screenshot: Screenshot, + monkeypatch: pytest.MonkeyPatch, +) -> None: + def _post_done(*_args, **_kwargs): + return SimpleNamespace(status_code=200, text="ok", json=lambda: {"content": [{"text": "done"}]}) + + monkeypatch.setattr( + "src.services.claude_api_client.requests.post", + _post_done, + ) + + assert client_implementation.validate_oauth_token() is True + response = client_implementation.send_multimodal_prompt("Describe", sample_screenshot) + assert response == "done" + + +def test_token_refresh_workflow_with_missing_oauth_config(tmp_path: Path) -> None: + client = AnthropicAPIClient(oauth_token_path=str(tmp_path / "missing.json")) + + with pytest.raises(OAuthConfigNotFoundError): + client.refresh_oauth_token() diff --git a/tests/contract/test_region_selector.py b/tests/contract/test_region_selector.py index a10c611..1defce1 100644 --- a/tests/contract/test_region_selector.py +++ b/tests/contract/test_region_selector.py @@ -1,257 +1,197 @@ -""" -Contract tests for IRegionSelector interface. +"""Executable contract tests for IRegionSelector using a deterministic fake implementation.""" -Verifies that all implementations of IRegionSelector adhere to the contract. -These tests are run against each concrete implementation to ensure consistency. -""" +from __future__ import annotations + +from dataclasses import dataclass import pytest + from src.interfaces.screenshot_service import IRegionSelector +from src.lib.exceptions import InvalidRegionError, RegionSelectionCancelledError, SelectionToolNotFoundError from src.models.entities import CaptureRegion -from src.lib.exceptions import ( - RegionSelectionCancelledError, - SelectionToolNotFoundError, - InvalidRegionError -) -class TestIRegionSelectorContract: - """ - Contract test suite for IRegionSelector interface. - - All implementations of IRegionSelector MUST pass these tests. - """ - - @pytest.fixture - def selector_implementation(self): - """ - Override this fixture in concrete test classes to provide the implementation. - - Example: - @pytest.fixture - def selector_implementation(self): - return SlurpRegionSelector() - """ - pytest.skip("Must be implemented by concrete test class") - - def test_interface_inheritance(self, selector_implementation): - """Test that implementation inherits from IRegionSelector.""" - assert isinstance(selector_implementation, IRegionSelector) - - def test_select_region_graphical_returns_capture_region(self, selector_implementation): - """Test that select_region_graphical() returns a CaptureRegion object.""" - # This test requires user interaction or mocking - # Skip in automated tests - pytest.skip("Requires user interaction or mocking") - - def test_select_region_graphical_default_monitor(self, selector_implementation): - """Test select_region_graphical() with default monitor parameter.""" - pytest.skip("Requires user interaction or mocking") - - def test_select_region_graphical_specific_monitor(self, selector_implementation): - """Test select_region_graphical() with specific monitor.""" - pytest.skip("Requires user interaction or mocking") - - def test_select_region_graphical_user_cancels(self, selector_implementation): - """Test that user cancellation raises RegionSelectionCancelledError.""" - pytest.skip("Requires user interaction or mocking") - - def test_select_region_graphical_tool_not_found(self, selector_implementation): - """Test that missing tool raises SelectionToolNotFoundError.""" - pytest.skip("Requires mocking tool detection") - - def test_select_region_coordinates_returns_capture_region(self, selector_implementation): - """Test that select_region_coordinates() returns a CaptureRegion object.""" - region = selector_implementation.select_region_coordinates( - x=100, - y=100, - width=400, - height=300, - monitor=0 - ) +@dataclass +class _Monitor: + width: int + height: int - assert isinstance(region, CaptureRegion) - assert region.x == 100 - assert region.y == 100 - assert region.width == 400 - assert region.height == 300 - assert region.monitor == 0 - assert region.selection_method == 'coordinates' - - def test_select_region_coordinates_validates_parameters(self, selector_implementation): - """Test that select_region_coordinates() validates input parameters.""" - # Valid parameters should work - region = selector_implementation.select_region_coordinates( - x=0, - y=0, - width=100, - height=100, - monitor=0 - ) - assert isinstance(region, CaptureRegion) - - def test_select_region_coordinates_negative_x_raises_error(self, selector_implementation): - """Test that negative x coordinate raises InvalidRegionError.""" - with pytest.raises(InvalidRegionError): - selector_implementation.select_region_coordinates( - x=-10, - y=0, - width=100, - height=100, - monitor=0 - ) - - def test_select_region_coordinates_negative_y_raises_error(self, selector_implementation): - """Test that negative y coordinate raises InvalidRegionError.""" - with pytest.raises(InvalidRegionError): - selector_implementation.select_region_coordinates( - x=0, - y=-10, - width=100, - height=100, - monitor=0 - ) - - def test_select_region_coordinates_zero_width_raises_error(self, selector_implementation): - """Test that zero width raises InvalidRegionError.""" - with pytest.raises(InvalidRegionError): - selector_implementation.select_region_coordinates( - x=0, - y=0, - width=0, - height=100, - monitor=0 - ) - - def test_select_region_coordinates_zero_height_raises_error(self, selector_implementation): - """Test that zero height raises InvalidRegionError.""" - with pytest.raises(InvalidRegionError): - selector_implementation.select_region_coordinates( - x=0, - y=0, - width=100, - height=0, - monitor=0 - ) - - def test_select_region_coordinates_negative_width_raises_error(self, selector_implementation): - """Test that negative width raises InvalidRegionError.""" - with pytest.raises(InvalidRegionError): - selector_implementation.select_region_coordinates( - x=0, - y=0, - width=-100, - height=100, - monitor=0 - ) - - def test_select_region_coordinates_negative_height_raises_error(self, selector_implementation): - """Test that negative height raises InvalidRegionError.""" - with pytest.raises(InvalidRegionError): - selector_implementation.select_region_coordinates( - x=0, - y=0, - width=100, - height=-100, - monitor=0 - ) - - def test_select_region_coordinates_large_values(self, selector_implementation): - """Test select_region_coordinates() with large coordinate values.""" - region = selector_implementation.select_region_coordinates( - x=5000, - y=3000, - width=1920, - height=1080, - monitor=0 - ) - assert region.x == 5000 - assert region.y == 3000 - assert region.width == 1920 - assert region.height == 1080 +class FakeRegionSelector(IRegionSelector): + """Deterministic selector implementation to validate contract behavior.""" - def test_select_region_coordinates_multiple_monitors(self, selector_implementation): - """Test select_region_coordinates() with different monitor indices.""" - region1 = selector_implementation.select_region_coordinates( - x=0, y=0, width=100, height=100, monitor=0 + def __init__(self) -> None: + self._monitors = {0: _Monitor(1920, 1080), 1: _Monitor(2560, 1440)} + self._graphical_mode = "ok" # ok | cancel | tool_missing + + def set_graphical_mode(self, mode: str) -> None: + self._graphical_mode = mode + + def select_region_graphical(self, monitor: int = 0) -> CaptureRegion: + self._ensure_monitor(monitor) + if self._graphical_mode == "cancel": + raise RegionSelectionCancelledError() + if self._graphical_mode == "tool_missing": + raise SelectionToolNotFoundError() + + return CaptureRegion( + x=20, + y=30, + width=640, + height=360, + monitor=monitor, + selection_method="graphical", ) - region2 = selector_implementation.select_region_coordinates( - x=0, y=0, width=100, height=100, monitor=1 + + def select_region_coordinates( + self, x: int, y: int, width: int, height: int, monitor: int = 0 + ) -> CaptureRegion: + monitor_info = self._ensure_monitor(monitor) + + region = CaptureRegion( + x=x, + y=y, + width=width, + height=height, + monitor=monitor, + selection_method="coordinates", ) - assert region1.monitor == 0 - assert region2.monitor == 1 + try: + region.validate(monitor_width=monitor_info.width, monitor_height=monitor_info.height) + except ValueError as exc: + raise InvalidRegionError(str(exc)) from exc + + return region + + def _ensure_monitor(self, monitor: int) -> _Monitor: + if monitor not in self._monitors: + raise InvalidRegionError(f"Monitor {monitor} is not available") + return self._monitors[monitor] - def test_select_region_coordinates_sets_selection_method(self, selector_implementation): - """Test that selection_method is set to 'coordinates'.""" - region = selector_implementation.select_region_coordinates( - x=0, y=0, width=100, height=100, monitor=0 - ) - assert region.selection_method == 'coordinates' +@pytest.fixture() +def selector_implementation() -> FakeRegionSelector: + return FakeRegionSelector() -class TestIRegionSelectorGraphical: - """ - Contract tests for graphical region selection. +def test_interface_inheritance(selector_implementation: FakeRegionSelector) -> None: + assert isinstance(selector_implementation, IRegionSelector) - These tests require mocking or user interaction. - """ - @pytest.fixture - def selector_implementation(self): - """Override in concrete test classes.""" - pytest.skip("Must be implemented by concrete test class") +def test_select_region_graphical_returns_capture_region( + selector_implementation: FakeRegionSelector, +) -> None: + region = selector_implementation.select_region_graphical() - def test_graphical_selection_returns_valid_region(self, selector_implementation): - """Test that graphical selection returns valid CaptureRegion.""" - pytest.skip("Requires user interaction or mocking") + assert isinstance(region, CaptureRegion) + assert region.selection_method == "graphical" + assert region.monitor == 0 + assert region.width > 0 + assert region.height > 0 - def test_graphical_selection_on_multimonitor(self, selector_implementation): - """Test graphical selection on multi-monitor setup.""" - pytest.skip("Requires user interaction or mocking") - def test_graphical_selection_respects_monitor_parameter(self, selector_implementation): - """Test that monitor parameter is respected in graphical selection.""" - pytest.skip("Requires user interaction or mocking") +def test_select_region_graphical_specific_monitor(selector_implementation: FakeRegionSelector) -> None: + region = selector_implementation.select_region_graphical(monitor=1) + assert region.monitor == 1 - def test_graphical_selection_handles_escape_key(self, selector_implementation): - """Test that pressing Escape during selection raises RegionSelectionCancelledError.""" - pytest.skip("Requires user interaction or mocking") - def test_graphical_selection_handles_window_close(self, selector_implementation): - """Test that closing selection window raises RegionSelectionCancelledError.""" - pytest.skip("Requires user interaction or mocking") +def test_select_region_graphical_user_cancels(selector_implementation: FakeRegionSelector) -> None: + selector_implementation.set_graphical_mode("cancel") + with pytest.raises(RegionSelectionCancelledError): + selector_implementation.select_region_graphical() + + +def test_select_region_graphical_tool_not_found(selector_implementation: FakeRegionSelector) -> None: + selector_implementation.set_graphical_mode("tool_missing") + with pytest.raises(SelectionToolNotFoundError): + selector_implementation.select_region_graphical() + + +def test_select_region_coordinates_returns_capture_region( + selector_implementation: FakeRegionSelector, +) -> None: + region = selector_implementation.select_region_coordinates( + x=100, + y=100, + width=400, + height=300, + monitor=0, + ) + + assert isinstance(region, CaptureRegion) + assert region.x == 100 + assert region.y == 100 + assert region.width == 400 + assert region.height == 300 + assert region.monitor == 0 + assert region.selection_method == "coordinates" + + +@pytest.mark.parametrize( + ("x", "y", "width", "height"), + [(-10, 0, 100, 100), (0, -10, 100, 100), (0, 0, 0, 100), (0, 0, 100, 0), (0, 0, -1, 10), (0, 0, 10, -1)], +) +def test_select_region_coordinates_invalid_values_raise_error( + selector_implementation: FakeRegionSelector, + x: int, + y: int, + width: int, + height: int, +) -> None: + with pytest.raises(InvalidRegionError): + selector_implementation.select_region_coordinates( + x=x, + y=y, + width=width, + height=height, + monitor=0, + ) + + +def test_select_region_coordinates_large_values_within_monitor_allowed( + selector_implementation: FakeRegionSelector, +) -> None: + region = selector_implementation.select_region_coordinates( + x=0, + y=0, + width=1920, + height=1080, + monitor=0, + ) + + assert region.width == 1920 + assert region.height == 1080 + + +def test_select_region_coordinates_out_of_bounds_detected( + selector_implementation: FakeRegionSelector, +) -> None: + with pytest.raises(InvalidRegionError): + selector_implementation.select_region_coordinates( + x=1900, + y=1000, + width=100, + height=100, + monitor=0, + ) -class TestIRegionSelectorErrorHandling: - """ - Contract tests for error handling in IRegionSelector implementations. - """ +def test_select_region_coordinates_multiple_monitors(selector_implementation: FakeRegionSelector) -> None: + region_primary = selector_implementation.select_region_coordinates(0, 0, 100, 100, monitor=0) + region_secondary = selector_implementation.select_region_coordinates(0, 0, 100, 100, monitor=1) - @pytest.fixture - def selector_implementation(self): - """Override in concrete test classes.""" - pytest.skip("Must be implemented by concrete test class") + assert region_primary.monitor == 0 + assert region_secondary.monitor == 1 - def test_coordinates_out_of_bounds_detected(self, selector_implementation): - """Test that out-of-bounds coordinates are detected (if implementation validates).""" - # Some implementations may validate against screen bounds - # Others may defer validation to capture stage - # This test documents expected behavior - pytest.skip("Validation behavior depends on implementation") - def test_invalid_monitor_index_handled(self, selector_implementation): - """Test handling of invalid monitor index.""" - # Some implementations may validate monitor exists - # Others may defer to capture stage - pytest.skip("Validation behavior depends on implementation") +def test_invalid_monitor_index_handled(selector_implementation: FakeRegionSelector) -> None: + with pytest.raises(InvalidRegionError): + selector_implementation.select_region_coordinates(0, 0, 10, 10, monitor=99) -# NOTE: Concrete test classes will inherit from these and provide actual implementations -# Example: -# class TestSlurpRegionSelector(TestIRegionSelectorContract): -# @pytest.fixture -# def selector_implementation(self): -# return SlurpRegionSelector() +def test_graphical_selection_invalid_monitor_index_handled( + selector_implementation: FakeRegionSelector, +) -> None: + with pytest.raises(InvalidRegionError): + selector_implementation.select_region_graphical(monitor=99) diff --git a/tests/contract/test_vision_service.py b/tests/contract/test_vision_service.py index a47119c..c9331ce 100644 --- a/tests/contract/test_vision_service.py +++ b/tests/contract/test_vision_service.py @@ -1,406 +1,207 @@ -""" -Contract tests for IVisionService interface. +"""Executable contract tests for IVisionService using VisionService with deterministic doubles.""" -Verifies that all implementations of IVisionService adhere to the contract. -These tests are run against each concrete implementation to ensure consistency. -""" +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path +from unittest.mock import Mock +from uuid import uuid4 import pytest -from uuid import UUID from src.interfaces.screenshot_service import IVisionService -from src.models.entities import CaptureRegion -from src.lib.exceptions import ( - VisionCommandError, - SessionAlreadyActiveError, - DisplayNotAvailableError -) - - -class TestIVisionServiceContract: - """ - Contract test suite for IVisionService interface. - - All implementations of IVisionService MUST pass these tests. - """ - - @pytest.fixture - def service_implementation(self): - """ - Override this fixture in concrete test classes to provide the implementation. - - Example: - @pytest.fixture - def service_implementation(self): - return VisionService(config_manager, capture, processor, api_client, ...) - """ - pytest.skip("Must be implemented by concrete test class") - - def test_interface_inheritance(self, service_implementation): - """Test that implementation inherits from IVisionService.""" - assert isinstance(service_implementation, IVisionService) - - def test_execute_vision_command_returns_string(self, service_implementation): - """Test that execute_vision_command() returns a string response.""" - prompt = "What do you see in this screenshot?" - - response = service_implementation.execute_vision_command(prompt) - - assert isinstance(response, str) - assert len(response) > 0 - - def test_execute_vision_command_with_empty_prompt(self, service_implementation): - """Test execute_vision_command() with empty prompt.""" - # Should still work - image-only analysis - response = service_implementation.execute_vision_command("") - - assert isinstance(response, str) - - def test_execute_vision_command_with_long_prompt(self, service_implementation): - """Test execute_vision_command() with very long prompt.""" - long_prompt = "Analyze this screenshot. " * 200 - - response = service_implementation.execute_vision_command(long_prompt) - - assert isinstance(response, str) - - def test_execute_vision_command_workflow(self, service_implementation): - """ - Test that execute_vision_command() follows the correct workflow: - 1. Capture full screen - 2. Apply privacy zones (if configured) - 3. Optimize image - 4. Send to Claude API - 5. Cleanup temp files - 6. Return response - """ - prompt = "Test vision command" - - # Execute command - response = service_implementation.execute_vision_command(prompt) - - # Should return valid response - assert isinstance(response, str) - assert len(response) > 0 - - # Note: Cleanup verification requires checking temp directory - # This is implementation-specific - - def test_execute_vision_area_command_returns_string(self, service_implementation): - """Test that execute_vision_area_command() returns a string response.""" - prompt = "What do you see in this region?" - - # Define a region - region = CaptureRegion( - x=100, - y=100, - width=400, - height=300, - monitor=0, - selection_method='coordinates' - ) - - response = service_implementation.execute_vision_area_command(prompt, region) - - assert isinstance(response, str) - assert len(response) > 0 - - def test_execute_vision_area_command_with_coordinates(self, service_implementation): - """Test execute_vision_area_command() with pre-defined coordinates.""" - region = CaptureRegion( - x=0, - y=0, - width=800, - height=600, - monitor=0, - selection_method='coordinates' - ) - - response = service_implementation.execute_vision_area_command("Analyze this area", region) - - assert isinstance(response, str) - - def test_execute_vision_area_command_without_region(self, service_implementation): - """Test execute_vision_area_command() without pre-defined region (graphical selection).""" - # When region=None, should trigger graphical selection - # This test may need to be mocked or skipped in headless environments - - try: - response = service_implementation.execute_vision_area_command("Test", region=None) - assert isinstance(response, str) - except (VisionCommandError, DisplayNotAvailableError): - # Expected in headless or when graphical tool not available - pytest.skip("Graphical selection not available in test environment") - - def test_execute_vision_area_command_workflow(self, service_implementation): - """ - Test that execute_vision_area_command() follows the correct workflow: - 1. Select region (graphical or coordinates) - 2. Capture selected region - 3. Apply privacy zones (if configured) - 4. Optimize image - 5. Send to Claude API - 6. Cleanup temp files - 7. Return response - """ - region = CaptureRegion( - x=50, - y=50, - width=500, - height=400, - monitor=0, - selection_method='coordinates' - ) - - response = service_implementation.execute_vision_area_command("Test area", region) - - assert isinstance(response, str) - assert len(response) > 0 - - def test_execute_vision_auto_command_returns_session_id(self, service_implementation): - """Test that execute_vision_auto_command() returns a session UUID.""" - session_id = service_implementation.execute_vision_auto_command(interval_seconds=30) - - assert isinstance(session_id, UUID) - - # Cleanup: stop the session - try: - service_implementation.execute_vision_stop_command() - except: - pass - - def test_execute_vision_auto_command_with_custom_interval(self, service_implementation): - """Test execute_vision_auto_command() with custom interval.""" - session_id = service_implementation.execute_vision_auto_command(interval_seconds=60) - - assert isinstance(session_id, UUID) - - # Cleanup - try: - service_implementation.execute_vision_stop_command() - except: - pass - - def test_execute_vision_auto_command_with_default_interval(self, service_implementation): - """Test execute_vision_auto_command() with default interval from config.""" - session_id = service_implementation.execute_vision_auto_command(interval_seconds=None) - - assert isinstance(session_id, UUID) - - # Cleanup - try: - service_implementation.execute_vision_stop_command() - except: - pass - - def test_execute_vision_auto_command_prevents_multiple_sessions(self, service_implementation): - """Test that only one monitoring session can be active at a time.""" - # Start first session - session_id1 = service_implementation.execute_vision_auto_command(interval_seconds=30) - assert isinstance(session_id1, UUID) - - # Try to start second session - should fail - with pytest.raises(SessionAlreadyActiveError): - service_implementation.execute_vision_auto_command(interval_seconds=30) - - # Cleanup - service_implementation.execute_vision_stop_command() - - def test_execute_vision_stop_command_stops_session(self, service_implementation): - """Test that execute_vision_stop_command() stops active session.""" - # Start session - session_id = service_implementation.execute_vision_auto_command(interval_seconds=30) - assert isinstance(session_id, UUID) - - # Stop session - service_implementation.execute_vision_stop_command() - - # Should be able to start new session now - session_id2 = service_implementation.execute_vision_auto_command(interval_seconds=30) - assert isinstance(session_id2, UUID) - - # Cleanup - service_implementation.execute_vision_stop_command() - - def test_execute_vision_stop_command_without_active_session_raises_error(self, service_implementation): - """Test that stopping non-existent session raises error.""" - # Ensure no active session - try: - service_implementation.execute_vision_stop_command() - except: - pass - - # Try to stop again - should raise error - with pytest.raises(VisionCommandError): - service_implementation.execute_vision_stop_command() - - def test_execute_vision_stop_command_workflow(self, service_implementation): - """ - Test that execute_vision_stop_command() follows correct workflow: - 1. Get active session - 2. Stop session - 3. Cleanup session resources - """ - # Start session - session_id = service_implementation.execute_vision_auto_command(interval_seconds=30) - - # Stop session - service_implementation.execute_vision_stop_command() - - # Verify session is stopped (can start new one) - session_id2 = service_implementation.execute_vision_auto_command(interval_seconds=30) - assert isinstance(session_id2, UUID) - assert session_id2 != session_id # Different session - - # Cleanup - service_implementation.execute_vision_stop_command() - - -class TestIVisionServiceErrorHandling: - """ - Contract tests for error handling in IVisionService implementations. - """ - - @pytest.fixture - def service_implementation(self): - """Override in concrete test classes.""" - pytest.skip("Must be implemented by concrete test class") - - def test_execute_vision_command_headless_raises_error(self, service_implementation, monkeypatch): - """Test that vision command in headless environment raises appropriate error.""" - # Mock headless environment - monkeypatch.delenv('DISPLAY', raising=False) - monkeypatch.delenv('WAYLAND_DISPLAY', raising=False) - - with pytest.raises((VisionCommandError, DisplayNotAvailableError)): - service_implementation.execute_vision_command("Test") - - def test_execute_vision_command_oauth_error_propagates(self, service_implementation): - """Test that OAuth errors are properly propagated.""" - # This test requires mocking the API client to simulate auth failure - pytest.skip("Requires mocking - implement in concrete class") - - def test_execute_vision_area_command_invalid_region_raises_error(self, service_implementation): - """Test that invalid region raises VisionCommandError.""" - # Region with negative coordinates - invalid_region = CaptureRegion( - x=-100, - y=-100, - width=200, - height=200, - monitor=0, - selection_method='coordinates' - ) - - with pytest.raises(VisionCommandError): - service_implementation.execute_vision_area_command("Test", invalid_region) - - def test_execute_vision_auto_command_invalid_interval_raises_error(self, service_implementation): - """Test that invalid interval raises VisionCommandError.""" - # Negative interval - with pytest.raises(VisionCommandError): - service_implementation.execute_vision_auto_command(interval_seconds=-1) - - # Zero interval - with pytest.raises(VisionCommandError): - service_implementation.execute_vision_auto_command(interval_seconds=0) - - -class TestIVisionServiceIntegration: - """ - Integration-style contract tests for complete workflows. - """ - - @pytest.fixture - def service_implementation(self): - """Override in concrete test classes.""" - pytest.skip("Must be implemented by concrete test class") - - @pytest.mark.slow - @pytest.mark.integration - def test_full_vision_workflow_end_to_end(self, service_implementation): - """ - Test complete /vision workflow from capture to response. - - This is an integration test requiring: - - Display available - - Screenshot tools installed - - Valid OAuth configuration - - Network connectivity - """ - try: - response = service_implementation.execute_vision_command( - "Describe what you see in this screenshot briefly." - ) - - assert isinstance(response, str) - assert len(response) > 10 # Should have meaningful response - - except (DisplayNotAvailableError, VisionCommandError) as e: - pytest.skip(f"Integration test skipped: {e}") - - @pytest.mark.slow - @pytest.mark.integration - def test_full_vision_area_workflow_end_to_end(self, service_implementation): - """Test complete /vision.area workflow with coordinates.""" - region = CaptureRegion( - x=0, - y=0, - width=800, - height=600, - monitor=0, - selection_method='coordinates' - ) - - try: - response = service_implementation.execute_vision_area_command( - "What is in this region?", - region - ) - - assert isinstance(response, str) - assert len(response) > 10 - - except (DisplayNotAvailableError, VisionCommandError) as e: - pytest.skip(f"Integration test skipped: {e}") - - @pytest.mark.slow - @pytest.mark.integration - def test_monitoring_session_lifecycle(self, service_implementation): - """Test complete monitoring session lifecycle: start -> run -> stop.""" - try: - # Start session - session_id = service_implementation.execute_vision_auto_command(interval_seconds=60) - assert isinstance(session_id, UUID) - - # Let it run briefly - import time - time.sleep(2) - - # Stop session - service_implementation.execute_vision_stop_command() - - except (DisplayNotAvailableError, VisionCommandError) as e: - pytest.skip(f"Integration test skipped: {e}") - - -# NOTE: Concrete test classes will inherit from these and provide actual implementations -# Example: -# class TestVisionService(TestIVisionServiceContract): -# @pytest.fixture -# def service_implementation(self): -# config_manager = ConfigurationManager() -# temp_manager = TempFileManager() -# capture = ScreenshotCaptureFactory.create() -# processor = PillowImageProcessor() -# api_client = AnthropicAPIClient() -# session_manager = MonitoringSessionManager() -# -# return VisionService( -# config_manager=config_manager, -# temp_manager=temp_manager, -# capture=capture, -# processor=processor, -# api_client=api_client, -# session_manager=session_manager -# ) +from src.lib.exceptions import SessionAlreadyActiveError, VisionCommandError +from src.models.entities import CaptureRegion, Configuration, MonitoringSession, Screenshot +from src.services.vision_service import VisionService + + +@pytest.fixture() +def sample_screenshot(tmp_path: Path) -> Screenshot: + file_path = tmp_path / "capture.png" + file_path.write_bytes(b"img") + size = file_path.stat().st_size + return Screenshot( + id=uuid4(), + timestamp=datetime.now(tz=UTC), + file_path=file_path, + format="png", + original_size_bytes=size, + optimized_size_bytes=size, + resolution=(1920, 1080), + source_monitor=0, + capture_method="test", + privacy_zones_applied=False, + ) + + +@pytest.fixture() +def configured_service(sample_screenshot: Screenshot) -> tuple[VisionService, dict[str, Mock]]: + config = Configuration() + config.privacy.prompt_first_use = False + config.privacy.enabled = False + config.screenshot.max_size_mb = 2.0 + + config_manager = Mock() + config_manager.load_config.return_value = config + + temp_manager = Mock() + capture = Mock() + capture.capture_full_screen.return_value = sample_screenshot + capture.capture_region.return_value = sample_screenshot + + processor = Mock() + processor.apply_privacy_zones.return_value = sample_screenshot + processor.optimize_image.return_value = sample_screenshot + + api_client = Mock() + api_client.send_multimodal_prompt.return_value = "analysis" + + region_selector = Mock() + region_selector.select_region_graphical.return_value = CaptureRegion( + x=10, + y=20, + width=300, + height=200, + monitor=0, + selection_method="graphical", + ) + + session_manager = Mock() + session_manager.start_session.return_value = MonitoringSession( + id=uuid4(), + started_at=datetime.now(tz=UTC), + interval_seconds=30, + ) + session_manager.get_active_session.return_value = MonitoringSession( + id=uuid4(), + started_at=datetime.now(tz=UTC), + interval_seconds=30, + ) + + service = VisionService( + config_manager=config_manager, + temp_manager=temp_manager, + capture=capture, + processor=processor, + api_client=api_client, + region_selector=region_selector, + session_manager=session_manager, + ) + + mocks = { + "config_manager": config_manager, + "temp_manager": temp_manager, + "capture": capture, + "processor": processor, + "api_client": api_client, + "region_selector": region_selector, + "session_manager": session_manager, + } + return service, mocks + + +def test_interface_inheritance(configured_service) -> None: + service, _ = configured_service + assert isinstance(service, IVisionService) + + +def test_execute_vision_command_returns_string(configured_service) -> None: + service, _ = configured_service + result = service.execute_vision_command("What do you see?") + assert isinstance(result, str) + assert result == "analysis" + + +def test_execute_vision_command_with_empty_prompt(configured_service) -> None: + service, mocks = configured_service + result = service.execute_vision_command("") + assert result == "analysis" + mocks["api_client"].send_multimodal_prompt.assert_called_once() + + +def test_execute_vision_command_with_long_prompt(configured_service) -> None: + service, _ = configured_service + result = service.execute_vision_command("Analyze " * 200) + assert isinstance(result, str) + + +def test_execute_vision_command_calls_cleanup(configured_service) -> None: + service, mocks = configured_service + service.execute_vision_command("cleanup") + mocks["temp_manager"].cleanup_temp_file.assert_called_once() + + +def test_execute_vision_area_command_returns_string_with_coordinates(configured_service) -> None: + service, mocks = configured_service + region = CaptureRegion(x=0, y=0, width=100, height=100, monitor=0, selection_method="coordinates") + + result = service.execute_vision_area_command("Area", region=region) + + assert result == "analysis" + mocks["capture"].capture_region.assert_called_once_with(region) + + +def test_execute_vision_area_command_without_region_uses_selector(configured_service) -> None: + service, mocks = configured_service + + result = service.execute_vision_area_command("Area") + + assert result == "analysis" + mocks["region_selector"].select_region_graphical.assert_called_once() + + +def test_execute_vision_area_command_invalid_region_raises_vision_error(configured_service) -> None: + service, mocks = configured_service + bad_region = CaptureRegion(x=-1, y=0, width=10, height=10, monitor=0, selection_method="coordinates") + mocks["capture"].capture_region.side_effect = ValueError("invalid") + + with pytest.raises(VisionCommandError): + service.execute_vision_area_command("x", bad_region) + + +def test_execute_vision_auto_command_returns_session_id(configured_service) -> None: + service, _ = configured_service + session_id = service.execute_vision_auto_command(interval_seconds=30) + assert session_id is not None + + +def test_execute_vision_auto_command_uses_default_interval(configured_service) -> None: + service, mocks = configured_service + service.execute_vision_auto_command(interval_seconds=None) + mocks["session_manager"].start_session.assert_called_once_with(30) + + +def test_execute_vision_auto_command_invalid_interval_raises_error(configured_service) -> None: + service, _ = configured_service + with pytest.raises(VisionCommandError, match="Interval must be positive"): + service.execute_vision_auto_command(interval_seconds=0) + + +def test_execute_vision_auto_command_prevents_multiple_sessions(configured_service) -> None: + service, mocks = configured_service + mocks["session_manager"].start_session.side_effect = SessionAlreadyActiveError("abc") + + with pytest.raises(VisionCommandError, match="already active"): + service.execute_vision_auto_command(interval_seconds=30) + + +def test_execute_vision_stop_command_stops_active_session(configured_service) -> None: + service, mocks = configured_service + service.execute_vision_stop_command() + mocks["session_manager"].stop_session.assert_called_once() + + +def test_execute_vision_stop_command_without_active_session_raises_error(configured_service) -> None: + service, mocks = configured_service + mocks["session_manager"].get_active_session.return_value = None + + with pytest.raises(VisionCommandError, match="No active monitoring session"): + service.execute_vision_stop_command() + + +def test_execute_vision_command_api_error_is_wrapped(configured_service) -> None: + service, mocks = configured_service + mocks["api_client"].send_multimodal_prompt.side_effect = RuntimeError("api down") + + with pytest.raises(VisionCommandError, match="Failed to execute vision command"): + service.execute_vision_command("hello")