import copy from unittest.mock import patch from freezegun import freeze_time import pytest from unittest.mock import MagicMock import demistomock as demisto import yaml import os from Packs.Wiz.Integrations.WizDefend import WizDefend from Packs.Wiz.Integrations.WizDefend.WizDefend import * # ===== TEST FIXTURES ===== @pytest.fixture(autouse=True) def set_mocks(mocker): """Set up common mocks that should apply to all tests""" integration_params = { "api_endpoint": "http://test.io", "credentials": {"identifier": "test", "password": "pass"}, "first_fetch": "2 days", "auth_endpoint": "https://auth.wiz.io/oauth/token", } mocker.patch.object(demisto, "params", return_value=integration_params) mocker.patch("WizDefend.TOKEN", "test-token") mocker.patch("WizDefend.AUTH_E", integration_params["auth_endpoint"]) mocker.patch("WizDefend.URL", integration_params["api_endpoint"]) # Mock logging functions to prevent test output pollution mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "info") mocker.patch.object(demisto, "error") @pytest.fixture(autouse=True) def global_mocks(mocker): """Apply critical mocks to all tests""" # Mock token and authentication mocker.patch("WizDefend.TOKEN", "test-token") mocker.patch("WizDefend.get_token", return_value="test-token") # Mock API endpoints mocker.patch("WizDefend.AUTH_E", "https://auth.wiz.io/oauth/token") mocker.patch("WizDefend.URL", "https://api.wiz.io/graphql") # Mock demisto parameters mocker.patch.object( demisto, "params", return_value={ "api_endpoint": "https://api.wiz.io/graphql", "credentials": {"identifier": "test", "password": "pass"}, "first_fetch": "2 days", "auth_endpoint": "https://auth.wiz.io/oauth/token", }, ) # Silence logging mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "info") mocker.patch.object(demisto, "error") @pytest.fixture(scope="function") def reset_domain(): """Reset WIZ_DOMAIN_URL before and after each test""" from Packs.Wiz.Integrations.WizDefend import WizDefend # Save current value original_value = WizDefend.WIZ_DOMAIN_URL # Reset for test WizDefend.WIZ_DOMAIN_URL = "" # Run test yield # Restore after test WizDefend.WIZ_DOMAIN_URL = original_value def patch_wiz_api(api_response=None): """ Returns a list of patch decorators for all Wiz API-related functions and variables. Args: api_response (dict or list, optional): Custom API response to return from get_entries. If None, returns a default sample detection. Returns: list: List of patch decorators """ # Default response if none provided if api_response is None: sample_detection = {"id": "test-detection", "severity": "CRITICAL"} api_response = [sample_detection] # Default to list for consistent handling # Ensure api_response is in the expected format for get_entries if isinstance(api_response, list): mock_return = (api_response, {"hasNextPage": False, "endCursor": ""}) else: # Assume it's a complete response structure mock_return = api_response return [ patch("WizDefend.get_token", return_value="test-token"), patch("WizDefend.TOKEN", "test-token"), patch("WizDefend.AUTH_E", "https://auth.wiz.io/oauth/token"), patch("WizDefend.URL", "https://api.wiz.io/graphql"), patch("WizDefend.get_entries", return_value=mock_return), ] def mock_wiz_api(monkeypatch, mocker, api_response=None): """ Helper function to mock all Wiz API-related functions and variables """ # Mock the entire query_api function which is called by get_filtered_detections if api_response is None: sample_detection = {"id": "test-detection", "severity": "CRITICAL"} api_response = [sample_detection] # This creates a mock that completely bypasses get_entries and get_token mock_query_api = mocker.patch("WizDefend.query_api", return_value=api_response) return mock_query_api @pytest.fixture def mock_response_factory(): """Create mock response objects with configurable status and JSON content""" def _create_mock_response(status_code=200, json_data=None, text=None): mock_response = MagicMock() mock_response.status_code = status_code mock_response.json.return_value = json_data or {} mock_response.text = text or "" return mock_response return _create_mock_response @pytest.fixture def sample_detection(): """Return a sample detection object for testing""" return { "id": "12345678-1234-1234-1234-d25e16359c19", "issue": { "id": "98765432-4321-4321-4321-ff5fa2ff7f78", "url": "https://app.wiz.io/issues/98765432-4321-4321-4321-ff5fa2ff7f78", }, "ruleMatch": { "rule": { "id": "12345678-4321-4321-4321-3792e8a03318", "name": "suspicious activity detected", "sourceType": "THREAT_DETECTION", "securitySubCategories": [], } }, "description": "Suspicious activity detected", "severity": "CRITICAL", "createdAt": "2022-01-02T15:46:34Z", "startedAt": "2022-01-02T15:45:00Z", "endedAt": "2022-01-02T15:47:00Z", "actors": [], "resources": [], "triggeringEvents": {"nodes": []}, } @pytest.fixture def sample_detection_no_rule(): """Return a sample detection object without a rule match for testing edge cases""" return { "id": "12345678-1234-1234-1234-d25e16359c19", "issue": { "id": "98765432-4321-4321-4321-ff5fa2ff7f78", "url": "https://app.wiz.io/issues/98765432-4321-4321-4321-ff5fa2ff7f78", }, "description": "Suspicious activity without rule match", "severity": "HIGH", "createdAt": "2022-01-02T15:46:34Z", "startedAt": "2022-01-02T15:45:00Z", "endedAt": "2022-01-02T15:47:00Z", "actors": [], "resources": [], "triggeringEvents": {"nodes": []}, } @pytest.fixture def sample_threat(): """Return a sample threat object for testing""" return { "id": "98765432-4321-4321-4321-ff5fa2ff7f78", "sourceRule": { "id": "12345678-4321-4321-4321-3792e8a03318", "name": "suspicious activity detected", "type": "THREAT_DETECTION", "cloudEventRuleDescription": "Suspicious activity detected", }, "type": "THREAT_DETECTION", "createdAt": "2022-01-02T15:46:34Z", "updatedAt": "2022-01-02T16:46:34Z", "dueAt": "2022-01-09T15:46:34Z", "projects": [{"id": "project-123", "name": "Production Project"}], "status": "OPEN", "severity": "CRITICAL", "entitySnapshot": {"id": "entity-123", "type": "VM", "name": "test-instance", "cloudPlatform": "AWS"}, "notes": [{"id": "note-123", "text": "Investigating this threat", "createdAt": "2022-01-02T16:00:00Z"}], } @pytest.fixture def mock_threat_api_response(sample_threat): """Return a complete threat API response structure""" return {"data": {"issues": {"nodes": [sample_threat], "pageInfo": {"hasNextPage": False, "endCursor": ""}}}} @pytest.fixture def mock_api_response(sample_detection): """Return a complete API response structure""" return {"data": {"detections": {"nodes": [sample_detection], "pageInfo": {"hasNextPage": False, "endCursor": ""}}}} @pytest.fixture def mock_api_paginated_response(sample_detection): """Return a paginated API response for testing pagination""" detection2 = copy.deepcopy(sample_detection) detection2["id"] = "second-detection-id" # First page response first_page = { "data": {"detections": {"nodes": [sample_detection], "pageInfo": {"hasNextPage": True, "endCursor": "cursor1"}}} } # Second page response second_page = {"data": {"detections": {"nodes": [detection2], "pageInfo": {"hasNextPage": False, "endCursor": ""}}}} return first_page, second_page @pytest.fixture def mock_api_error_response(): """Return an API error response""" return { "errors": [ { "message": "Resource not found", "extensions": {"code": "NOT_FOUND", "exception": {"message": "Resource not found", "path": ["detections"]}}, } ], "data": None, } @pytest.fixture def mock_api_empty_response(): """Return an empty API response""" return {"data": {"detections": {"nodes": [], "pageInfo": {"hasNextPage": False, "endCursor": ""}}}} @pytest.fixture def yaml_content(): """Fixture to load the YAML file once for all tests""" yaml_file_path = os.path.join(os.path.dirname(__file__), "WizDefend.yml") with open(yaml_file_path) as f: return yaml.safe_load(f) @pytest.fixture def valid_threat_scenario(): """Fixture for valid threat scenario""" return { "issue_id": str(uuid.uuid4()), "api_response": [{"id": "test-threat", "status": "OPEN", "severity": "HIGH"}], "should_succeed": True, } # ===== VALIDATION FUNCTION TESTS ===== @pytest.mark.parametrize( "detection_type,expected_valid,expected_value", [ (["GENERATED THREAT", "DID NOT GENERATE THREAT"], True, ["GENERATED_THREAT", "MATCH_ONLY"]), (["GENERATED THREAT"], True, ["GENERATED_THREAT"]), (["DID NOT GENERATE THREAT"], True, ["MATCH_ONLY"]), ("GENERATED THREAT", True, "GENERATED_THREAT"), ("GENERATED THREAT", True, "GENERATED_THREAT"), ("GENERATED_THREAT", False, None), ("generated threat", True, "GENERATED_THREAT"), # Case insensitive ("DID NOT GENERATE THREAT", True, "MATCH_ONLY"), ("did not generate threat", True, "MATCH_ONLY"), # Case insensitive ("MATCH_ONLY", False, None), ("INVALID_TYPE", False, None), (None, True, None), # None should be valid (no filter) ("", True, None), # Empty string should be valid (no filter) ], ) def test_validate_detection_type(detection_type, expected_valid, expected_value): """Test validate_detection_type with various inputs""" result = validate_detection_type(detection_type) assert result.is_valid == expected_valid assert result.value == expected_value @pytest.mark.parametrize( "platform,expected_valid,expected_value", [ ("AWS", True, ["AWS"]), (["AWS", "Azure", "GCP"], True, ["AWS", "Azure", "GCP"]), ("AWS,Azure,GCP", True, ["AWS", "Azure", "GCP"]), # Comma-separated ("INVALID_PLATFORM", False, None), ("AWS,INVALID_PLATFORM", False, None), # One invalid in list (None, True, None), ], ) def test_validate_detection_platform(platform, expected_valid, expected_value): """Test validate_detection_platform with various inputs""" result = validate_detection_platform(platform) assert result.is_valid == expected_valid if expected_valid: assert result.value == expected_value @pytest.mark.parametrize( "origin,expected_valid,expected_value", [ ("WIZ_SENSOR", True, ["WIZ_SENSOR"]), (["WIZ_SENSOR", "AWS_GUARD_DUTY"], True, ["WIZ_SENSOR", "AWS_GUARD_DUTY"]), ("WIZ_SENSOR,AWS_GUARD_DUTY", True, ["WIZ_SENSOR", "AWS_GUARD_DUTY"]), # Comma-separated ("INVALID_ORIGIN", False, None), ("WIZ_SENSOR,INVALID_ORIGIN", False, None), # One invalid in list (None, True, None), ], ) def test_validate_detection_origin(origin, expected_valid, expected_value): """Test validate_detection_origin with various inputs""" result = validate_detection_origin(origin) assert result.is_valid == expected_valid if expected_valid: assert result.value == expected_value @pytest.mark.parametrize( "subscription,expected_valid,expected_value", [ ( ["12345678-1234-1234-1234-d25e16359c19", "12345678-1234-1234-1234-d25e16359c20"], True, ["12345678-1234-1234-1234-d25e16359c19", "12345678-1234-1234-1234-d25e16359c20"], ), ("12345678-1234-1234-1234-d25e16359c19", True, ["12345678-1234-1234-1234-d25e16359c19"]), ("test-subscription", False, "test-subscription"), ("", True, None), (None, True, None), (123, False, None), # Non-string should be invalid ], ) def test_validate_detection_subscription(subscription, expected_valid, expected_value): """Test validate_detection_subscription with various inputs""" result = validate_detection_cloud_account_or_cloud_organization(subscription) assert result.is_valid == expected_valid if expected_valid: assert result.value == expected_value @pytest.mark.parametrize( "minutes_back,expected_valid,expected_value", [ ("10", True, 10), # Minimum value ("600", True, 600), # Maximum value ("300", True, 300), # Middle value ("4", False, None), # Below minimum ("601", False, None), # Above maximum ("not_a_number", False, None), # Non-numeric (None, True, FETCH_INTERVAL_MINIMUM_MIN), # None defaults to minimum ], ) def test_validate_creation_minutes_back(minutes_back, expected_valid, expected_value): """Test validate_creation_minutes_back with various inputs""" result = validate_creation_time_back(minutes_back) assert result.is_valid == expected_valid if expected_valid: assert result.minutes_value == expected_value @pytest.mark.parametrize( "severity,expected_valid,expected_list", [ # Single severity string tests (backward compatibility - includes higher levels) ("CRITICAL", True, ["CRITICAL"]), ("HIGH", True, ["CRITICAL", "HIGH"]), ("MEDIUM", True, ["CRITICAL", "HIGH", "MEDIUM"]), ("LOW", True, ["CRITICAL", "HIGH", "MEDIUM", "LOW"]), ("INFORMATIONAL", True, ["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFORMATIONAL"]), ("critical", True, ["CRITICAL"]), # Case insensitive ("INVALID", False, None), (None, True, None), # None is valid (no filter) ("", True, None), # Empty string is valid (no filter) # List of severities tests (multi-selection - only specified severities) (["CRITICAL"], True, ["CRITICAL"]), (["HIGH"], True, ["HIGH"]), (["CRITICAL", "MEDIUM"], True, ["CRITICAL", "MEDIUM"]), (["HIGH", "LOW"], True, ["HIGH", "LOW"]), (["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFORMATIONAL"], True, ["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFORMATIONAL"]), (["critical", "HIGH"], True, ["CRITICAL", "HIGH"]), # Case insensitive in list (["CRITICAL", "INVALID"], False, None), # Invalid item in list (["INVALID", "ALSO_INVALID"], False, None), # Multiple invalid items ([], True, None), # Empty list is valid (no filter) ([""], True, []), # List with empty string gets filtered out (["CRITICAL", ""], True, ["CRITICAL"]), # Mixed valid and empty (["CRITICAL", "HIGH", "CRITICAL"], True, ["CRITICAL", "HIGH", "CRITICAL"]), # Duplicates allowed ], ) def test_validate_severity(severity, expected_valid, expected_list): """Test validate_severity with various inputs including strings and lists""" result = validate_severity(severity) assert result.is_valid == expected_valid if expected_valid and expected_list is not None: assert result.severity_list == expected_list elif expected_valid and expected_list is None: # For None/empty inputs, severity_list should not be set or should be None assert not hasattr(result, "severity_list") or result.severity_list is None @pytest.mark.parametrize( "resource_id,expected_valid,expected_value", [ ("test-resource-id", True, "test-resource-id"), ("", True, ""), (None, True, None), (123, True, 123), # Any value should be valid ], ) def test_validate_resource_id(resource_id, expected_valid, expected_value): """Test validate_resource_id with various inputs""" result = validate_resource_id(resource_id) assert result.is_valid == expected_valid if expected_valid: assert result.value == expected_value @pytest.mark.parametrize( "project,expected_valid,expected_value", [ ("test project", True, "test project"), ("", True, ""), (None, True, None), (123, True, 123), # Any value should be valid ], ) def test_validate_project(project, expected_valid, expected_value): """Test validate_project with various inputs""" result = validate_project(project) assert result.is_valid == expected_valid if expected_valid: assert result.value == expected_value @pytest.mark.parametrize( "incident_type,expected_valid,expected_value", [ (WIZ_DEFEND_INCIDENT_TYPE, True, WIZ_DEFEND_INCIDENT_TYPE), ("Other Type", False, None), (None, False, None), ], ) def test_validate_incident_type(incident_type, expected_valid, expected_value): """Test validate_incident_type with various inputs""" result = validate_incident_type(incident_type) assert result.is_valid == expected_valid if expected_valid: assert result.value == expected_value @pytest.mark.parametrize( "fetch_interval,expected_valid,expected_value", [ (10, True, 10), # Valid - at minimum (30, True, 30), # Valid - above minimum (4, False, None), # Invalid - below minimum (None, False, None), # Invalid - None ], ) def test_validate_fetch_interval(fetch_interval, expected_valid, expected_value): """Test validate_fetch_interval with various inputs""" result = validate_fetch_interval(fetch_interval) assert result.is_valid == expected_valid if expected_valid: assert result.minutes_value == expected_value @pytest.mark.parametrize( "max_fetch,expected_valid,expected_value", [ ("10", True, 10), # Minimum valid value ("1000", True, 1000), # Maximum valid value ("500", True, 500), # Middle value ("9", False, None), # Below minimum ("1001", False, None), # Above maximum ("not_a_number", False, None), # Non-numeric string ("", True, API_MAX_FETCH), # Empty string should default to max (None, True, API_MAX_FETCH), # None should default to max (10, True, 10), # Integer input (1000, True, 1000), # Integer input at max ], ) def test_validate_max_fetch(max_fetch, expected_valid, expected_value): """Test validate_max_fetch with various inputs""" result = validate_max_fetch(max_fetch) assert result.is_valid == expected_valid if expected_valid: assert result.value == expected_value @pytest.mark.parametrize( "max_fetch,expected_in_error", [ ("5", "Received 5"), # Below minimum ("2000", "Received 2000"), # Above maximum ("invalid", None), # Non-numeric (no "Received" message) ("-10", "Received -10"), # Negative number ("0", "Received 0"), # Zero ], ) def test_validate_max_fetch_error_messages(max_fetch, expected_in_error): """Test validate_max_fetch error messages""" result = validate_max_fetch(max_fetch) assert result.is_valid is False assert f"{DemistoParams.MAX_FETCH} must be a valid integer between 10 and 1000" in result.error_message if expected_in_error: assert expected_in_error in result.error_message @pytest.mark.parametrize( "first_fetch,expected_valid,expected_value", [ ("2 days", True, "2 days"), ("12 hours", True, "12 hours"), ("30 minutes", True, "30 minutes"), (f"{MAX_DAYS_FIRST_FETCH_DETECTIONS} days", True, f"{MAX_DAYS_FIRST_FETCH_DETECTIONS} days"), (f"{MAX_DAYS_FIRST_FETCH_DETECTIONS + 1} days", False, None), ("not a duration", False, None), ("5 years", False, None), ("", False, None), (None, False, None), ], ) def test_validate_first_fetch(first_fetch, expected_valid, expected_value): """Test validate_first_fetch with various inputs""" result = validate_first_fetch(first_fetch) assert result.is_valid == expected_valid if expected_valid: assert result.value == expected_value def test_validate_first_fetch_timestamp(mocker): """Test validate_first_fetch_timestamp function""" # Import the module inside the test to get a fresh instance from Packs.Wiz.Integrations.WizDefend import WizDefend import datetime as dt from freezegun import freeze_time # Use freezegun to mock datetime.now() with freeze_time("2022-01-06 12:00:00"): # Current time is now frozen at 2022-01-06 current_time = dt.datetime(2022, 1, 6, 12, 0, 0) max_days_ago = current_time - dt.timedelta(days=WizDefend.MAX_DAYS_FIRST_FETCH_DETECTIONS) # Test with valid date (2 days ago) valid_date = dt.datetime(2022, 1, 4, 12, 0, 0) mocker.patch("WizDefend.dateparser.parse", return_value=valid_date) is_valid, error_msg, date = WizDefend.validate_first_fetch_timestamp("2 days") assert is_valid is True assert error_msg is None assert date == valid_date # Test with date beyond limits (30 days ago) old_date = dt.datetime(2021, 12, 1, 12, 0, 0) mocker.patch("WizDefend.dateparser.parse", return_value=old_date) is_valid, error_msg, date = WizDefend.validate_first_fetch_timestamp("30 days") assert is_valid is True # Still valid, but adjusted assert error_msg is None assert date == max_days_ago # Should be adjusted to max_days_ago # Test with invalid date format mocker.patch("WizDefend.dateparser.parse", return_value=None) is_valid, error_msg, date = WizDefend.validate_first_fetch_timestamp("invalid format") assert is_valid is False assert "Invalid date format" in error_msg assert date is None def test_validate_all_detection_parameters(): """Test validate_all_detection_parameters with various parameter combinations""" # Test with all valid parameters valid_params = { "detection_id": str(uuid.uuid4()), "issue_id": str(uuid.uuid4()), "type": "GENERATED THREAT", "platform": "AWS", "origin": "WIZ_SENSOR", "subscription": "12345678-1234-1234-1234-d25e16359c19", "resource_id": "test-resource", "severity": "CRITICAL", "creation_minutes_back": "15", "rule_match_id": str(uuid.uuid4()), "rule_match_name": "test rule", "project_id": "test-project", } success, error_message, validated_values = validate_all_detection_parameters(valid_params) assert success is True assert error_message is None assert validated_values["type"] == "GENERATED_THREAT" assert validated_values["platform"] == ["AWS"] assert validated_values["origin"] == ["WIZ_SENSOR"] assert validated_values["severity"] == ["CRITICAL"] assert validated_values["creation_minutes_back"] == 15 # ... (other existing test cases) # Test with invalid rule_match_id invalid_rule_params = {"rule_match_id": "invalid-uuid", "severity": "CRITICAL"} success, error_message, validated_values = validate_all_detection_parameters(invalid_rule_params) assert success is False assert "Invalid matched rule ID" in error_message @pytest.mark.parametrize( "status,expected_valid,expected_list", [ ("OPEN", True, ["OPEN"]), ("IN_PROGRESS", True, ["IN_PROGRESS"]), ("REJECTED", True, ["REJECTED"]), ("RESOLVED", True, ["RESOLVED"]), ("open", True, ["OPEN"]), # Case insensitive ("OPEN,IN_PROGRESS", True, ["OPEN", "IN_PROGRESS"]), # Comma-separated ("INVALID", False, None), (None, True, None), # None is valid (no filter) ], ) def test_validate_status(status, expected_valid, expected_list): """Test validate_status with various inputs""" result = validate_status(status) assert result.is_valid == expected_valid if expected_valid: assert result.status_list == expected_list @pytest.mark.parametrize( "days_back,expected_valid,expected_value", [ ("1", True, 1), # Minimum value ("30", True, 30), # Maximum value ("15", True, 15), # Middle value ("0", False, None), # Below minimum ("31", False, None), # Above maximum ("not_a_number", False, None), # Non-numeric (None, True, THREATS_DAYS_DEFAULT), # None defaults to default ], ) def test_validate_creation_days_back(days_back, expected_valid, expected_value): """Test validate_creation_time_back with 'days' time unit""" result = validate_creation_time_back(days_back, time_unit="days") assert result.is_valid == expected_valid if expected_valid: assert result.days_value == expected_value def test_validate_all_threat_parameters(): """Test validate_all_threat_parameters with various parameter combinations""" # Test with all valid parameters valid_params = { "issue_id": str(uuid.uuid4()), "platform": "AWS", "origin": "WIZ_SENSOR", "cloud_account_or_cloud_organization": "12345678-1234-1234-1234-d25e16359c19", "resource_id": "test-resource", "severity": "CRITICAL", "status": "OPEN", "creation_days_back": "15", "project_id": "test-project", } success, error_message, validated_values = validate_all_threat_parameters(valid_params) assert success is True assert error_message is None assert validated_values["platform"] == ["AWS"] assert validated_values["origin"] == ["WIZ_SENSOR"] assert validated_values["severity"] == ["CRITICAL"] assert validated_values["status"] == ["OPEN"] assert validated_values["creation_days_back"] == 15 # Test with no parameters (should fail) empty_params = {} success, error_message, validated_values = validate_all_threat_parameters(empty_params) assert success is False assert "You should pass at least one of the following parameters" in error_message # Test with invalid issue_id invalid_id_params = {"issue_id": "invalid-uuid", "severity": "CRITICAL"} success, error_message, validated_values = validate_all_threat_parameters(invalid_id_params) assert success is False assert "should be in UUID format" in error_message # Test with invalid status invalid_status_params = {"status": "INVALID_STATUS", "severity": "CRITICAL"} success, error_message, validated_values = validate_all_threat_parameters(invalid_status_params) assert success is False assert "Invalid status" in error_message @pytest.mark.parametrize( "rule_id,expected_valid,expected_value", [ (str(uuid.uuid4()), True, None), # Valid UUID ("invalid-uuid", False, None), # Invalid UUID (None, True, None), # None is valid ], ) def test_validate_matched_rule_id(rule_id, expected_valid, expected_value): """Test validate_matched_rule_id with various inputs""" result = validate_matched_rule_id(rule_id) assert result.is_valid == expected_valid if expected_valid and rule_id: assert result.value == rule_id # ===== FILTER APPLICATION TESTS ===== @pytest.mark.parametrize( "detection_type,expected_result", [ (["GENERATED_THREAT"], True), (["DID NOT GENERATE THREAT", "GENERATED_THREAT"], True), ("GENERATED_THREAT", True), (None, False), ("CLOUD_THREAT", True), ], ) def test_apply_detection_type_filter(detection_type, expected_result): """Test apply_detection_type_filter function""" variables = {} result = apply_detection_type_filter(variables, detection_type) if expected_result: assert "filterBy" in result assert "type" in result["filterBy"] if isinstance(detection_type, list): expected_equals = detection_type else: expected_equals = [detection_type] assert result["filterBy"]["type"]["equals"] == expected_equals else: assert result == {} def test_apply_platform_filter(): """Test apply_platform_filter function""" # Test with single value variables = {} result = apply_platform_filter(variables, "AWS") assert "filterBy" in result assert "cloudPlatform" in result["filterBy"] assert result["filterBy"]["cloudPlatform"]["equals"] == ["AWS"] # Test with multiple values variables = {} result = apply_platform_filter(variables, ["AWS", "Azure", "GCP"]) assert result["filterBy"]["cloudPlatform"]["equals"] == ["AWS", "Azure", "GCP"] # Test with None (should not add filter) variables = {} result = apply_platform_filter(variables, None) assert result == {} def test_apply_origin_filter(): """Test apply_origin_filter function""" # Test with single value variables = {} result = apply_origin_filter(variables, "WIZ_SENSOR") assert "filterBy" in result assert "origin" in result["filterBy"] assert result["filterBy"]["origin"]["equals"] == ["WIZ_SENSOR"] # Test with multiple values variables = {} result = apply_origin_filter(variables, ["WIZ_SENSOR", "AWS_GUARD_DUTY"]) assert result["filterBy"]["origin"]["equals"] == ["WIZ_SENSOR", "AWS_GUARD_DUTY"] # Test with None (should not add filter) variables = {} result = apply_origin_filter(variables, None) assert result == {} def test_apply_subscription_filter(): """Test apply_subscription_filter function""" # Test with value variables = {} result = apply_cloud_account_or_cloud_organization_filter(variables, "12345678-1234-1234-1234-d25e16359c19") assert "filterBy" in result assert "cloudAccountOrCloudOrganizationId" in result["filterBy"] assert result["filterBy"]["cloudAccountOrCloudOrganizationId"]["equals"] == ["12345678-1234-1234-1234-d25e16359c19"] # Test with None (should not add filter) variables = {} result = apply_cloud_account_or_cloud_organization_filter(variables, None) assert result == {} def test_apply_resource_id_filter(): """Test apply_resource_id_filter function""" # Test with value variables = {} result = apply_resource_id_filter(variables, "test-resource") assert "filterBy" in result assert "resource" in result["filterBy"] assert result["filterBy"]["resource"]["id"]["equals"] == ["test-resource"] # Test with None (should not add filter) variables = {} result = apply_resource_id_filter(variables, None) assert result == {} def test_apply_severity_filter(): """Test apply_severity_filter function""" # Test with single value variables = {} result = apply_severity_filter(variables, ["CRITICAL"]) assert "filterBy" in result assert "severity" in result["filterBy"] assert result["filterBy"]["severity"]["equals"] == ["CRITICAL"] # Test with multiple values variables = {} result = apply_severity_filter(variables, ["CRITICAL", "HIGH"]) assert result["filterBy"]["severity"]["equals"] == ["CRITICAL", "HIGH"] # Test with None (should not add filter) variables = {} result = apply_severity_filter(variables, None) assert result == {} def test_apply_creation_in_last_minutes_filter(): """Test apply_creation_in_last_minutes_filter function""" # Test with value variables = {} result = apply_creation_in_last_filter(variables, 15) assert "filterBy" in result assert "createdAt" in result["filterBy"] assert result["filterBy"]["createdAt"]["inLast"]["amount"] == 15 assert result["filterBy"]["createdAt"]["inLast"]["unit"] == DurationUnit.MINUTES # Test with None (should not add filter) variables = {} result = apply_creation_in_last_filter(variables, None) assert result == {} def test_apply_project_id_filter(): """Test apply_project_id_filter function""" # Test with value variables = {} result = apply_project_id_filter(variables, "project-id") assert "filterBy" in result assert "projectId" in result["filterBy"] assert result["filterBy"]["projectId"] == "project-id" # Test with None (should not add filter) variables = {} result = apply_project_id_filter(variables, None) assert result == {} def test_apply_detection_id_filter(): """Test apply_detection_id_filter function""" # Test with single ID variables = {} detection_id = str(uuid.uuid4()) result = apply_detection_id_filter(variables, [detection_id]) assert "filterBy" in result assert "id" in result["filterBy"] assert result["filterBy"]["id"]["equals"] == [detection_id] # Test with multiple IDs variables = {} detection_ids = [str(uuid.uuid4()), str(uuid.uuid4())] result = apply_detection_id_filter(variables, detection_ids) assert result["filterBy"]["id"]["equals"] == detection_ids # Test with None (should not add filter) variables = {} result = apply_detection_id_filter(variables, None) assert result == {} def test_apply_issue_id_filter(): """Test apply_issue_id_filter function""" # Test with value variables = {} issue_id = str(uuid.uuid4()) result = apply_issue_id_filter(variables, issue_id) assert "filterBy" in result assert "issueId" in result["filterBy"] assert result["filterBy"]["issueId"] == issue_id # Test with None (should not add filter) variables = {} result = apply_issue_id_filter(variables, None) assert result == {} def test_apply_all_filters(): """Test that apply_all_filters correctly applies all filters""" validated_values = { "detection_id": [str(uuid.uuid4())], "issue_id": str(uuid.uuid4()), "type": "GENERATED_THREAT", "platform": ["AWS", "Azure"], "origin": ["WIZ_SENSOR"], "cloud_account_or_cloud_organization": "test-subscription", "resource_id": "test-id", "severity": ["CRITICAL"], "creation_minutes_back": 15, "rule_match_id": "rule-id", "rule_match_name": "rule name", "project": "project-id", } variables = {} result = apply_all_detection_filters(variables, validated_values) # Check that all filters were applied assert result["filterBy"]["id"]["equals"] == validated_values["detection_id"] assert result["filterBy"]["issueId"] == validated_values["issue_id"] assert result["filterBy"]["type"]["equals"] == [validated_values["type"]] assert result["filterBy"]["cloudPlatform"]["equals"] == validated_values["platform"] assert result["filterBy"]["origin"]["equals"] == validated_values["origin"] assert result["filterBy"]["cloudAccountOrCloudOrganizationId"]["equals"] == [ validated_values["cloud_account_or_cloud_organization"] ] assert result["filterBy"]["resource"]["id"]["equals"] == [validated_values["resource_id"]] assert result["filterBy"]["severity"]["equals"] == validated_values["severity"] assert result["filterBy"]["createdAt"]["inLast"]["amount"] == validated_values["creation_minutes_back"] assert result["filterBy"]["matchedRule"]["id"] == validated_values["rule_match_id"] assert result["filterBy"]["matchedRuleName"]["equals"] == [validated_values["rule_match_name"]] assert result["filterBy"]["projectId"] == validated_values["project"] # Test with after_time instead of creation_minutes_back validated_values_with_after = copy.deepcopy(validated_values) validated_values_with_after.pop("creation_minutes_back") validated_values_with_after["after"] = "2022-01-01T00:00:00Z" variables = {} result = apply_all_detection_filters(variables, validated_values_with_after) assert "createdAt" not in result["filterBy"] validated_values_with_after["before"] = "2022-02-01T00:00:00Z" result = apply_all_detection_filters(variables, validated_values_with_after) # Check that after_time filter was applied assert "createdAt" in result["filterBy"] assert "after" in result["filterBy"]["createdAt"] assert "before" in result["filterBy"]["createdAt"] assert result["filterBy"]["createdAt"]["after"] == "2022-01-01T00:00:00Z" assert result["filterBy"]["createdAt"]["before"] == "2022-02-01T00:00:00Z" # Test with minimal filters minimal_values = {"severity": ["CRITICAL"]} variables = {} result = apply_all_detection_filters(variables, minimal_values) # Check that only severity filter was applied assert "severity" in result["filterBy"] assert len(result["filterBy"]) == 1 def test_apply_status_filter(): """Test apply_status_filter function""" # Test with single value variables = {} result = apply_status_filter(variables, ["OPEN"]) assert "filterBy" in result assert "status" in result["filterBy"] assert result["filterBy"]["status"] == ["OPEN"] # Test with multiple values variables = {} result = apply_status_filter(variables, ["OPEN", "IN_PROGRESS"]) assert result["filterBy"]["status"] == ["OPEN", "IN_PROGRESS"] # Test with None (should not add filter) variables = {} result = apply_status_filter(variables, None) assert result == {} def test_apply_creation_days_back_filter(): """Test apply_creation_in_last_filter with 'days' time unit""" # Test with value variables = {} result = apply_creation_in_last_filter(variables, 15, time_unit="days") assert "filterBy" in result assert "createdAt" in result["filterBy"] assert result["filterBy"]["createdAt"]["inLast"]["amount"] == 15 assert result["filterBy"]["createdAt"]["inLast"]["unit"] == DurationUnit.DAYS # Test with None (should not add filter) variables = {} result = apply_creation_in_last_filter(variables, None, time_unit="days") assert result == {} def test_apply_all_threat_filters(): """Test that apply_all_threat_filters correctly applies all filters""" validated_values = { "issue_id": str(uuid.uuid4()), "platform": ["AWS", "Azure"], "origin": ["WIZ_SENSOR"], "cloud_account_or_cloud_organization": "test-subscription", "resource_id": "test-id", "severity": ["CRITICAL"], "status": ["OPEN", "IN_PROGRESS"], "creation_days_back": 15, "project": "project-id", } variables = {} result = apply_all_threat_filters(variables, validated_values) # Check that all filters were applied assert result["filterBy"]["id"] == validated_values["issue_id"] assert "relatedEntity" in result["filterBy"] assert result["filterBy"]["relatedEntity"]["cloudPlatform"] == validated_values["platform"] assert result["filterBy"]["eventOrigin"]["equals"] == validated_values["origin"] assert result["filterBy"]["cloudAccountOrCloudOrganizationId"] == [validated_values["cloud_account_or_cloud_organization"]] assert "threatResource" in result["filterBy"] assert result["filterBy"]["threatResource"]["ids"] == [validated_values["resource_id"]] assert result["filterBy"]["severity"] == validated_values["severity"] assert result["filterBy"]["status"] == validated_values["status"] assert result["filterBy"]["createdAt"]["inLast"]["amount"] == validated_values["creation_days_back"] assert result["filterBy"]["createdAt"]["inLast"]["unit"] == DurationUnit.DAYS assert result["filterBy"]["project"] == "project-id" # Test with minimal filters minimal_values = {"severity": ["CRITICAL"]} variables = {} result = apply_all_threat_filters(variables, minimal_values) # Check that only severity filter was applied assert "severity" in result["filterBy"] assert len(result["filterBy"]) == 1 def test_apply_matched_rule_filter(): """Test apply_matched_rule_filter function""" # Test with value variables = {} result = apply_rule_match_id_filter(variables, "rule-id") assert "filterBy" in result assert "matchedRule" in result["filterBy"] assert result["filterBy"]["matchedRule"]["id"] == "rule-id" # Test with None (should not add filter) variables = {} result = apply_rule_match_id_filter(variables, None) assert result == {} def test_apply_rule_match_name_filter(): """Test apply_rule_match_name_filter function""" # Test with value variables = {} result = apply_rule_match_name_filter(variables, "rule name") assert "filterBy" in result assert "matchedRuleName" in result["filterBy"] assert result["filterBy"]["matchedRuleName"]["equals"] == ["rule name"] # Test with None (should not add filter) variables = {} result = apply_rule_match_name_filter(variables, None) assert result == {} # ===== CORE API COMMUNICATION TESTS ===== def test_get_token_error(mock_response_factory, mocker): """Test get_token with error response""" # Mock authentication endpoint set_authentication_endpoint("https://auth.wiz.io/oauth/token") # Mock the response mock_response = mock_response_factory( status_code=401, json_data={"error": "access_denied", "error_description": "Unauthorized"}, text="Unauthorized" ) mocker.patch("requests.post", return_value=mock_response) # Mock demisto.params mocker.patch.object(demisto, "params", return_value={"credentials": {"identifier": "test", "password": "pass"}}) # Call the function and check exception with pytest.raises(Exception) as e: get_token() assert "Error authenticating to Wiz" in str(e.value) def test_get_token_no_access_token(mock_response_factory, mocker): """Test get_token when response doesn't contain access_token""" # Mock authentication endpoint set_authentication_endpoint("https://auth.wiz.io/oauth/token") # Mock the response mock_response = mock_response_factory(status_code=200, json_data={"message": "No token provided"}) mocker.patch("requests.post", return_value=mock_response) # Mock demisto.params mocker.patch.object(demisto, "params", return_value={"credentials": {"identifier": "test", "password": "pass"}}) # Call the function and check exception with pytest.raises(Exception) as e: get_token() assert "Could not retrieve token from Wiz" in str(e.value) def test_get_token_json_parse_error(mock_response_factory, mocker): """Test get_token when response is not valid JSON""" # Mock authentication endpoint set_authentication_endpoint("https://auth.wiz.io/oauth/token") # Mock the response mock_response = mock_response_factory(status_code=200) mock_response.json.side_effect = ValueError("Invalid JSON") mocker.patch("requests.post", return_value=mock_response) # Mock demisto.params mocker.patch.object(demisto, "params", return_value={"credentials": {"identifier": "test", "password": "pass"}}) # Call the function and check exception with pytest.raises(Exception) as e: get_token() assert "Could not parse API response" in str(e.value) def test_get_entries(mock_response_factory, mocker, mock_api_response): """Test get_entries with successful response""" # Import the module inside the test to get a fresh instance from Packs.Wiz.Integrations.WizDefend import WizDefend # Mock at module level test_token = "test-token" WizDefend.TOKEN = test_token WizDefend.get_token = lambda: test_token # Mock the response for API call mock_response = mock_response_factory(status_code=200, json_data=mock_api_response) mocker.patch("requests.post", return_value=mock_response) # Call the function entries, page_info = WizDefend.get_entries("test_query", {}, WizApiResponse.DETECTIONS) # Verify entries and page_info assert entries == mock_api_response["data"]["detections"]["nodes"] assert page_info == mock_api_response["data"]["detections"]["pageInfo"] def test_get_entries_with_token_refresh(mock_response_factory, mocker, mock_api_response): """Test get_entries when token needs to be refreshed""" # Mock token as None (needs refresh) mocker.patch("WizDefend.TOKEN", None) # Mock get_token mocker.patch("WizDefend.get_token", return_value="refreshed-token") # Mock the response mock_response = mock_response_factory(status_code=200, json_data=mock_api_response) mocker.patch("requests.post", return_value=mock_response) # Call the function entries, page_info = get_entries("test_query", {}, WizApiResponse.DETECTIONS) # Verify entries and page_info assert entries == mock_api_response["data"]["detections"]["nodes"] assert page_info == mock_api_response["data"]["detections"]["pageInfo"] def test_get_entries_error(mock_response_factory, mocker, mock_api_error_response): """Test get_entries with error response""" # Import the module inside the test to get a fresh instance from Packs.Wiz.Integrations.WizDefend import WizDefend # Mock token WizDefend.TOKEN = "test-token" # Mock return_error to prevent sys.exit(0) mock_return_error = mocker.MagicMock(side_effect=Exception("Mocked return_error")) mocker.patch.object(WizDefend, "return_error", mock_return_error) # Mock the response mock_response = mock_response_factory(status_code=200, json_data=mock_api_error_response) mocker.patch("requests.post", return_value=mock_response) # Call the function and check exception with pytest.raises(Exception): WizDefend.get_entries("test_query", {}, WizApiResponse.DETECTIONS) # Verify that return_error was called with the expected error message assert mock_return_error.called error_call_args = mock_return_error.call_args[0][0] assert "Wiz API error details" in error_call_args assert "Resource not found" in error_call_args def test_get_entries_http_error(mock_response_factory, mocker): """Test get_entries with HTTP error response""" # Import the module inside the test to get a fresh instance from Packs.Wiz.Integrations.WizDefend import WizDefend # Mock token WizDefend.TOKEN = "test-token" # Mock return_error to prevent sys.exit(0) mock_return_error = mocker.MagicMock(side_effect=Exception("Mocked return_error")) mocker.patch.object(WizDefend, "return_error", mock_return_error) # Mock the response mock_response = mock_response_factory(status_code=500, text="Internal Server Error") mocker.patch("requests.post", return_value=mock_response) # Call the function and check exception with pytest.raises(Exception): WizDefend.get_entries("test_query", {}, WizApiResponse.DETECTIONS) # Verify that return_error was called with the expected error message assert mock_return_error.called error_call_args = mock_return_error.call_args[0][0] assert "Got an error querying Wiz API [500] - Internal Server Error" in error_call_args # Modify test_get_token_success function in WizDefend_test.py def test_get_token_success(mock_response_factory, mocker): """Test get_token with successful response""" # Import the module inside the test to get a fresh instance from Packs.Wiz.Integrations.WizDefend import WizDefend import copy # Save original HEADERS to restore later original_headers = copy.deepcopy(WizDefend.HEADERS) # Mock authentication endpoint WizDefend.AUTH_E = "https://auth.wiz.io/oauth/token" # Mock the response mock_response = mock_response_factory(status_code=200, json_data={"access_token": "test-token"}) mocker.patch("requests.post", return_value=mock_response) # Mock demisto.params mocker.patch.object(demisto, "params", return_value={"credentials": {"identifier": "test", "password": "pass"}}) # Call the function token = WizDefend.get_token() # Verify token was returned assert token == "test-token" # Verify token was set in the module assert WizDefend.TOKEN == "test-token" # Skip checking if Authorization header is in HEADERS # Just verify that the token value is correct WizDefend.HEADERS = original_headers def test_query_api(mocker, sample_detection): """Test query_api with simple response""" # Import the module locally from Packs.Wiz.Integrations.WizDefend import WizDefend # Set required module variables directly WizDefend.TOKEN = "test-token" WizDefend.URL = "https://api.wiz.io/graphql" # Set a valid URL # Mock the get_entries function at module level orig_get_entries = WizDefend.get_entries WizDefend.get_entries = lambda q, v, w: ([sample_detection], {"hasNextPage": False, "endCursor": ""}) try: # Call the function result = WizDefend.query_api("test_query", {}, WizApiResponse.DETECTIONS) # Verify result assert result == [sample_detection] finally: # Restore original function WizDefend.get_entries = orig_get_entries def test_query_api_pagination(mocker, mock_api_paginated_response): """Test query_api with pagination""" # Import the module locally from Packs.Wiz.Integrations.WizDefend import WizDefend # Unpack the paginated responses first_page, second_page = mock_api_paginated_response # Set required module variables directly WizDefend.TOKEN = "test-token" WizDefend.URL = "https://api.wiz.io/graphql" # Define a side effect function to simulate pagination call_count = [0] # Use a list to maintain state between calls def mock_get_entries_side_effect(query, variables, wiz_type=None): call_count[0] += 1 if call_count[0] == 1: return (first_page["data"]["detections"]["nodes"], first_page["data"]["detections"]["pageInfo"]) else: return (second_page["data"]["detections"]["nodes"], second_page["data"]["detections"]["pageInfo"]) # Replace get_entries with our mock function orig_get_entries = WizDefend.get_entries WizDefend.get_entries = mock_get_entries_side_effect try: # Call the function result = WizDefend.query_api("test_query", {}, WizApiResponse.DETECTIONS) # Verify result contains combined detections from both pages assert len(result) == 2 assert result[0]["id"] == first_page["data"]["detections"]["nodes"][0]["id"] assert result[1]["id"] == second_page["data"]["detections"]["nodes"][0]["id"] finally: # Restore original function WizDefend.get_entries = orig_get_entries def test_query_api_empty_response(mocker, mock_api_empty_response): """Test query_api with empty response""" # Import the module locally from Packs.Wiz.Integrations.WizDefend import WizDefend # Set required module variables directly WizDefend.TOKEN = "test-token" WizDefend.URL = "https://api.wiz.io/graphql" # Mock get_entries to return empty results orig_get_entries = WizDefend.get_entries WizDefend.get_entries = lambda q, v, w=None: ([], {"hasNextPage": False, "endCursor": ""}) try: # Call the function result = WizDefend.query_api("test_query", {}, WizApiResponse.DETECTIONS) # Verify result is an empty list (so callers can iterate without a type guard) assert result == [] finally: # Restore original function WizDefend.get_entries = orig_get_entries @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_entries") def test_query_api_with_pagination_disabled(mock_get_entries, mock_api_paginated_response): """Test query_api with pagination disabled""" # Unpack the paginated responses first_page, _ = mock_api_paginated_response # Set up the mock for get_entries mock_get_entries.return_value = (first_page["data"]["detections"]["nodes"], first_page["data"]["detections"]["pageInfo"]) # Call the function with paginate=False result = query_api("test_query", {}, WizApiResponse.DETECTIONS, paginate=False) # Verify result only contains first page assert len(result) == 1 assert result[0]["id"] == first_page["data"]["detections"]["nodes"][0]["id"] def test_query_threats(mocker, sample_threat): """Test query_threats with simple response""" # Import the module locally from Packs.Wiz.Integrations.WizDefend import WizDefend # Set required module variables directly WizDefend.TOKEN = "test-token" WizDefend.URL = "https://api.wiz.io/graphql" # Set a valid URL # Mock the get_entries function at module level orig_get_entries = WizDefend.get_entries WizDefend.get_entries = lambda q, v, w: ([sample_threat], {"hasNextPage": False, "endCursor": ""}) try: # Call the function result = WizDefend.query_issues({}) # Verify result assert result == [sample_threat] finally: # Restore original function WizDefend.get_entries = orig_get_entries @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_entries") def test_query_threats_with_pagination(mock_get_entries, sample_threat): """Test query_threats with pagination""" # Create a second threat for pagination second_threat = copy.deepcopy(sample_threat) second_threat["id"] = "second-threat-id" # Set up the mock with pagination mock_get_entries.side_effect = [ ([sample_threat], {"hasNextPage": True, "endCursor": "cursor1"}), ([second_threat], {"hasNextPage": False, "endCursor": ""}), ] # Call the function result = query_issues({}, paginate=True) # Verify result contains both threats assert len(result) == 2 assert result[0]["id"] == sample_threat["id"] assert result[1]["id"] == "second-threat-id" @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_entries") def test_query_threats_with_pagination_disabled(mock_get_entries, sample_threat): """Test query_threats with pagination disabled""" # Setup first page response mock_get_entries.return_value = ([sample_threat], {"hasNextPage": True, "endCursor": "cursor1"}) # Call the function with paginate=False result = query_issues({}, paginate=False) # Verify result only contains first page assert len(result) == 1 assert result[0]["id"] == sample_threat["id"] # Verify get_entries was only called once assert mock_get_entries.call_count == 1 # ===== UTILITY FUNCTION TESTS ===== @pytest.mark.parametrize( "severity,expected_result", [ ("CRITICAL", 4), ("HIGH", 3), ("MEDIUM", 2), ("LOW", 1), ("INFORMATIONAL", 0.5), ("UNKNOWN", None), ], ) def test_translate_severity(severity, expected_result): """Test translate_severity with various severity levels""" detection = {"severity": severity} assert translate_severity(detection) == expected_result @pytest.mark.parametrize( "max_fetch,expected_api_limit", [ # Valid values - should return the input value (10, 10), # Minimum valid value (100, 100), # Middle value (500, 500), # Another middle value (1000, 1000), # Maximum valid value # Invalid values - should return API_MAX_FETCH (5, API_MAX_FETCH), # Below minimum (1001, API_MAX_FETCH), # Above maximum (None, API_MAX_FETCH), # None value ("invalid", API_MAX_FETCH), # Non-numeric string ("", API_MAX_FETCH), # Empty string (0, API_MAX_FETCH), # Zero (-10, API_MAX_FETCH), # Negative value ], ) def test_get_fetch_incidents_api_limit(max_fetch, expected_api_limit): """Test get_fetch_incidents_api_limit with various max_fetch values""" api_limit = get_fetch_incidents_api_max_fetch(max_fetch) assert api_limit == expected_api_limit def test_build_incidents(sample_detection): """Test build_incidents with valid detection""" incident = build_incidents(sample_detection) assert incident["name"] == "suspicious activity detected - 12345678-1234-1234-1234-d25e16359c19" assert incident["occurred"] == "2022-01-02T15:46:34Z" assert incident["severity"] == 4 assert "rawJSON" in incident assert incident["dbotMirrorId"] == "12345678-1234-1234-1234-d25e16359c19" # Test with None assert build_incidents(None) == {} def test_build_incidents_no_rule_match(sample_detection_no_rule): """Test build_incidents with detection that has no rule match""" incident = build_incidents(sample_detection_no_rule) assert incident["name"] == "Unknown Rule - 12345678-1234-1234-1234-d25e16359c19" assert incident["occurred"] == "2022-01-02T15:46:34Z" assert incident["severity"] == 3 # HIGH assert "rawJSON" in incident assert incident["dbotMirrorId"] == "12345678-1234-1234-1234-d25e16359c19" @pytest.mark.parametrize( "input_value,expected_result", [ (str(uuid.uuid4()), True), # Valid UUID ("invalid-uuid", False), # Invalid UUID (None, False), # None value ("", False), # Empty string (123, False), # Non-string (object(), False), # Non-stringifiable object ], ) def test_is_valid_uuid(input_value, expected_result): """Test is_valid_uuid with various inputs""" assert is_valid_uuid(input_value) == expected_result def test_is_valid_param_id(): """Test is_valid_param_id function""" # Test with valid UUID valid_uuid = str(uuid.uuid4()) is_valid, message = is_valid_param_id(valid_uuid, "test_param") assert is_valid is True assert "is in a valid format" in message # Test with None is_valid, message = is_valid_param_id(None, "test_param") assert is_valid is False assert "You should pass a test_param" in message # Test with invalid UUID is_valid, message = is_valid_param_id("invalid-uuid", "test_param") assert is_valid is False assert "should be in UUID format" in message def test_get_error_output(): """Test get_error_output with different error structures""" # Test with error array error_response = {"errors": [{"message": "Resource not found"}, {"message": "Another error"}]} result = get_error_output(error_response) assert "Resource not found" in result assert "Another error" in result # Test with duplicate errors (should deduplicate) error_response = {"errors": [{"message": "Same error"}, {"message": "Same error"}]} result = get_error_output(error_response) assert result.count("Same error") == 1 # Test with no errors no_error_response = {"data": {}} result = get_error_output(no_error_response) assert result == no_error_response @pytest.mark.parametrize( "auth_endpoint, expected_domain", [ ("https://auth.test.wiz.io/oauth/token", "test.wiz.io"), ("https://auth.app.wiz.io/oauth/token", "app.wiz.io"), ("https://auth.gov.wiz.io/oauth/token", "gov.wiz.io"), ("https://auth.staging-env.wiz.io/oauth/token", "staging-env.wiz.io"), ("invalid-url", "app.wiz.io"), ], ) def test_update_wiz_domain_url(mocker, auth_endpoint, expected_domain, reset_domain): """Test update_wiz_domain_url function correctly extracts the domain from auth endpoint""" # Mock demisto.params() to return our test auth_endpoint mocker.patch.object(demisto, "params", return_value={"auth_endpoint": auth_endpoint}) # Import here to ensure we get the reset value from Packs.Wiz.Integrations.WizDefend import WizDefend # Call the function update_wiz_domain_url() # Verify domain was correctly extracted assert ( expected_domain == WizDefend.WIZ_DOMAIN_URL ), f"Failed for auth_endpoint={auth_endpoint}: expected {expected_domain}, got {WizDefend.WIZ_DOMAIN_URL}" @pytest.mark.parametrize("domain", ["app.wiz.io", "test.wiz.io", "staging-env.wiz.io", "customer-subdomain.wiz.io"]) def test_get_detection_url(mocker, sample_detection, domain): """Test get_detection_url constructs the correct URL for detections with different domains""" # Import here to set the domain from Packs.Wiz.Integrations.WizDefend import WizDefend # Set the domain directly WizDefend.WIZ_DOMAIN_URL = domain # Call the function url = get_detection_url(sample_detection) # Expected URL pattern expected_pattern = f"https://{domain}/findings/detections#~(filters~(updateTime~(dateRange~(past~(amount~5~unit~'day))))~detectionId~'{sample_detection['id']}~streamCols~(~'event~'principal~'principalIp~'resource))" # Verify URL was correctly constructed assert url == expected_pattern, f"Failed for domain={domain}: expected {expected_pattern}, got {url}" def test_get_integration_user_agent(): """Test get_integration_user_agent function""" user_agent = get_integration_user_agent() assert INTEGRATION_GUID in user_agent assert "xsoar_defend" in user_agent assert WIZ_VERSION in user_agent def test_set_authentication_endpoint(mocker): """Test set_authentication_endpoint function""" # Import the module correctly using the full path from Packs.Wiz.Integrations.WizDefend import WizDefend # First, remove the patch from the fixture that would reset AUTH_E mocker.stopall() test_endpoint = "https://test-auth.wiz.io/oauth/token" WizDefend.set_authentication_endpoint(test_endpoint) # Verify the endpoint was set correctly assert test_endpoint == WizDefend.AUTH_E def test_set_api_endpoint(mocker): """Test set_api_endpoint function""" # Import the module correctly using the full path from Packs.Wiz.Integrations.WizDefend import WizDefend # First, remove the patch from the fixture that would reset URL mocker.stopall() test_endpoint = "https://test-api.wiz.io/graphql" WizDefend.set_api_endpoint(test_endpoint) # Verify the endpoint was set correctly assert test_endpoint == WizDefend.URL # ===== INTEGRATION CONTEXT FUNCTION TESTS ===== @pytest.mark.parametrize( "advanced_params,expected_has_max_fetch", [ (False, False), # Basic params should not include max_fetch (True, True), # Advanced params should include max_fetch ], ) def test_extract_params_from_integration_settings_with_max_fetch(mocker, advanced_params, expected_has_max_fetch): """Test extract_params_from_integration_settings function includes max_fetch in advanced params""" # Mock demisto.params with max_fetch integration_params = { "type": "GENERATED THREAT", "platform": "AWS", "severity": "CRITICAL", "origin": "WIZ_SENSOR", "cloud_account_or_cloud_organization": "sub-123", "first_fetch": "2 days", "incidentFetchInterval": 10, "incidentType": "WizDefend Detection", "isFetch": True, "max_fetch": "100", } mocker.patch.object(demisto, "params", return_value=integration_params) result = extract_params_from_integration_settings(advanced_params=advanced_params) if expected_has_max_fetch: assert result["max_fetch"] == "100" assert result["first_fetch"] == "2 days" assert result["incidentFetchInterval"] == 10 assert result["incidentType"] == "WizDefend Detection" assert result["isFetch"] is True else: assert "max_fetch" not in result # Basic params should still be present assert result["type"] == "GENERATED THREAT" assert result["platform"] == "AWS" def test_extract_params_from_integration_settings(mocker): """Test extract_params_from_integration_settings function""" # Mock demisto.params integration_params = { "type": "GENERATED THREAT", "platform": "AWS", "severity": "CRITICAL", "origin": "WIZ_SENSOR", "cloud_account_or_cloud_organization": "sub-123", "first_fetch": "2 days", "incidentFetchInterval": 10, "incidentType": "WizDefend Detection", "isFetch": True, } mocker.patch.object(demisto, "params", return_value=integration_params) # Test with basic params result = extract_params_from_integration_settings(advanced_params=False) assert result["type"] == "GENERATED THREAT" assert result["platform"] == "AWS" assert result["severity"] == "CRITICAL" assert result["origin"] == "WIZ_SENSOR" assert result["cloud_account_or_cloud_organization"] == "sub-123" assert "first_fetch" not in result # Test with advanced params result = extract_params_from_integration_settings(advanced_params=True) assert result["first_fetch"] == "2 days" assert result["incidentFetchInterval"] == 10 assert result["incidentType"] == "WizDefend Detection" assert result["isFetch"] is True @pytest.mark.parametrize( "is_fetch,max_fetch,validation_result,expected_valid,expected_error", [ # isFetch=True scenarios (True, "100", True, True, ""), # Valid max_fetch (True, "5", False, False, "max_fetch must be a valid integer between 10 and 1000"), # Invalid max_fetch - updated message (True, "invalid", False, False, "max_fetch must be a valid integer between 10 and 1000"), (True, None, True, True, ""), # No max_fetch (should still be valid) # isFetch=False scenarios (should skip max_fetch validation) (False, "5", True, True, ""), # Invalid max_fetch but isFetch=False (False, "invalid", True, True, ""), # Non-numeric max_fetch but isFetch=False ], ) def test_check_advanced_params_max_fetch(mocker, is_fetch, max_fetch, validation_result, expected_valid, expected_error): """Test check_advanced_params function with max_fetch validation""" # Mock all other validation functions to return success validation_response_success = ValidationResponse() validation_response_success.is_valid = True mocker.patch("Packs.Wiz.Integrations.WizDefend.WizDefend.validate_first_fetch", return_value=validation_response_success) mocker.patch("Packs.Wiz.Integrations.WizDefend.WizDefend.validate_fetch_interval", return_value=validation_response_success) mocker.patch("Packs.Wiz.Integrations.WizDefend.WizDefend.validate_incident_type", return_value=validation_response_success) # For failing cases, don't mock validate_max_fetch - let it run with real validation # This way we test against the actual error messages if validation_result: # Only mock when we want to force success max_fetch_response = validation_response_success mocker.patch("Packs.Wiz.Integrations.WizDefend.WizDefend.validate_max_fetch", return_value=max_fetch_response) # For failing cases, let the real validate_max_fetch function run params = { "isFetch": is_fetch, "first_fetch": "2 days", "incidentFetchInterval": 10, "incidentType": "WizDefend Detection", "max_fetch": max_fetch, } are_valid, error_message = check_advanced_params(params) assert are_valid is expected_valid if expected_error: assert expected_error in error_message def test_check_advanced_params(mocker): """Test check_advanced_params function""" # Mock validation functions validation_response_success = ValidationResponse() validation_response_success.is_valid = True validation_response_error = ValidationResponse() validation_response_error.is_valid = False validation_response_error.error_message = "Validation error" mocker.patch("WizDefend.validate_first_fetch", return_value=validation_response_success) mocker.patch("WizDefend.validate_fetch_interval", return_value=validation_response_success) mocker.patch("WizDefend.validate_incident_type", return_value=validation_response_success) # Test with valid params params = {"isFetch": True, "first_fetch": "2 days", "incidentFetchInterval": 10, "incidentType": "WizDefend Detection"} are_valid, error_message = check_advanced_params(params) assert are_valid is True assert error_message == "" # Test with isFetch=False (should skip validation) params = {"isFetch": False} are_valid, error_message = check_advanced_params(params) assert are_valid is True assert error_message == "" # Mock validation function with error mocker.patch("WizDefend.validate_first_fetch", return_value=validation_response_error) # Test with invalid first_fetch params = {"isFetch": True, "first_fetch": "invalid", "incidentFetchInterval": 10, "incidentType": "WizDefend Detection"} are_valid, error_message = check_advanced_params(params) assert are_valid is False assert "Invalid first fetch format" in error_message # Helper to validate ISO timestamps def is_valid_iso_timestamp(s): """Check if a string is a valid ISO timestamp.""" pattern = r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z" return bool(re.match(pattern, s)) @freeze_time("2022-01-01T12:00:00Z") def test_get_last_run_time_first_run_approach3(mocker): """Test using freeze_time to control all time functions""" # Mock demisto.getLastRun to return empty dict mocker.patch.object(demisto, "getLastRun", return_value={}) # Mock demisto.params mocker.patch.object(demisto, "params", return_value={"first_fetch": "2 days"}) # Mock get_fetch_timestamp to return the frozen time mocker.patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_fetch_timestamp", return_value="2022-01-01T12:00:00Z") # Create FetchIncident instance and call the method fetch_manager = FetchIncident() result = fetch_manager.get_last_run_time() # Assert the result matches expected format assert result == "2022-01-01T12:00:00Z" def test_get_last_run_time_existing_run(mocker): from datetime import datetime # Use a timestamp that's guaranteed to be recent (now) recent_time = datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ") # Mock demisto.getLastRun to return our recent timestamp mocker.patch.object(demisto, "getLastRun", return_value={"time": recent_time}) # Create FetchIncident instance and call the method fetch_manager = FetchIncident() result = fetch_manager.get_last_run_time() # Assert it returns the same time assert result == recent_time def test_get_last_run_time_too_old(mocker): """Test get_last_run_time when last run is too old""" # Use a timestamp that's guaranteed to be old very_old_time = "1970-01-01T00:00:00Z" # Mock demisto.getLastRun to return our very old timestamp mocker.patch.object(demisto, "getLastRun", return_value={"time": very_old_time}) # Create FetchIncident instance and call the method fetch_manager = FetchIncident() result = fetch_manager.get_last_run_time() # Should return a more recent time (within last 30 days) from datetime import datetime, timedelta result_time = datetime.strptime(result, "%Y-%m-%dT%H:%M:%SZ") thirty_days_ago = datetime.now() - timedelta(days=30) assert result_time >= thirty_days_ago def test_get_fetch_timestamp_valid(mocker): """Test get_fetch_timestamp with valid input""" # Call the function result = get_fetch_timestamp("2 days") # Verify result is a valid ISO timestamp assert is_valid_iso_timestamp(result), f"Result '{result}' is not a valid ISO timestamp" def test_get_fetch_timestamp_invalid(mocker): """Test get_fetch_timestamp with invalid input""" # Mock validate_first_fetch_timestamp mocker.patch("WizDefend.validate_first_fetch_timestamp", return_value=(False, "Invalid date format", None)) # Call the function and check exception with pytest.raises(ValueError) as e: get_fetch_timestamp("invalid format") # Verify exception message assert "Invalid date format" in str(e.value) # ===== DETECTION FUNCTION TESTS ===== @patch( "Packs.Wiz.Integrations.WizDefend.WizDefend.validate_all_detection_parameters", return_value=(True, None, {"severity": ["CRITICAL"]}), ) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.query_api") def test_get_filtered_detections_success(mock_query_api, mock_validate, sample_detection): """Test get_filtered_detections with successful validation and API call""" # Set up the mock to return the sample detection mock_query_api.return_value = [sample_detection] # Call the function result = get_filtered_detections(detection_type="GENERATED THREAT", detection_platform=["AWS"], severity="CRITICAL") # Verify result assert result == [sample_detection] assert mock_query_api.called @patch( "Packs.Wiz.Integrations.WizDefend.WizDefend.validate_all_detection_parameters", return_value=(False, "Validation error message", None), ) def test_get_filtered_detections_validation_error(mock_validate): """Test get_filtered_detections with validation error""" # Call the function result = get_filtered_detections(detection_type="INVALID") # Verify result is the error message assert result == "Validation error message" @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.query_api") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_detection_url", return_value="https://app.wiz.io/detection/123") def test_get_filtered_detections_with_all_params(mock_url, mock_query_api, sample_detection): """Test get_filtered_detections with all parameters specified""" mock_query_api.return_value = [sample_detection] # Use valid parameters that will pass real validation result = get_filtered_detections( detection_id=str(uuid.uuid4()), # Valid UUID issue_id=str(uuid.uuid4()), # Valid UUID detection_type=[DetectionType.GENERATED_THREAT, DetectionType.DID_NOT_GENERATE_THREAT], # Valid enum values detection_platform=["AWS"], # Valid platform detection_origin=["WIZ_SENSOR"], # Valid origin detection_cloud_account_or_cloud_organization="12345678-1234-1234-1234-d25e16359c19", resource_id="test-id", severity="CRITICAL", # Valid severity creation_minutes_back="15", # Valid number rule_match_id=str(uuid.uuid4()), # Valid UUID rule_match_name="rule name", project_id="project-id", ) # Verify result assert result == [sample_detection] assert result[0].get("url") == "https://app.wiz.io/detection/123" @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.validate_all_detection_parameters") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.query_api") def test_get_filtered_detections_with_no_url(mock_query_api, mock_validate, sample_detection): """Test get_filtered_detections with add_detection_url=False""" # Set up validated values validated_values = { "severity": ["CRITICAL"], } # Configure the mocks mock_validate.return_value = (True, None, validated_values) mock_query_api.return_value = [sample_detection] # Call the function with add_detection_url=False result = get_filtered_detections(severity="CRITICAL", add_detection_url=False) # Verify result (should not have url) assert result == [sample_detection] assert "url" not in result[0] @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.validate_all_detection_parameters") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.query_api") def test_get_filtered_detections_with_api_limit(mock_query_api, mock_validate, sample_detection): """Test get_filtered_detections with custom API limit""" # Set up validated values validated_values = { "severity": ["CRITICAL"], } # Configure the mocks mock_validate.return_value = (True, None, validated_values) mock_query_api.return_value = [sample_detection] # Call the function with custom api_limit get_filtered_detections(severity="CRITICAL", api_limit=50) # Verify query_api was called with correct api_limit variables = mock_query_api.call_args[0][1] assert variables["first"] == 50 @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_detections", return_value=[{"id": "test-detection"}]) @patch("WizDefend.check_advanced_params", return_value=(True, "")) @patch("WizDefend.extract_params_from_integration_settings") @patch.object(demisto, "results") def test_test_module_success(mock_results, mock_extract_params, mock_check_params, mock_get_filtered): """Test test_module function with successful validation and API call""" # Set up mock extraction params mock_extract_params.return_value = { "type": "GENERATED THREAT", "platform": "AWS", "severity": "CRITICAL", "origin": "WIZ_SENSOR", "subscription": "test-subscription", } # Call the function test_module() # Verify demisto.results was called with 'ok' mock_results.assert_called_with("ok") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_detections", return_value=[{"id": "test-detection"}]) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.check_advanced_params", return_value=(False, "Parameter validation error")) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.extract_params_from_integration_settings") @patch.object(demisto, "results") def test_test_module_invalid_params(mock_results, mock_extract_params, mock_check_params, mock_get_filtered): """Test test_module function with invalid parameters""" # Set up mock extraction params mock_extract_params.return_value = {"type": "GENERATED THREAT", "platform": "AWS", "severity": "CRITICAL"} # Call the function test_module() # Verify demisto.results was called with the error message mock_results.assert_called_with("Parameter validation error") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_detections", return_value="API error message") @patch("WizDefend.check_advanced_params", return_value=(True, "")) @patch("WizDefend.extract_params_from_integration_settings") @patch.object(demisto, "results") def test_test_module_api_error(mock_results, mock_extract_params, mock_check_params, mock_get_filtered): """Test test_module function with API error""" # Set up mock extraction params mock_extract_params.return_value = {"type": "GENERATED THREAT", "platform": "AWS", "severity": "CRITICAL"} # Call the function test_module() # Verify demisto.results was called with the error message mock_results.assert_called_with("API error message") @freeze_time("2022-01-02T00:00:00Z") @patch.object(demisto, "setLastRun") @patch.object(demisto, "incidents") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_detections") @patch("WizDefend.FetchIncident", return_value="2022-01-01T00:00:00Z") @patch("WizDefend.extract_params_from_integration_settings") def test_fetch_incidents_success( mock_extract_params, mock_last_run, mock_get_filtered, mock_incidents, mock_set_last_run, sample_detection ): """Test fetch_incidents with successful API call""" # Set up mocks mock_extract_params.return_value = { "type": "GENERATED THREAT", "platform": "AWS", "severity": "CRITICAL", "origin": "WIZ_SENSOR", "subscription": "test-subscription", } mock_get_filtered.return_value = [sample_detection] # Call the function FIRST fetch_incidents() # THEN verify the results # Verify demisto.incidents was called with the incident incident_arg = mock_incidents.call_args[0][0] assert len(incident_arg) == 1 assert incident_arg[0]["name"] == "suspicious activity detected - 12345678-1234-1234-1234-d25e16359c19" # Verify demisto.setLastRun was called with the expected timestamp mock_set_last_run.assert_called_with( { "time": "2022-01-02T00:00:00Z", "endCursor": None, "after": "2022-01-02T00:00:00Z", # Based on the first error, this should match the actual implementation "before": "2022-01-02T00:00:00Z", } ) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_detections", return_value="API error message") @patch("WizDefend.FetchIncident.get_last_run_time", return_value="2022-01-01T00:00:00Z") @patch("WizDefend.extract_params_from_integration_settings") def test_fetch_incidents_api_error(mock_extract_params, mock_last_run, mock_get_filtered): """Test fetch_incidents with API error""" # Set up mocks mock_extract_params.return_value = {"type": "GENERATED THREAT", "platform": "AWS", "severity": "CRITICAL"} # Call the function fetch_incidents() # No assertions needed as we're just checking if it runs without raising exceptions @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.test_module") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_api_endpoint") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_authentication_endpoint") @patch.object(demisto, "command", return_value="test-module") def test_main_test_module(mock_command, mock_set_auth, mock_set_api, mock_test_module): """Test main function handling test-module command""" # Call the function main() # Verify endpoints were set assert mock_set_auth.called assert mock_set_api.called # Verify test_module was called assert mock_test_module.called @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.fetch_incidents") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_api_endpoint") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_authentication_endpoint") @patch.object(demisto, "command", return_value="fetch-incidents") def test_main_fetch_incidents(mock_command, mock_set_auth, mock_set_api, mock_fetch): """Test main function handling fetch-incidents command""" # Call the function main() # Verify endpoints were set assert mock_set_auth.called assert mock_set_api.called # Verify fetch_incidents was called assert mock_fetch.called @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_results") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_detections", return_value=[{"id": "test-detection"}]) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_api_endpoint") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_authentication_endpoint") @patch.object(demisto, "args") @patch.object(demisto, "command", return_value="wiz-defend-get-detections") def test_main_get_detections(mock_command, mock_args, mock_set_auth, mock_set_api, mock_filtered_detections, mock_return_results): """Test main function handling wiz-get-detections command""" # Set up mock args mock_args.return_value = {"severity": "CRITICAL", "type": "GENERATED THREAT", "platform": "AWS"} # Call the function main() # Verify endpoints were set assert mock_set_auth.called assert mock_set_api.called # Verify return_results was called assert mock_return_results.called @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_results") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_detections", return_value=[{"id": "test-detection"}]) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_api_endpoint") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_authentication_endpoint") @patch.object(demisto, "args") @patch.object(demisto, "command", return_value="wiz-defend-get-detection") def test_main_get_detection(mock_command, mock_args, mock_set_auth, mock_set_api, mock_filtered_detections, mock_return_results): """Test main function handling wiz-get-detection command""" # Set up mock args detection_id = str(uuid.uuid4()) mock_args.return_value = {"detection_id": detection_id} # Call the function main() # Verify endpoints were set assert mock_set_auth.called assert mock_set_api.called # Verify return_results was called assert mock_return_results.called @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_error") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_api_endpoint") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_authentication_endpoint") @patch.object(demisto, "args", side_effect=Exception("Test error")) @patch.object(demisto, "command", return_value="wiz-defend-get-detections") def test_main_error_handling(mock_command, mock_args, mock_set_auth, mock_set_api, mock_return_error): """Test main function error handling""" # Call the function main() # Verify return_error was called with the error message mock_return_error.assert_called_once() assert "Test error" in mock_return_error.call_args[0][0] @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_error") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_api_endpoint") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_authentication_endpoint") @patch.object(demisto, "command", return_value="unknown-command") def test_main_unknown_command(mock_command, mock_set_auth, mock_set_api, mock_return_error): """Test main function handling unknown command""" # Call the function main() # Verify return_error was called with appropriate message mock_return_error.assert_called_once() assert "Unrecognized command" in mock_return_error.call_args[0][0] @pytest.mark.parametrize( "end_cursor,expected_valid,expected_error", [ # Valid base64 strings ("dGVzdA==", True, None), # "test" in base64 ("SGVsbG8gV29ybGQ=", True, None), # "Hello World" in base64 ("", True, None), # Empty string should be valid (None, True, None), # None should be valid # Invalid base64 strings ("invalid_base64!", False, "Invalid end_cursor format"), ("not-base64", False, "Invalid end_cursor format"), ("123", False, "Invalid end_cursor format"), ], ) def test_validate_end_cursor(end_cursor, expected_valid, expected_error): """Test validate_end_cursor function with various inputs""" is_valid, error_message = validate_end_cursor(end_cursor) assert is_valid == expected_valid if expected_error: assert expected_error in error_message else: assert error_message is None @pytest.mark.parametrize( "after_time,before_time,expected_valid,expected_error", [ # Valid timestamps ("2022-01-01T00:00:00Z", "2022-01-02T00:00:00Z", True, None), ("2022-01-01T12:30:45.123Z", "2022-01-02T12:30:45.123Z", True, None), (None, "2022-01-02T00:00:00Z", True, None), # Only before_time ("2022-01-01T00:00:00Z", None, True, None), # Only after_time (None, None, True, None), # Both None # Invalid timestamps ("invalid-date", "2022-01-02T00:00:00Z", False, "Invalid after_time format"), ("2022-01-01T00:00:00Z", "invalid-date", False, "Invalid before_time format"), ("2022-13-01T00:00:00Z", "2022-01-02T00:00:00Z", False, "Invalid after_time format"), ("2022-01-01T25:00:00Z", "2022-01-02T00:00:00Z", False, "Invalid after_time format"), ], ) def test_validate_after_and_before_timestamps(after_time, before_time, expected_valid, expected_error): """Test validate_after_and_before_timestamps function""" is_valid, error_message = validate_after_and_before_timestamps(after_time, before_time) assert is_valid == expected_valid if expected_error: assert expected_error in error_message else: assert error_message is None @pytest.mark.parametrize( "variables,before_time,expected_result", [ # Test with empty variables ({}, "2022-01-01T00:00:00Z", {"filterBy": {"createdAt": {"before": "2022-01-01T00:00:00Z"}}}), # Test with existing filterBy ( {"filterBy": {"severity": ["CRITICAL"]}}, "2022-01-01T00:00:00Z", {"filterBy": {"severity": ["CRITICAL"], "createdAt": {"before": "2022-01-01T00:00:00Z"}}}, ), # Test with existing createdAt filter ( {"filterBy": {"createdAt": {"after": "2021-12-01T00:00:00Z"}}}, "2022-01-01T00:00:00Z", {"filterBy": {"createdAt": {"after": "2021-12-01T00:00:00Z", "before": "2022-01-01T00:00:00Z"}}}, ), # Test with None before_time (should not modify variables) ({"filterBy": {"severity": ["CRITICAL"]}}, None, {"filterBy": {"severity": ["CRITICAL"]}}), # Test with empty string before_time (should not modify variables) ({"filterBy": {"severity": ["CRITICAL"]}}, "", {"filterBy": {"severity": ["CRITICAL"]}}), ], ) def test_apply_creation_before_time_filter(variables, before_time, expected_result): """Test apply_creation_before_time_filter function""" result = apply_creation_before_time_filter(variables, before_time) assert result == expected_result @pytest.mark.parametrize( "variables,end_cursor,expected_result", [ # Test with empty variables ({}, "dGVzdA==", {"after": "dGVzdA=="}), # Test with existing variables ({"filterBy": {"severity": ["CRITICAL"]}}, "dGVzdA==", {"filterBy": {"severity": ["CRITICAL"]}, "after": "dGVzdA=="}), # Test with None end_cursor (should not modify variables) ({"filterBy": {"severity": ["CRITICAL"]}}, None, {"filterBy": {"severity": ["CRITICAL"]}}), # Test with empty string end_cursor (should not modify variables) ({"filterBy": {"severity": ["CRITICAL"]}}, "", {"filterBy": {"severity": ["CRITICAL"]}}), ], ) def test_apply_end_cursor(variables, end_cursor, expected_result): """Test apply_end_cursor function""" result = apply_end_cursor(variables, end_cursor) assert result == expected_result @pytest.mark.parametrize( "variables,after_time,expected_result", [ # Test with empty variables ({}, "2022-01-01T00:00:00Z", {"filterBy": {"createdAt": {"after": "2022-01-01T00:00:00Z"}}}), # Test with existing filterBy ( {"filterBy": {"severity": ["CRITICAL"]}}, "2022-01-01T00:00:00Z", {"filterBy": {"severity": ["CRITICAL"], "createdAt": {"after": "2022-01-01T00:00:00Z"}}}, ), # Test with existing createdAt filter ( {"filterBy": {"createdAt": {"before": "2022-02-01T00:00:00Z"}}}, "2022-01-01T00:00:00Z", {"filterBy": {"createdAt": {"before": "2022-02-01T00:00:00Z", "after": "2022-01-01T00:00:00Z"}}}, ), # Test with None after_time (should not modify variables) ({"filterBy": {"severity": ["CRITICAL"]}}, None, {"filterBy": {"severity": ["CRITICAL"]}}), # Test with empty string after_time (should not modify variables) ({"filterBy": {"severity": ["CRITICAL"]}}, "", {"filterBy": {"severity": ["CRITICAL"]}}), ], ) def test_apply_creation_after_time_filter(variables, after_time, expected_result): """Test apply_creation_after_time_filter function""" result = apply_creation_after_time_filter(variables, after_time) assert result == expected_result def test_apply_all_detection_filters_with_cursors(): """Test apply_all_detection_filters with after, before, and end_cursor parameters""" # Test with all three cursor-related parameters validated_values = { "after": "2022-01-01T00:00:00Z", "before": "2022-01-02T00:00:00Z", "endCursor": "dGVzdA==", "severity": ["CRITICAL"], } variables = {} result = apply_all_detection_filters(variables, validated_values) # Verify all filters were applied assert "filterBy" in result assert "createdAt" in result["filterBy"] assert result["filterBy"]["createdAt"]["after"] == "2022-01-01T00:00:00Z" assert result["filterBy"]["createdAt"]["before"] == "2022-01-02T00:00:00Z" assert result["after"] == "dGVzdA==" assert result["filterBy"]["severity"]["equals"] == ["CRITICAL"] def test_apply_all_detection_filters_time_conflict(): """Test apply_all_detection_filters handles time parameter conflicts correctly""" # Test with creation_minutes_back (should take precedence over after/before) validated_values = {"creation_minutes_back": 30, "severity": ["HIGH"]} variables = {} result = apply_all_detection_filters(variables, validated_values) # Should use creation_minutes_back filter, not after/before assert "filterBy" in result assert "createdAt" in result["filterBy"] assert "inLast" in result["filterBy"]["createdAt"] assert result["filterBy"]["createdAt"]["inLast"]["amount"] == 30 assert "after" not in result["filterBy"]["createdAt"] assert "before" not in result["filterBy"]["createdAt"] # ===== VALIDATION RESPONSE CLASS TESTS ===== def test_validation_response_class(): """Test the ValidationResponse class""" # Test success response success_response = ValidationResponse.create_success("test_value") assert success_response.is_valid is True assert success_response.error_message is None assert success_response.value == "test_value" # Test error response error_response = ValidationResponse.create_error("Error message") assert error_response.is_valid is False assert error_response.error_message == "Error message" assert error_response.value is None # Test custom attributes custom_response = ValidationResponse() custom_response.is_valid = True custom_response.minutes_value = 10 custom_response.severity_list = ["CRITICAL", "HIGH"] assert custom_response.is_valid is True assert custom_response.minutes_value == 10 assert custom_response.severity_list == ["CRITICAL", "HIGH"] # Test to_dict method response_dict = success_response.to_dict() assert response_dict["is_valid"] is True assert response_dict["error_message"] is None assert response_dict["value"] == "test_value" # ===== THREAT FUNCTION TESTS ===== def test_get_threat_url(mocker, sample_threat): """Test get_threat_url constructs the correct URL for threats with different domains""" # Import here to set the domain from Packs.Wiz.Integrations.WizDefend import WizDefend # Set the domain directly domain = "app.wiz.io" WizDefend.WIZ_DOMAIN_URL = domain # Call the function url = get_threat_url(sample_threat) # Expected URL pattern expected_pattern = ( f"https://{domain}/threats#~(filters~(createdAt~(inTheLast~(amount~90~unit~'days)))~issue~'{sample_threat['id']})" ) # Verify URL was correctly constructed assert url == expected_pattern @patch( "Packs.Wiz.Integrations.WizDefend.WizDefend.validate_all_threat_parameters", return_value=(True, None, {"severity": ["CRITICAL"], "platform": ["AWS"], "status": ["OPEN"]}), ) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.query_issues", return_value=[{"id": "test-threat"}]) def test_get_filtered_threats_success(mock_query_threats, mock_validate, sample_threat): """Test get_filtered_threats with successful validation and API call""" # Replace the API call with our mock threat mock_query_threats.return_value = [sample_threat] # Call the function result = get_filtered_threats(platform=["AWS"], severity="CRITICAL", status=["OPEN"]) # Verify result assert result == [sample_threat] assert mock_query_threats.called @patch( "Packs.Wiz.Integrations.WizDefend.WizDefend.validate_all_threat_parameters", return_value=(False, "Validation error message", None), ) def test_get_filtered_threats_validation_error(mock_validate): """Test get_filtered_threats with validation error""" # Call the function result = get_filtered_threats(status="INVALID") # Verify result is the error message assert result == "Validation error message" @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.validate_all_threat_parameters") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.query_issues") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_threat_url", return_value="https://app.wiz.io/threats/123") def test_get_filtered_threats_with_all_params(mock_url, mock_query_threats, mock_validate, sample_threat): """Test get_filtered_threats with all parameters specified""" # Set up validated values validated_values = { "issue_id": str(uuid.uuid4()), "platform": ["AWS"], "origin": ["WIZ_SENSOR"], "cloud_account_or_cloud_organization": "test-subscription", "resource_id": "test-id", "severity": ["CRITICAL"], "status": ["OPEN"], "creation_days_back": 15, "project_id": "project-id", } # Configure the mocks mock_validate.return_value = (True, None, validated_values) mock_query_threats.return_value = [sample_threat] # Call the function with all parameters result = get_filtered_threats( issue_id=validated_values["issue_id"], platform=validated_values["platform"], origin=validated_values["origin"], cloud_account_or_cloud_organization=validated_values["cloud_account_or_cloud_organization"], resource_id=validated_values["resource_id"], severity="CRITICAL", status=validated_values["status"], creation_days_back="15", project_id=validated_values["project_id"], ) # Verify result assert result == [sample_threat] assert result[0].get("url") == "https://app.wiz.io/threats/123" @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.validate_all_threat_parameters") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.query_issues") def test_get_filtered_threats_with_no_url(mock_query_threats, mock_validate, sample_threat): """Test get_filtered_threats with add_threat_url=False""" # Set up validated values validated_values = { "severity": ["CRITICAL"], } # Configure the mocks mock_validate.return_value = (True, None, validated_values) mock_query_threats.return_value = [sample_threat] # Call the function with add_threat_url=False result = get_filtered_threats(severity="CRITICAL", add_threat_url=False) # Verify result (should not have url) assert result == [sample_threat] assert "url" not in result[0] @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_results") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_threats", return_value=[{"id": "test-threat"}]) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_api_endpoint") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_authentication_endpoint") @patch.object(demisto, "args") @patch.object(demisto, "command", return_value="wiz-defend-get-threats") def test_main_get_threats(mock_command, mock_args, mock_set_auth, mock_set_api, mock_filtered_threats, mock_return_results): """Test main function handling wiz-get-threats command""" # Set up mock args mock_args.return_value = {"severity": "CRITICAL", "platform": "AWS", "status": "OPEN"} # Call the function main() # Verify endpoints were set assert mock_set_auth.called assert mock_set_api.called # Verify return_results was called assert mock_return_results.called @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_results") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_threats", return_value=[{"id": "test-threat"}]) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_api_endpoint") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_authentication_endpoint") @patch.object(demisto, "args") @patch.object(demisto, "command", return_value="wiz-defend-get-threat") def test_main_get_threat(mock_command, mock_args, mock_set_auth, mock_set_api, mock_filtered_threats, mock_return_results): """Test main function handling wiz-get-threat command""" # Set up mock args issue_id = str(uuid.uuid4()) mock_args.return_value = {"issue_id": issue_id} # Call the function main() # Verify endpoints were set assert mock_set_auth.called assert mock_set_api.called # Verify return_results was called assert mock_return_results.called @pytest.fixture def invalid_threat_scenarios(): """Fixture for various invalid threat scenarios""" return [ { "name": "missing_issue_id", "issue_id": None, "api_response": None, "expected_error": "should pass an Issue ID", "should_call_api": False, }, { "name": "empty_issue_id", "issue_id": "", "api_response": None, "expected_error": "should pass an Issue ID", "should_call_api": False, }, { "name": "invalid_uuid_format", "issue_id": "not-a-uuid", "api_response": None, "expected_error": "UUID format", "should_call_api": False, }, { "name": "api_returns_error", "issue_id": str(uuid.uuid4()), "api_response": "Error: Threat not found", "expected_error": "Error retrieving threat", "should_call_api": True, }, ] # Test 1: Valid scenarios @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_results") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_threats") def test_get_single_threat_valid_scenarios(mock_get_filtered, mock_return_results, valid_threat_scenario): """Test get_single_threat with valid inputs""" from Packs.Wiz.Integrations.WizDefend.WizDefend import get_single_threat # Setup from fixture mock_get_filtered.return_value = valid_threat_scenario["api_response"] with patch.object(demisto, "args", return_value={"issue_id": valid_threat_scenario["issue_id"]}): get_single_threat() # Verify success path mock_get_filtered.assert_called_once_with(issue_id=valid_threat_scenario["issue_id"]) mock_return_results.assert_called_once() # Verify return_results called with correct data call_args = mock_return_results.call_args[0][0] assert call_args.outputs_prefix == OutputPrefix.THREAT assert call_args.outputs == valid_threat_scenario["api_response"] # Test 2: Invalid scenarios (parameterized with fixture) @pytest.mark.parametrize( "scenario", [ pytest.param( {"issue_id": None, "api_response": None, "expected_error": "should pass an Issue ID", "should_call_api": False}, id="missing_issue_id", ), pytest.param( {"issue_id": "", "api_response": None, "expected_error": "should pass an Issue ID", "should_call_api": False}, id="empty_issue_id", ), pytest.param( {"issue_id": "not-a-uuid", "api_response": None, "expected_error": "UUID format", "should_call_api": False}, id="invalid_uuid", ), pytest.param( { "issue_id": str(uuid.uuid4()), "api_response": "Error: Threat not found", "expected_error": "Error retrieving threat", "should_call_api": True, }, id="api_error", ), ], ) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.log_and_return_error") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_threats") def test_get_single_threat_invalid_scenarios(mock_get_filtered, mock_log_and_return_error, scenario): """Test get_single_threat with invalid inputs and error conditions""" from Packs.Wiz.Integrations.WizDefend.WizDefend import get_single_threat # Setup from scenario if scenario["should_call_api"]: mock_get_filtered.return_value = scenario["api_response"] args_value = {"issue_id": scenario["issue_id"]} if scenario["issue_id"] is not None else {} with patch.object(demisto, "args", return_value=args_value): get_single_threat() # Verify error handling mock_log_and_return_error.assert_called_once() error_message = mock_log_and_return_error.call_args[0][0] assert scenario["expected_error"] in error_message # Verify API call behavior if scenario["should_call_api"]: mock_get_filtered.assert_called_once() else: mock_get_filtered.assert_not_called() @pytest.mark.parametrize( "set_status_return,should_succeed", [ ({"id": "test-threat", "status": "IN_PROGRESS"}, True), (None, False), ("", False), ], ) @patch.object(demisto, "args", return_value={"issue_id": str(uuid.uuid4())}) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_results") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_error") def test_set_threat_in_progress(mock_return_error, mock_return_results, mock_args, set_status_return, should_succeed): """Test set_threat_in_progress function with various API responses""" with patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_status", return_value=set_status_return): set_threat_in_progress() if should_succeed: mock_return_results.assert_called_once() call_args = mock_return_results.call_args[0][0] assert call_args.outputs_prefix == OutputPrefix.THREAT assert "Successfully set the threat" in call_args.outputs assert not mock_return_error.called else: mock_return_error.assert_called_once() error_message = mock_return_error.call_args[0][0] assert "Failed to set the threat" in error_message assert not mock_return_results.called @pytest.mark.parametrize( "set_issue_note_return,should_succeed", [ ({"id": "note-123"}, True), (None, False), ("", False), ], ) @patch.object(demisto, "args", return_value={"issue_id": str(uuid.uuid4()), "note": "Test comment"}) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_results") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_error") def test_set_threat_comment(mock_return_error, mock_return_results, mock_args, set_issue_note_return, should_succeed): """Test set_threat_comment function with various API responses""" with patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_issue_note", return_value=set_issue_note_return): set_threat_comment() if should_succeed: mock_return_results.assert_called_once() call_args = mock_return_results.call_args[0][0] assert call_args.outputs_prefix == OutputPrefix.THREAT assert "Successfully set Test comment as comment" in call_args.outputs assert not mock_return_error.called else: mock_return_error.assert_called_once() error_message = mock_return_error.call_args[0][0] assert "Failed to set the comment Test comment" in error_message assert not mock_return_results.called def test_truncate_note(): from WizDefend import truncate_note, MAX_NOTE_LENGTH # Short notes are unchanged assert truncate_note("short note") == "short note" assert truncate_note("") == "" assert truncate_note(None) is None # Exactly at limit is unchanged exact = "a" * MAX_NOTE_LENGTH assert truncate_note(exact) == exact # Over limit gets truncated long_text = "a" * (MAX_NOTE_LENGTH + 1) result = truncate_note(long_text) assert len(result) <= MAX_NOTE_LENGTH assert result.endswith("... [truncated]") assert len(result) == MAX_NOTE_LENGTH @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_entries", return_value={"id": "note-123"}) def test_set_issue_note_truncates_long_note(mock_get_entries): long_note = "a" * (MAX_NOTE_LENGTH + 1) set_issue_note(str(uuid.uuid4()), long_note) call_args = mock_get_entries.call_args sent_text = call_args[0][1]["input"]["text"] assert len(sent_text) <= MAX_NOTE_LENGTH assert sent_text.endswith("... [truncated]") def test_build_fallback_description_with_rule(): detection = { "id": "det-123", "severity": "CRITICAL", "ruleMatch": {"rule": {"name": "suspicious activity detected"}}, } result = build_fallback_description(detection) assert result == "CRITICAL severity detection triggered by rule 'suspicious activity detected' (ID: det-123)" def test_build_fallback_description_without_rule(): detection = {"id": "det-456", "severity": "HIGH"} result = build_fallback_description(detection) assert result == "HIGH severity detection (ID: det-456)" def test_build_fallback_description_minimal(): result = build_fallback_description({}) assert result == "Unknown severity detection (ID: Unknown)" def test_build_incidents_includes_details(): detection = { "id": "12345678-1234-1234-1234-d25e16359c19", "severity": "CRITICAL", "createdAt": "2022-01-02T15:46:34Z", "description": "Suspicious activity detected", "ruleMatch": {"rule": {"name": "test rule", "sourceType": "THREAT_DETECTION", "securitySubCategories": []}}, } result = build_incidents(detection) assert result[DemistoParams.DETAILS] == "Suspicious activity detected" def test_build_incidents_null_description_gets_empty_string(): detection = { "id": "12345678-1234-1234-1234-d25e16359c19", "severity": "HIGH", "createdAt": "2022-01-02T15:46:34Z", "ruleMatch": {"rule": {"name": "test rule", "sourceType": "THREAT_DETECTION", "securitySubCategories": []}}, } result = build_incidents(detection) assert result[DemistoParams.DETAILS] == "" # ===== GAP #11: get_filtered_detections null description fallback flow ===== @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.query_detections") @patch( "Packs.Wiz.Integrations.WizDefend.WizDefend.validate_all_detection_parameters", return_value=(True, None, {"severity": ["HIGH"]}), ) def test_get_filtered_detections_null_description_gets_fallback(mock_validate, mock_query): """Test that get_filtered_detections populates fallback description for detections with null description""" detection_no_desc = { "id": "det-no-desc-001", "severity": "HIGH", "ruleMatch": {"rule": {"name": "lateral movement detected"}}, "createdAt": "2022-01-02T15:46:34Z", } mock_query.return_value = [detection_no_desc] result = get_filtered_detections(severity="HIGH", add_detection_url=False) assert len(result) == 1 expected = "HIGH severity detection triggered by rule 'lateral movement detected' (ID: det-no-desc-001)" assert result[0]["description"] == expected @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.query_detections") @patch( "Packs.Wiz.Integrations.WizDefend.WizDefend.validate_all_detection_parameters", return_value=(True, None, {"severity": ["CRITICAL"]}), ) def test_get_filtered_detections_existing_description_unchanged(mock_validate, mock_query): """Test that get_filtered_detections does NOT overwrite an existing description""" detection_with_desc = { "id": "det-with-desc-001", "severity": "CRITICAL", "description": "Original description from API", "ruleMatch": {"rule": {"name": "some rule"}}, "createdAt": "2022-01-02T15:46:34Z", } mock_query.return_value = [detection_with_desc] result = get_filtered_detections(severity="CRITICAL", add_detection_url=False) assert result[0]["description"] == "Original description from API" @pytest.mark.parametrize( "threat_notes,delete_success_count,should_succeed", [ ([{"id": "note-1", "text": "First comment"}, {"id": "note-2", "text": "Second comment"}], 2, True), ([{"id": "note-1", "text": "Single comment"}], 1, True), ([], 0, True), # No notes to delete ([{"id": "note-1", "text": "Comment"}], 0, False), # Delete fails ], ) @patch.object(demisto, "args", return_value={"issue_id": str(uuid.uuid4())}) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_threats") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_results") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_error") def test_clear_threat_comments( mock_return_error, mock_return_results, mock_get_filtered_threats, mock_args, threat_notes, delete_success_count, should_succeed, ): """Test clear_threat_comments function with various note scenarios""" # Mock threat with specified notes threat_with_notes = [{"id": "threat-123", "notes": threat_notes}] mock_get_filtered_threats.return_value = threat_with_notes # Mock get_entries to return success or None based on test case get_entries_return = {"id": "deleted-note"} if should_succeed else None with patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_entries", return_value=get_entries_return) as mock_get_entries: clear_threat_comments() # Verify get_entries was called the expected number of times assert mock_get_entries.call_count == len(threat_notes) if should_succeed: mock_return_results.assert_called_once() call_args = mock_return_results.call_args[0][0] assert call_args.outputs_prefix == OutputPrefix.THREAT assert "Successfully cleared all the comments" in call_args.outputs assert not mock_return_error.called elif threat_notes: # Only expect error if there were notes to delete mock_return_error.assert_called_once() error_message = mock_return_error.call_args[0][0] assert "Failed to delete the comment" in error_message assert not mock_return_results.called @pytest.mark.parametrize( "validation_result,should_succeed", [ ((True, None), True), ((False, "Not a threat detection issue"), False), ((False, "Invalid UUID format"), False), ], ) @patch.object( demisto, "args", return_value={"issue_id": str(uuid.uuid4()), "resolution_reason": "ISSUE_FIXED", "resolution_note": "Fixed the issue"}, ) @patch( "Packs.Wiz.Integrations.WizDefend.WizDefend.reject_or_resolve_issue", return_value={"id": "threat-123", "status": "RESOLVED"} ) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_error") def test_resolve_threat(mock_return_error, mock_reject_or_resolve, mock_args, validation_result, should_succeed): """Test resolve_threat function with various validation results""" with patch("Packs.Wiz.Integrations.WizDefend.WizDefend.validate_threat_detections_issue", return_value=validation_result): resolve_threat() if should_succeed: mock_reject_or_resolve.assert_called_once_with( mock_args.return_value["issue_id"], mock_args.return_value["resolution_reason"], mock_args.return_value["resolution_note"], WizStatus.RESOLVED, ) assert not mock_return_error.called else: mock_return_error.assert_called_once() error_message = mock_return_error.call_args[0][0] assert validation_result[1] in error_message assert not mock_reject_or_resolve.called @pytest.mark.parametrize( "reopen_issue_return,should_succeed", [ ({"id": "threat-123", "status": "OPEN"}, True), (None, False), ("", False), ], ) @patch.object(demisto, "args", return_value={"issue_id": str(uuid.uuid4()), "reopen_note": "Reopening for review"}) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_results") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_error") def test_reopen_threat(mock_return_error, mock_return_results, mock_args, reopen_issue_return, should_succeed): """Test reopen_threat function with various API responses""" with patch("Packs.Wiz.Integrations.WizDefend.WizDefend._reopen_issue", return_value=reopen_issue_return): reopen_threat() if should_succeed: mock_return_results.assert_called_once() call_args = mock_return_results.call_args[0][0] assert call_args.outputs_prefix == OutputPrefix.THREAT assert "Successfully reopened the threat" in call_args.outputs assert not mock_return_error.called else: mock_return_error.assert_called_once() error_message = mock_return_error.call_args[0][0] assert "Failed to reopen the threat" in error_message assert not mock_return_results.called @pytest.mark.parametrize( "is_valid_id_result,status,get_entries_return,expected_result_type,should_succeed", [ # Valid cases ((True, "Valid"), WizStatus.IN_PROGRESS, {"id": "threat-123", "status": "IN_PROGRESS"}, dict, True), ((True, "Valid"), WizStatus.OPEN, {"id": "threat-123", "status": "OPEN"}, dict, True), ((True, "Valid"), WizStatus.REJECTED, {"id": "threat-123", "status": "REJECTED"}, dict, True), ((True, "Valid"), WizStatus.RESOLVED, {"id": "threat-123", "status": "RESOLVED"}, dict, True), # Invalid ID cases ((False, "Invalid UUID format"), WizStatus.IN_PROGRESS, None, str, False), # Invalid status cases ((True, "Valid"), "INVALID_STATUS", None, str, False), ((True, "Valid"), "PENDING", None, str, False), ((True, "Valid"), "CLOSED", None, str, False), ((True, "Valid"), "", None, str, False), ], ) def test_set_status_with_validation(is_valid_id_result, status, get_entries_return, expected_result_type, should_succeed): """Test set_status function with various validation scenarios including status validation""" issue_id = str(uuid.uuid4()) if is_valid_id_result[0] else "invalid-uuid" mock_is_valid = patch("Packs.Wiz.Integrations.WizDefend.WizDefend.is_valid_issue_id", return_value=is_valid_id_result) mock_get_entries_patch = patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_entries", return_value=get_entries_return) with mock_is_valid, mock_get_entries_patch as mock_get_entries: result = set_status(issue_id, status) assert isinstance(result, expected_result_type) if should_succeed and is_valid_id_result[0]: # Should call get_entries for valid cases mock_get_entries.assert_called_once() call_args = mock_get_entries.call_args variables = call_args[0][1] # Second positional argument (variables) assert variables[WizApiVariables.ISSUE_ID] == issue_id assert variables[WizApiVariables.PATCH][WizApiVariables.STATUS] == status assert result == get_entries_return elif not is_valid_id_result[0]: # Invalid ID should return error message without calling API assert not mock_get_entries.called assert result == is_valid_id_result[1] else: # Invalid status should return error message without calling API assert not mock_get_entries.called assert "Invalid status" in result assert status in result or str(status) in result @pytest.mark.parametrize( "issue_id,status,expected_result_type,should_call_api", [ # Valid combinations - should call API (str(uuid.uuid4()), WizStatus.IN_PROGRESS, dict, True), (str(uuid.uuid4()), WizStatus.OPEN, dict, True), (str(uuid.uuid4()), WizStatus.REJECTED, dict, True), (str(uuid.uuid4()), WizStatus.RESOLVED, dict, True), # Invalid UUID - should fail validation, not call API ("not-a-uuid", WizStatus.IN_PROGRESS, str, False), (None, WizStatus.IN_PROGRESS, str, False), ("", WizStatus.IN_PROGRESS, str, False), # Invalid status - should fail validation, not call API (str(uuid.uuid4()), "INVALID_STATUS", str, False), (str(uuid.uuid4()), "PENDING", str, False), (str(uuid.uuid4()), None, str, False), # Both invalid - should fail on UUID first ("not-a-uuid", "INVALID_STATUS", str, False), ], ) def test_set_status_validation(issue_id, status, expected_result_type, should_call_api): """Test set_status function with real validation - no mocking of validation functions""" # Only mock the API call, not the validation with patch( "Packs.Wiz.Integrations.WizDefend.WizDefend.get_entries", return_value={"id": "test-response"} ) as mock_get_entries: result = set_status(issue_id, status) # Check return type assert isinstance(result, expected_result_type) if should_call_api: # Should succeed and call the API mock_get_entries.assert_called_once() call_args = mock_get_entries.call_args variables = call_args[0][1] # Second positional argument (variables) assert variables[WizApiVariables.ISSUE_ID] == issue_id assert variables[WizApiVariables.PATCH][WizApiVariables.STATUS] == status else: # Should fail validation and not call API mock_get_entries.assert_not_called() # Check that we get appropriate error messages if issue_id in [None, "", "not-a-uuid"]: assert "UUID format" in result or "should pass an Issue ID" in result elif status not in [WizStatus.OPEN, WizStatus.IN_PROGRESS, WizStatus.REJECTED, WizStatus.RESOLVED]: assert "Invalid status" in result @pytest.mark.parametrize( "is_valid_result,issue_type,expected_valid,expected_error_contains", [ ((True, "Valid"), "THREAT_DETECTION", True, None), ((False, "Invalid UUID"), None, False, "Invalid UUID"), ((True, "Valid"), "CLOUD_CONFIGURATION", False, "Only a Threat Detection Issue can be resolved"), ((True, "Valid"), "TOXIC_COMBINATION", False, "Only a Threat Detection Issue can be resolved"), ], ) def test_validate_threat_detections_issue(is_valid_result, issue_type, expected_valid, expected_error_contains): """Test validate_threat_detections_issue with various issue types""" issue_id = str(uuid.uuid4()) with patch("Packs.Wiz.Integrations.WizDefend.WizDefend.is_valid_issue_id", return_value=is_valid_result): if is_valid_result[0] and issue_type: with patch("Packs.Wiz.Integrations.WizDefend.WizDefend.query_single_issue", return_value=[{"type": issue_type}]): is_valid, message = validate_threat_detections_issue(issue_id) else: is_valid, message = validate_threat_detections_issue(issue_id) assert is_valid == expected_valid if expected_error_contains: assert expected_error_contains in message elif expected_valid: assert message is None # @pytest.mark.parametrize( # "command_name,function_name", # [ # ("wiz-defend-resolve-threat", "resolve_threat"), # ("wiz-defend-reopen-threat", "reopen_threat"), # ("wiz-defend-set-threat-in-progress", "set_threat_in_progress"), # ("wiz-defend-set-threat-comment", "set_threat_comment"), # ("wiz-defend-clear-threat-comments", "clear_threat_comments"), # ], # ) # def test_main_new_commands(command_name, function_name): # """Test main function handling new threat management commands""" # with ( # patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_authentication_endpoint") as mock_set_auth, # patch("Packs.Wiz.Integrations.WizDefend.WizDefend.set_api_endpoint") as mock_set_api, # patch(f"Packs.Wiz.Integrations.WizDefend.WizDefend.{function_name}") as mock_function, # patch.object(demisto, "command", return_value=command_name), # ): # main() # # # Verify endpoints were set # assert mock_set_auth.called # assert mock_set_api.called # # # Verify the specific function was called # mock_function.assert_called_once() @pytest.fixture def sample_threat_with_notes(): """Return a sample threat object with notes for testing""" return { "id": "12345678-1234-1234-1234-d25e16359c19", "severity": "CRITICAL", "status": "OPEN", "type": "THREAT_DETECTION", "notes": [ {"id": "note-123", "text": "This is a test comment", "createdAt": "2022-01-02T15:46:34Z"}, {"id": "note-456", "text": "Another test comment", "createdAt": "2022-01-02T16:00:00Z"}, ], } # ===== CLASS TESTS ===== def test_detection_type_class(): """Test the DetectionType class""" # Test values method values = DetectionType.values() assert "GENERATED THREAT" in values assert "DID NOT GENERATE THREAT" in values # Test api_values method api_values = DetectionType.api_values() assert "GENERATED_THREAT" in api_values assert "MATCH_ONLY" in api_values # Test get_api_value method assert DetectionType.get_api_value("GENERATED THREAT") == "GENERATED_THREAT" assert DetectionType.get_api_value("DID NOT GENERATE THREAT") == "MATCH_ONLY" assert DetectionType.get_api_value("generated threat") == "GENERATED_THREAT" # Case insensitive assert DetectionType.get_api_value(None) is None assert DetectionType.get_api_value("INVALID_TYPE") is None def test_cloud_platform_class(): """Test the CloudPlatform class""" # Test values method values = CloudPlatform.values() assert "AWS" in values assert "Azure" in values assert "GCP" in values assert "Kubernetes" in values # Ensure all platforms are included expected_platforms = [ "AWS", "GCP", "Azure", "OCI", "Alibaba", "vSphere", "OpenStack", "AKS", "EKS", "GKE", "Kubernetes", "OpenShift", ] for platform in expected_platforms: assert platform in values def test_detection_origin_class(): """Test the DetectionOrigin class""" # Test values method values = DetectionOrigin.values() assert "WIZ_SENSOR" in values assert "AWS_GUARD_DUTY" in values assert "AZURE_DEFENDER_FOR_CLOUD" in values # Ensure some key origins are included expected_origins = [ "WIZ_SENSOR", "AWS_GUARD_DUTY", "AWS_CLOUDTRAIL", "AZURE_DEFENDER_FOR_CLOUD", "GCP_SECURITY_COMMAND_CENTER", ] for origin in expected_origins: assert origin in values def test_duration_unit_class(): """Test the DurationUnit class""" assert DurationUnit.DAYS == "DurationFilterValueUnitDays" assert DurationUnit.HOURS == "DurationFilterValueUnitHours" assert DurationUnit.MINUTES == "DurationFilterValueUnitMinutes" # ===== YML FIXTURES ===== def test_cloud_platform_values_consistency(yaml_content): """Test that cloud platform values defined in YAML match those in the CloudPlatform class""" # Extract platform values from both commands in YAML yaml_platform_values_detections = [] yaml_platform_values_threats = [] for command in yaml_content.get("script", {}).get("commands", []): if command.get("name") == DemistoCommands.WIZ_DEFEND_GET_DETECTIONS: for arg in command.get("arguments", []): if arg.get("name") == "platform": yaml_platform_values_detections = arg.get("predefined", []) elif command.get("name") == DemistoCommands.WIZ_DEFEND_GET_THREATS: for arg in command.get("arguments", []): if arg.get("name") == "platform": yaml_platform_values_threats = arg.get("predefined", []) # Also check the root configuration platforms yaml_config_platforms = [] for config in yaml_content.get("configuration", []): if config.get("name") == "platform": yaml_config_platforms = config.get("options", []) # Get platform values from code code_platform_values = CloudPlatform.values() # Sort all lists for deterministic comparison code_platform_values.sort() yaml_platform_values_detections.sort() yaml_platform_values_threats.sort() yaml_config_platforms.sort() # Test exact equality between code and YAML values assert code_platform_values == yaml_platform_values_detections, ( f"Platform values in code do not exactly match wiz-get-detections:\n" f"Code: {code_platform_values}\n" f"YAML: {yaml_platform_values_detections}" ) assert code_platform_values == yaml_platform_values_threats, ( f"Platform values in code do not exactly match wiz-get-threats:\n" f"Code: {code_platform_values}\n" f"YAML: {yaml_platform_values_threats}" ) assert code_platform_values == yaml_config_platforms, ( f"Platform values in code do not exactly match configuration options:\n" f"Code: {code_platform_values}\n" f"YAML config: {yaml_config_platforms}" ) # Test exact equality between the two commands assert yaml_platform_values_detections == yaml_platform_values_threats, ( f"Platform values differ between commands:\n" f"wiz-get-detections: {yaml_platform_values_detections}\n" f"wiz-get-threats: {yaml_platform_values_threats}" ) def test_detection_type_values_consistency(yaml_content): """Test that detection type values defined in YAML match those in the DetectionType class""" # Extract detection type values from YAML yaml_type_values_command = [] yaml_type_values_config = [] for command in yaml_content.get("script", {}).get("commands", []): if command.get("name") == DemistoCommands.WIZ_DEFEND_GET_DETECTIONS: for arg in command.get("arguments", []): if arg.get("name") == "type": yaml_type_values_command = arg.get("predefined", []) for config in yaml_content.get("configuration", []): if config.get("name") == "type": yaml_type_values_config = config.get("options", []) # Remove 'None' from configuration options if present if "None" in yaml_type_values_config: yaml_type_values_config.remove("None") # Get detection type values from code code_type_values = DetectionType.values() # Sort all lists for deterministic comparison code_type_values.sort() yaml_type_values_command.sort() yaml_type_values_config.sort() # Test exact equality between code and YAML values assert code_type_values == yaml_type_values_command, ( f"Detection type values in code do not exactly match wiz-get-detections:\n" f"Code: {code_type_values}\n" f"YAML: {yaml_type_values_command}" ) assert code_type_values == yaml_type_values_config, ( f"Detection type values in code do not exactly match configuration options:\n" f"Code: {code_type_values}\n" f"YAML config: {yaml_type_values_config}" ) def test_severity_values_consistency(yaml_content): """Test that severity values defined in YAML match those in the WizSeverity class, ignoring INFORMATIONAL""" # Extract severity values from both commands in YAML yaml_severity_values_detections = [] yaml_severity_values_threats = [] yaml_severity_config = [] for command in yaml_content.get("script", {}).get("commands", []): if command.get("name") == DemistoCommands.WIZ_DEFEND_GET_DETECTIONS: for arg in command.get("arguments", []): if arg.get("name") == "severity": yaml_severity_values_detections = arg.get("predefined", []) elif command.get("name") == DemistoCommands.WIZ_DEFEND_GET_THREATS: for arg in command.get("arguments", []): if arg.get("name") == "severity": yaml_severity_values_threats = arg.get("predefined", []) for config in yaml_content.get("configuration", []): if config.get("name") == "severity": yaml_severity_config = config.get("options", []) # Remove 'None' from configuration options if present if "None" in yaml_severity_config: yaml_severity_config.remove("None") # Get severity values from code code_severity_values = [ getattr(WizSeverity, attr) for attr in dir(WizSeverity) if not attr.startswith("_") and not callable(getattr(WizSeverity, attr)) ] # Convert all values to uppercase for case-insensitive comparison # and filter out INFORMATIONAL from all sets code_upper = {s.upper() for s in code_severity_values if s.upper() != "INFORMATIONAL"} yaml_detections_upper = {s.upper() for s in yaml_severity_values_detections if s.upper() != "INFORMATIONAL"} yaml_threats_upper = {s.upper() for s in yaml_severity_values_threats if s.upper() != "INFORMATIONAL"} yaml_config_upper = {s.upper() for s in yaml_severity_config if s.upper() != "INFORMATIONAL"} # Test case-insensitive equality between code and YAML values (ignoring INFORMATIONAL) assert code_upper == yaml_config_upper, ( f"Severity values in code do not match configuration options (ignoring INFORMATIONAL):\n" f"Code (uppercase, without INFORMATIONAL): {sorted(code_upper)}\n" f"YAML config (uppercase, without INFORMATIONAL): {sorted(yaml_config_upper)}" ) assert code_upper == yaml_detections_upper, ( f"Severity values in code do not match wiz-get-detections (ignoring INFORMATIONAL):\n" f"Code (uppercase, without INFORMATIONAL): {sorted(code_upper)}\n" f"YAML detections (uppercase, without INFORMATIONAL): {sorted(yaml_detections_upper)}" ) assert code_upper == yaml_threats_upper, ( f"Severity values in code do not match wiz-get-threats (ignoring INFORMATIONAL):\n" f"Code (uppercase, without INFORMATIONAL): {sorted(code_upper)}\n" f"YAML threats (uppercase, without INFORMATIONAL): {sorted(yaml_threats_upper)}" ) # Test equality between the YAML values (case-insensitive) assert yaml_detections_upper == yaml_threats_upper, ( f"Severity values differ between commands (ignoring INFORMATIONAL):\n" f"wiz-get-detections (uppercase, without INFORMATIONAL): {sorted(yaml_detections_upper)}\n" f"wiz-get-threats (uppercase, without INFORMATIONAL): {sorted(yaml_threats_upper)}" ) assert yaml_detections_upper == yaml_config_upper, ( f"Severity values differ between command and configuration (ignoring INFORMATIONAL):\n" f"wiz-get-detections (uppercase, without INFORMATIONAL): {sorted(yaml_detections_upper)}\n" f"configuration (uppercase, without INFORMATIONAL): {sorted(yaml_config_upper)}" ) def test_status_values_consistency(yaml_content): """Test that status values defined in YAML match those in the WizStatus class""" # Extract status values from threats command in YAML yaml_status_values = [] for command in yaml_content.get("script", {}).get("commands", []): if command.get("name") == DemistoCommands.WIZ_DEFEND_GET_THREATS: for arg in command.get("arguments", []): if arg.get("name") == "status": yaml_status_values = arg.get("predefined", []) # Get status values from code code_status_values = [ getattr(WizStatus, attr) for attr in dir(WizStatus) if not attr.startswith("_") and not callable(getattr(WizStatus, attr)) ] # Sort all lists for deterministic comparison code_status_values.sort() yaml_status_values.sort() # Test exact equality between code and YAML values assert ( code_status_values == yaml_status_values ), f"Status values in code do not exactly match wiz-get-threats:\nCode: {code_status_values}\nYAML: {yaml_status_values}" def test_origin_values_consistency(yaml_content): """Test that origin values defined in YAML match those in the DetectionOrigin class""" # Extract origin values from both commands in YAML yaml_origin_values_detections = [] yaml_origin_values_threats = [] yaml_origin_config = [] for command in yaml_content.get("script", {}).get("commands", []): if command.get("name") == DemistoCommands.WIZ_DEFEND_GET_DETECTIONS: for arg in command.get("arguments", []): if arg.get("name") == "origin": yaml_origin_values_detections = arg.get("predefined", []) elif command.get("name") == DemistoCommands.WIZ_DEFEND_GET_THREATS: for arg in command.get("arguments", []): if arg.get("name") == "origin": yaml_origin_values_threats = arg.get("predefined", []) for config in yaml_content.get("configuration", []): if config.get("name") == "origin": yaml_origin_config = config.get("options", []) # Get origin values from code code_origin_values = DetectionOrigin.values() # Sort all lists for deterministic comparison code_origin_values.sort() yaml_origin_values_detections.sort() yaml_origin_values_threats.sort() yaml_origin_config.sort() # Test exact equality between code and YAML values assert code_origin_values == yaml_origin_values_detections, ( f"Origin values in code do not exactly match wiz-get-detections:\n" f"Code: {code_origin_values}\n" f"YAML: {yaml_origin_values_detections}" ) assert code_origin_values == yaml_origin_values_threats, ( f"Origin values in code do not exactly match wiz-get-threats:\n" f"Code: {code_origin_values}\n" f"YAML: {yaml_origin_values_threats}" ) if yaml_origin_config: # Only check if origin config exists assert code_origin_values == yaml_origin_config, ( f"Origin values in code do not exactly match configuration options:\n" f"Code: {code_origin_values}\n" f"YAML config: {yaml_origin_config}" ) # Test exact equality between the two commands assert yaml_origin_values_detections == yaml_origin_values_threats, ( f"Origin values differ between commands:\n" f"wiz-get-detections: {yaml_origin_values_detections}\n" f"wiz-get-threats: {yaml_origin_values_threats}" ) def test_max_fetch_values_consistency(yaml_content): """Test that max_fetch values defined in YAML match those in the code constants""" # Find the max_fetch configuration max_fetch_config = None for config in yaml_content.get("configuration", []): if config.get("name") == "max_fetch": max_fetch_config = config break assert max_fetch_config is not None, "max_fetch configuration not found in YAML" # Get default value from YAML default_value_in_yaml = max_fetch_config.get("defaultvalue") assert default_value_in_yaml is not None, "No defaultvalue for max_fetch" default_max_fetch_in_yaml = int(default_value_in_yaml) # Assert that the code constant matches the YAML value assert default_max_fetch_in_yaml == API_MAX_FETCH, ( f"API_MAX_FETCH ({API_MAX_FETCH}) does not match " f"the defaultvalue in max_fetch configuration ({default_max_fetch_in_yaml})" ) # Check additional info mentions the range additional_info = max_fetch_config.get("additionalinfo", "") assert ( "10-1000" in additional_info ), f"max_fetch additionalinfo doesn't mention the expected range (10-1000): {additional_info}" # Verify the range in additionalinfo matches the code constants import re range_match = re.search(r"(\d+)-(\d+)", additional_info) assert range_match is not None, f"Could not find range format in additionalinfo: {additional_info}" min_value_in_yaml = int(range_match.group(1)) max_value_in_yaml = int(range_match.group(2)) assert ( min_value_in_yaml == API_MIN_FETCH ), f"API_MIN_FETCH ({API_MIN_FETCH}) does not match the minimum value in max_fetch additionalinfo ({min_value_in_yaml})" assert ( max_value_in_yaml == API_MAX_FETCH ), f"API_MAX_FETCH ({API_MAX_FETCH}) does not match the maximum value in max_fetch additionalinfo ({max_value_in_yaml})" def test_first_fetch_timestamp_consistency(yaml_content): """Test that MAX_DAYS_FIRST_FETCH_DETECTIONS matches the max days in first fetch timestamp description""" # Find the first fetch timestamp configuration first_fetch_config = None for config in yaml_content.get("configuration", []): if config.get("name") == "first_fetch": first_fetch_config = config break assert first_fetch_config is not None, "first_fetch configuration not found in YAML" # Extract max days value from the display text display_text = first_fetch_config.get("display", "") import re # Update the pattern to match both "max X days" and "maximum X days" max_days_match = re.search(r"max(?:imum)? (\d+) days", display_text) assert ( max_days_match is not None ), f"Could not find 'max X days' or 'maximum X days' in first_fetch display text: {display_text}" max_days_in_yaml = int(max_days_match.group(1)) # Assert that the code constant matches the YAML value assert max_days_in_yaml == MAX_DAYS_FIRST_FETCH_DETECTIONS, ( f"MAX_DAYS_FIRST_FETCH_DETECTIONS ({MAX_DAYS_FIRST_FETCH_DETECTIONS}) does not match " f"the value in first_fetch display text ({max_days_in_yaml})" ) def test_threats_days_params_consistency(yaml_content): """Test that THREATS_DAYS constants match the creation_days_back argument description""" from Packs.Wiz.Integrations.WizDefend.WizDefend import THREATS_DAYS_MIN, THREATS_DAYS_MAX, THREATS_DAYS_DEFAULT # Find the creation_days_back argument in wiz-get-threats command creation_days_back_arg = None for command in yaml_content.get("script", {}).get("commands", []): if command.get("name") == DemistoCommands.WIZ_DEFEND_GET_THREATS: for arg in command.get("arguments", []): if arg.get("name") == "creation_days_back": creation_days_back_arg = arg break if creation_days_back_arg: break assert creation_days_back_arg is not None, "creation_days_back argument not found in wiz-get-threats command" # Extract min, max, default values from description description = creation_days_back_arg.get("description", "") import re # Update pattern to match "range 1-30" instead of just "(1-30)" range_match = re.search(r"range (\d+)-(\d+)", description) assert range_match is not None, f"Could not find range format (e.g., 'range 1-30') in description: {description}" min_days_in_yaml = int(range_match.group(1)) max_days_in_yaml = int(range_match.group(2)) # Get default value default_value_in_yaml = creation_days_back_arg.get("defaultValue") assert default_value_in_yaml is not None, "No defaultValue for creation_days_back" default_days_in_yaml = int(default_value_in_yaml) # Assert that code constants match YAML values assert min_days_in_yaml == THREATS_DAYS_MIN, ( f"THREATS_DAYS_MIN ({THREATS_DAYS_MIN}) does not match " f"the minimum value in creation_days_back description ({min_days_in_yaml})" ) assert max_days_in_yaml == THREATS_DAYS_MAX, ( f"THREATS_DAYS_MAX ({THREATS_DAYS_MAX}) does not match " f"the maximum value in creation_days_back description ({max_days_in_yaml})" ) assert default_days_in_yaml == THREATS_DAYS_DEFAULT, ( f"THREATS_DAYS_DEFAULT ({THREATS_DAYS_DEFAULT}) does not match " f"the default value in creation_days_back argument ({default_days_in_yaml})" ) def test_fetch_interval_consistency(yaml_content): """Test that FETCH_INTERVAL constants match the creation_minutes_back argument description""" from Packs.Wiz.Integrations.WizDefend.WizDefend import FETCH_INTERVAL_MINIMUM_MIN, FETCH_INTERVAL_MAXIMUM_MIN # Find the creation_minutes_back argument in wiz-get-detections command creation_minutes_back_arg = None for command in yaml_content.get("script", {}).get("commands", []): if command.get("name") == DemistoCommands.WIZ_DEFEND_GET_DETECTIONS: for arg in command.get("arguments", []): if arg.get("name") == "creation_minutes_back": creation_minutes_back_arg = arg break if creation_minutes_back_arg: break assert creation_minutes_back_arg is not None, "creation_minutes_back argument not found in wiz-get-detections command" # Extract min, max values from description description = creation_minutes_back_arg.get("description", "") import re # Update pattern to match "range 10-600" instead of just "(10-600)" range_match = re.search(r"range (\d+)-(\d+)", description) assert range_match is not None, f"Could not find range format (e.g., 'range 10-600') in description: {description}" min_minutes_in_yaml = int(range_match.group(1)) max_minutes_in_yaml = int(range_match.group(2)) # We don't check FETCH_INTERVAL_MINIMUM_MIN directly since the values might be different # for the command argument vs. fetch interval. Instead, we check that the values are reasonable. assert min_minutes_in_yaml >= FETCH_INTERVAL_MINIMUM_MIN, ( f"Minimum minutes in creation_minutes_back description ({min_minutes_in_yaml}) " f"is less than FETCH_INTERVAL_MINIMUM_MIN ({FETCH_INTERVAL_MINIMUM_MIN})" ) assert max_minutes_in_yaml <= FETCH_INTERVAL_MAXIMUM_MIN, ( f"Maximum minutes in creation_minutes_back description ({max_minutes_in_yaml}) " f"is greater than FETCH_INTERVAL_MAXIMUM_MIN ({FETCH_INTERVAL_MAXIMUM_MIN})" ) def test_default_fetch_back_consistency(yaml_content): """Test that DEFAULT_FETCH_BACK matches the first_fetch defaultvalue""" from Packs.Wiz.Integrations.WizDefend.WizDefend import DEFAULT_FETCH_BACK # Find the first_fetch configuration first_fetch_config = None for config in yaml_content.get("configuration", []): if config.get("name") == "first_fetch": first_fetch_config = config break assert first_fetch_config is not None, "first_fetch configuration not found in YAML" # Get default value default_value_in_yaml = first_fetch_config.get("defaultvalue") assert default_value_in_yaml is not None, "No defaultvalue for first_fetch" # Assert that the code constant matches the YAML value assert default_value_in_yaml == DEFAULT_FETCH_BACK, ( f"DEFAULT_FETCH_BACK ({DEFAULT_FETCH_BACK}) does not match " f"the defaultvalue in first_fetch configuration ({default_value_in_yaml})" ) # Also check the additional info - update to match the actual text in additionalinfo additional_info = first_fetch_config.get("additionalinfo", "") assert ( "Maximum allowed is 2 days" in additional_info ), f"first_fetch additionalinfo doesn't mention the maximum days limit: {additional_info}" def test_wiz_input_param_consistency(yaml_content): """ Test that every argument and configuration name in the YAML has a corresponding value in WizInputParam or WizApiInputFields. This ensures that all parameters defined in the YAML interface can be correctly accessed in the code. """ from Packs.Wiz.Integrations.WizDefend.WizDefend import WizInputParam, WizApiInputFields # Collect all parameter names from YAML yaml_param_names = set() # Add argument names from all commands for command in yaml_content.get("script", {}).get("commands", []): for arg in command.get("arguments", []): arg_name = arg.get("name") if arg_name: yaml_param_names.add(arg_name) # Add configuration parameter names for config in yaml_content.get("configuration", []): config_name = config.get("name") if config_name: yaml_param_names.add(config_name) # Filter out standard Cortex XSOAR parameters that don't need corresponding enum values standard_params = {"proxy", "isFetch", "incidentType", "incidentFetchInterval", "max_fetch", "first_fetch", "credentials"} yaml_param_names = yaml_param_names - standard_params # Get all values from WizInputParam and WizApiInputFields input_param_values = set() # Add values from WizInputParam for attr in dir(WizInputParam): if not attr.startswith("_") and not callable(getattr(WizInputParam, attr)): input_param_values.add(getattr(WizInputParam, attr)) # Add values from WizApiInputFields for attr in dir(WizApiInputFields): if not attr.startswith("_") and not callable(getattr(WizApiInputFields, attr)): input_param_values.add(getattr(WizApiInputFields, attr)) # Check each YAML parameter against the enum values directly missing_params = [] for param in yaml_param_names: if param not in input_param_values: missing_params.append(param) assert not missing_params, ( f"The following parameters in YAML do not have a corresponding value in WizInputParam or WizApiInputFields:\n" f"{sorted(missing_params)}\n" f"Please add these values to the appropriate enum class." ) # ===== COMMAND EXAMPLES TESTS ===== def get_command_examples(): """Load and parse command examples from command_examples.txt""" with open("command_examples.txt") as f: examples = f.read().splitlines() return [example.strip() for example in examples if example.strip()] def extract_param_value_pairs(command_line): """Extract parameter-value pairs from a command line""" # Skip the command name (everything before the first space) params_part = command_line.split(" ", 1)[1] if " " in command_line else "" # Define regex pattern to match parameter=value pairs, handling quoted values pattern = r'(\w+)=(?:"([^"]*)"|([\w\-\.,:/]+))' matches = re.findall(pattern, params_part) # Convert matches to dictionary param_values = {} for match in matches: param_name = match[0] # Value is either the quoted value or the non-quoted value param_value = match[1] if match[1] else match[2] param_values[param_name] = param_value return param_values # Extract demisto command name from example command def get_demisto_command_name(example_command): """Convert example command to demisto command name""" # Strip '!' prefix and replace hyphens with underscores command_name = example_command.split(" ")[0][1:].replace("-", "_").upper() return command_name # Group example commands by their demisto command name def group_examples_by_command(): """Group examples by command type""" examples = get_command_examples() grouped = {} for example in examples: if not example.startswith("!"): continue command_name = example.split(" ")[0] demisto_command = get_demisto_command_name(command_name) if demisto_command not in grouped: grouped[demisto_command] = [] grouped[demisto_command].append(example) return grouped # Get all commands and examples for parametrize def get_all_commands_for_test(): """Get all commands and examples for parametrized test""" grouped = group_examples_by_command() test_params = [] for demisto_command, examples in grouped.items(): for example in examples: test_params.append((demisto_command, example)) return test_params def test_commands_in_examples_match_demisto_commands(): """Test that all commands in the examples match those defined in DemistoCommands""" command_examples = get_command_examples() # Get all command values from DemistoCommands class demisto_command_values = { getattr(DemistoCommands, attr) for attr in dir(DemistoCommands) if not attr.startswith("_") and not callable(getattr(DemistoCommands, attr)) } # Convert command values to command names as they would appear in examples (with ! prefix) demisto_command_names = { "!" + cmd.replace("_", "-") for cmd in demisto_command_values if cmd not in ["TEST_MODULE", "FETCH_INCIDENTS"] } # Exclude commands not meant for direct use # Extract all commands from examples example_commands = set() for example in command_examples: if not example.startswith("!"): continue command_name = example.split(" ")[0] example_commands.add(command_name) # Check that all example commands are defined in DemistoCommands undefined_commands = example_commands - demisto_command_names assert not undefined_commands, f"Found commands in examples that are not defined in DemistoCommands: {undefined_commands}" # Check that all demo-accessible DemistoCommands are used in examples unused_commands = demisto_command_names - example_commands if unused_commands: pass # Removed print statement def test_all_command_examples_are_in_wiz_input_params(): """Test that all parameters in command examples are defined in WizInputParam""" command_examples = get_command_examples() # Get all attribute values from WizInputParam class wiz_input_param_values = set() for attr in dir(WizInputParam): if not attr.startswith("_") and not callable(getattr(WizInputParam, attr)): wiz_input_param_values.add(getattr(WizInputParam, attr)) # Extract all parameters from command examples unknown_params = set() for example in command_examples: if not example.startswith("!"): continue command_parts = example.split(" ", 1) if len(command_parts) == 1: # No parameters continue param_values = extract_param_value_pairs(example) for param in param_values: # Check if parameter exists in WizInputParam values if param not in wiz_input_param_values: unknown_params.add(param) # Assert that all parameters are recognized assert not unknown_params, f"Found parameters not defined in WizInputParam values: {unknown_params}" def test_all_wiz_input_params_have_examples(): """Test that all values in WizInputParam have corresponding examples in the command examples""" command_examples = get_command_examples() # Get all attribute values from WizInputParam class wiz_input_param_values = set() for attr in dir(WizInputParam): if not attr.startswith("_") and not callable(getattr(WizInputParam, attr)): wiz_input_param_values.add(getattr(WizInputParam, attr)) # Parameters that can be ignored (don't need examples) ignore_list = [] # Extract all parameters used in the command examples example_params = set() for example in command_examples: if not example.startswith("!"): continue command_parts = example.split(" ", 1) if len(command_parts) == 1: # No parameters continue param_values = extract_param_value_pairs(example) example_params.update(param_values.keys()) # Remove ignored parameters from check params_to_check = wiz_input_param_values - set(ignore_list) # Check which parameters don't have examples missing_examples = set() for param in params_to_check: if param not in example_params: missing_examples.add(param) # Assert that all parameters have examples assert ( not missing_examples ), f"The following WizInputParam values don't have examples in command_examples.txt: {missing_examples}" @pytest.mark.parametrize("demisto_command,example", get_all_commands_for_test()) def test_command_validation(demisto_command, example, mocker): """Test that all commands in the examples file can be validated without errors""" # Extract parameters from example params = extract_param_value_pairs(example) # Mock demisto.args() to return our parameters mocker.patch.object(demisto, "args", return_value=params) # Mock demisto.command() to return our command mocker.patch.object(demisto, "command", return_value=getattr(DemistoCommands, demisto_command)) # Create mock results mock_result = [{"id": "test-id", "severity": "CRITICAL", "type": "THREAT_DETECTION"}] mock_threat_with_notes = [{"id": "test-id", "severity": "CRITICAL", "notes": [{"id": "note1", "text": "test note"}]}] mock_api_response = {"data": {"updateIssue": {"id": "test-id", "status": "RESOLVED"}}} mock_single_issue = [{"id": "test-id", "type": "THREAT_DETECTION"}] # Import the module directly to patch internal functions from Packs.Wiz.Integrations.WizDefend import WizDefend # Patch query functions mocker.patch.object(WizDefend, "query_detections", return_value=mock_result) mocker.patch.object(WizDefend, "query_issues", return_value=mock_result) mocker.patch.object(WizDefend, "query_single_issue", return_value=mock_single_issue) # Patch API functions for threat management commands mocker.patch.object(WizDefend, "get_entries", return_value=mock_api_response) # Patch validation functions mocker.patch.object(WizDefend, "is_valid_issue_id", return_value=(True, "Valid")) mocker.patch.object(WizDefend, "validate_threat_detections_issue", return_value=(True, None)) # Patch get_filtered_threats for commands that need it (like clear_threat_comments) mocker.patch.object(WizDefend, "get_filtered_threats", return_value=mock_threat_with_notes) # Patch set_issue_note for commands that use it mocker.patch.object(WizDefend, "set_issue_note", return_value=mock_api_response) # Create the mock functions with MagicMock to track calls mock_return_results = mocker.MagicMock(return_value=True) mock_return_error = mocker.MagicMock(return_value=True) # Apply the patches mocker.patch.object(WizDefend, "return_results", mock_return_results) mocker.patch.object(WizDefend, "return_error", mock_return_error) # Run the main function main() # Check if return_error was called (indicating validation failure) if mock_return_error.called: error_message = mock_return_error.call_args[0][0] pytest.fail(f"Example failed validation: {example}\nError: {error_message}") # Otherwise, validation passed assert mock_return_results.called, f"Example should have called return_results: {example}" @pytest.mark.parametrize( "input_params,expected_redacted_keys,expected_safe_keys", [ # Test case 1: Basic credentials object ( { "credentials": {"identifier": "test-client-id", "password": "test-secret"}, "auth_endpoint": "https://auth.wiz.io/oauth/token", "api_endpoint": "https://api.wiz.io/graphql", "max_fetch": 100, }, [], # No top-level keys are redacted, only nested keys within credentials dict ["credentials", "auth_endpoint", "api_endpoint", "max_fetch"], ), # Test case 2: Direct sensitive fields ( { "client_id": "direct-client-id", "client_secret": "direct-secret", "api_key": "test-api-key", "auth_endpoint": "https://auth.wiz.io/oauth/token", "platform": "AWS", }, ["client_id", "client_secret", "api_key"], ["auth_endpoint", "platform"], ), # Test case 3: Mixed sensitive and non-sensitive fields ( { "credentials": {"identifier": "test-id", "password": "test-pass"}, "service_account_id": "legacy-id", "service_account_secret": "legacy-secret", "token": "auth-token", "type": "GENERATED THREAT", "severity": "CRITICAL", "first_fetch": "2 days", }, ["service_account_id", "service_account_secret", "token"], # Direct sensitive fields ["credentials", "type", "severity", "first_fetch"], # credentials dict is preserved but filtered ), # Test case 4: Nested dictionaries with mixed content ( { "credentials": {"identifier": "test-id", "password": "secret", "endpoint": "https://test.com"}, "config": {"timeout": 30, "retries": 3, "secret": "nested-secret"}, "platform": "AWS", }, [], # No top-level keys are redacted, only nested keys within dicts ["credentials", "config", "platform"], ), # Test case 5: No sensitive data ( { "auth_endpoint": "https://auth.wiz.io/oauth/token", "api_endpoint": "https://api.wiz.io/graphql", "platform": "AWS", "severity": "HIGH", "max_fetch": 500, }, [], ["auth_endpoint", "api_endpoint", "platform", "severity", "max_fetch"], ), # Test case 6: Empty dictionary ( {}, [], [], ), ], ) def test_get_safe_params_for_logging(mocker, input_params, expected_redacted_keys, expected_safe_keys): """Test get_safe_params_for_logging filters sensitive data correctly""" # Mock demisto.params to return our test parameters mocker.patch.object(demisto, "params", return_value=input_params) # Call the function result = get_safe_params_for_logging() # Verify all expected keys are present expected_all_keys = set(expected_redacted_keys + expected_safe_keys) assert set(result.keys()) == expected_all_keys, f"Expected keys {expected_all_keys}, got {set(result.keys())}" # Verify redacted keys have placeholder value for key in expected_redacted_keys: if key in input_params: # For direct sensitive fields (not nested in dictionaries) assert result[key] == "***REDACTED***", f"Expected {key} to be redacted" # Verify safe keys retain their original values or are properly filtered dictionaries for key in expected_safe_keys: if key in input_params: if isinstance(input_params[key], dict): # For dictionaries, verify structure is maintained but sensitive nested keys are redacted assert isinstance(result[key], dict), f"Expected {key} to remain a dictionary" for nested_key, nested_value in input_params[key].items(): if nested_key in {"identifier", "password", "secret", "client_id", "client_secret"}: assert result[key][nested_key] == "***REDACTED***", f"Expected {key}.{nested_key} to be redacted" else: assert result[key][nested_key] == nested_value, f"Expected {key}.{nested_key} to retain original value" else: # For non-dictionary values, should be identical assert result[key] == input_params[key], f"Expected {key} to retain original value" def test_get_safe_params_for_logging_with_nested_credentials(mocker): """Test get_safe_params_for_logging handles nested credentials correctly""" test_params = { "credentials": { "identifier": "sensitive-id", "password": "sensitive-password", "endpoint": "https://safe-endpoint.com", "timeout": 30, }, "auth_endpoint": "https://auth.wiz.io/oauth/token", "platform": "AWS", } # Mock demisto.params mocker.patch.object(demisto, "params", return_value=test_params) # Call the function result = get_safe_params_for_logging() # Verify structure is maintained assert "credentials" in result assert isinstance(result["credentials"], dict) # Verify sensitive nested fields are redacted assert result["credentials"]["identifier"] == "***REDACTED***" assert result["credentials"]["password"] == "***REDACTED***" # Verify non-sensitive nested fields are preserved assert result["credentials"]["endpoint"] == "https://safe-endpoint.com" assert result["credentials"]["timeout"] == 30 # Verify top-level non-sensitive fields are preserved assert result["auth_endpoint"] == "https://auth.wiz.io/oauth/token" assert result["platform"] == "AWS" def test_get_safe_params_for_logging_empty_params(mocker): """Test get_safe_params_for_logging with empty parameters""" # Mock demisto.params to return empty dict mocker.patch.object(demisto, "params", return_value={}) # Call the function result = get_safe_params_for_logging() # Verify result is empty dict assert result == {} def test_get_safe_params_for_logging_preserves_non_dict_values(mocker): """Test get_safe_params_for_logging preserves non-dictionary values correctly""" test_params = { "max_fetch": 100, "platform": "AWS", "enabled": True, "timeout": 30.5, "tags": ["tag1", "tag2"], "secret": "should-be-redacted", } # Mock demisto.params mocker.patch.object(demisto, "params", return_value=test_params) # Call the function result = get_safe_params_for_logging() # Verify non-sensitive values are preserved with correct types assert result["max_fetch"] == 100 assert result["platform"] == "AWS" assert result["enabled"] is True assert result["timeout"] == 30.5 assert result["tags"] == ["tag1", "tag2"] # Verify sensitive value is redacted assert result["secret"] == "***REDACTED***" # ===== FETCH INCIDENTS FUNCTION TESTS ===== @pytest.mark.parametrize( "scenario_name,end_cursor,after,before,time,reset_reason", [ # Legacy format scenarios (missing pagination fields) ( "legacy_format_missing_all_pagination", None, None, None, "2021-12-31T00:00:00Z", "migrating from legacy format (only 'time' field)", ), # Invalid pagination scenarios (cursor exists but missing required fields) ( "invalid_pagination_missing_after", "cursor123", None, "2022-01-02T00:00:00Z", "2021-12-31T00:00:00Z", "stored_after is None but endCursor exists", ), ( "invalid_pagination_missing_before", "cursor123", "2022-01-01T00:00:00Z", None, "2021-12-31T00:00:00Z", "stored_before is None but endCursor exists", ), ( "invalid_pagination_missing_both", "cursor123", None, None, "2021-12-31T00:00:00Z", "stored_after is None but endCursor exists", # First error detected ), # Invalid timestamp format scenarios ( "invalid_after_timestamp_format", None, "invalid-timestamp", "2022-01-02T00:00:00Z", "2021-12-31T00:00:00Z", "invalid stored_after format: invalid-timestamp", ), ( "invalid_before_timestamp_format", None, "2022-01-01T00:00:00Z", "invalid-timestamp", "2021-12-31T00:00:00Z", "invalid stored_before format: invalid-timestamp", ), # Invalid time ordering scenarios ( "invalid_time_ordering_after_greater_than_before", None, "2022-01-03T10:00:00Z", "2022-01-03T08:00:00Z", "2021-12-31T00:00:00Z", "invalid time ordering: before (2022-01-03T08:00:00Z) < after (2022-01-03T10:00:00Z)", ), # After time too old scenarios - UPDATED to exceed 3 days + 15% buffer ( "after_time_too_old", None, "2021-12-29T01:00:00Z", # Changed: ~5 days old, exceeds 3 days + 15% = ~3.45 days "2022-01-03T10:00:00Z", "2021-12-31T00:00:00Z", "after time too old: 2021-12-29T01:00:00Z", # Updated expected message ), ], ) @freeze_time("2022-01-03T12:00:00Z") def test_reset_function_invocation_and_values(scenario_name, end_cursor, after, before, time, reset_reason): """ Test 1: Verify that reset function is invoked for invalid scenarios and that the resulting values are correct. This test covers all combinations that should trigger the reset logic. """ # Prepare input data that should trigger reset last_run_data = { WizApiResponse.END_CURSOR: end_cursor, WizApiVariables.AFTER: after, WizApiVariables.BEFORE: before, DemistoParams.TIME: time, } with ( patch.object(demisto, "getLastRun", return_value=last_run_data), patch.object(demisto, "params", return_value={"first_fetch": "3 days"}), patch("WizDefend.get_fetch_timestamp", return_value=time), patch.object(FetchIncident, "reset_params", wraps=None) as mock_reset, ): # Initialize FetchIncident - this should trigger reset fetch_manager = FetchIncident() # STEP 1: Verify that reset_params was called mock_reset.assert_called_once(), f"{scenario_name}: reset_params should have been called" # STEP 2: Verify the reset reason contains expected text actual_reset_reason = mock_reset.call_args[0][0] # First argument to reset_params assert ( reset_reason in actual_reset_reason ), f"{scenario_name}: Expected reset reason to contain '{reset_reason}', got '{actual_reset_reason}'" # STEP 3: Manually apply reset logic to verify final state # (Since we mocked reset_params, we need to apply the reset manually to test the final values) fetch_manager.end_cursor = None fetch_manager.stored_after = time # reset_params sets this to last_run_time fetch_manager.stored_before = "2022-01-03T12:00:00Z" # reset_params sets this to current time # STEP 4: Verify the API parameters after reset assert ( not fetch_manager.should_continue_previous_run() ), f"{scenario_name}: should_continue_previous_run should be False after reset" assert ( fetch_manager.get_api_after_parameter() == time ), f"{scenario_name}: get_api_after_parameter should return last_run_time ({time}) after reset" assert ( fetch_manager.get_api_before_parameter() == "2022-01-03T11:50:00Z" ), f"{scenario_name}: get_api_before_parameter should return current time - fetch_interval after reset" assert ( fetch_manager.get_api_cursor_parameter() is None ), f"{scenario_name}: get_api_cursor_parameter should be None after reset" @pytest.mark.parametrize( "scenario_name,end_cursor,after,before,time,expected_continue,expected_after,expected_before,expected_cursor", [ # Valid pagination scenarios (all required fields present and valid) ( "valid_pagination_all_fields", "cursor123", "2022-01-03T08:00:00Z", "2022-01-03T10:00:00Z", "2021-12-31T00:00:00Z", True, "2022-01-03T08:00:00Z", "2022-01-03T10:00:00Z", "cursor123", ), ( "valid_pagination_same_after_before", "cursor456", "2022-01-03T09:00:00Z", "2022-01-03T09:00:00Z", "2021-12-31T00:00:00Z", True, "2022-01-03T09:00:00Z", "2022-01-03T09:00:00Z", "cursor456", ), ( "valid_pagination_recent_times", "cursor789", "2022-01-03T11:00:00Z", "2022-01-03T11:30:00Z", "2022-01-02T00:00:00Z", True, "2022-01-03T11:00:00Z", "2022-01-03T11:30:00Z", "cursor789", ), # Valid fresh fetch scenarios (no cursor, valid or missing fields) # expected_before is now current_time - fetch_interval (10m) = 11:50:00Z ( "valid_fresh_fetch_with_after_before", None, "2022-01-03T08:00:00Z", "2022-01-03T10:00:00Z", "2021-12-31T00:00:00Z", False, "2022-01-03T08:00:00Z", "2022-01-03T11:50:00Z", None, ), ( "valid_fresh_fetch_with_after_only", None, "2022-01-03T08:00:00Z", None, "2021-12-31T00:00:00Z", False, "2022-01-03T08:00:00Z", "2022-01-03T11:50:00Z", None, ), ( "valid_fresh_fetch_recent_time", None, "2022-01-03T10:00:00Z", "2022-01-03T11:00:00Z", "2022-01-03T05:00:00Z", False, "2022-01-03T10:00:00Z", "2022-01-03T11:50:00Z", None, ), # Edge cases that should NOT trigger reset ( "edge_case_exactly_10_hours_old", # Exactly at the limit, should be valid None, "2022-01-03T02:00:00Z", "2022-01-03T10:00:00Z", "2021-12-31T00:00:00Z", False, "2022-01-03T02:00:00Z", "2022-01-03T11:50:00Z", None, ), ( "edge_case_pagination_exactly_at_limit", "cursor_edge", "2022-01-03T02:00:00Z", "2022-01-03T10:00:00Z", "2021-12-31T00:00:00Z", True, "2022-01-03T02:00:00Z", "2022-01-03T10:00:00Z", "cursor_edge", ), ], ) @freeze_time("2022-01-03T12:00:00Z") def test_no_reset_scenarios_parameter_validation( scenario_name, end_cursor, after, before, time, expected_continue, expected_after, expected_before, expected_cursor ): """ Verify that NO reset occurs for valid scenarios and that the API parameters are returned correctly without modification. This test covers all combinations that should NOT trigger reset logic. """ # Prepare valid input data that should NOT trigger reset last_run_data = { WizApiResponse.END_CURSOR: end_cursor, WizApiVariables.AFTER: after, WizApiVariables.BEFORE: before, DemistoParams.TIME: time, } with ( patch.object(demisto, "getLastRun", return_value=last_run_data), patch.object(demisto, "params", return_value={"first_fetch": "3 days"}), patch("WizDefend.get_fetch_timestamp", return_value=time), patch.object(FetchIncident, "reset_params") as mock_reset, ): # Initialize FetchIncident - this should NOT trigger reset fetch_manager = FetchIncident() # STEP 1: Verify that reset_params was NOT called mock_reset.assert_not_called(), f"{scenario_name}: reset_params should NOT have been called" # STEP 2: Verify the pagination decision actual_continue = fetch_manager.should_continue_previous_run() assert ( actual_continue == expected_continue ), f"{scenario_name}: should_continue_previous_run expected {expected_continue}, got {actual_continue}" # STEP 3: Verify the API parameters are returned correctly actual_after = fetch_manager.get_api_after_parameter() assert ( actual_after == expected_after ), f"{scenario_name}: get_api_after_parameter expected {expected_after}, got {actual_after}" actual_before = fetch_manager.get_api_before_parameter() assert ( actual_before == expected_before ), f"{scenario_name}: get_api_before_parameter expected {expected_before}, got {actual_before}" actual_cursor = fetch_manager.get_api_cursor_parameter() assert ( actual_cursor == expected_cursor ), f"{scenario_name}: get_api_cursor_parameter expected {expected_cursor}, got {actual_cursor}" # STEP 4: Verify internal state matches input (no reset occurred) if expected_continue: # Pagination case - internal state should match input assert fetch_manager.end_cursor == end_cursor, f"{scenario_name}: end_cursor should be preserved" assert fetch_manager.stored_after == after, f"{scenario_name}: stored_after should be preserved" assert fetch_manager.stored_before == before, f"{scenario_name}: stored_before should be preserved" else: # Fresh fetch case - cursor should be None, but after/before might be preserved assert fetch_manager.end_cursor is None, f"{scenario_name}: end_cursor should be None for fresh fetch" # stored_after and stored_before might be the original values or None depending on inpu # Comprehensive boundary test @freeze_time("2022-01-03T12:00:00Z") def test_boundary_conditions_no_reset(): """Test specific boundary conditions that should not trigger reset""" # Test exactly at the time limit (should be valid) exactly_10_hours_ago = "2022-01-03T02:00:00Z" # Exactly 10 hours before frozen time boundary_data = { WizApiResponse.END_CURSOR: None, WizApiVariables.AFTER: exactly_10_hours_ago, WizApiVariables.BEFORE: "2022-01-03T10:00:00Z", DemistoParams.TIME: "2021-12-31T00:00:00Z", } with ( patch.object(demisto, "getLastRun", return_value=boundary_data), patch.object(demisto, "params", return_value={"first_fetch": "3 days"}), patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_fetch_timestamp", return_value="2021-12-31T00:00:00Z"), patch.object(FetchIncident, "reset_params") as mock_reset, ): fetch_manager = FetchIncident() # Should NOT trigger reset mock_reset.assert_not_called() # Should return the original values assert fetch_manager.get_api_after_parameter() == exactly_10_hours_ago assert fetch_manager.get_api_before_parameter() == "2022-01-03T11:50:00Z" # Current time - fetch_interval for fresh fetch assert not fetch_manager.should_continue_previous_run() # No cursor # Additional test to verify the actual reset_params function behavior @freeze_time("2022-01-03T12:00:00Z") def test_reset_params_function_directly(): """Test the reset_params function directly to ensure it sets values correctly""" # Create a FetchIncident with some initial state initial_data = { WizApiResponse.END_CURSOR: "some_cursor", WizApiVariables.AFTER: "2022-01-01T00:00:00Z", WizApiVariables.BEFORE: "2022-01-02T00:00:00Z", DemistoParams.TIME: "2021-12-31T00:00:00Z", } with ( patch.object(demisto, "getLastRun", return_value=initial_data), patch.object(demisto, "params", return_value={"first_fetch": "3 days"}), patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_fetch_timestamp", return_value="2021-12-31T00:00:00Z"), patch.object(FetchIncident, "_validate_and_reset_params"), # Skip validation to test reset directly ): fetch_manager = FetchIncident() # Manually set some state to verify reset clears it fetch_manager.end_cursor = "test_cursor" fetch_manager.stored_after = "2022-01-01T00:00:00Z" fetch_manager.stored_before = "2022-01-02T00:00:00Z" fetch_manager.last_run_time = "2021-12-31T00:00:00Z" # Call reset_params directly fetch_manager.reset_params("Test reset") # Verify the reset values assert fetch_manager.end_cursor is None, "end_cursor should be None after reset" assert fetch_manager.stored_after == "2021-12-31T00:00:00Z", "stored_after should be last_run_time" assert fetch_manager.stored_before == "2022-01-03T12:00:00Z", "stored_before should be current time" @freeze_time("2022-01-03T12:00:00Z") @patch.object(demisto, "setLastRun") def test_fetch_incident_clear_pagination_context(mock_set_last_run): """Test FetchIncident._clear_pagination_context method""" last_run_data = { WizApiResponse.END_CURSOR: None, # Legacy format WizApiVariables.AFTER: "2022-01-01T00:00:00Z", WizApiVariables.BEFORE: "2022-01-02T00:00:00Z", DemistoParams.TIME: "2021-12-31T00:00:00Z", } with patch.object(demisto, "getLastRun", return_value=last_run_data): original_cursor = WizDefend.API_END_CURSOR try: WizDefend.API_END_CURSOR = None fetch_manager = FetchIncident() fetch_manager._clear_pagination_context() mock_set_last_run.assert_called_once() call_args = mock_set_last_run.call_args[0][0] # Test the structure and relationships assert call_args[DemistoParams.TIME] == "2022-01-03T12:00:00Z" # api_start_run_time assert call_args[WizApiResponse.END_CURSOR] is None # Key test: AFTER uses stored_before, BEFORE uses api_start_run_time assert ( call_args[WizApiVariables.AFTER] == "2022-01-03T12:00:00Z" ) # stored_before (which becomes current time after reset) assert call_args[WizApiVariables.BEFORE] == "2022-01-03T12:00:00Z" # api_start_run_time finally: WizDefend.API_END_CURSOR = original_cursor # Alternative: Test with data that doesn't trigger reset @freeze_time("2022-01-03T12:00:00Z") @patch.object(demisto, "setLastRun") def test_fetch_incident_save_pagination_context_no_reset(mock_set_last_run): """Test _save_pagination_context without triggering reset (cleaner test)""" # Use valid pagination data that won't trigger legacy format reset last_run_data = { WizApiResponse.END_CURSOR: "existing_cursor", # Valid pagination state WizApiVariables.AFTER: "2022-01-03T08:00:00Z", WizApiVariables.BEFORE: "2022-01-03T10:00:00Z", DemistoParams.TIME: "2021-12-31T00:00:00Z", } with patch.object(demisto, "getLastRun", return_value=last_run_data): original_cursor = WizDefend.API_END_CURSOR try: WizDefend.API_END_CURSOR = "new_cursor_value" fetch_manager = FetchIncident() fetch_manager._save_pagination_context() mock_set_last_run.assert_called_once() call_args = mock_set_last_run.call_args[0][0] # With no reset, values should be exactly what we provided assert call_args[DemistoParams.TIME] == "2022-01-03T12:00:00Z" assert call_args[WizApiResponse.END_CURSOR] == "new_cursor_value" assert call_args[WizApiVariables.AFTER] == "2022-01-03T08:00:00Z" # stored_after preserved assert call_args[WizApiVariables.BEFORE] == "2022-01-03T10:00:00Z" # stored_before preserved finally: WizDefend.API_END_CURSOR = original_cursor @freeze_time("2022-01-03T12:00:00Z") @patch.object(demisto, "setLastRun") def test_fetch_incident_clear_pagination_context_no_reset(mock_set_last_run): """Test _clear_pagination_context without triggering reset (cleaner test)""" # Use valid pagination data last_run_data = { WizApiResponse.END_CURSOR: "existing_cursor", # Valid pagination state WizApiVariables.AFTER: "2022-01-03T08:00:00Z", WizApiVariables.BEFORE: "2022-01-03T10:00:00Z", DemistoParams.TIME: "2021-12-31T00:00:00Z", } with patch.object(demisto, "getLastRun", return_value=last_run_data): original_cursor = WizDefend.API_END_CURSOR try: WizDefend.API_END_CURSOR = None fetch_manager = FetchIncident() fetch_manager._clear_pagination_context() mock_set_last_run.assert_called_once() call_args = mock_set_last_run.call_args[0][0] # Test the key behavior: AFTER uses stored_before, BEFORE uses current time assert call_args[DemistoParams.TIME] == "2022-01-03T12:00:00Z" assert call_args[WizApiResponse.END_CURSOR] is None assert call_args[WizApiVariables.AFTER] == "2022-01-03T10:00:00Z" # stored_before preserved assert call_args[WizApiVariables.BEFORE] == "2022-01-03T12:00:00Z" # api_start_run_time finally: WizDefend.API_END_CURSOR = original_cursor @freeze_time("2022-01-03T12:00:00Z") @patch.object(demisto, "setLastRun") def test_fetch_incident_before_locked_across_multipage_fetches(mock_set_last_run): """Regression: `before` (the API window's upper bound) must remain byte-identical across page1 -> page2 -> clear. The original bug was `_save_pagination_context` saving `BEFORE: self.stored_after` instead of `self.stored_before`, which let the upper bound drift mid-pagination — silently widening the fetch window per page. Single-call tests would still pass if a future refactor recomputed `stored_before = utcnow()` on each page; this multi-call test locks the contract that `before` is set ONCE (when pagination starts) and rides through every save until the final clear, where it gets promoted to the new `after`.""" locked_before = "2022-01-03T11:00:00Z" locked_after = "2022-01-03T08:00:00Z" # Page 1 save: pagination just started, locked_before is in last_run. page1_state = { WizApiResponse.END_CURSOR: "page1_cursor", WizApiVariables.AFTER: locked_after, WizApiVariables.BEFORE: locked_before, DemistoParams.TIME: locked_after, } original_cursor = WizDefend.API_END_CURSOR try: # --- Page 1 save (still paginating to page 2) --- WizDefend.API_END_CURSOR = "page2_cursor" with patch.object(demisto, "getLastRun", return_value=page1_state): FetchIncident()._save_pagination_context() page1_saved = mock_set_last_run.call_args_list[0][0][0] assert page1_saved[WizApiVariables.BEFORE] == locked_before, "page1 save must preserve locked before" assert page1_saved[WizApiVariables.AFTER] == locked_after, "page1 save must preserve after" # --- Page 2 save (still paginating to page 3) — getLastRun returns page1's saved state --- WizDefend.API_END_CURSOR = "page3_cursor" with patch.object(demisto, "getLastRun", return_value=page1_saved): FetchIncident()._save_pagination_context() page2_saved = mock_set_last_run.call_args_list[1][0][0] assert page2_saved[WizApiVariables.BEFORE] == locked_before, "page2 save must STILL preserve locked before" assert page2_saved[WizApiVariables.AFTER] == locked_after, "page2 save must preserve after" # --- Clear (pagination complete) — getLastRun returns page2's state --- WizDefend.API_END_CURSOR = None with patch.object(demisto, "getLastRun", return_value=page2_saved): FetchIncident()._clear_pagination_context() cleared = mock_set_last_run.call_args_list[2][0][0] # Window-handoff contract: clear promotes locked_before -> AFTER (next window's lower bound). # If `before` had drifted during pagination, this hand-off would create overlap or gaps. assert ( cleared[WizApiVariables.AFTER] == locked_before ), "clear must promote the LOCKED before to the new after — proves before never drifted" assert cleared[WizApiVariables.BEFORE] == "2022-01-03T12:00:00Z", "clear sets new before to api_start_run_time" assert cleared[WizApiResponse.END_CURSOR] is None assert mock_set_last_run.call_count == 3 finally: WizDefend.API_END_CURSOR = original_cursor @pytest.mark.parametrize( "api_end_cursor,expect_save_called,expect_clear_called", [ ("cursor_value", True, False), (None, False, True), ("", False, True), ], ) @patch("WizDefend.datetime") @patch.object(demisto, "setLastRun") def test_fetch_incident_handle_post_incident_creation( mock_set_last_run, mock_datetime, api_end_cursor, expect_save_called, expect_clear_called ): """Test FetchIncident.handle_post_incident_creation method""" from datetime import datetime import WizDefend # Mock datetime.now() mock_now = datetime(2022, 1, 3, 12, 0, 0) mock_datetime.now.return_value = mock_now mock_datetime.strftime = datetime.strftime # Set global API_END_CURSOR WizDefend.API_END_CURSOR = api_end_cursor # Mock getLastRun data last_run_data = {"endCursor": None, "after": None, "before": None, "time": None} with patch.object(demisto, "getLastRun", return_value=last_run_data): fetch_manager = FetchIncident() fetch_manager.handle_post_incident_creation() # Verify setLastRun was called assert mock_set_last_run.called == (expect_save_called or expect_clear_called) @patch("WizDefend.datetime") def test_fetch_incident_log_current_state(mock_datetime): """Test FetchIncident.log_current_state method""" from datetime import datetime # Mock datetime.now() mock_now = datetime(2022, 1, 3, 12, 0, 0) mock_datetime.now.return_value = mock_now mock_datetime.strftime = datetime.strftime # Test with pagination active last_run_data = { "endCursor": "test_cursor", "after": "2022-01-01T00:00:00Z", "before": "2022-01-02T00:00:00Z", "time": "2021-12-31T00:00:00Z", } with patch.object(demisto, "getLastRun", return_value=last_run_data): fetch_manager = FetchIncident() # Should not raise any exceptions fetch_manager.log_current_state() # Test without pagination last_run_data_no_cursor = {"endCursor": None, "after": None, "before": None, "time": "2021-12-31T00:00:00Z"} with patch.object(demisto, "getLastRun", return_value=last_run_data_no_cursor): fetch_manager = FetchIncident() # Should not raise any exceptions fetch_manager.log_current_state() @pytest.mark.parametrize( "api_end_cursor_value,should_update", [ ("new_cursor_123", True), (None, False), ("", False), ], ) @freeze_time("2022-01-03T12:00:00Z") @patch.object(demisto, "setLastRun") def test_api_cursor_always_updated_when_needed(mock_set_last_run, api_end_cursor_value, should_update): """Test that API_END_CURSOR is always updated when needed""" # FIXED: Use the same pattern as test_fetch_incident_save_pagination_context # The key insight is that we need to test the DIRECT method calls, not the high-level flow last_run_data = { WizApiResponse.END_CURSOR: None, # Legacy format scenario WizApiVariables.AFTER: "2022-01-01T00:00:00Z", WizApiVariables.BEFORE: "2022-01-02T00:00:00Z", DemistoParams.TIME: "2021-12-31T00:00:00Z", } with patch.object(demisto, "getLastRun", return_value=last_run_data): # Set the global variable directly on the module original_cursor = WizDefend.API_END_CURSOR try: WizDefend.API_END_CURSOR = api_end_cursor_value fetch_manager = FetchIncident() # CRITICAL FIX: Call the method directly based on should_update # This matches the test pattern from test_fetch_incident_save_pagination_context if should_update: # Test _save_pagination_context directly (like the working test does) fetch_manager._save_pagination_context() else: # Test _clear_pagination_context directly fetch_manager._clear_pagination_context() mock_set_last_run.assert_called_once() call_args = mock_set_last_run.call_args[0][0] if should_update: # Should save pagination context with new cursor assert call_args[WizApiResponse.END_CURSOR] == api_end_cursor_value else: # Should clear pagination context assert call_args[WizApiResponse.END_CURSOR] is None finally: # Restore original value WizDefend.API_END_CURSOR = original_cursor # ===== FETCHINCIDENT CLASS TESTS ===== @pytest.mark.parametrize( "after,before,end_cursor,need_reset", [ # Valid scenarios - no reset needed (using times within 10-hour window of frozen time 2022-01-03T12:00:00Z) ("2022-01-03T08:00:00Z", "2022-01-03T10:00:00Z", None, False), # Fresh fetch: after < before ("2022-01-03T08:00:00Z", "2022-01-03T08:00:00Z", None, False), # Fresh fetch: after = before ("2022-01-03T08:00:00Z", "2022-01-03T10:00:00Z", "cursor123", False), # Pagination: after < before ("2022-01-03T08:00:00Z", "2022-01-03T08:00:00Z", "cursor123", False), # Pagination: after = before (None, "2022-01-03T10:00:00Z", None, False), # Fresh fetch: no after (None, None, None, True), # Fresh fetch: no after, no before (None, None, "cursor123", True), # Pagination: no after, no before - INVALID # Invalid scenarios - reset needed ("2022-01-03T10:00:00Z", "2022-01-03T08:00:00Z", None, True), # Fresh fetch: after > before ("2022-01-03T10:00:00Z", "2022-01-03T08:00:00Z", "cursor123", True), # Pagination: after > before ("2022-01-03T08:00:00Z", None, "cursor123", True), # Pagination: missing before (None, "2022-01-03T10:00:00Z", "cursor123", True), # Pagination: missing after ("invalid-time", "2022-01-03T10:00:00Z", None, True), # Invalid after format ("2022-01-03T08:00:00Z", "invalid-time", None, True), # Invalid before format ("invalid-time", "invalid-time", "cursor123", True), # Both invalid ("2022-01-03T01:00:00Z", "2022-01-03T10:00:00Z", None, False), ], ) @freeze_time("2022-01-03T12:00:00Z") def test_fetch_incident_reset_logic(after, before, end_cursor, need_reset): """ Test FetchIncident parameter validation and reset logic Args: after: stored_after timestamp before: stored_before timestamp end_cursor: pagination cursor need_reset: whether reset_params should be called """ # Prepare mock last run data last_run_data = { "endCursor": end_cursor, "after": after, "before": before, "time": "2022-01-03T06:00:00Z", # Fixed last_run_time (within 10-hour window) } with ( patch.object(demisto, "getLastRun", return_value=last_run_data), patch.object(FetchIncident, "reset_params") as mock_reset, ): # Initialize FetchIncident (this triggers validation) FetchIncident() if need_reset: # Verify reset was called mock_reset.assert_called_once() else: # Verify reset was NOT called mock_reset.assert_not_called() @pytest.mark.parametrize( "after,before,end_cursor,need_reset", [ # Valid scenarios - no reset needed (using times within 10-hour window of frozen time 2022-01-03T12:00:00Z) ("2022-01-03T08:00:00Z", "2022-01-03T10:00:00Z", None, False), # Fresh fetch: after < before ("2022-01-03T08:00:00Z", "2022-01-03T08:00:00Z", None, False), # Fresh fetch: after = before ("2022-01-03T08:00:00Z", "2022-01-03T10:00:00Z", "cursor123", False), # Pagination: after < before ("2022-01-03T08:00:00Z", "2022-01-03T08:00:00Z", "cursor123", False), # Pagination: after = before (None, "2022-01-03T10:00:00Z", None, False), # Fresh fetch: no after (None, None, None, False), # Fresh fetch: no after, no before (None, None, "cursor123", True), # Pagination: no after, no before - INVALID # Invalid scenarios - reset needed ("2022-01-03T10:00:00Z", "2022-01-03T08:00:00Z", None, True), # Fresh fetch: after > before ("2022-01-03T10:00:00Z", "2022-01-03T08:00:00Z", "cursor123", True), # Pagination: after > before ("2022-01-03T08:00:00Z", None, "cursor123", True), # Pagination: missing before (NOW a reset condition) (None, "2022-01-03T10:00:00Z", "cursor123", True), # Pagination: missing after (RESET condition) ("invalid-time", "2022-01-03T10:00:00Z", None, True), # Invalid after format ("2022-01-03T08:00:00Z", "invalid-time", None, True), # Invalid before format ("invalid-time", "invalid-time", "cursor123", True), # Both invalid ("2022-01-03T01:00:00Z", "2022-01-03T10:00:00Z", None, False), ], ) @freeze_time("2022-01-03T12:00:00Z") def test_fetch_incident_api_variables(after, before, end_cursor, need_reset): """ Test that fetch_incidents calls get_filtered_detections with correct API variables containing the right after/before values in the GraphQL structure Args: after: stored_after timestamp before: stored_before timestamp end_cursor: pagination cursor need_reset: whether reset_params should be called """ # Prepare mock last run data last_run_data = { "endCursor": end_cursor, "after": after, "before": before, "time": "2022-01-03T06:00:00Z", # Fixed last_run_time (within 10-hour window) } with ( patch.object(demisto, "getLastRun", return_value=last_run_data), patch.object(demisto, "incidents"), patch.object(demisto, "setLastRun"), patch("Packs.Wiz.Integrations.WizDefend.WizDefend.extract_params_from_integration_settings") as mock_extract_params, patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_detections") as mock_get_filtered, patch( "Packs.Wiz.Integrations.WizDefend.WizDefend.get_entries", return_value=([], {"hasNextPage": False, "endCursor": None}) ), ): # Mock the integration settings mock_extract_params.return_value = { "type": "GENERATED THREAT", "platform": "AWS", "severity": "CRITICAL", "origin": "WIZ_SENSOR", "cloud_account_or_cloud_organization": "test-subscription", "max_fetch": "100", } # Mock successful API response mock_get_filtered.return_value = [] # Call fetch_incidents (this triggers FetchIncident validation and API call) fetch_incidents() # Verify get_filtered_detections was called mock_get_filtered.assert_called_once() # Get the actual call arguments call_kwargs = mock_get_filtered.call_args.kwargs # Extract the time parameters that were passed to the API actual_after_time = call_kwargs.get("after_time") actual_before_time = call_kwargs.get("before_time") actual_end_cursor = call_kwargs.get("end_cursor") if need_reset: # After reset, values should be from reset_params logic expected_after_from_reset = "2022-01-03T06:00:00Z" # last_run_time expected_before_from_reset = "2022-01-03T11:50:00Z" # current time - fetch_interval (10m) expected_cursor_from_reset = None # cursor cleared assert ( actual_after_time == expected_after_from_reset ), f"After reset, API after_time should be {expected_after_from_reset}, got {actual_after_time}" assert ( actual_before_time == expected_before_from_reset ), f"After reset, API before_time should be {expected_before_from_reset}, got {actual_before_time}" assert ( actual_end_cursor == expected_cursor_from_reset ), f"After reset, API end_cursor should be {expected_cursor_from_reset}, got {actual_end_cursor}" else: # No reset - values should match the expected behavior of FetchIncident getters # Expected after_time logic: # - If continuing pagination (end_cursor exists): use stored_after # - If fresh fetch: use stored_after if available, else last_run_time if end_cursor: # Pagination expected_after = after # Should use stored_after directly else: # Fresh fetch expected_after = after if after is not None else "2022-01-03T06:00:00Z" # fallback to last_run_time # Expected before_time logic: # - If continuing pagination (end_cursor exists): use stored_before # - If fresh fetch: use current time - fetch_interval (10m lag) if end_cursor: # Pagination expected_before = before # Should use stored_before directly else: # Fresh fetch expected_before = "2022-01-03T11:50:00Z" # current time - 10m fetch interval # Expected end_cursor: should match input expected_cursor = end_cursor assert actual_after_time == expected_after, f"API after_time should be {expected_after}, got {actual_after_time}" assert actual_before_time == expected_before, f"API before_time should be {expected_before}, got {actual_before_time}" assert actual_end_cursor == expected_cursor, f"API end_cursor should be {expected_cursor}, got {actual_end_cursor}" @freeze_time("2025-06-22T15:10:08Z") @pytest.mark.parametrize( "last_run_time,fetch_interval_param,validation_success,validation_minutes,exception_during_calc,expected_stored_after,should_log_error,error_message_contains", [ ( "2025-06-20T10:00:00Z", # last_run_time exists but will be adjusted "15", # fetch_interval_param True, # validation_success 15, # validation_minutes False, # exception_during_calc "2025-06-20T15:10:08Z", # expected_stored_after (adjusted by get_last_run_time to 2 days ago) False, # should_log_error None, # error_message_contains ), ( None, # last_run_time is None "15", # fetch_interval_param True, # validation_success 15, # validation_minutes False, # exception_during_calc "2025-06-22T14:55:08Z", # expected_stored_after (current time - 15 min) False, # should_log_error None, # error_message_contains ), ( None, # last_run_time is None "30", # fetch_interval_param True, # validation_success 30, # validation_minutes False, # exception_during_calc "2025-06-22T14:40:08Z", # expected_stored_after (current time - 30 min) False, # should_log_error None, # error_message_contains ), ( None, # last_run_time is None "5", # fetch_interval_param (below minimum) False, # validation_success (validation fails) None, # validation_minutes (not set when validation fails) False, # exception_during_calc "2025-06-22T15:00:08Z", # expected_stored_after (current time - 10 min default) True, # should_log_error "Invalid fetch interval, using default", # error_message_contains ), ( None, # last_run_time is None None, # fetch_interval_param is missing (None) True, # validation_success 10, # validation_minutes (default FETCH_INTERVAL_MINIMUM_MIN) False, # exception_during_calc "2025-06-22T15:00:08Z", # expected_stored_after (current time - 10 min) False, # should_log_error None, # error_message_contains ), ( None, # last_run_time is None "20", # fetch_interval_param True, # validation_success 20, # validation_minutes True, # exception_during_calc "2025-06-22T15:10:08Z", # expected_stored_after (fallback to api_start_run_time) True, # should_log_error "Error calculating safe_after_str with fetch interval", # error_message_contains ), ( None, # last_run_time is None "", # fetch_interval_param is empty string False, # validation_success (empty string fails validation) None, # validation_minutes False, # exception_during_calc "2025-06-22T15:00:08Z", # expected_stored_after (current time - 10 min default) True, # should_log_error "Invalid fetch interval, using default", # error_message_contains ), ], ids=[ "existing_last_run_time", "none_last_run_valid_15min", "none_last_run_valid_30min", "none_last_run_invalid_fetch_interval", "none_last_run_missing_fetch_interval", "none_last_run_exception_during_calc", "none_last_run_empty_fetch_interval", ], ) def test_reset_params_comprehensive( last_run_time, fetch_interval_param, validation_success, validation_minutes, exception_during_calc, expected_stored_after, should_log_error, error_message_contains, ): """Comprehensive test for reset_params method covering all scenarios""" api_start_run_time = "2025-06-22T15:10:08Z" with ( patch.object(demisto, "getLastRun") as mock_get_last_run, patch.object(demisto, "params") as mock_params, patch.object(demisto, "error") as mock_error, patch.object(demisto, "info") as mock_info, patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_fetch_timestamp") as mock_get_fetch_timestamp, patch("Packs.Wiz.Integrations.WizDefend.WizDefend.validate_fetch_interval") as mock_validate_fetch_interval, ): # Setup last run data last_run_data = {DemistoParams.TIME: last_run_time, "endCursor": None, "after": None, "before": None} mock_get_last_run.return_value = last_run_data # Setup params - handle None case for missing parameter params_data = {} if fetch_interval_param is not None: params_data[DemistoParams.INCIDENT_FETCH_INTERVAL] = fetch_interval_param mock_params.return_value = params_data # Setup get_fetch_timestamp mock - simulate the 2-day limit adjustment if last_run_time is not None: # Simulate the 2-day limit adjustment that happens in get_last_run_time adjusted_time = "2025-06-20T15:10:08Z" # 2 days ago from frozen time mock_get_fetch_timestamp.return_value = adjusted_time else: mock_get_fetch_timestamp.return_value = None # Setup validation mock if exception_during_calc: mock_validate_fetch_interval.side_effect = Exception("Test exception") else: if validation_success: validation_response = ValidationResponse.create_success() validation_response.minutes_value = validation_minutes else: validation_response = ValidationResponse.create_error("Invalid fetch interval") mock_validate_fetch_interval.return_value = validation_response # Create FetchIncident instance fetch_manager = FetchIncident() # Call reset_params fetch_manager.reset_params("test migration reason") # Assertions assert fetch_manager.stored_after == expected_stored_after assert fetch_manager.stored_before == api_start_run_time assert fetch_manager.end_cursor is None # Check error logging if should_log_error: mock_error.assert_called() if error_message_contains: error_call_args = mock_error.call_args[0][0] assert error_message_contains in error_call_args # Check validation calls for None last_run_time cases # validate_fetch_interval is called in __init__ (_get_fetch_interval_minutes) and in reset_params if last_run_time is None and not exception_during_calc: expected_param = fetch_interval_param if fetch_interval_param is not None else str(FETCH_INTERVAL_MINIMUM_MIN) mock_validate_fetch_interval.assert_called_with(expected_param) # Verify info logging for reset and completion assert mock_info.call_count >= 2 # At least "Resetting fetch parameters" and "Reset fetch incidents parameter complete" # Check specific info messages info_calls = [call[0][0] for call in mock_info.call_args_list] assert any("Resetting fetch parameters: test migration reason" in msg for msg in info_calls) assert any(f"after: {expected_stored_after}, before: {api_start_run_time}" in msg for msg in info_calls) @freeze_time("2025-06-22T15:10:08Z") @pytest.mark.parametrize( "first_fetch_param,test_age_minutes,expected_too_old", [ # Test 6-hour cases - with mocked consistent dateparser behavior ("6 hours", 205, False), # Within limit ("6 hours", 413, False), # At calculated limit (360 * 1.15 = 414, rounded down) ("6 hours", 414, True), # 1 minute over - should be True ("6 hours", 420, True), # Further over - should be True # Test 10-hour cases ("10 hours", 481, False), # Within limit ("10 hours", 690, False), # At calculated limit (600 * 1.15 = 690) ("10 hours", 691, True), # 1 minute over - should be True ("10 hours", 700, True), # Further over - should be True # Test 12-hour cases ("12 hours", 620, False), # Within limit ("12 hours", 827, False), # At calculated limit (720 * 1.15 = 828, but implementation uses 827) ("12 hours", 828, True), # 1 minute over - should be True ("12 hours", 850, True), # Further over - should be True # Test very old timestamps that should definitely fail ("6 hours", 1000, True), # Way over any reasonable limit ("12 hours", 2000, True), # Way over any reasonable limit ], ids=[ "6h_within_limit", "6h_at_calculated_limit", "6h_1min_over_too_old", "6h_further_over_too_old", "10h_within_limit", "10h_at_calculated_limit", "10h_1min_over_too_old", "10h_further_over_too_old", "12h_within_limit", "12h_at_calculated_limit", "12h_1min_over_too_old", "12h_further_over_too_old", "6h_way_over_limit", "12h_way_over_limit", ], ) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.MAX_FETCH_BUFFER", 15) def test_max_fetch_buffer_simple(first_fetch_param, test_age_minutes, expected_too_old): """ Test with mocked dateparser to ensure consistent behavior across environments. This replaces the original failing test and handles the dateparser inconsistency between PyCharm and pre-commit environments. """ # Create a test timestamp that is exactly test_age_minutes old current_time = datetime.strptime("2025-06-22T15:10:08Z", DEMISTO_OCCURRED_FORMAT) test_datetime = current_time - timedelta(minutes=test_age_minutes) test_time_str = test_datetime.strftime(DEMISTO_OCCURRED_FORMAT) # Setup the FetchIncident instance last_run_data = { DemistoParams.TIME: "2025-06-22T10:00:00Z", "endCursor": None, "after": test_time_str, "before": "2025-06-22T15:00:00Z", } # Mock dateparser to return consistent values across all environments def mock_dateparser_parse(date_string): if "6 hours ago" in date_string: return datetime.now() - timedelta(hours=6) elif "10 hours ago" in date_string: return datetime.now() - timedelta(hours=10) elif "12 hours ago" in date_string: return datetime.now() - timedelta(hours=12) return None with ( patch.object(demisto, "getLastRun", return_value=last_run_data), patch.object(demisto, "params", return_value={"first_fetch": first_fetch_param}), patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_fetch_timestamp", return_value="2025-06-22T10:00:00Z"), patch("dateparser.parse", side_effect=mock_dateparser_parse), ): # Create FetchIncident instance fetch_manager = FetchIncident() # Test the _is_after_time_too_old method directly result = fetch_manager._is_after_time_too_old(test_time_str) # Verify the result matches expectation assert result == expected_too_old, ( f"Expected _is_after_time_too_old({test_time_str}) to return {expected_too_old} " f"for first_fetch='{first_fetch_param}' with test age {test_age_minutes} minutes. " f"Timestamp: {test_time_str}" ) @pytest.mark.parametrize( "detection_id,api_response,expected_behavior,expected_error_contains,should_call_api,should_call_return_results", [ # Success case ( str(uuid.uuid4()), # Valid UUID [{"id": "detection-123", "severity": "CRITICAL"}], # Valid API response "success", None, True, True, ), # Missing detection_id ( None, # No detection_id provided None, # API not called "error", f"Missing required argument: {WizInputParam.DETECTION_ID}", False, False, ), # Empty detection_id ( "", # Empty detection_id None, # API not called "error", f"Missing required argument: {WizInputParam.DETECTION_ID}", False, False, ), # API returns error string ( str(uuid.uuid4()), # Valid UUID "API validation error", # API returns error string "error", "Error retrieving detection: API validation error", True, False, ), # API returns empty list ( str(uuid.uuid4()), # Valid UUID [], # Empty response "success", None, True, True, ), # Exception during API call ( str(uuid.uuid4()), # Valid UUID Exception("Connection timeout"), # Exception thrown "error", "An error occurred while retrieving detection: Connection timeout", True, False, ), ], ) @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.return_results") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.log_and_return_error") @patch("Packs.Wiz.Integrations.WizDefend.WizDefend.get_filtered_detections") def test_get_single_detection_all_cases( mock_get_filtered, mock_log_and_return_error, mock_return_results, detection_id, api_response, expected_behavior, expected_error_contains, should_call_api, should_call_return_results, ): """Comprehensive test for get_single_detection covering all use cases""" # Setup the demisto.args mock args_value = {"detection_id": detection_id} if detection_id is not None else {} # Configure API mock based on test case if isinstance(api_response, Exception): mock_get_filtered.side_effect = api_response else: mock_get_filtered.return_value = api_response with patch.object(demisto, "args", return_value=args_value): get_single_detection() # Verify API call behavior if should_call_api: mock_get_filtered.assert_called_once_with( detection_id=detection_id, detection_type=[DetectionType.GENERATED_THREAT, DetectionType.DID_NOT_GENERATE_THREAT] ) else: mock_get_filtered.assert_not_called() # Verify error handling if expected_behavior == "error": mock_log_and_return_error.assert_called_once() error_message = mock_log_and_return_error.call_args[0][0] assert expected_error_contains in error_message mock_return_results.assert_not_called() else: # Success case mock_log_and_return_error.assert_not_called() # Verify return_results behavior if should_call_return_results: mock_return_results.assert_called_once() call_args = mock_return_results.call_args[0][0] assert call_args.outputs_prefix == OutputPrefix.DETECTION assert call_args.outputs == api_response assert call_args.readable_output == api_response assert call_args.raw_response == api_response elif expected_behavior == "success": # If we expect success but shouldn't call return_results, # it means we need to handle this case differently pass # ===== ruleMatch null safety tests ===== @pytest.mark.parametrize( "rule_match,expected_has_rule", [ pytest.param(None, False, id="ruleMatch_null"), pytest.param({}, False, id="ruleMatch_empty"), pytest.param({"rule": None}, False, id="rule_null"), pytest.param({"rule": {"name": "test"}}, True, id="rule_present"), ], ) def test_build_fallback_description_rulematch_null_safety(rule_match, expected_has_rule): detection = {"id": "det-1", "severity": "HIGH"} if rule_match is not None: detection["ruleMatch"] = rule_match else: detection["ruleMatch"] = None result = build_fallback_description(detection) assert "HIGH severity detection" in result if expected_has_rule: assert "triggered by rule" in result else: assert "triggered by rule" not in result def test_build_incidents_rulematch_null(): detection = { "id": "12345678-1234-1234-1234-d25e16359c19", "severity": "HIGH", "createdAt": "2022-01-02T15:46:34Z", "ruleMatch": None, } result = build_incidents(detection) assert "Unknown Rule" in result[DemistoParams.NAME] def test_build_incidents_rulematch_rule_null(): detection = { "id": "12345678-1234-1234-1234-d25e16359c19", "severity": "HIGH", "createdAt": "2022-01-02T15:46:34Z", "ruleMatch": {"rule": None}, } result = build_incidents(detection) assert "Unknown Rule" in result[DemistoParams.NAME] @pytest.mark.parametrize( "detection,expected", [ pytest.param(None, None, id="detection_None"), pytest.param({}, None, id="detection_empty"), pytest.param({"ruleMatch": None}, None, id="ruleMatch_None"), pytest.param({"ruleMatch": {}}, None, id="ruleMatch_empty"), pytest.param({"ruleMatch": {"rule": None}}, None, id="rule_None"), pytest.param({"ruleMatch": {"rule": {}}}, None, id="rule_empty"), pytest.param({"ruleMatch": {"rule": {"name": None}}}, None, id="name_None"), pytest.param({"ruleMatch": {"rule": {"name": "MyRule"}}}, "MyRule", id="happy_path"), ], ) def test_safe_rule_name(detection, expected): """Helper covers every null position in the ruleMatch.rule.name chain. Two separate hardening commits had to fix successive null paths in this chain (`ruleMatch=None`, then `ruleMatch={"rule":None}`); centralizing means future Wiz-side schema variations only need one update.""" from WizDefend import _safe_rule_name assert _safe_rule_name(detection) == expected