diff --git a/src/purple_mcp/libs/alerts/models.py b/src/purple_mcp/libs/alerts/models.py index b1d6a45..1905bb5 100644 --- a/src/purple_mcp/libs/alerts/models.py +++ b/src/purple_mcp/libs/alerts/models.py @@ -609,6 +609,8 @@ class AIInvestigation(BaseModel): restriction_reason: Reason the investigation was restricted, if applicable. """ + model_config = _camel_case_model_config + alert_id: str = Field(alias="alertId") result: str | None = None status: str | None = None diff --git a/tests/unit/libs/alerts/helpers/__init__.py b/tests/unit/libs/alerts/helpers/__init__.py index 4e98ff9..602ba32 100644 --- a/tests/unit/libs/alerts/helpers/__init__.py +++ b/tests/unit/libs/alerts/helpers/__init__.py @@ -1,13 +1,8 @@ """Test helpers for alerts unit tests.""" -import json from typing import TypeVar -from unittest.mock import AsyncMock -import pytest - -from purple_mcp.libs.alerts import Alert, AlertConnection, AlertNote, PageInfo -from purple_mcp.type_defs import JsonDict +from purple_mcp.libs.alerts import PageInfo T = TypeVar("T") @@ -15,32 +10,6 @@ class MockAlertsClientBuilder: """Factory for creating mock alerts clients with common responses.""" - @staticmethod - def create_mock( - method_name: str, - return_value: object = None, - side_effect: Exception | None = None, - ) -> AsyncMock: - """Create a mock client with specified method behavior. - - Args: - method_name: Name of the method to mock - return_value: Value to return when method is called - side_effect: Exception to raise when method is called - - Returns: - AsyncMock client with configured method - """ - mock_client = AsyncMock() - method = getattr(mock_client, method_name) - - if side_effect: - method.side_effect = side_effect - else: - method.return_value = return_value - - return mock_client - @staticmethod def create_empty_connection(connection_type: type[T]) -> T: """Create an empty connection response. @@ -60,243 +29,3 @@ def create_empty_connection(connection_type: type[T]) -> T: endCursor=None, ), ) - - @staticmethod - def create_alert_connection(alerts: list[Alert] | None = None) -> AlertConnection: - """Create an AlertConnection with provided alerts. - - Args: - alerts: List of alerts to include in edges - - Returns: - AlertConnection with alerts as edges - """ - if alerts is None: - return MockAlertsClientBuilder.create_empty_connection(AlertConnection) - - from purple_mcp.libs.alerts.models import AlertEdge - - edges = [AlertEdge(node=alert, cursor=f"cursor-{alert.id}") for alert in alerts] - - return AlertConnection( - edges=edges, - pageInfo=PageInfo( - hasNextPage=len(edges) > 0, - hasPreviousPage=False, - startCursor=edges[0].cursor if edges else None, - endCursor=edges[-1].cursor if edges else None, - ), - ) - - -class JSONAssertions: - """Helper methods for common JSON response assertions.""" - - @staticmethod - def assert_connection_response(result: str, expected_edges: int | None = None) -> JsonDict: - """Assert that a JSON response has connection structure. - - Args: - result: JSON string response - expected_edges: Expected number of edges (if specified) - - Returns: - Parsed JSON data - - Raises: - AssertionError: If response doesn't match expectations - """ - data: JsonDict = json.loads(result) - assert "edges" in data, "Response missing 'edges' field" - assert "page_info" in data, "Response missing 'page_info' field" - - if expected_edges is not None: - edges = data["edges"] - assert isinstance(edges, list), "edges field must be a list" - actual_edges = len(edges) - assert actual_edges == expected_edges, ( - f"Expected {expected_edges} edges, got {actual_edges}" - ) - - return data - - @staticmethod - def assert_alert_response(result: str, alert_id: str | None = None) -> JsonDict: - """Assert that a JSON response contains valid alert data. - - Args: - result: JSON string response - alert_id: Expected alert ID (if specified) - - Returns: - Parsed JSON data - - Raises: - AssertionError: If response doesn't match expectations - """ - data: JsonDict = json.loads(result) - assert "id" in data, "Response missing 'id' field" - assert "severity" in data, "Response missing 'severity' field" - assert "name" in data, "Response missing 'name' field" - - if alert_id is not None: - assert data["id"] == alert_id, f"Expected alert ID {alert_id}, got {data['id']}" - - return data - - @staticmethod - def assert_note_response(result: str, note_id: str | None = None) -> JsonDict: - """Assert that a JSON response contains valid note data. - - Args: - result: JSON string response - note_id: Expected note ID (if specified) - - Returns: - Parsed JSON data - - Raises: - AssertionError: If response doesn't match expectations - """ - data: JsonDict = json.loads(result) - assert "id" in data, "Response missing 'id' field" - assert "text" in data, "Response missing 'text' field" - assert "created_at" in data, "Response missing 'created_at' field" - - if note_id is not None: - assert data["id"] == note_id, f"Expected note ID {note_id}, got {data['id']}" - - return data - - @staticmethod - def assert_error_message( - exc_info: pytest.ExceptionInfo[BaseException], - expected_message: str, - expected_cause_message: str | None = None, - ) -> None: - """Assert that an exception contains expected message and optionally validates cause. - - Args: - exc_info: pytest.ExceptionInfo instance - expected_message: Expected error message substring - expected_cause_message: Optional expected underlying cause - message substring. If provided, validates exception - chaining. - - Raises: - AssertionError: If message not found in exception or cause - validation fails. - """ - actual_message = str(exc_info.value) - assert expected_message in actual_message, ( - f"Expected error message to contain '{expected_message}', but got: '{actual_message}'" - ) - - # If cause message expected, validate exception chaining - if expected_cause_message is not None: - cause = exc_info.value.__cause__ - assert cause is not None, ( - "Expected exception to have underlying cause, but __cause__ was None." - ) - assert expected_cause_message in str(cause) - - @staticmethod - def assert_null_response(result: str) -> None: - """Assert that a JSON response is null. - - Args: - result: JSON string response - - Raises: - AssertionError: If response is not null - """ - data = json.loads(result) - assert data is None, f"Expected null response, got: {data}" - - -class AlertsTestData: - """Common test data for alerts tests.""" - - @staticmethod - def create_test_alert( - alert_id: str = "alert-123", - name: str = "Test Alert", - severity: str = "HIGH", - status: str = "NEW", - ) -> Alert: - """Create a test alert with default or custom values. - - Args: - alert_id: Alert ID - name: Alert name - severity: Alert severity - status: Alert status - - Returns: - Alert instance for testing - """ - from purple_mcp.libs.alerts import Severity, Status - - return Alert( - id=alert_id, - name=name, - severity=Severity(severity), - status=Status(status), - detectedAt="2024-01-01T00:00:00Z", - ) - - @staticmethod - def create_test_note( - note_id: str = "note-123", - text: str = "Test note", - alert_id: str = "alert-123", - ) -> AlertNote: - """Create a test note with default or custom values. - - Args: - note_id: Note ID - text: Note text - alert_id: Associated alert ID - - Returns: - AlertNote instance for testing - """ - return AlertNote( - id=note_id, - text=text, - createdAt="2024-01-01T00:00:00Z", - alertId=alert_id, - ) - - @staticmethod - def create_page_info( - has_next: bool = False, - has_prev: bool = False, - start_cursor: str | None = None, - end_cursor: str | None = None, - ) -> JsonDict: - """Create page info dict for connection responses. - - Args: - has_next: Whether there's a next page - has_prev: Whether there's a previous page - start_cursor: Start cursor value - end_cursor: End cursor value - - Returns: - Page info dictionary - """ - return { - "hasNextPage": has_next, - "hasPreviousPage": has_prev, - "startCursor": start_cursor, - "endCursor": end_cursor, - } - - -# Export all helpers -__all__ = [ - "AlertsTestData", - "JSONAssertions", - "MockAlertsClientBuilder", -] diff --git a/tests/unit/libs/alerts/helpers/base.py b/tests/unit/libs/alerts/helpers/base.py deleted file mode 100644 index 017b9dc..0000000 --- a/tests/unit/libs/alerts/helpers/base.py +++ /dev/null @@ -1,232 +0,0 @@ -"""Base test class for alerts functionality.""" - -import json -from collections.abc import Callable, Mapping -from typing import Any, Protocol, TypeVar -from unittest.mock import AsyncMock, Mock - -import pytest - -from purple_mcp.libs.alerts import ( - Alert, - AlertConnection, - AlertHistoryConnection, - AlertNote, - AlertNoteConnection, -) -from purple_mcp.libs.alerts.models import PageInfo -from purple_mcp.type_defs import JsonDict -from tests.unit.libs.alerts.helpers import JSONAssertions, MockAlertsClientBuilder - -T = TypeVar("T") - - -class ToolFunction(Protocol): - """Protocol for tool functions that can be tested. - - Uses Any for maximum flexibility since test helpers need to work with - diverse tool function signatures (get_alert(str), search_alerts(list[JsonDict], int, str), etc.) - and don't require type safety for passed-through arguments. - """ - - async def __call__(self, *args: Any, **kwargs: Any) -> str: - """Execute the tool with the given arguments.""" - ... - - -class AlertsTestBase: - """Base class with common test utilities for alerts tests.""" - - @staticmethod - async def assert_tool_success( - tool_func: ToolFunction, - mock_get_client: Mock, - mock_response: str - | JsonDict - | list[JsonDict] - | Alert - | AlertConnection - | AlertNote - | AlertNoteConnection - | AlertHistoryConnection - | None, - expected_client_method: str, - expected_client_args: JsonDict | None = None, - tool_args: JsonDict | None = None, - response_validator: Callable[[str], None] | None = None, - ) -> str: - """Test successful tool execution with common assertions. - - Args: - tool_func: The tool function to test - mock_get_client: Mock for _get_alerts_client - mock_response: Response to return from mock client - expected_client_method: Expected method to be called on client - expected_client_args: Expected arguments for client method (None = just check called) - tool_args: Arguments to pass to tool function - response_validator: Optional function to validate response - - Returns: - The tool function's response - - Example: - await self.assert_tool_success( - alerts.get_alert, - mock_get_client, - mock_alert, - "get_alert", - None, # Just verify method was called - {"alert_id": "123"} - ) - """ - # Setup mock - mock_client = MockAlertsClientBuilder.create_mock(expected_client_method, mock_response) - mock_get_client.return_value = mock_client - - # Execute tool - result = await tool_func(**(tool_args or {})) - - # Verify client method was called - method = getattr(mock_client, expected_client_method) - if expected_client_args is not None: - method.assert_called_once_with(**expected_client_args) - else: - method.assert_called_once() - - # Validate response if validator provided - if response_validator: - response_validator(result) - - return result - - @staticmethod - async def assert_tool_error( - tool_func: ToolFunction, - mock_get_client: Mock, - mock_side_effect: Exception, - expected_client_method: str, - expected_error_message: str, - expected_client_args: JsonDict | None = None, - tool_args: JsonDict | None = None, - ) -> None: - """Test tool error handling with common assertions. - - Args: - tool_func: The tool function to test - mock_get_client: Mock for _get_alerts_client - mock_side_effect: Exception to raise from mock client - expected_client_method: Expected method to be called on client - expected_error_message: Expected error message substring - expected_client_args: Expected arguments to be passed to client method - tool_args: Arguments to pass to tool function - - Example: - await self.assert_tool_error( - alerts.get_alert, - mock_get_client, - AlertsClientError("Network error"), - "get_alert", - "Failed to retrieve alert", - {"alert_id": "123"} - ) - """ - # Setup mock to raise error - mock_client = MockAlertsClientBuilder.create_mock( - expected_client_method, side_effect=mock_side_effect - ) - mock_get_client.return_value = mock_client - - # Execute and expect error - with pytest.raises(RuntimeError) as exc_info: - await tool_func(**(tool_args or {})) - - JSONAssertions.assert_error_message(exc_info, expected_error_message) - - @staticmethod - async def assert_tool_validation_error( - tool_func: ToolFunction, - tool_args: Mapping[str, object], - expected_error: str, - ) -> None: - """Test tool parameter validation. - - Args: - tool_func: The tool function to test - tool_args: Invalid arguments to pass to tool - expected_error: Expected validation error message - - Example: - await self.assert_tool_validation_error( - alerts.list_alerts, - {"first": 0}, - "first must be between 1 and 100" - ) - """ - with pytest.raises(ValueError) as exc_info: - await tool_func(**tool_args) - - JSONAssertions.assert_error_message(exc_info, expected_error) - - @staticmethod - def assert_json_response_equals( - result: str, expected: str | JsonDict | list[JsonDict] - ) -> None: - """Assert JSON response equals expected value. - - Args: - result: JSON string response - expected: Expected value (will be JSON encoded for comparison) - """ - actual = json.loads(result) - expected_json = json.loads(json.dumps(expected, default=str)) - assert actual == expected_json, f"Expected {expected_json}, got {actual}" - - @staticmethod - def create_mock_with_connection( - method_name: str, - connection_type: type[AlertConnection], - items: list[Alert] | None = None, - ) -> AsyncMock: - """Create a mock client that returns a connection response. - - Args: - method_name: Client method name - connection_type: Type of connection to create - items: Items to include in connection edges - - Returns: - Configured mock client - - Example: - mock_client = self.create_mock_with_connection( - "list_alerts", - AlertConnection, - [test_alert1, test_alert2] - ) - """ - if items: - # Create connection with items - from purple_mcp.libs.alerts.models import AlertEdge, AlertHistoryEdge, AlertNoteEdge - - edge_class = { - "AlertConnection": AlertEdge, - "AlertNoteConnection": AlertNoteEdge, - "AlertHistoryConnection": AlertHistoryEdge, - }.get(connection_type.__name__, AlertEdge) - - edges = [edge_class(node=item, cursor=f"cursor-{item.id}") for item in items] - - connection = connection_type( - edges=edges, - pageInfo=PageInfo( - hasNextPage=False, - hasPreviousPage=False, - startCursor=edges[0].cursor if edges else None, - endCursor=edges[-1].cursor if edges else None, - ), - ) - else: - # Empty connection - connection = MockAlertsClientBuilder.create_empty_connection(connection_type) - - return MockAlertsClientBuilder.create_mock(method_name, connection) diff --git a/tests/unit/libs/alerts/test_client.py b/tests/unit/libs/alerts/test_client.py index 1987b8d..0b0b313 100644 --- a/tests/unit/libs/alerts/test_client.py +++ b/tests/unit/libs/alerts/test_client.py @@ -757,3 +757,76 @@ async def test_get_history_with_pagination( assert result.page_info.has_previous_page is True assert result.total_count == 0 + + +class TestGetAlertInvestigationReport: + """Test get_alert_investigation_report method.""" + + @pytest.mark.asyncio + async def test_successful_get_report( + self, config: AlertsConfig, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Test successful investigation report retrieval.""" + client = AlertsClient(config) + response_data: JsonDict = { + "aiInvestigations": [ + { + "alertId": "alert-123", + "result": "# Investigation Report\nNo threats found.", + "status": "COMPLETED", + "verdict": "FALSE_POSITIVE", + "timestamp": "2024-01-01T00:00:00Z", + "purpleAiStatus": "DONE", + "investigationStep": None, + } + ] + } + + mock_execute_qry = AsyncMock(return_value=response_data) + monkeypatch.setattr( + client, AlertsClient.execute_compatible_query.__name__, mock_execute_qry + ) + + result = await client.get_alert_investigation_report("alert-123") + + assert result is not None + assert result.alert_id == "alert-123" + assert result.result == "# Investigation Report\nNo threats found." + assert result.status == "COMPLETED" + assert result.verdict == "FALSE_POSITIVE" + assert result.timestamp == "2024-01-01T00:00:00Z" + assert result.purple_ai_status == "DONE" + + @pytest.mark.asyncio + async def test_returns_none_when_empty_list( + self, config: AlertsConfig, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Test that None is returned when no investigation exists.""" + client = AlertsClient(config) + response_data: JsonDict = {"aiInvestigations": []} + + mock_execute_qry = AsyncMock(return_value=response_data) + monkeypatch.setattr( + client, AlertsClient.execute_compatible_query.__name__, mock_execute_qry + ) + + result = await client.get_alert_investigation_report("alert-123") + + assert result is None + + @pytest.mark.asyncio + async def test_returns_none_when_key_missing( + self, config: AlertsConfig, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Test that None is returned when response lacks aiInvestigations key.""" + client = AlertsClient(config) + response_data: JsonDict = {} + + mock_execute_qry = AsyncMock(return_value=response_data) + monkeypatch.setattr( + client, AlertsClient.execute_compatible_query.__name__, mock_execute_qry + ) + + result = await client.get_alert_investigation_report("alert-123") + + assert result is None diff --git a/tests/unit/libs/test_alerts_field_selection.py b/tests/unit/libs/test_alerts_field_selection.py deleted file mode 100644 index 6bd1787..0000000 --- a/tests/unit/libs/test_alerts_field_selection.py +++ /dev/null @@ -1,146 +0,0 @@ -"""Unit tests for alerts library dynamic field selection.""" - -from purple_mcp.libs.alerts.client import ALLOWED_ALERT_FIELDS, DEFAULT_ALERT_FIELDS -from purple_mcp.libs.graphql_utils import build_node_fields - - -class TestAlertsDefaultFields: - """Test that default alert fields are properly defined.""" - - def test_not_empty(self) -> None: - """Test that DEFAULT_ALERT_FIELDS is not empty.""" - assert len(DEFAULT_ALERT_FIELDS) > 0 - - def test_contains_required_fields(self) -> None: - """Test that DEFAULT_ALERT_FIELDS contains essential fields.""" - required_fields = ["id", "severity", "status", "name"] - field_string = " ".join(DEFAULT_ALERT_FIELDS) - - for field in required_fields: - assert field in field_string - - -class TestAlertsBuildNodeFields: - """Test build_node_fields with alert field defaults.""" - - def test_with_none_returns_defaults(self) -> None: - """Test that passing None returns all default fields.""" - result = build_node_fields(None, DEFAULT_ALERT_FIELDS) - - # Should contain all default fields - for field in DEFAULT_ALERT_FIELDS: - assert field in result - - # Should have proper indentation - assert " id" in result - - def test_with_minimal_fields(self) -> None: - """Test building query with minimal fields.""" - result = build_node_fields(["id"], DEFAULT_ALERT_FIELDS) - - assert " id" in result - assert "severity" not in result - assert "status" not in result - - def test_with_custom_fields(self) -> None: - """Test building query with custom field selection.""" - custom_fields = ["id", "severity", "status", "name"] - result = build_node_fields(custom_fields, DEFAULT_ALERT_FIELDS) - - for field in custom_fields: - assert field in result - - # Should not contain other fields - assert "description" not in result - assert "detectedAt" not in result - - def test_with_nested_objects(self) -> None: - """Test that nested object fields are included correctly.""" - fields = ["id", "asset { id name type }"] - result = build_node_fields(fields, DEFAULT_ALERT_FIELDS) - - assert "id" in result - assert "asset { id name type }" in result - - -class TestAlertsAutoExpansion: - """Test auto-expansion of nested objects for alerts.""" - - def test_auto_expand_asset(self) -> None: - """Test that 'asset' expands to full fragment.""" - result = build_node_fields(["id", "asset"], DEFAULT_ALERT_FIELDS) - - assert " id" in result - assert " asset { id name type }" in result - # Should not contain just "asset" alone - lines = result.split("\n") - assert " asset" not in lines - - def test_auto_expand_assignee(self) -> None: - """Test that 'assignee' expands to full fragment (no id in alerts API).""" - result = build_node_fields(["id", "assignee"], DEFAULT_ALERT_FIELDS) - - assert " id" in result - # Assignee in alerts API uses userId (not id), so no auto-prepend - assert " assignee { userId email fullName }" in result - - def test_auto_expand_detection_source(self) -> None: - """Test that 'detectionSource' expands to full fragment.""" - result = build_node_fields(["id", "detectionSource"], DEFAULT_ALERT_FIELDS) - - assert " id" in result - assert " detectionSource { product vendor }" in result - - def test_explicit_fragment_not_expanded(self) -> None: - """Test that explicit fragments are used as-is, not expanded.""" - # Request only asset.id using explicit fragment - result = build_node_fields(["id", "asset { id }"], DEFAULT_ALERT_FIELDS) - - assert " id" in result - assert " asset { id }" in result - # Should NOT expand to the full default - assert "asset { id name type }" not in result - - def test_mixed_simple_and_nested_fields(self) -> None: - """Test mixing simple fields with auto-expanded nested fields.""" - result = build_node_fields(["id", "severity", "asset", "status"], DEFAULT_ALERT_FIELDS) - - assert " id" in result - assert " severity" in result - assert " asset { id name type }" in result - assert " status" in result - - def test_multiple_nested_objects_auto_expand(self) -> None: - """Test multiple nested objects can be auto-expanded together.""" - result = build_node_fields( - ["id", "asset", "assignee", "detectionSource"], DEFAULT_ALERT_FIELDS - ) - - assert " asset { id name type }" in result - # Assignee in alerts API uses userId (not id), so no auto-prepend - assert " assignee { userId email fullName }" in result - # detectionSource doesn't have id in the schema, so no id prepended - assert " detectionSource { product vendor }" in result - - -class TestAlertsDataSourcesHandling: - """Test special handling of dataSources field for alerts.""" - - def test_data_sources_field_allowed(self) -> None: - """Test that dataSources field is in the allowlist and can be requested.""" - # dataSources is in the allowlist for custom field selection - result = build_node_fields(["id", "dataSources"], ALLOWED_ALERT_FIELDS) - - assert " id" in result - assert " dataSources" in result - - def test_data_sources_not_in_defaults(self) -> None: - """Test that dataSources is NOT in DEFAULT_ALERT_FIELDS to avoid conflicts.""" - # dataSources should NOT be in default fields because it's added via template substitution - assert "dataSources" not in DEFAULT_ALERT_FIELDS - field_string = " ".join(DEFAULT_ALERT_FIELDS) - assert "dataSources" not in field_string - - def test_data_sources_in_allowlist(self) -> None: - """Test that dataSources IS in ALLOWED_ALERT_FIELDS for custom field selection.""" - assert "dataSources" in ALLOWED_ALERT_FIELDS diff --git a/tests/unit/test_alerts_dos_protection.py b/tests/unit/test_alerts_dos_protection.py deleted file mode 100644 index 3369e72..0000000 --- a/tests/unit/test_alerts_dos_protection.py +++ /dev/null @@ -1,248 +0,0 @@ -"""Tests for DoS protection in alerts tools.""" - -import json -from unittest.mock import Mock, patch - -import pytest -from pydantic import JsonValue - -from purple_mcp.libs.alerts import AlertConnection -from purple_mcp.tools import alerts -from purple_mcp.type_defs import JsonDict -from tests.unit.libs.alerts.helpers.base import AlertsTestBase - - -class TestSearchAlertsDoSProtection(AlertsTestBase): - """Test DoS protection for search_alerts tool.""" - - @pytest.mark.asyncio - async def test_too_many_filters(self) -> None: - """Test that too many filters are rejected.""" - # Create 51 filters (over the limit of 50) - filters: list[dict[str, JsonValue]] = [] - for i in range(51): - filters.append( - {"fieldId": f"field{i}", "filterType": "string_equals", "value": f"value{i}"} - ) - - tool_args = {"filters": json.dumps(filters)} - await self.assert_tool_validation_error( - alerts.search_alerts, - tool_args, - "Too many filters: 51. Maximum allowed: 50", - ) - - @pytest.mark.asyncio - async def test_exactly_max_filters_allowed(self) -> None: - """Test that exactly 50 filters (at the limit) are allowed.""" - # Create exactly 50 filters (at the limit) - filters: list[dict[str, JsonValue]] = [] - for i in range(50): - filters.append( - {"fieldId": f"field{i}", "filterType": "string_equals", "value": f"value{i}"} - ) - - # Mock the client to avoid actual API calls - with patch("purple_mcp.tools.alerts._get_alerts_client") as mock_get_client: - mock_client = self.create_mock_with_connection("search_alerts", AlertConnection) - mock_get_client.return_value = mock_client - - # Should not raise an error - result = await alerts.search_alerts(filters=json.dumps(filters)) - assert result is not None - - @pytest.mark.parametrize( - ("filter_dict", "expected_error"), - [ - pytest.param( - {"fieldId": "severity", "filterType": "string_in", "values": ["HIGH"] * 101}, - "Filter 0 has too many values: 101. Maximum allowed: 100", - id="string-in-too-many-values", - ), - pytest.param( - {"fieldId": "priority", "filterType": "int_in", "values": list(range(101))}, - "Filter 0 has too many values: 101. Maximum allowed: 100", - id="int-in-too-many-values", - ), - pytest.param( - {"fieldId": "description", "filterType": "fulltext", "values": ["term"] * 101}, - "Filter 0 has too many values: 101. Maximum allowed: 100", - id="fulltext-too-many-values", - ), - pytest.param( - { - "fieldId": "status", - "filterType": "string_in", - "values": ["OPEN"] * 101, - "isNegated": True, - }, - "Filter 0 has too many values: 101. Maximum allowed: 100", - id="negated-string-in-too-many-values", - ), - ], - ) - @pytest.mark.asyncio - async def test_filter_with_too_many_values( - self, filter_dict: JsonDict, expected_error: str - ) -> None: - """Test that filters with too many values are rejected.""" - await self.assert_tool_validation_error( - alerts.search_alerts, {"filters": json.dumps([filter_dict])}, expected_error - ) - - @pytest.mark.parametrize( - "filter_dict", - [ - pytest.param( - {"fieldId": "severity", "filterType": "string_in", "values": ["HIGH"] * 100}, - id="string-in-max-values", - ), - pytest.param( - {"fieldId": "priority", "filterType": "int_in", "values": list(range(100))}, - id="int-in-max-values", - ), - pytest.param( - {"fieldId": "description", "filterType": "fulltext", "values": ["term"] * 100}, - id="fulltext-max-values", - ), - pytest.param( - { - "fieldId": "status", - "filterType": "string_in", - "values": ["OPEN"] * 100, - "isNegated": True, - }, - id="negated-string-in-max-values", - ), - ], - ) - @patch("purple_mcp.tools.alerts._get_alerts_client") - @pytest.mark.asyncio - async def test_filter_with_exactly_max_values_allowed( - self, mock_get_client: Mock, filter_dict: JsonDict - ) -> None: - """Test that filters with exactly 100 values (at the limit) are allowed.""" - mock_client = self.create_mock_with_connection("search_alerts", AlertConnection) - mock_get_client.return_value = mock_client - - # Should not raise an error - result = await alerts.search_alerts(filters=json.dumps([filter_dict])) - assert result is not None - - @pytest.mark.asyncio - async def test_early_validation_catches_oversized_values_array(self) -> None: - """Test that the early validation in search_alerts catches oversized arrays.""" - # This tests the early validation before filter parsing - filter_with_oversized_values: JsonDict = { - "field": "severity", - "operator": "IN", - "values": ["HIGH"] * 101, # This should be caught by early validation - } - - await self.assert_tool_validation_error( - alerts.search_alerts, - {"filters": json.dumps([filter_with_oversized_values])}, - "Filter 0 has too many values: 101. Maximum allowed: 100", - ) - - @pytest.mark.asyncio - async def test_mixed_valid_and_invalid_filters(self) -> None: - """Test scenario with some valid filters and one invalid filter.""" - filters: list[dict[str, JsonValue]] = [ - # Valid filters - {"fieldId": "severity", "filterType": "string_equals", "value": "HIGH"}, - {"fieldId": "status", "filterType": "string_equals", "value": "OPEN"}, - # Invalid filter with too many values - {"fieldId": "tags", "filterType": "string_in", "values": ["tag"] * 101}, - ] - - await self.assert_tool_validation_error( - alerts.search_alerts, - {"filters": json.dumps(filters)}, - "Filter 2 has too many values: 101. Maximum allowed: 100", - ) - - @patch("purple_mcp.tools.alerts._get_alerts_client") - @pytest.mark.asyncio - async def test_complex_valid_scenario(self, mock_get_client: Mock) -> None: - """Test a complex but valid scenario with many filters and values.""" - mock_client = self.create_mock_with_connection("search_alerts", AlertConnection) - mock_get_client.return_value = mock_client - - # Create a complex but valid scenario - filters: list[dict[str, JsonValue]] = [] - - # Add 25 filters with single values - for i in range(25): - filters.append( - {"fieldId": f"field{i}", "filterType": "string_equals", "value": f"value{i}"} - ) - - # Add 5 filters with exactly 100 values each - for i in range(5): - filter_dict: dict[str, JsonValue] = { - "fieldId": f"multi_field{i}", - "filterType": "string_in", - "values": [f"value{j}" for j in range(100)], - } - filters.append(filter_dict) - - # Should not raise an error (30 filters total, all within limits) - result = await alerts.search_alerts(filters=json.dumps(filters), first=50) - assert result is not None - - @patch("purple_mcp.tools.alerts._get_alerts_client") - @pytest.mark.asyncio - async def test_non_list_values_not_affected(self, mock_get_client: Mock) -> None: - """Test that non-list values in filters are not affected by validation.""" - mock_client = self.create_mock_with_connection("search_alerts", AlertConnection) - mock_get_client.return_value = mock_client - - # Filters with non-list values should work fine - # Filters with non-list values should work fine - filters: list[dict[str, JsonValue]] = [ - {"fieldId": "severity", "filterType": "string_equals", "value": "HIGH"}, - {"fieldId": "priority", "filterType": "int_equals", "value": 5}, - {"fieldId": "isResolved", "filterType": "boolean_equals", "value": False}, - ] - - result = await alerts.search_alerts(filters=json.dumps(filters)) - assert result is not None - - @patch("purple_mcp.tools.alerts._get_alerts_client") - @pytest.mark.asyncio - async def test_empty_filters_list_allowed(self, mock_get_client: Mock) -> None: - """Test that empty filters list is allowed.""" - mock_client = self.create_mock_with_connection("search_alerts", AlertConnection) - mock_get_client.return_value = mock_client - - result = await alerts.search_alerts(filters=json.dumps([])) - assert result is not None - - @patch("purple_mcp.tools.alerts._get_alerts_client") - @pytest.mark.asyncio - async def test_none_filters_allowed(self, mock_get_client: Mock) -> None: - """Test that None filters is allowed.""" - mock_client = self.create_mock_with_connection("search_alerts", AlertConnection) - mock_get_client.return_value = mock_client - - result = await alerts.search_alerts(filters=None) - assert result is not None - - -class TestDoSProtectionConstants: - """Test DoS protection constants are correctly defined.""" - - def test_constants_exist(self) -> None: - """Test that DoS protection constants exist and have expected values.""" - assert hasattr(alerts, "MAX_FILTERS_COUNT") - assert hasattr(alerts, "MAX_FILTER_VALUES_COUNT") - assert alerts.MAX_FILTERS_COUNT == 50 - assert alerts.MAX_FILTER_VALUES_COUNT == 100 - - def test_constants_are_integers(self) -> None: - """Test that constants are proper integers.""" - assert isinstance(alerts.MAX_FILTERS_COUNT, int) - assert isinstance(alerts.MAX_FILTER_VALUES_COUNT, int) - assert alerts.MAX_FILTERS_COUNT > 0 - assert alerts.MAX_FILTER_VALUES_COUNT > 0 diff --git a/tests/unit/tools/test_alerts_investigation_report.py b/tests/unit/tools/test_alerts_investigation_report.py new file mode 100644 index 0000000..da2447d --- /dev/null +++ b/tests/unit/tools/test_alerts_investigation_report.py @@ -0,0 +1,100 @@ +"""Tests for the get_alert_investigation_report tool.""" + +import json +from unittest.mock import AsyncMock, create_autospec + +import pytest + +from purple_mcp.libs.alerts import AIInvestigation, AlertsClient +from purple_mcp.libs.alerts.exceptions import AlertsGraphQLError +from purple_mcp.tools import alerts + + +class TestGetAlertInvestigationReport: + """Test get_alert_investigation_report tool.""" + + @pytest.mark.asyncio + async def test_returns_report_json(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test successful report retrieval returns JSON-serialised model.""" + fake_report = AIInvestigation( + alertId="alert-123", + result="# Report\nNo threats found.", + status="COMPLETED", + verdict="FALSE_POSITIVE", + timestamp="2024-01-01T00:00:00Z", + purpleAiStatus="DONE", + ) + + mock_client = create_autospec(AlertsClient, spec_set=True, instance=True) + mock_client.get_alert_investigation_report = AsyncMock(return_value=fake_report) + monkeypatch.setattr(alerts, alerts._get_alerts_client.__name__, lambda: mock_client) + + result = await alerts.get_alert_investigation_report("alert-123") + + mock_client.get_alert_investigation_report.assert_called_once_with(alert_id="alert-123") + data = json.loads(result) + assert data["alertId"] == "alert-123" + assert data["result"] == "# Report\nNo threats found." + assert data["status"] == "COMPLETED" + assert data["verdict"] == "FALSE_POSITIVE" + assert data["purpleAiStatus"] == "DONE" + + @pytest.mark.asyncio + async def test_returns_null_json_when_no_report(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that None from the client is serialised as JSON null.""" + mock_client = create_autospec(AlertsClient, spec_set=True, instance=True) + mock_client.get_alert_investigation_report = AsyncMock(return_value=None) + monkeypatch.setattr(alerts, alerts._get_alerts_client.__name__, lambda: mock_client) + + result = await alerts.get_alert_investigation_report("alert-456") + + assert json.loads(result) is None + + @pytest.mark.asyncio + async def test_excludes_none_fields(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that None-valued optional fields are excluded from the output.""" + fake_report = AIInvestigation( + alertId="alert-789", + result="# Report", + status="COMPLETED", + verdict="TRUE_POSITIVE", + ) + + mock_client = create_autospec(AlertsClient, spec_set=True, instance=True) + mock_client.get_alert_investigation_report = AsyncMock(return_value=fake_report) + monkeypatch.setattr(alerts, alerts._get_alerts_client.__name__, lambda: mock_client) + + result = await alerts.get_alert_investigation_report("alert-789") + + data = json.loads(result) + assert "timestamp" not in data + assert "purpleAiStatus" not in data + assert "investigationStep" not in data + + @pytest.mark.asyncio + async def test_graphql_error_raises_runtime_error( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Test that a GraphQL error is wrapped in RuntimeError.""" + mock_client = create_autospec(AlertsClient, spec_set=True, instance=True) + mock_client.get_alert_investigation_report = AsyncMock( + side_effect=AlertsGraphQLError("Dummy Error") + ) + monkeypatch.setattr(alerts, alerts._get_alerts_client.__name__, lambda: mock_client) + + with pytest.raises( + RuntimeError, match=r"Failed to retrieve investigation report for alert alert-999" + ): + await alerts.get_alert_investigation_report("alert-999") + + @pytest.mark.asyncio + async def test_value_error_re_raised(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that ValueError is re-raised without wrapping.""" + mock_client = create_autospec(AlertsClient, spec_set=True, instance=True) + mock_client.get_alert_investigation_report = AsyncMock( + side_effect=ValueError("invalid alert_id") + ) + monkeypatch.setattr(alerts, alerts._get_alerts_client.__name__, lambda: mock_client) + + with pytest.raises(ValueError, match=r"invalid alert_id"): + await alerts.get_alert_investigation_report("") diff --git a/tests/unit/tools/test_fields_validation.py b/tests/unit/tools/test_fields_validation.py new file mode 100644 index 0000000..ac4f2fe --- /dev/null +++ b/tests/unit/tools/test_fields_validation.py @@ -0,0 +1,61 @@ +"""Tests for the shared bounded fields-parameter validator.""" + +import json + +import pytest + +from purple_mcp.tools.fields_validation import ( + MAX_FIELD_LENGTH, + MAX_FIELDS_COUNT, + MAX_FIELDS_JSON_LENGTH, + parse_fields_parameter, +) + + +def test_none_returns_none() -> None: + """Test that None input returns None.""" + assert parse_fields_parameter(None) is None + + +def test_valid_list_parsed() -> None: + """Test that valid JSON field list is parsed correctly.""" + assert parse_fields_parameter(json.dumps(["id", "name"])) == ["id", "name"] + + +def test_too_many_fields_rejected() -> None: + """Test that field lists exceeding MAX_FIELDS_COUNT are rejected.""" + payload = json.dumps([f"f{i}" for i in range(MAX_FIELDS_COUNT + 1)]) + with pytest.raises(ValueError, match="Too many fields"): + parse_fields_parameter(payload) + + +def test_payload_too_large_rejected() -> None: + """Test that payloads exceeding MAX_FIELDS_JSON_LENGTH are rejected.""" + payload = "x" * (MAX_FIELDS_JSON_LENGTH + 1) + with pytest.raises(ValueError, match="too large"): + parse_fields_parameter(payload) + + +def test_field_too_long_rejected() -> None: + """Test that individual fields exceeding MAX_FIELD_LENGTH are rejected.""" + payload = json.dumps(["a" * (MAX_FIELD_LENGTH + 1)]) + with pytest.raises(ValueError, match="too long"): + parse_fields_parameter(payload) + + +def test_non_list_rejected() -> None: + """Test that non-list JSON structures are rejected.""" + with pytest.raises(ValueError, match="must be an array"): + parse_fields_parameter(json.dumps({"not": "a list"})) + + +def test_non_string_element_rejected() -> None: + """Test that non-string field elements are rejected.""" + with pytest.raises(ValueError, match="must be strings"): + parse_fields_parameter(json.dumps(["ok", 123])) + + +def test_invalid_json_rejected() -> None: + """Test that malformed JSON is rejected.""" + with pytest.raises(ValueError, match="Invalid JSON"): + parse_fields_parameter("{not json")