""" Comprehensive tests for reporting system. Tests report generation, export, validation, and integration. """ import pytest import uuid import json import csv import tempfile import shutil from datetime import datetime, timedelta from pathlib import Path from unittest.mock import Mock, patch, AsyncMock from reporting.report_generator import get_report_generator, ReportGenerator from reporting.export_utils import get_report_exporter, ReportExporter from reporting.report_validator import get_report_validator, ReportValidator from reporting.file_manager import get_file_manager, ReportFileManager from reporting.report_integration import get_report_integration_service, ReportIntegrationService from schemas.report_schema import ( ReportGenerationRequest, ReportGenerationResult, ReportValidationRequest, ReportValidationResult, ReportFormat, ReportType, ReportStatus, FullReport, SummaryReport, CSVReportData, ReportMetadata ) class TestReportGenerator: """Test report generator functionality.""" @pytest.fixture def mock_experiment_manager(self): """Create mock experiment manager.""" manager = Mock() manager.get_experiment = Mock() return manager @pytest.fixture def mock_audit_manager(self): """Create mock audit manager.""" manager = Mock() manager.get_audit_trail = Mock() return manager @pytest.fixture def report_generator(self, mock_experiment_manager, mock_audit_manager): """Create report generator with mock dependencies.""" return ReportGenerator(mock_experiment_manager, mock_audit_manager) def test_generate_full_report(self, report_generator, mock_experiment_manager, mock_audit_manager): """Test full report generation.""" run_id = str(uuid.uuid4()) # Mock experiment mock_experiment = Mock() mock_experiment.run_id = run_id mock_experiment.experiment_name = "Test Experiment" mock_experiment.description = "Test Description" mock_experiment.status = Mock() mock_experiment.status.value = "completed" mock_experiment.created_at = datetime.utcnow() mock_experiment.started_at = datetime.utcnow() mock_experiment.completed_at = datetime.utcnow() mock_experiment.config_snapshot = Mock() mock_experiment.config_snapshot.model_name = "test-model" mock_experiment.config_snapshot.dataset_name = "test-dataset" mock_experiment.config_snapshot.dataset_version = "v1" mock_experiment.config_snapshot.attack_types = ["jailbreak", "injection"] mock_experiment.config_snapshot.prompt_count = 10 mock_experiment.config_snapshot.max_iterations = 5 mock_experiment.config_snapshot.mutation_enabled = True mock_experiment.result_summary = Mock() mock_experiment.result_summary.execution_time_ms = 5000 mock_experiment.result_summary.total_attacks = 10 mock_experiment.result_summary.successful_attacks = 3 mock_experiment.result_summary.failed_attacks = 7 mock_experiment.result_summary.success_rate = 0.3 mock_experiment.result_summary.robustness_score = 0.8 mock_experiment.result_summary.risk_score = 0.2 mock_experiment.full_result = { "score_card": { "robustness_score": 0.8, "risk_score": 0.2, "hallucination_rate": 0.05, "toxicity_rate": 0.02, "confidence_score": 0.9 } } mock_experiment_manager.get_experiment.return_value = mock_experiment # Mock audit trail mock_audit_trail = Mock() mock_audit_trail.run_id = run_id mock_audit_trail.status = Mock() mock_audit_trail.status.value = "verified" mock_audit_trail.integrity_level = Mock() mock_audit_trail.integrity_level.value = "high" mock_audit_trail.config_record = Mock() mock_audit_trail.config_record.config_dict = {"model_name": "test-model"} mock_audit_trail.config_record.config_hash = "abcd1234" * 8 mock_audit_trail.result_record = Mock() mock_audit_trail.result_record.result_data = {"score_card": {"robustness_score": 0.8}} mock_audit_trail.result_record.result_checksum = "efgh5678" * 8 mock_audit_trail.audit_metadata = Mock() mock_audit_trail.audit_metadata.combined_hash = "ijkl9012" * 8 mock_audit_trail.verified_at = datetime.utcnow() mock_audit_trail.replay_count = 0 mock_audit_trail.last_replay_timestamp = None mock_audit_trail.replay_status = Mock() mock_audit_trail.replay_status.value = "not_attempted" mock_audit_manager.get_audit_trail.return_value = mock_audit_trail # Generate report request = ReportGenerationRequest( run_id=run_id, report_type=ReportType.FULL, format=ReportFormat.JSON, requested_by="test@example.com" ) result = report_generator.generate_report(request) assert result.success is True assert result.report_id is not None assert result.status == ReportStatus.COMPLETED assert result.error_message is None assert result.generation_duration_ms >= 0 def test_generate_summary_report(self, report_generator, mock_experiment_manager, mock_audit_manager): """Test summary report generation.""" run_id = str(uuid.uuid4()) # Setup mocks (similar to full report test) mock_experiment = Mock() mock_experiment.run_id = run_id mock_experiment.config_snapshot = Mock() mock_experiment.config_snapshot.model_name = "test-model" mock_experiment.config_snapshot.dataset_name = "test-dataset" mock_experiment.config_snapshot.attack_types = ["jailbreak"] mock_experiment.config_snapshot.prompt_count = 10 mock_experiment.config_snapshot.max_iterations = 5 mock_experiment.config_snapshot.mutation_enabled = True mock_experiment.status = Mock() mock_experiment.status.value = "completed" mock_experiment.created_at = datetime.utcnow() mock_experiment.result_summary = Mock() mock_experiment.result_summary.robustness_score = 0.8 mock_experiment.result_summary.risk_score = 0.2 mock_experiment.result_summary.success_rate = 0.3 mock_experiment.full_result = { "score_card": { "robustness_score": 0.8, "risk_score": 0.2, "hallucination_rate": 0.05 } } mock_experiment_manager.get_experiment.return_value = mock_experiment mock_audit_trail = Mock() mock_audit_trail.run_id = run_id mock_audit_trail.status = Mock() mock_audit_trail.status.value = "verified" mock_audit_trail.integrity_level = Mock() mock_audit_trail.integrity_level.value = "high" mock_audit_trail.config_record = Mock() mock_audit_trail.config_record.config_hash = "abcd1234" * 8 mock_audit_trail.result_record = Mock() mock_audit_trail.result_record.result_checksum = "efgh5678" * 8 mock_audit_trail.audit_metadata = Mock() mock_audit_trail.audit_metadata.combined_hash = "ijkl9012" * 8 mock_audit_manager.get_audit_trail.return_value = mock_audit_trail # Generate summary report request = ReportGenerationRequest( run_id=run_id, report_type=ReportType.SUMMARY, format=ReportFormat.JSON, requested_by="test@example.com" ) result = report_generator.generate_report(request) assert result.success is True assert result.report_id is not None # Check cached report cached_report = report_generator.get_report(result.report_id) assert cached_report is not None assert 'data' in cached_report def test_generate_report_experiment_not_found(self, report_generator, mock_experiment_manager): """Test report generation when experiment not found.""" run_id = str(uuid.uuid4()) # Mock experiment not found mock_experiment_manager.get_experiment.return_value = None request = ReportGenerationRequest( run_id=run_id, report_type=ReportType.FULL, format=ReportFormat.JSON ) result = report_generator.generate_report(request) assert result.success is False assert result.error_message is not None assert "not found" in result.error_message def test_generate_report_audit_not_found(self, report_generator, mock_experiment_manager, mock_audit_manager): """Test report generation when audit trail not found.""" run_id = str(uuid.uuid4()) # Mock experiment exists mock_experiment = Mock() mock_experiment.run_id = run_id mock_experiment_manager.get_experiment.return_value = mock_experiment # Mock audit trail not found mock_audit_manager.get_audit_trail.return_value = None request = ReportGenerationRequest( run_id=run_id, report_type=ReportType.FULL, format=ReportFormat.JSON ) result = report_generator.generate_report(request) assert result.success is False assert result.error_message is not None assert "not found" in result.error_message def test_list_reports(self, report_generator): """Test listing reports.""" # Generate some reports first run_ids = [str(uuid.uuid4()) for _ in range(3)] for run_id in run_ids: # Mock experiment and audit mock_experiment = Mock() mock_experiment.run_id = run_id mock_experiment.config_snapshot = Mock() mock_experiment.config_snapshot.model_name = "test-model" mock_experiment.status = Mock() mock_experiment.status.value = "completed" mock_experiment.created_at = datetime.utcnow() mock_audit_trail = Mock() mock_audit_trail.run_id = run_id mock_audit_trail.status = Mock() mock_audit_trail.status.value = "verified" mock_audit_trail.config_record = Mock() mock_audit_trail.config_record.config_hash = "abcd1234" * 8 mock_audit_trail.result_record = Mock() mock_audit_trail.result_record.result_checksum = "efgh5678" * 8 with patch.object(report_generator.experiment_manager, 'get_experiment', return_value=mock_experiment), \ patch.object(report_generator.audit_manager, 'get_audit_trail', return_value=mock_audit_trail): request = ReportGenerationRequest( run_id=run_id, report_type=ReportType.SUMMARY, format=ReportFormat.JSON ) result = report_generator.generate_report(request) assert result.success is True # List all reports all_reports = report_generator.list_reports() assert len(all_reports) == 3 # List reports for specific run_id specific_reports = report_generator.list_reports(run_id=run_ids[0]) assert len(specific_reports) == 1 assert specific_reports[0].run_id == run_ids[0] class TestReportExporter: """Test report exporter functionality.""" @pytest.fixture def temp_reports_dir(self): """Create temporary reports directory.""" temp_dir = Path(tempfile.mkdtemp()) yield temp_dir shutil.rmtree(temp_dir) @pytest.fixture def report_exporter(self, temp_reports_dir): """Create report exporter with temporary directory.""" return ReportExporter(str(temp_reports_dir)) def test_export_json_report(self, report_exporter): """Test JSON report export.""" run_id = str(uuid.uuid4()) report_data = { "experiment": { "run_id": run_id, "model_name": "test-model", "status": "completed", "created_at": datetime.utcnow().isoformat() }, "audit": { "run_id": run_id, "config_hash": "abcd1234" * 8, "result_checksum": "efgh5678" * 8, "audit_status": "verified" } } file_path = report_exporter.export_json_report(report_data, run_id, ReportType.FULL) assert Path(file_path).exists() assert file_path.endswith('.json') # Verify file content with open(file_path, 'r', encoding='utf-8') as f: exported_data = json.load(f) assert 'export_metadata' in exported_data assert 'report_data' in exported_data assert exported_data['report_data']['experiment']['run_id'] == run_id assert exported_data['export_metadata']['export_format'] == 'json' assert exported_data['export_metadata']['report_type'] == 'full' def test_export_csv_summary(self, report_exporter): """Test CSV summary export.""" run_id = str(uuid.uuid4()) report_data = { "experiment": { "run_id": run_id, "model_name": "test-model", "dataset_name": "test-dataset", "dataset_version": "v1", "attack_types": ["jailbreak", "injection"], "status": "completed", "created_at": datetime.utcnow().isoformat(), "execution_time_ms": 5000, "total_attacks": 10, "successful_attacks": 3, "failed_attacks": 7, "success_rate": 0.3, "robustness_score": 0.8, "risk_score": 0.2, "hallucination_rate": 0.05, "toxicity_rate": 0.02, "confidence_score": 0.9 }, "audit": { "run_id": run_id, "config_hash": "abcd1234" * 8, "result_checksum": "efgh5678" * 8, "audit_status": "verified", "integrity_level": "high", "confidence_level": "high" } } file_path = report_exporter.export_csv_summary(report_data, run_id) assert Path(file_path).exists() assert file_path.endswith('.csv') # Verify CSV content with open(file_path, 'r', newline='', encoding='utf-8') as f: reader = csv.DictReader(f) rows = list(reader) assert len(rows) >= 1 assert rows[0]['run_id'] == run_id assert rows[0]['model_name'] == 'test-model' assert rows[0]['attack_types'] == 'jailbreak,injection' # Check metadata file metadata_path = Path(file_path).with_suffix('.json') assert metadata_path.exists() with open(metadata_path, 'r', encoding='utf-8') as f: metadata = json.load(f) assert 'export_metadata' in metadata assert metadata['export_metadata']['export_format'] == 'csv' def test_export_csv_detailed(self, report_exporter): """Test detailed CSV export.""" run_id = str(uuid.uuid4()) report_data = { "experiment": { "run_id": run_id, "model_name": "test-model", "status": "completed", "created_at": datetime.utcnow().isoformat(), "robustness_score": 0.8, "risk_score": 0.2, "confidence_score": 0.9, "hallucination_rate": 0.05, "toxicity_rate": 0.02, "total_attacks": 10, "successful_attacks": 3, "failed_attacks": 7, "success_rate": 0.3 }, "audit": { "run_id": run_id, "integrity_level": "high", "confidence_level": "high", "replay_count": 0, "verification_timestamp": datetime.utcnow().isoformat(), "confidence_score": 0.95 } } file_path = report_exporter.export_csv_detailed(report_data, run_id) assert Path(file_path).exists() assert file_path.endswith('.csv') # Verify CSV content with open(file_path, 'r', newline='', encoding='utf-8') as f: reader = csv.DictReader(f) rows = list(reader) assert len(rows) >= 1 # Check for different metric types metric_types = set(row['metric_type'] for row in rows) assert 'basic_info' in metric_types assert 'score' in metric_types or 'attack' in metric_types def test_export_batch_reports(self, report_exporter): """Test batch report export.""" run_ids = [str(uuid.uuid4()) for _ in range(3)] reports = [] for run_id in run_ids: report_data = { "experiment": { "run_id": run_id, "model_name": "test-model", "status": "completed" }, "audit": { "run_id": run_id, "config_hash": "abcd1234" * 8, "result_checksum": "efgh5678" * 8 } } reports.append(report_data) file_paths = report_exporter.export_batch_reports(reports, ReportFormat.JSON) assert len(file_paths) == 3 for file_path in file_paths: assert Path(file_path).exists() assert file_path.endswith('.json') def test_get_report_file_info(self, report_exporter): """Test getting file information.""" run_id = str(uuid.uuid4()) report_data = {"experiment": {"run_id": run_id}} file_path = report_exporter.export_json_report(report_data, run_id) file_info = report_exporter.get_report_file_info(file_path) assert 'file_path' in file_info assert 'file_name' in file_info assert 'file_size_bytes' in file_info assert 'file_checksum' in file_info assert file_info['file_extension'] == '.json' def test_list_reports(self, report_exporter): """Test listing reports.""" run_ids = [str(uuid.uuid4()) for _ in range(3)] # Create some reports for run_id in run_ids: report_data = {"experiment": {"run_id": run_id}} report_exporter.export_json_report(report_data, run_id) # List all reports all_reports = report_exporter.list_reports() assert len(all_reports) == 3 # List reports for specific run_id specific_reports = report_exporter.list_reports(run_id=run_ids[0]) assert len(specific_reports) == 1 assert run_ids[0] in specific_reports[0]['file_name'] # List reports by format json_reports = report_exporter.list_reports(format=ReportFormat.JSON) assert len(json_reports) == 3 csv_reports = report_exporter.list_reports(format=ReportFormat.CSV) assert len(csv_reports) == 0 def test_delete_report(self, report_exporter): """Test deleting reports.""" run_id = str(uuid.uuid4()) report_data = {"experiment": {"run_id": run_id}} file_path = report_exporter.export_json_report(report_data, run_id) assert Path(file_path).exists() # Delete report deleted = report_exporter.delete_report(file_path) assert deleted is True assert not Path(file_path).exists() # Try to delete non-existent file deleted_again = report_exporter.delete_report(file_path) assert deleted_again is False class TestReportValidator: """Test report validator functionality.""" @pytest.fixture def temp_reports_dir(self): """Create temporary reports directory.""" temp_dir = Path(tempfile.mkdtemp()) yield temp_dir shutil.rmtree(temp_dir) @pytest.fixture def report_validator(self, temp_reports_dir): """Create report validator with temporary directory.""" return ReportValidator(str(temp_reports_dir)) def test_verify_json_report(self, report_validator): """Test JSON report validation.""" run_id = str(uuid.uuid4()) # Create valid JSON report report_data = { "export_metadata": { "exported_at": datetime.utcnow().isoformat(), "export_format": "json", "report_type": "full", "run_id": run_id, "file_version": "1.0" }, "report_data": { "experiment": { "run_id": run_id, "model_name": "test-model", "attack_types": ["jailbreak"], "status": "completed", "created_at": datetime.utcnow().isoformat() }, "audit": { "run_id": run_id, "config_hash": "abcd1234" * 8, "result_checksum": "efgh5678" * 8, "audit_status": "verified" }, "config_snapshot": {"model_name": "test-model"}, "full_result": {"score_card": {"robustness_score": 0.8}} } } file_path = Path(temp_reports_dir) / f"test_report_{run_id}.json" with open(file_path, 'w', encoding='utf-8') as f: json.dump(report_data, f) request = ReportValidationRequest(file_path=str(file_path)) result = report_validator.verify_report(request) assert result.is_valid is True assert result.format_valid is True assert result.integrity_valid is True assert result.not_expired is True assert len(result.issues) == 0 def test_verify_csv_report(self, report_validator): """Test CSV report validation.""" run_id = str(uuid.uuid4()) # Create valid CSV report csv_content = """run_id,model_name,dataset_name,attack_types,created_at,status,total_attacks,robustness_score,risk_score,config_hash,result_checksum,audit_status test-run,test-model,test-dataset,jailbreak,injection,2023-01-01T00:00:00,completed,10,0.8,0.2,abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234,efgh5678efgh5678efgh5678efgh5678efgh5678efgh5678efgh5678efgh5678,verified""" file_path = Path(temp_reports_dir) / f"test_report_{run_id}.csv" with open(file_path, 'w', encoding='utf-8') as f: f.write(csv_content) request = ReportValidationRequest(file_path=str(file_path)) result = report_validator.verify_report(request) assert result.is_valid is True assert result.format_valid is True assert result.integrity_valid is True assert len(result.issues) == 0 def test_verify_invalid_json(self, report_validator): """Test validation of invalid JSON.""" file_path = Path(temp_reports_dir) / "invalid.json" with open(file_path, 'w', encoding='utf-8') as f: f.write('{"invalid": json}') # Invalid JSON syntax request = ReportValidationRequest(file_path=str(file_path)) result = report_validator.verify_report(request) assert result.is_valid is False assert result.format_valid is False assert "Invalid JSON" in result.issues[0] def test_verify_missing_file(self, report_validator): """Test validation of missing file.""" request = ReportValidationRequest(file_path="non_existent_file.json") result = report_validator.verify_report(request) assert result.is_valid is False assert result.format_valid is False assert "not found" in result.issues[0] def test_verify_batch_reports(self, report_validator): """Test batch report validation.""" run_ids = [str(uuid.uuid4()) for _ in range(3)] file_paths = [] # Create valid reports for run_id in run_ids: report_data = { "export_metadata": { "exported_at": datetime.utcnow().isoformat(), "export_format": "json", "run_id": run_id }, "report_data": { "experiment": {"run_id": run_id}, "audit": {"run_id": run_id} } } file_path = Path(temp_reports_dir) / f"test_report_{run_id}.json" with open(file_path, 'w', encoding='utf-8') as f: json.dump(report_data, f) file_paths.append(str(file_path)) results = report_validator.verify_batch_reports(file_paths) assert len(results) == 3 for result in results: assert result.is_valid is True def test_validate_report_format(self, report_validator): """Test report format validation.""" run_id = str(uuid.uuid4()) # Create valid report report_data = { "export_metadata": { "exported_at": datetime.utcnow().isoformat(), "export_format": "json", "run_id": run_id }, "report_data": { "experiment": {"run_id": run_id}, "audit": {"run_id": run_id} } } file_path = Path(temp_reports_dir) / f"test_report_{run_id}.json" with open(file_path, 'w', encoding='utf-8') as f: json.dump(report_data, f) validation_info = report_validator.validate_report_format(str(file_path)) assert validation_info['is_valid'] is True assert validation_info['format'] == 'json' assert validation_info['integrity_verified'] is True def test_get_validation_summary(self, report_validator): """Test validation summary generation.""" # Create some validation results results = [] for i in range(5): result = Mock(spec=ReportValidationResult) result.is_valid = i < 3 # First 3 are valid result.issues = [] if i < 3 else [f"Error {i}"] result.warnings = [f"Warning {i}"] results.append(result) summary = report_validator.get_validation_summary(results) assert summary['total_reports'] == 5 assert summary['valid_reports'] == 3 assert summary['invalid_reports'] == 2 assert summary['validation_rate'] == 0.6 assert summary['total_issues'] == 2 assert summary['total_warnings'] == 5 class TestFileManager: """Test file manager functionality.""" @pytest.fixture def temp_reports_dir(self): """Create temporary reports directory.""" temp_dir = Path(tempfile.mkdtemp()) yield temp_dir shutil.rmtree(temp_dir) @pytest.fixture def file_manager(self, temp_reports_dir): """Create file manager with temporary directory.""" return ReportFileManager(str(temp_reports_dir)) def test_get_unique_filename(self, file_manager): """Test unique filename generation.""" run_id = str(uuid.uuid4()) filename1 = file_manager.get_unique_filename(run_id, "full", "json") filename2 = file_manager.get_unique_filename(run_id, "full", "json") assert filename1 != filename2 assert run_id in filename1 assert "full" in filename1 assert filename1.endswith(".json") def test_store_file(self, file_manager): """Test file storage.""" content = '{"test": "content"}' filename = "test_file.json" file_path = file_manager.store_file(content, filename, "json") assert Path(file_path).exists() assert file_path.name == filename # Verify content with open(file_path, 'r', encoding='utf-8') as f: stored_content = f.read() assert stored_content == content def test_get_file_info(self, file_manager): """Test file information retrieval.""" content = '{"test": "content"}' filename = "test_file.json" file_path = file_manager.store_file(content, filename, "json") file_info = file_manager.get_file_info(file_path) assert 'file_path' in file_info assert 'file_name' in file_info assert 'file_size_bytes' in file_info assert 'file_size_mb' in file_info assert 'created_at' in file_info assert 'file_extension' in file_info assert file_info['file_extension'] == '.json' assert file_info['is_readable'] is True def test_list_files(self, file_manager): """Test file listing.""" # Store some files files = [ ("file1.json", '{"test": 1}'), ("file2.csv", "col1,col2\nval1,val2"), ("file3.json", '{"test": 3}') ] for filename, content in files: format = "json" if filename.endswith('.json') else "csv" file_manager.store_file(content, filename, format) # List all files all_files = file_manager.list_files() assert len(all_files) == 3 # List by format json_files = file_manager.list_files(format="json") assert len(json_files) == 2 csv_files = file_manager.list_files(format="csv") assert len(csv_files) == 1 def test_delete_file(self, file_manager): """Test file deletion.""" content = '{"test": "content"}' filename = "test_delete.json" file_path = file_manager.store_file(content, filename, "json") assert Path(file_path).exists() # Delete file deleted = file_manager.delete_file(file_path) assert deleted is True assert not Path(file_path).exists() # Try to delete again deleted_again = file_manager.delete_file(file_path) assert deleted_again is False def test_archive_file(self, file_manager): """Test file archiving.""" content = '{"test": "content"}' filename = "test_archive.json" file_path = file_manager.store_file(content, filename, "json") original_path = Path(file_path) # Archive file archived = file_manager.archive_file(file_path) assert archived is True assert not original_path.exists() # Check archive directory archive_dir = file_manager.reports_dir / "archive" archive_files = list(archive_dir.glob("*")) assert len(archive_files) == 1 def test_get_storage_stats(self, file_manager): """Test storage statistics.""" # Store some files for i in range(3): content = f'{{"test": {i}}}' filename = f"test_stats_{i}.json" file_manager.store_file(content, filename, "json") stats = file_manager.get_storage_stats() assert 'total_files' in stats assert 'total_size_bytes' in stats assert 'total_size_mb' in stats assert 'max_storage_mb' in stats assert 'storage_usage_percent' in stats assert 'format_distribution' in stats assert stats['total_files'] == 3 assert 'json' in stats['format_distribution'] def test_check_storage_health(self, file_manager): """Test storage health check.""" health = file_manager.check_storage_health() assert 'status' in health assert 'issues' in health assert 'recommendations' in health assert 'storage_stats' in health assert health['status'] in ['healthy', 'warning', 'critical', 'error'] class TestReportIntegration: """Test report integration service.""" @pytest.fixture def mock_experiment_manager(self): """Create mock experiment manager.""" manager = Mock() manager.get_experiment = Mock() return manager @pytest.fixture def mock_audit_manager(self): """Create mock audit manager.""" manager = Mock() manager.get_audit_trail = Mock() return manager @pytest.fixture def integration_service(self, mock_experiment_manager, mock_audit_manager): """Create integration service with mock dependencies.""" return ReportIntegrationService() def test_generate_comprehensive_report(self, integration_service): """Test comprehensive report generation with integration.""" run_id = str(uuid.uuid4()) # Mock experiment and audit (similar to previous tests) with patch.object(integration_service.experiment_manager, 'get_experiment') as mock_get_exp, \ patch.object(integration_service.audit_manager, 'get_audit_trail') as mock_get_audit: # Setup mocks mock_experiment = Mock() mock_experiment.run_id = run_id mock_experiment.config_snapshot = Mock() mock_experiment.config_snapshot.model_name = "test-model" mock_experiment.status = Mock() mock_experiment.status.value = "completed" mock_experiment.created_at = datetime.utcnow() mock_get_exp.return_value = mock_experiment mock_audit_trail = Mock() mock_audit_trail.run_id = run_id mock_audit_trail.status = Mock() mock_audit_trail.status.value = "verified" mock_audit_trail.config_record = Mock() mock_audit_trail.config_record.config_hash = "abcd1234" * 8 mock_audit_trail.result_record = Mock() mock_audit_trail.result_record.result_checksum = "efgh5678" * 8 mock_get_audit.return_value = mock_audit_trail # Generate report result = integration_service.generate_comprehensive_report( run_id, ReportType.FULL, ReportFormat.JSON, "test@example.com" ) assert result.success is True assert result.report_id is not None def test_generate_comprehensive_report_experiment_not_found(self, integration_service): """Test comprehensive report generation when experiment not found.""" run_id = str(uuid.uuid4()) with patch.object(integration_service.experiment_manager, 'get_experiment', return_value=None): result = integration_service.generate_comprehensive_report(run_id) assert result.success is False assert "not found" in result.error_message def test_generate_batch_reports(self, integration_service): """Test batch report generation.""" run_ids = [str(uuid.uuid4()) for _ in range(3)] with patch.object(integration_service.experiment_manager, 'get_experiment') as mock_get_exp, \ patch.object(integration_service.audit_manager, 'get_audit_trail') as mock_get_audit: # Setup mocks for all runs for run_id in run_ids: mock_experiment = Mock() mock_experiment.run_id = run_id mock_experiment.config_snapshot = Mock() mock_experiment.config_snapshot.model_name = "test-model" mock_experiment.status = Mock() mock_experiment.status.value = "completed" mock_audit_trail = Mock() mock_audit_trail.run_id = run_id mock_audit_trail.status = Mock() mock_audit_trail.status.value = "verified" mock_audit_trail.config_record = Mock() mock_audit_trail.config_record.config_hash = "abcd1234" * 8 mock_audit_trail.result_record = Mock() mock_audit_trail.result_record.result_checksum = "efgh5678" * 8 mock_get_exp.side_effect = [Mock(run_id=run_id, config_snapshot=Mock(), status=Mock(value="completed")) for run_id in run_ids] mock_get_audit.side_effect = [Mock(run_id=run_id, status=Mock(value="verified")) for run_id in run_ids] results = integration_service.generate_batch_reports(run_ids) assert len(results) == 3 for result in results: assert result.success is True def test_verify_report_integrity(self, integration_service): """Test report integrity verification.""" # Create temporary report file with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: report_data = { "export_metadata": { "exported_at": datetime.utcnow().isoformat(), "export_format": "json", "run_id": "test-run-id" }, "report_data": { "experiment": {"run_id": "test-run-id"}, "audit": {"run_id": "test-run-id"} } } json.dump(report_data, f) temp_file_path = f.name try: with patch.object(integration_service.experiment_manager, 'get_experiment', return_value=Mock()), \ patch.object(integration_service.audit_manager, 'get_audit_trail', return_value=Mock()): result = integration_service.verify_report_integrity(temp_file_path) assert 'file_path' in result assert 'validation_result' in result assert 'cross_validation' in result assert 'integrity_score' in result assert 'overall_valid' in result finally: Path(temp_file_path).unlink() def test_get_run_report_summary(self, integration_service): """Test getting run report summary.""" run_id = str(uuid.uuid4()) with patch.object(integration_service.experiment_manager, 'get_experiment') as mock_get_exp, \ patch.object(integration_service.audit_manager, 'get_audit_trail') as mock_get_audit, \ patch.object(integration_service.report_exporter, 'list_reports', return_value=[]): # Setup mocks mock_experiment = Mock() mock_experiment.run_id = run_id mock_experiment.experiment_name = "Test Experiment" mock_experiment.status = Mock() mock_experiment.status.value = "completed" mock_experiment.created_at = datetime.utcnow() mock_experiment.config_snapshot = Mock() mock_experiment.config_snapshot.model_name = "test-model" mock_experiment.config_snapshot.dataset_name = "test-dataset" mock_experiment.config_snapshot.attack_types = ["jailbreak"] mock_get_exp.return_value = mock_experiment mock_audit_trail = Mock() mock_audit_trail.run_id = run_id mock_audit_trail.status = Mock() mock_audit_trail.status.value = "verified" mock_audit_trail.integrity_level = Mock() mock_audit_trail.integrity_level.value = "high" mock_audit_trail.config_record = Mock() mock_audit_trail.config_record.config_hash = "abcd1234" * 8 mock_audit_trail.result_record = Mock() mock_audit_trail.result_record.result_checksum = "efgh5678" * 8 mock_get_audit.return_value = mock_audit_trail summary = integration_service.get_run_report_summary(run_id) assert 'run_id' in summary assert 'experiment_info' in summary assert 'audit_info' in summary assert 'report_summary' in summary assert summary['experiment_info']['run_id'] == run_id assert summary['audit_info']['run_id'] == run_id def test_validate_run_completeness(self, integration_service): """Test run completeness validation.""" run_id = str(uuid.uuid4()) with patch.object(integration_service.experiment_manager, 'get_experiment') as mock_get_exp, \ patch.object(integration_service.audit_manager, 'get_audit_trail') as mock_get_audit, \ patch.object(integration_service.report_exporter, 'list_reports', return_value=[]): # Setup complete mocks mock_experiment = Mock() mock_experiment.status = Mock() mock_experiment.status.value = "completed" mock_get_exp.return_value = mock_experiment mock_audit_trail = Mock() mock_audit_trail.status = Mock() mock_audit_trail.status.value = "verified" mock_get_audit.return_value = mock_audit_trail validation = integration_service.validate_run_completeness(run_id) assert 'run_id' in validation assert 'completeness_score' in validation assert 'overall_complete' in validation assert 'components' in validation assert validation['components']['experiment_complete'] is True assert validation['components']['audit_complete'] is True class TestGlobalFunctions: """Test global convenience functions.""" def test_get_report_generator(self): """Test getting global report generator.""" generator = get_report_generator() assert generator is not None assert isinstance(generator, ReportGenerator) def test_get_report_exporter(self): """Test getting global report exporter.""" exporter = get_report_exporter() assert exporter is not None assert isinstance(exporter, ReportExporter) def test_get_report_validator(self): """Test getting global report validator.""" validator = get_report_validator() assert validator is not None assert isinstance(validator, ReportValidator) def test_get_file_manager(self): """Test getting global file manager.""" manager = get_file_manager() assert manager is not None assert isinstance(manager, ReportFileManager) def test_get_report_integration_service(self): """Test getting global integration service.""" service = get_report_integration_service() assert service is not None assert isinstance(service, ReportIntegrationService) if __name__ == "__main__": pytest.main([__file__])