Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 9 additions & 5 deletions server/secops/secops_mcp/tools/security_rules.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")

Expand Down Expand Up @@ -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(
Expand Down
42 changes: 41 additions & 1 deletion server/secops/tests/test_secops_tools_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down Expand Up @@ -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