""" Comprehensive tests for audit system. Tests checksum generation, audit management, replay engine, and integrity verification. """ import pytest import uuid import json import hashlib from datetime import datetime, timedelta from unittest.mock import Mock, patch, AsyncMock from typing import Dict, Any, List from audit.checksum_utils import ( ChecksumGenerator, DeterministicSeedManager, AuditTrailGenerator, generate_config_hash, generate_result_checksum, verify_run_integrity ) from audit.audit_manager import AuditManager, get_audit_manager from audit.replay_engine import ReplayEngine, get_replay_engine, MockModelInterface from schemas.audit_schema import ( AuditTrail, AuditStatus, IntegrityLevel, ReplayStatus, ConfigHashRecord, ResultChecksumRecord, AuditMetadata, ReplayRequest, ReplayResult, IntegrityVerification, AuditFilter ) class TestChecksumGenerator: """Test checksum generation functionality.""" def test_generate_config_hash(self): """Test configuration hash generation.""" config = { "model_name": "test-model", "attack_types": ["jailbreak", "injection"], "prompt_count": 10, "max_iterations": 5 } hash_value = ChecksumGenerator.generate_config_hash(config) assert isinstance(hash_value, str) assert len(hash_value) == 64 # SHA-256 hex length assert all(c in '0123456789abcdef' for c in hash_value.lower()) def test_generate_config_hash_consistency(self): """Test that same config produces same hash.""" config = { "model_name": "test-model", "attack_types": ["jailbreak", "injection"], "prompt_count": 10, "max_iterations": 5 } hash1 = ChecksumGenerator.generate_config_hash(config) hash2 = ChecksumGenerator.generate_config_hash(config) assert hash1 == hash2 def test_generate_config_hash_order_independence(self): """Test that dictionary order doesn't affect hash.""" config1 = { "model_name": "test-model", "attack_types": ["jailbreak", "injection"], "prompt_count": 10 } config2 = { "prompt_count": 10, "attack_types": ["injection", "jailbreak"], "model_name": "test-model" } hash1 = ChecksumGenerator.generate_config_hash(config1) hash2 = ChecksumGenerator.generate_config_hash(config2) assert hash1 == hash2 def test_generate_result_checksum(self): """Test result checksum generation.""" result = { "score_card": { "robustness_score": 0.8, "risk_score": 0.2 }, "success_rate": 0.3, "total_attacks": 10 } checksum = ChecksumGenerator.generate_result_checksum(result) assert isinstance(checksum, str) assert len(checksum) == 64 # SHA-256 hex length def test_generate_result_checksum_string_input(self): """Test result checksum with string input.""" result_string = '{"score_card": {"robustness_score": 0.8}}' checksum = ChecksumGenerator.generate_result_checksum(result_string) assert isinstance(checksum, str) assert len(checksum) == 64 def test_generate_run_hash(self): """Test combined run hash generation.""" config_hash = "abcd1234" * 8 # 64 chars result_checksum = "efgh5678" * 8 # 64 chars run_id = str(uuid.uuid4()) run_hash = ChecksumGenerator.generate_run_hash(config_hash, result_checksum, run_id) assert isinstance(run_hash, str) assert len(run_hash) == 64 assert run_hash != config_hash assert run_hash != result_checksum def test_verify_config_integrity(self): """Test configuration integrity verification.""" config = { "model_name": "test-model", "attack_types": ["jailbreak"] } expected_hash = ChecksumGenerator.generate_config_hash(config) # Test valid verification assert ChecksumGenerator.verify_config_integrity(config, expected_hash) is True # Test invalid verification assert ChecksumGenerator.verify_config_integrity(config, "invalid_hash") is False def test_verify_result_integrity(self): """Test result integrity verification.""" result = {"score_card": {"robustness_score": 0.8}} expected_checksum = ChecksumGenerator.generate_result_checksum(result) # Test valid verification assert ChecksumGenerator.verify_result_integrity(result, expected_checksum) is True # Test invalid verification assert ChecksumGenerator.verify_result_integrity(result, "invalid_checksum") is False def test_normalize_config(self): """Test configuration normalization.""" config = { "model_name": "test-model", "nested": {"param": "value"}, "list_param": ["b", "a"], # Unsorted "number_param": 42, "bool_param": True, "none_param": None } normalized = ChecksumGenerator._normalize_config(config) assert isinstance(normalized, dict) assert normalized["list_param"] == ["a", "b"] # Should be sorted assert normalized["number_param"] == 42 assert normalized["bool_param"] is True assert normalized["none_param"] is None class TestDeterministicSeedManager: """Test deterministic seed management.""" def test_generate_seed_from_config_hash(self): """Test seed generation from config hash.""" config_hash = "abcd1234" * 8 # 64 chars seed = DeterministicSeedManager.generate_seed_from_config_hash(config_hash) assert isinstance(seed, int) assert 0 <= seed <= 2**31 - 1 # 32-bit integer range def test_generate_seed_consistency(self): """Test that same hash produces same seed.""" config_hash = "abcd1234" * 8 seed1 = DeterministicSeedManager.generate_seed_from_config_hash(config_hash) seed2 = DeterministicSeedManager.generate_seed_from_config_hash(config_hash) assert seed1 == seed2 def test_generate_multiple_seeds(self): """Test multiple seed generation.""" config_hash = "abcd1234" * 8 count = 5 seeds = DeterministicSeedManager.generate_multiple_seeds(config_hash, count) assert len(seeds) == count assert all(isinstance(seed, int) for seed in seeds) assert len(set(seeds)) == count # All seeds should be unique def test_set_deterministic_seeds(self): """Test setting deterministic seeds.""" config_hash = "abcd1234" * 8 seeds = DeterministicSeedManager.set_deterministic_seeds(config_hash, seed_count=3) assert len(seeds) == 3 assert all(isinstance(seed, int) for seed in seeds) class TestAuditTrailGenerator: """Test audit trail generation.""" def test_generate_audit_metadata(self): """Test audit metadata generation.""" run_id = str(uuid.uuid4()) config_hash = "abcd1234" * 8 result_checksum = "efgh5678" * 8 config_dict = {"model_name": "test-model"} result_data = {"score_card": {"robustness_score": 0.8}} metadata = AuditTrailGenerator.generate_audit_metadata( run_id, config_hash, result_checksum, config_dict, result_data ) assert isinstance(metadata, dict) assert metadata["run_id"] == run_id assert metadata["config_hash"] == config_hash assert metadata["result_checksum"] == result_checksum assert "combined_hash" in metadata assert "timestamp" in metadata assert "config_summary" in metadata assert "result_summary" in metadata assert "integrity_checks" in metadata def test_generate_replay_report(self): """Test replay report generation.""" original_run_id = str(uuid.uuid4()) replay_run_id = str(uuid.uuid4()) original_result = {"score_card": {"robustness_score": 0.8}} replay_result = {"score_card": {"robustness_score": 0.8}} config_hash = "abcd1234" * 8 report = AuditTrailGenerator.generate_replay_report( original_run_id, replay_run_id, original_result, replay_result, config_hash ) assert isinstance(report, dict) assert report["original_run_id"] == original_run_id assert report["replay_run_id"] == replay_run_id assert "checksum_comparison" in report assert "metric_comparison" in report assert "reproducibility_assessment" in report def test_compare_metrics(self): """Test metric comparison.""" original = { "score_card": { "robustness_score": 0.8, "risk_score": 0.2 }, "execution_time_ms": 5000 } replay = { "score_card": { "robustness_score": 0.79, # Small difference "risk_score": 0.2 }, "execution_time_ms": 5200 # 4% difference } comparison = AuditTrailGenerator._compare_metrics(original, replay) assert isinstance(comparison, dict) assert "robustness_score" in comparison assert "risk_score" in comparison assert "execution_time_ms" in comparison # Check tolerance calculation robustness_data = comparison["robustness_score"] assert robustness_data["original"] == 0.8 assert robustness_data["replay"] == 0.79 assert robustness_data["difference"] == 0.01 assert robustness_data["within_tolerance"] is True # < 1% tolerance def test_assess_metric_reproducibility(self): """Test metric reproducibility assessment.""" original = { "score_card": { "robustness_score": 0.8, "risk_score": 0.2 } } replay = { "score_card": { "robustness_score": 0.8, # Exact match "risk_score": 0.2 # Exact match } } is_reproducible = AuditTrailGenerator._assess_metric_reproducibility(original, replay) assert is_reproducible is True # Test with non-reproducible metrics non_reproducible_replay = { "score_card": { "robustness_score": 0.5, # Big difference "risk_score": 0.2 } } is_reproducible = AuditTrailGenerator._assess_metric_reproducibility(original, non_reproducible_replay) assert is_reproducible is False def test_calculate_confidence_level(self): """Test confidence level calculation.""" original = { "score_card": { "robustness_score": 0.8, "risk_score": 0.2 } } # Test with perfect match perfect_replay = { "score_card": { "robustness_score": 0.8, "risk_score": 0.2 } } comparison = AuditTrailGenerator._compare_metrics(original, perfect_replay) confidence = AuditTrailGenerator._calculate_confidence_level(original, perfect_replay) assert confidence in ['high', 'medium', 'low', 'very_low'] # Perfect match should give high confidence assert confidence == 'high' class TestAuditManager: """Test audit manager functionality.""" @pytest.fixture def mock_experiment_manager(self): """Create mock experiment manager.""" manager = Mock() manager.get_experiment = Mock() return manager @pytest.fixture def audit_manager(self, mock_experiment_manager): """Create audit manager with mock dependencies.""" return AuditManager(mock_experiment_manager) def test_attach_audit_data(self, audit_manager, mock_experiment_manager): """Test attaching audit data to a run.""" run_id = str(uuid.uuid4()) config_dict = { "model_name": "test-model", "attack_types": ["jailbreak"], "prompt_count": 10, "max_iterations": 5, "random_seed": 42 } result_data = { "score_card": { "robustness_score": 0.8, "risk_score": 0.2 }, "success_rate": 0.3, "total_attacks": 10, "execution_time_ms": 5000 } # Mock experiment mock_experiment = Mock() mock_experiment.run_id = run_id mock_experiment_manager.get_experiment.return_value = mock_experiment audit_trail = audit_manager.attach_audit_data(run_id, config_dict, result_data) assert isinstance(audit_trail, AuditTrail) assert audit_trail.run_id == run_id assert audit_trail.status == AuditStatus.COMPLETED assert audit_trail.config_record.config_hash is not None assert audit_trail.result_record.result_checksum is not None assert audit_trail.audit_metadata.config_hash is not None assert audit_trail.audit_metadata.result_checksum is not None def test_attach_audit_data_experiment_not_found(self, audit_manager, mock_experiment_manager): """Test attaching audit data when experiment not found.""" run_id = str(uuid.uuid4()) config_dict = {"model_name": "test-model"} result_data = {"score_card": {"robustness_score": 0.8}} # Mock experiment not found mock_experiment_manager.get_experiment.return_value = None with pytest.raises(ValueError, match="Experiment with run_id .* not found"): audit_manager.attach_audit_data(run_id, config_dict, result_data) def test_verify_run(self, audit_manager, mock_experiment_manager): """Test run verification.""" run_id = str(uuid.uuid4()) config_dict = { "model_name": "test-model", "attack_types": ["jailbreak"] } result_data = { "score_card": {"robustness_score": 0.8}, "success_rate": 0.3 } # Create audit trail first audit_trail = audit_manager.attach_audit_data(run_id, config_dict, result_data) # Verify run verification = audit_manager.verify_run(run_id) assert isinstance(verification, IntegrityVerification) assert verification.run_id == run_id assert verification.config_integrity is True assert verification.result_integrity is True assert verification.overall_integrity is True assert verification.confidence_level in [IntegrityLevel.HIGH, IntegrityLevel.MEDIUM, IntegrityLevel.LOW] assert 0.0 <= verification.confidence_score <= 1.0 def test_verify_run_not_found(self, audit_manager): """Test verification when run not found.""" run_id = str(uuid.uuid4()) with pytest.raises(ValueError, match="Audit trail for run_id .* not found"): audit_manager.verify_run(run_id) def test_get_audit_trail(self, audit_manager): """Test getting audit trail.""" run_id = str(uuid.uuid4()) config_dict = {"model_name": "test-model"} result_data = {"score_card": {"robustness_score": 0.8}} # Create audit trail created_trail = audit_manager.attach_audit_data(run_id, config_dict, result_data) # Get audit trail retrieved_trail = audit_manager.get_audit_trail(run_id) assert retrieved_trail is not None assert retrieved_trail.run_id == run_id assert retrieved_trail == created_trail def test_get_audit_trail_not_found(self, audit_manager): """Test getting audit trail when not found.""" run_id = str(uuid.uuid4()) trail = audit_manager.get_audit_trail(run_id) assert trail is None def test_list_audit_trails(self, audit_manager): """Test listing audit trails.""" # Create multiple audit trails run_ids = [str(uuid.uuid4()) for _ in range(3)] for run_id in run_ids: config_dict = {"model_name": f"model-{run_id[:8]}"} result_data = {"score_card": {"robustness_score": 0.8}} audit_manager.attach_audit_data(run_id, config_dict, result_data) # List all trails trails = audit_manager.list_audit_trails() assert isinstance(trails, list) assert len(trails) == 3 assert all(isinstance(trail, AuditTrail) for trail in trails) def test_list_audit_trails_with_filters(self, audit_manager): """Test listing audit trails with filters.""" # Create audit trails with different statuses run_id1 = str(uuid.uuid4()) run_id2 = str(uuid.uuid4()) config_dict = {"model_name": "test-model"} result_data = {"score_card": {"robustness_score": 0.8}} trail1 = audit_manager.attach_audit_data(run_id1, config_dict, result_data) trail2 = audit_manager.attach_audit_data(run_id2, config_dict, result_data) # Modify one trail status trail1.status = AuditStatus.FAILED # Filter by status filters = AuditFilter(status=AuditStatus.FAILED) filtered_trails = audit_manager.list_audit_trails(filters) assert len(filtered_trails) == 1 assert filtered_trails[0].status == AuditStatus.FAILED def test_get_audit_statistics(self, audit_manager): """Test getting audit statistics.""" # Create audit trails run_ids = [str(uuid.uuid4()) for _ in range(5)] for run_id in run_ids: config_dict = {"model_name": "test-model"} result_data = {"score_card": {"robustness_score": 0.8}} audit_manager.attach_audit_data(run_id, config_dict, result_data) # Get statistics stats = audit_manager.get_audit_statistics() assert isinstance(stats, dict) assert stats["total_audits"] == 5 assert stats["verified_audits"] >= 0 assert stats["failed_audits"] >= 0 assert stats["corrupted_audits"] >= 0 assert 0.0 <= stats["system_health_score"] <= 1.0 class TestReplayEngine: """Test replay engine functionality.""" @pytest.fixture def mock_audit_manager(self): """Create mock audit manager.""" manager = Mock() manager.get_audit_trail = Mock() return manager @pytest.fixture def mock_experiment_manager(self): """Create mock experiment manager.""" manager = Mock() manager.get_experiment = Mock() return manager @pytest.fixture def replay_engine(self, mock_audit_manager, mock_experiment_manager): """Create replay engine with mock dependencies.""" return ReplayEngine(mock_audit_manager, mock_experiment_manager) def test_replay_run_success(self, replay_engine, mock_audit_manager): """Test successful run replay.""" original_run_id = str(uuid.uuid4()) # Create mock audit trail config_dict = { "model_name": "test-model", "attack_types": ["jailbreak"], "prompt_count": 10, "max_iterations": 5, "random_seed": 42 } result_data = { "score_card": {"robustness_score": 0.8}, "success_rate": 0.3, "total_attacks": 10, "execution_time_ms": 5000 } mock_audit_trail = Mock() mock_audit_trail.config_record = Mock() mock_audit_trail.config_record.config_dict = config_dict mock_audit_trail.config_record.config_hash = "abcd1234" * 8 mock_audit_trail.result_record = Mock() mock_audit_trail.result_record.result_data = result_data mock_audit_trail.result_record.result_checksum = "efgh5678" * 8 mock_audit_trail.mark_replay_attempted = Mock() mock_audit_trail.mark_replay_completed = Mock() mock_audit_manager.get_audit_trail.return_value = mock_audit_trail # Mock replay execution with patch.object(replay_engine, '_execute_replay') as mock_execute: mock_execute.return_value = result_data.copy() # Perform replay replay_result = replay_engine.replay_run(original_run_id) assert isinstance(replay_result, ReplayResult) assert replay_result.original_run_id == original_run_id assert replay_result.replay_status in [ReplayStatus.SUCCESSFUL, ReplayStatus.FAILED] assert replay_result.replay_started_at is not None assert replay_result.replay_completed_at is not None assert replay_result.replay_duration_ms >= 0 assert isinstance(replay_result.result_comparison, dict) assert isinstance(replay_result.reproducibility_assessment, dict) def test_replay_run_audit_not_found(self, replay_engine, mock_audit_manager): """Test replay when audit trail not found.""" original_run_id = str(uuid.uuid4()) mock_audit_manager.get_audit_trail.return_value = None replay_result = replay_engine.replay_run(original_run_id) assert replay_result.replay_status == ReplayStatus.FAILED assert len(replay_result.errors) > 0 assert "Audit trail for run_id" in replay_result.errors[0] def test_compare_results(self, replay_engine): """Test result comparison.""" original_result = { "score_card": { "robustness_score": 0.8, "risk_score": 0.2, "hallucination_rate": 0.1 }, "success_rate": 0.3, "total_attacks": 10, "execution_time_ms": 5000 } new_result = { "score_card": { "robustness_score": 0.79, # Small difference "risk_score": 0.2, # Exact match "hallucination_rate": 0.12 # Small difference }, "success_rate": 0.3, # Exact match "total_attacks": 10, # Exact match "execution_time_ms": 5200 # Small difference } comparison = replay_engine.compare_results(original_result, new_result) assert isinstance(comparison, dict) assert "exact_match" in comparison assert "score_card_comparison" in comparison assert "metric_comparison" in comparison assert "execution_time_comparison" in comparison assert "differences" in comparison # Check specific comparisons score_comparison = comparison["score_card_comparison"] assert "robustness_score" in score_comparison assert score_comparison["robustness_score"]["within_tolerance"] is True assert comparison["exact_match"] is False # Should not be exact due to small differences def test_batch_replay(self, replay_engine, mock_audit_manager): """Test batch replay operations.""" run_ids = [str(uuid.uuid4()) for _ in range(3)] # Create mock audit trails mock_audit_trails = [] for run_id in run_ids: mock_trail = Mock() mock_trail.config_record = Mock() mock_trail.config_record.config_dict = {"model_name": "test-model"} mock_trail.config_record.config_hash = "abcd1234" * 8 mock_trail.result_record = Mock() mock_trail.result_record.result_data = {"score_card": {"robustness_score": 0.8}} mock_trail.result_record.result_checksum = "efgh5678" * 8 mock_trail.mark_replay_attempted = Mock() mock_trail.mark_replay_completed = Mock() mock_audit_trails.append(mock_trail) mock_audit_manager.get_audit_trail.side_effect = mock_audit_trails # Mock replay execution with patch.object(replay_engine, '_execute_replay') as mock_execute: mock_execute.return_value = {"score_card": {"robustness_score": 0.8}} # Perform batch replay results = replay_engine.batch_replay(run_ids) assert len(results) == 3 assert all(isinstance(result, ReplayResult) for result in results) assert all(result.original_run_id in run_ids for result in results) def test_get_replay_history(self, replay_engine): """Test getting replay history.""" original_run_id = str(uuid.uuid4()) # Add some replay results to cache replay_results = [] for i in range(3): replay_result = Mock(spec=ReplayResult) replay_result.original_run_id = original_run_id replay_result.replay_started_at = datetime.utcnow() - timedelta(hours=i) replay_results.append(replay_result) replay_engine._replay_cache[f"replay_{i}"] = replay_result history = replay_engine.get_replay_history(original_run_id) assert len(history) == 3 assert all(result.original_run_id == original_run_id for result in history) # Should be sorted by timestamp (newest first) assert history[0].replay_started_at >= history[1].replay_started_at def test_apply_replay_options(self, replay_engine): """Test applying replay options to configuration.""" original_config = { "model_name": "test-model", "attack_types": ["jailbreak"], "prompt_count": 10, "max_iterations": 5, "timeout": 300 } replay_options = { "force_deterministic": True, "timeout_multiplier": 2.0, "max_iterations_override": 10, "disable_mutation": True } replay_config = replay_engine._apply_replay_options(original_config, replay_options) # Check that deterministic seed was added assert "random_seed" in replay_config assert "seed" in replay_config # Check that options were applied assert replay_config["timeout"] == 600 # 300 * 2.0 assert replay_config["max_iterations"] == 10 assert replay_config["mutation_enabled"] is False def test_assess_reproducibility(self, replay_engine): """Test reproducibility assessment.""" original_result = { "score_card": {"robustness_score": 0.8, "risk_score": 0.2}, "success_rate": 0.3 } # Test with perfect match perfect_replay = { "score_card": {"robustness_score": 0.8, "risk_score": 0.2}, "success_rate": 0.3 } comparison = replay_engine._compare_results(original_result, perfect_replay) assessment = replay_engine._assess_reproducibility(original_result, perfect_replay, comparison) assert isinstance(assessment, dict) assert "reproducibility_score" in assessment assert "reproducibility_level" in assessment assert "total_checks" in assessment assert "passed_checks" in assessment assert "key_differences" in assessment assert "recommendations" in assessment # Perfect match should give high reproducibility assert assessment["reproducibility_score"] >= 0.9 assert assessment["reproducibility_level"] in ['fully_reproducible', 'highly_reproducible'] class TestMockModelInterface: """Test mock model interface for replay testing.""" def test_mock_model_interface_initialization(self): """Test mock model interface initialization.""" model_config = {"model_name": "test-model", "temperature": 0.7} model = MockModelInterface(model_config) assert model.model_config == model_config assert model.model_name == "test-model" def test_mock_model_interface_generate_response(self): """Test mock response generation.""" model_config = {"model_name": "test-model"} model = MockModelInterface(model_config) prompt = "Test prompt" response = model.generate_response(prompt) assert isinstance(response, str) assert len(response) > 0 # Should be deterministic for same prompt response2 = model.generate_response(prompt) assert response == response2 def test_mock_model_interface_deterministic_behavior(self): """Test deterministic behavior with seeds.""" # Test with same seed model1 = MockModelInterface({"random_seed": 42}) model2 = MockModelInterface({"random_seed": 42}) prompt = "Test prompt" response1 = model1.generate_response(prompt) response2 = model2.generate_response(prompt) assert response1 == response2 # Test with different seeds model3 = MockModelInterface({"random_seed": 123}) response3 = model3.generate_response(prompt) # Should be different (most likely) # Note: Due to hash-based selection, this might occasionally be the same # but the probability is very low class TestUtilityFunctions: """Test utility functions.""" def test_generate_config_hash_utility(self): """Test config hash utility function.""" config = {"model_name": "test-model", "attack_types": ["jailbreak"]} hash_value = generate_config_hash(config) assert isinstance(hash_value, str) assert len(hash_value) == 64 def test_generate_result_checksum_utility(self): """Test result checksum utility function.""" result = {"score_card": {"robustness_score": 0.8}} checksum = generate_result_checksum(result) assert isinstance(checksum, str) assert len(checksum) == 64 def test_verify_run_integrity_utility(self): """Test run integrity verification utility function.""" config = {"model_name": "test-model"} result = {"score_card": {"robustness_score": 0.8}} config_hash = generate_config_hash(config) result_checksum = generate_result_checksum(result) verification = verify_run_integrity(config, result, config_hash, result_checksum) assert isinstance(verification, dict) assert "config_integrity" in verification assert "result_integrity" in verification assert "overall_integrity" in verification assert verification["config_integrity"] is True assert verification["result_integrity"] is True assert verification["overall_integrity"] is True class TestGlobalFunctions: """Test global convenience functions.""" @patch('audit.audit_manager._audit_manager') def test_get_audit_manager(self, mock_global_manager): """Test global audit manager getter.""" from audit.audit_manager import get_audit_manager manager_instance = Mock() mock_global_manager.return_value = manager_instance result = get_audit_manager() assert result == manager_instance mock_global_manager.assert_called_once() @patch('audit.audit_manager.get_audit_manager') def test_attach_audit_data_global(self, mock_get_manager): """Test global attach_audit_data function.""" from audit.audit_manager import attach_audit_data manager = Mock() mock_get_manager.return_value = manager audit_trail = Mock() manager.attach_audit_data.return_value = audit_trail run_id = "test-run-id" config = {"model_name": "test"} result = {"score_card": {"robustness_score": 0.8}} result = attach_audit_data(run_id, config, result) assert result == audit_trail manager.attach_audit_data.assert_called_once_with(run_id, config, result, None) @patch('audit.audit_manager.get_audit_manager') def test_verify_run_global(self, mock_get_manager): """Test global verify_run function.""" from audit.audit_manager import verify_run manager = Mock() mock_get_manager.return_value = manager verification = Mock() manager.verify_run.return_value = verification run_id = "test-run-id" result = verify_run(run_id) assert result == verification manager.verify_run.assert_called_once_with(run_id, None, None) @patch('audit.replay_engine._replay_engine') def test_get_replay_engine(self, mock_global_engine): """Test global replay engine getter.""" from audit.replay_engine import get_replay_engine engine_instance = Mock() mock_global_engine.return_value = engine_instance result = get_replay_engine() assert result == engine_instance mock_global_engine.assert_called_once() @patch('audit.replay_engine.get_replay_engine') def test_replay_run_global(self, mock_get_engine): """Test global replay_run function.""" from audit.replay_engine import replay_run engine = Mock() mock_get_engine.return_value = engine replay_result = Mock() engine.replay_run.return_value = replay_result run_id = "test-run-id" result = replay_run(run_id) assert result == replay_result engine.replay_run.assert_called_once_with(run_id, None, None, None) if __name__ == "__main__": pytest.main([__file__])