import importlib.util import json import sys import types import unittest from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import Mock, patch ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) sys.modules.setdefault("setuphelpers", types.ModuleType("setuphelpers")) SPEC = importlib.util.spec_from_file_location("defender_setup", ROOT / "setup.py") defender_setup = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(defender_setup) NOW = datetime(2026, 10, 27, 12, 0, tzinfo=timezone.utc) def defender_state(**overrides): state = { "running_mode": "Normal", "antivirus_enabled": True, "realtime_enabled": True, "behavior_monitor_enabled": True, "ioav_enabled": True, "signature_age": 0, "signature_version": "synthetic-signature-version", "signature_last_updated": "2026-10-27T11:00:00Z", "engine_version": "synthetic-engine-version", "service_enabled": True, } state.update(overrides) return state def make_threat(threat_id="100", severity_id=1, name="Synthetic threat", is_active=False, source="ThreatSeverityID", did_threat_execute=False): return { "threat_id": str(threat_id), "threat_name": name, "threat_severity_id": severity_id, "severity_source": source, "is_active": is_active, "did_threat_execute": did_threat_execute, } def make_detection( detection_id="detection-1", threat_id="100", status_id=1, detected_hours_ago=1, changed_hours_ago=None, error_code=0, action_success=True, resources=None, ): detected = NOW - timedelta(hours=detected_hours_ago) changed = NOW - timedelta(hours=changed_hours_ago) if changed_hours_ago is not None else detected return { "detection_id": detection_id, "threat_id": str(threat_id), "status_id": status_id, "detected_at": detected.isoformat().replace("+00:00", "Z"), "changed_at": changed.isoformat().replace("+00:00", "Z"), "remediation_at": None, "error_code": error_code, "action_success": action_success, "execution_status_id": 0, "resources": resources or [], } def make_data(detections=None, threats=None, defender=None, collection_complete=True, **extra): detection_rows = detections if isinstance(detections, list) else ([] if detections is None else [detections]) threat_rows = threats if isinstance(threats, list) else ([] if threats is None else [threats]) return { "collection_complete": collection_complete, "history_detection_count": len(detection_rows), "returned_detection_count": len(detection_rows), "defender": defender_state(**(defender or {})), "threats": threat_rows, "detections": detection_rows, **extra, } class TestDefenderClassification(unittest.TestCase): def build(self, data, rules=None): return defender_setup.build_audit_snapshot(data, now=NOW, rules=rules) def test_no_events_with_operational_defender_is_ok(self): snapshot = self.build(make_data()) self.assertEqual(snapshot["summary"]["audit_status"], "OK") self.assertEqual(snapshot["summary"]["max_event_level_6h"], 0) self.assertEqual(snapshot["summary"]["max_active_level"], 0) self.assertEqual(snapshot["summary"]["current_risk_level"], 0) self.assertEqual(snapshot["detections"], []) def test_low_quarantined_threat_has_residual_risk_and_remains_active(self): data = make_data( [make_detection(status_id=3)], [make_threat(severity_id=1, is_active=False)], ) snapshot = self.build(data) item = snapshot["active_threats"][0] self.assertEqual(item["event_level"], 3) # 4 - 3 + 2 active self.assertEqual(item["status"], "Quarantined") self.assertEqual(snapshot["summary"]["audit_status"], "OK") def test_severe_removed_is_historical_event_not_current_error(self): data = make_data( [make_detection(status_id=4)], [make_threat(severity_id=5, is_active=False)], ) snapshot = self.build(data) self.assertEqual(snapshot["summary"]["max_event_level_6h"], 9) self.assertEqual(snapshot["summary"]["max_active_level"], 0) self.assertEqual(snapshot["summary"]["current_risk_level"], 0) self.assertEqual(snapshot["summary"]["audit_status"], "OK") self.assertEqual(snapshot["summary"]["resolved_count_6h"], 1) def test_removed_threat_with_execution_observed_remains_unresolved_risk(self): data = make_data( [make_detection(status_id=4)], [make_threat(severity_id=5, is_active=False, did_threat_execute=True)], ) snapshot = self.build(data) self.assertEqual(snapshot["active_threats"][0]["active_reason"], "execution_observed") self.assertEqual(snapshot["active_threats"][0]["event_level"], 11) # 12 - 3 removed + 2 observed self.assertEqual(snapshot["summary"]["audit_status"], "ERROR") def test_current_executing_state_keeps_removed_detection_active(self): event = make_detection(status_id=4) event["execution_status_id"] = 3 snapshot = self.build(make_data([event], [make_threat(severity_id=2, is_active=False)])) self.assertEqual(snapshot["active_threats"][0]["active_reason"], "execution_allowed_or_running") def test_severe_detected_threat_is_error(self): data = make_data( [make_detection(status_id=1)], [make_threat(severity_id=5, is_active=True)], ) snapshot = self.build(data) self.assertEqual(snapshot["summary"]["max_active_level"], 14) self.assertEqual(snapshot["summary"]["current_risk_level"], 14) self.assertEqual(snapshot["summary"]["audit_status"], "ERROR") def test_failed_remediation_adds_failure_and_active_modifiers(self): data = make_data( [make_detection(status_id=103, action_success=False)], [make_threat(severity_id=1, is_active=False)], ) snapshot = self.build(data) self.assertEqual(snapshot["active_threats"][0]["event_level"], 9) # 4 + 3 + 2 self.assertEqual(snapshot["summary"]["audit_status"], "WARNING") def test_action_success_false_adds_failure_modifier_even_before_failed_status_rollup(self): data = make_data( [make_detection(status_id=1, action_success=False)], [make_threat(severity_id=1, is_active=True)], ) snapshot = self.build(data) self.assertEqual(snapshot["detections"][0]["event_level"], 9) # 4 + 2 active + 3 failure def test_explicitly_allowed_threat_is_scored_and_correlated(self): data = make_data( [make_detection(status_id=5)], [make_threat(severity_id=2, is_active=False)], ) snapshot = self.build(data) self.assertEqual(snapshot["active_threats"][0]["event_level"], 11) # 6 + 3 + 2 self.assertEqual(snapshot["summary"]["audit_status"], "ERROR") self.assertIn("explicitly_allowed", [item["rule"] for item in snapshot["correlations"]]) def test_repeated_detections_raise_score_and_correlation(self): events = [make_detection("detection-a"), make_detection("detection-b", detected_hours_ago=2)] snapshot = self.build(make_data(events, [make_threat(severity_id=1, is_active=True)])) self.assertEqual(len(snapshot["detections"]), 2) self.assertTrue(all(item["event_level"] == 8 for item in snapshot["detections"])) self.assertIn("same_threat_repeated", [item["rule"] for item in snapshot["correlations"]]) self.assertEqual(snapshot["summary"]["audit_status"], "WARNING") def test_repeated_remediation_failures_correlate_once_per_detection(self): events = [ make_detection("failure-a", status_id=103, detected_hours_ago=1, action_success=False), make_detection("failure-b", status_id=103, detected_hours_ago=2, action_success=False), ] snapshot = self.build(make_data(events, [make_threat(severity_id=1, is_active=False)])) failures = [item for item in snapshot["correlations"] if item["rule"] == "repeated_remediation_failures"] self.assertEqual(len(failures), 1) self.assertEqual(failures[0]["count"], 2) self.assertEqual(failures[0]["threat_ids"], ["100"]) def test_old_active_threat_remains_visible_outside_six_hour_window(self): event = make_detection(detected_hours_ago=10, changed_hours_ago=10) snapshot = self.build(make_data([event], [make_threat(severity_id=5, is_active=True)])) self.assertEqual(snapshot["summary"]["detection_count_6h"], 0) self.assertEqual(snapshot["summary"]["max_event_level_6h"], 0) self.assertEqual(snapshot["summary"]["active_count"], 1) self.assertEqual(snapshot["summary"]["current_risk_level"], 14) self.assertEqual(snapshot["summary"]["audit_status"], "ERROR") def test_realtime_disabled_in_normal_mode_forces_error(self): snapshot = self.build(make_data(defender={"realtime_enabled": False})) self.assertEqual(snapshot["summary"]["current_risk_level"], 2) self.assertEqual(snapshot["summary"]["audit_status"], "ERROR") self.assertIn("Realtime protection disabled", snapshot["summary"]["protection_issues"]) def test_passive_mode_does_not_create_protection_false_positive(self): snapshot = self.build(make_data(defender={"running_mode": "Passive", "realtime_enabled": False})) self.assertEqual(snapshot["summary"]["current_risk_level"], 0) self.assertEqual(snapshot["summary"]["audit_status"], "OK") def test_collection_error_is_fail_closed(self): snapshot = defender_setup.build_collection_error_snapshot("powershell_timeout") self.assertFalse(snapshot["collection_complete"]) self.assertEqual(snapshot["summary"]["current_risk_level"], 16) self.assertEqual(snapshot["summary"]["audit_status"], "ERROR") def test_missing_history_arrays_is_incomplete_and_error(self): snapshot = self.build({"defender": defender_state()}) self.assertFalse(snapshot["collection_complete"]) self.assertEqual(snapshot["summary"]["audit_status"], "ERROR") self.assertEqual(snapshot["summary"]["current_risk_level"], 16) def test_singleton_powershell_objects_are_normalized_as_one_row(self): data = make_data( detections=make_detection(status_id=1), threats=make_threat(severity_id=1, is_active=True), ) snapshot = self.build(data) self.assertTrue(snapshot["collection_complete"]) self.assertEqual(len(snapshot["detections"]), 1) self.assertEqual(snapshot["detections"][0]["threat_id"], "100") def test_null_empty_collections_are_treated_as_empty_not_failed(self): data = make_data() data["detections"] = None data["threats"] = None snapshot = self.build(data) self.assertTrue(snapshot["collection_complete"]) self.assertEqual(snapshot["detections"], []) self.assertEqual(snapshot["summary"]["audit_status"], "OK") def test_same_threatid_with_distinct_detectionids_remains_two_incidents(self): detections = [ make_detection("distinct-a", detected_hours_ago=1), make_detection("distinct-b", detected_hours_ago=2), make_detection("distinct-a", detected_hours_ago=1), # duplicate record of same detection ] snapshot = self.build(make_data(detections, [make_threat(severity_id=1, is_active=True)])) self.assertEqual(len(snapshot["detections"]), 2) self.assertEqual(snapshot["summary"]["detection_count_6h"], 2) self.assertIn("same_threat_repeated", [item["rule"] for item in snapshot["correlations"]]) def test_old_detection_with_recent_status_change_enters_window(self): event = make_detection(status_id=6, detected_hours_ago=20, changed_hours_ago=1) snapshot = self.build(make_data([event], [make_threat(severity_id=5, is_active=False)])) self.assertEqual(snapshot["summary"]["detection_count_6h"], 1) self.assertEqual(snapshot["summary"]["max_event_level_6h"], 10) self.assertEqual(snapshot["summary"]["current_risk_level"], 0) self.assertEqual(snapshot["summary"]["audit_status"], "OK") def test_legacy_catalog_severityid_is_normalized_to_documented_enum(self): data = make_data( [make_detection(status_id=1)], [make_threat(severity_id=3, source="SeverityID", is_active=True)], ) snapshot = self.build(data) self.assertEqual(snapshot["detections"][0]["severity_id"], 4) self.assertEqual(snapshot["detections"][0]["severity"], "High") self.assertEqual(snapshot["summary"]["max_active_level"], 11) def test_severe_status_mapping_uses_threatseverityid_five(self): self.assertEqual( defender_setup.classify_detection({"threat_severity_id": 5, "status_id": 2}), 9, ) self.assertEqual( defender_setup.classify_detection({"threat_severity_id": 4, "status_id": 2}), 6, ) def test_two_distinct_threats_correlate_and_protection_disabled_is_combined(self): events = [ make_detection("threat-a", threat_id="100", status_id=1), make_detection("threat-b", threat_id="200", status_id=1, detected_hours_ago=2), ] threats = [ make_threat("100", severity_id=5, is_active=True, did_threat_execute=True), make_threat("200", severity_id=1, is_active=True), ] snapshot = self.build(make_data(events, threats, defender={"realtime_enabled": False})) rules = [item["rule"] for item in snapshot["correlations"]] self.assertIn("multiple_threats", rules) self.assertIn("severe_with_protection_disabled", rules) severe = next(item for item in snapshot["detections"] if item["threat_id"] == "100") self.assertEqual(severe["event_level"], 16) # 12 + 2 active + 2 realtime disabled self.assertTrue(severe["did_threat_execute"]) def test_resources_are_reduced_to_types_and_counts(self): event = make_detection(resources=[r"file:_C:\Users\synthetic\private.txt", "registry:synthetic"]) snapshot = self.build(make_data([event], [make_threat(is_active=True)])) serialized = json.dumps(snapshot) self.assertEqual(snapshot["detections"][0]["resource_count"], 2) self.assertEqual(snapshot["detections"][0]["resource_kinds"], ["file", "registry"]) self.assertNotIn("synthetic\\private.txt", serialized) self.assertNotIn("Users", serialized) def test_wapt_status_uses_current_risk_not_historical_event_maximum(self): data = make_data( [make_detection(status_id=4)], [make_threat(severity_id=5, is_active=False)], ) snapshot = self.build(data) self.assertEqual(snapshot["summary"]["max_event_level_6h"], 9) self.assertEqual(snapshot["summary"]["audit_status"], "OK") def test_classifier_modifiers_are_configurable_and_clamped(self): rules = { "active_modifier": 20, "status_modifiers": {1: 20}, } self.assertEqual( defender_setup.classify_detection({"threat_severity_id": 5, "status_id": 1}, rules=rules), 16, ) def test_wapt_thresholds_are_ok_warning_error(self): self.assertEqual(defender_setup._wapt_status(6, [], True, False, True, 0, 0), "OK") self.assertEqual(defender_setup._wapt_status(7, [], True, False, True, 0, 0), "WARNING") self.assertEqual(defender_setup._wapt_status(10, [], True, False, True, 0, 0), "ERROR") class TestDefenderCollectionAndWaptAdapter(unittest.TestCase): def test_powershell_timeout_is_a_safe_collection_error(self): with patch.object(defender_setup.subprocess, "run", side_effect=defender_setup.subprocess.TimeoutExpired("powershell", 90)): with self.assertRaises(defender_setup.DefenderCollectionError) as raised: defender_setup.powershell_json("synthetic script", timeout=90) self.assertEqual(raised.exception.code, "powershell_timeout") def test_powershell_json_accepts_bom_and_never_uses_executionpolicy_bypass(self): payload = {"collection_complete": True, "defender": {}, "threats": [], "detections": []} completed = types.SimpleNamespace(returncode=0, stdout=("\ufeff" + json.dumps(payload)).encode("utf-8"), stderr=b"") with patch.object(defender_setup.subprocess, "run", return_value=completed) as run: result = defender_setup.powershell_json("synthetic script") args = run.call_args.args[0] self.assertEqual(result, payload) self.assertNotIn("-ExecutionPolicy", args) self.assertNotIn("Bypass", args) self.assertEqual(run.call_args.kwargs["timeout"], defender_setup.POWERSHELL_TIMEOUT_SECONDS) def test_audit_timeout_returns_error_and_stores_failed_snapshot(self): wapt = Mock() with patch.object(defender_setup, "WAPT", wapt, create=True), patch.object( defender_setup, "collect_defender", side_effect=defender_setup.DefenderCollectionError("powershell_timeout") ): result = defender_setup.audit() self.assertEqual(result, "ERROR") stored = [call.args for call in wapt.write_audit_data_if_changed.call_args_list] snapshot_call = next(args for args in stored if args[1] == "snapshot") self.assertFalse(snapshot_call[2]["collection_complete"]) self.assertEqual(snapshot_call[2]["summary"]["audit_status"], "ERROR") def test_wapt_audit_api_uses_official_signature_and_json_value(self): wapt = Mock() value = {"schema_version": 2, "detections": [], "correlations": []} with patch.object(defender_setup, "WAPT", wapt, create=True): defender_setup.save_audit("snapshot", value) wapt.write_audit_data_if_changed.assert_called_once_with( "Microsoft Defender", "snapshot", value, max_count=512, keep_days=90, ) def test_audit_hook_returns_wapt_status_and_writes_v2_snapshot(self): wapt = Mock() with patch.object(defender_setup, "WAPT", wapt, create=True), patch.object( defender_setup, "collect_defender", return_value=make_data() ): result = defender_setup.audit() self.assertEqual(result, "OK") written_keys = [call.args[1] for call in wapt.write_audit_data_if_changed.call_args_list] self.assertEqual(written_keys, ["snapshot", "summary", "detections", "active_threats", "correlations"]) snapshot_call = wapt.write_audit_data_if_changed.call_args_list[0] self.assertEqual(snapshot_call.args[2]["schema_version"], 2) def test_control_declares_windows_wapt_26_and_15_minute_audit(self): fields = {} for line in (ROOT / "WAPT" / "control").read_text(encoding="utf-8").splitlines(): if line.strip() and not line.lstrip().startswith("#"): key, value = line.split(":", 1) fields[key.strip()] = value.strip() self.assertEqual(fields["package"], "comit-audit-defender") self.assertEqual(fields["min_wapt_version"], "2.6") self.assertEqual(fields["audit_schedule"], "15m") self.assertEqual(fields["target_os"], "windows") def test_powershell_collection_is_local_read_only_and_does_not_emit_raw_resources(self): script = defender_setup.POWERSHELL_SCRIPT self.assertIn("Get-MpComputerStatus", script) self.assertIn("Get-MpThreatDetection", script) self.assertIn("Get-MpThreat -ErrorAction Stop", script) self.assertNotIn("Invoke-WebRequest", script) self.assertNotIn("Remove-MpThreat", script) self.assertNotIn("Set-MpPreference", script) self.assertNotIn("resources = @($_.Resources)", script) self.assertNotIn("Select-Object -First 100", script) if __name__ == "__main__": unittest.main()