diff --git a/server/secops/secops_mcp/tools/security_rules.py b/server/secops/secops_mcp/tools/security_rules.py index f410b644..ff55eedc 100644 --- a/server/secops/secops_mcp/tools/security_rules.py +++ b/server/secops/secops_mcp/tools/security_rules.py @@ -14,11 +14,11 @@ """Security Operations MCP tools for security rules.""" import logging +from datetime import datetime, timedelta, timezone from typing import Any, Dict, Optional from secops_mcp.server import get_chronicle_client, server - # Configure logging logger = logging.getLogger("secops-mcp") @@ -630,12 +630,16 @@ async def test_rule( chronicle = get_chronicle_client(project_id, customer_id, region) # Define time range for testing - from datetime import datetime, timedelta, timezone - - end_time = datetime.now(timezone.utc) + current_time = datetime.now(timezone.utc) + # Buffer back to the start of the current hour + end_time = current_time.replace(minute=0, second=0, microsecond=0) start_time = end_time - timedelta(hours=hours_back) - logger.info(f"Rule test time range: {start_time} to {end_time}") + logger.info( + "Rule test time range: %s to %s (buffered to start of current hour)", + start_time, + end_time, + ) # Test the rule test_results = chronicle.run_rule_test( diff --git a/server/secops/tests/test_secops_tools_unit.py b/server/secops/tests/test_secops_tools_unit.py index 68027e9e..77703560 100644 --- a/server/secops/tests/test_secops_tools_unit.py +++ b/server/secops/tests/test_secops_tools_unit.py @@ -45,7 +45,7 @@ def wrapper(func): from secops_mcp.tools.search import search_udm from secops_mcp.tools.udm_search import export_udm_search_csv from secops_mcp.tools.security_events import search_security_events -from secops_mcp.tools.security_rules import get_rule_detections +from secops_mcp.tools.security_rules import get_rule_detections, test_rule as secops_test_rule @pytest.fixture def mock_chronicle_client(): @@ -299,3 +299,43 @@ async def test_export_csv_invalid_date(mock_get_client): assert isinstance(result, str) assert "Error parsing date format" in result assert "yesterday" in result + + +# ========================================================================= +# Tests for test_rule (PR #143) +# ========================================================================= + +@pytest.mark.asyncio +async def test_test_rule_buffers_end_time_to_start_of_hour(mock_get_client): + """Test test_rule buffers end_time to start of current hour.""" + mock_get_client.run_rule_test.return_value = [ + {"type": "detection", "detection": {"ruleName": "test_rule", "id": "d-1"}}, + {"type": "progress", "percentDone": 100}, + ] + + result = await secops_test_rule( + rule_text="rule test { condition: true }", + project_id="my-proj", + customer_id="my-cust", + region="us", + hours_back=24, + ) + + # Verify run_rule_test was invoked + mock_get_client.run_rule_test.assert_called_once() + _, kwargs = mock_get_client.run_rule_test.call_args + + # Check end_time has minute, second, microsecond set to 0 (buffered to start of hour) + end_time = kwargs["end_time"] + start_time = kwargs["start_time"] + assert end_time.minute == 0 + assert end_time.second == 0 + assert end_time.microsecond == 0 + + # Check start_time is 24 hours before end_time + assert end_time - start_time == timedelta(hours=24) + + # Check output contains detection summary + assert "Total Detections: 1" in result + assert "Rule successfully detected 1 event(s)" in result +