import pytest import io import numpy as np from unittest.mock import patch, MagicMock def test_detect_unauthorized(client): """Test that detection endpoints require authorization.""" response = client.post('/api/detect') assert response.status_code == 401 assert 'Unauthorized' in response.get_json()['error'] response = client.post('/api/detect/deepfake') assert response.status_code == 401 assert 'Unauthorized' in response.get_json()['error'] def test_detect_no_image(client): """Test /api/detect with authorized session but no image.""" with client.session_transaction() as sess: sess['logged_in'] = True sess['role'] = 'admin' response = client.post('/api/detect') assert response.status_code == 400 assert 'No image provided' in response.get_json()['error'] def test_detect_success(client, app): """Test /api/detect successfully detects morphing.""" with client.session_transaction() as sess: sess['logged_in'] = True sess['role'] = 'admin' # Create a mock image file data = { 'image': (io.BytesIO(b"dummy image data"), 'test.jpg'), 'mode': 'fast', 'detection_type': 'morph' } with patch('src.routes.detection.secure_filename', return_value='test.jpg'), \ patch('src.routes.detection.os.path.join', return_value='static/uploads/test.jpg'), \ patch('src.routes.detection.open', create=True) as mock_open: # Mocking file save mock_file = MagicMock() data['image'] = (mock_file, 'test.jpg') response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 200 json_data = response.get_json() assert json_data['is_morphed'] is True assert json_data['confidence'] == 0.88 assert json_data['tier'] == 'Tier1_Fast' app.mg_api.detect_morph.assert_called_once() def test_detect_deepfake_standalone_success(client, app): """Test /api/detect/deepfake standalone endpoint.""" with client.session_transaction() as sess: sess['logged_in'] = True data = { 'image': (io.BytesIO(b"dummy image data"), 'test.jpg') } with patch('src.routes.detection.secure_filename', return_value='test.jpg'), \ patch('src.routes.detection.os.path.join', return_value='static/uploads/test.jpg'): response = client.post('/api/detect/deepfake', data=data, content_type='multipart/form-data') assert response.status_code == 200 json_data = response.get_json() assert json_data['is_deepfake'] is True assert json_data['deepfake_score'] == 0.95 assert 'artifacts' in json_data app.deepfake_detector.detect.assert_called_once() def test_detect_invalid_file_extension(client): """Test /api/detect returns 400 for invalid file extensions.""" with client.session_transaction() as sess: sess['logged_in'] = True data = { 'image': (io.BytesIO(b"dummy data"), 'test.txt') } response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 400 assert 'Invalid file type' in response.get_json()['error'] def test_detect_invalid_mode(client): """Test /api/detect returns 400 for invalid modes.""" with client.session_transaction() as sess: sess['logged_in'] = True data = { 'image': (io.BytesIO(b"dummy data"), 'test.jpg'), 'mode': 'super_deep' } response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 400 assert 'Invalid mode' in response.get_json()['error'] def test_detect_v1_success(client, app): """Test /api/v1/detect successfully detects morphing.""" with client.session_transaction() as sess: sess['logged_in'] = True data = { 'image': (io.BytesIO(b"dummy image data"), 'test.jpg'), 'mode': 'fast' } with patch('src.routes.detection.secure_filename', return_value='test.jpg'), \ patch('src.routes.detection.os.path.join', return_value='static/uploads/test.jpg'): response = client.post('/api/v1/detect', data=data, content_type='multipart/form-data') assert response.status_code == 200 json_data = response.get_json() assert json_data['is_morphed'] is True def test_ctm_veto_override_tier1(client, app, monkeypatch): """Test Tier 1 overrules CTM when Tier 1 confidence > 0.90 and CTM disagrees with low confidence < 0.85.""" monkeypatch.setenv('CTM_ENABLED', 'true') app.mg_api.detect_morph.return_value = { 'is_morphed': True, 'confidence': 0.95, 'processing_time': 0.05 } with client.session_transaction() as sess: sess['logged_in'] = True data = { 'image': (io.BytesIO(b"dummy image data"), 'veto_test.jpg'), 'mode': 'deep_forensics', 'detection_type': 'morph' } with patch('src.routes.detection.secure_filename', return_value='veto_test.jpg'), \ patch('src.routes.detection.os.path.join', return_value='static/uploads/veto_test.jpg'), \ patch('src.models.ctm_forensic_agent.get_ctm_agent') as mock_get_ctm: mock_ctm_agent = MagicMock() mock_ctm_agent.analyze.return_value = MagicMock( is_morphed=False, confidence=0.65, forensic_steps=16, attention_regions=['nose', 'eyes'], evidence_video_path=None ) mock_get_ctm.return_value = mock_ctm_agent response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 200 json_data = response.get_json() assert json_data['is_morphed'] is True assert json_data['confidence'] == 0.95 assert json_data['tier'] == 'Tier1_Fast_Overruled_CTM' assert json_data['forensic_steps'] == 16 assert json_data['attention_regions'] == ['nose', 'eyes'] def test_ctm_veto_boundary_conditions(client, app, monkeypatch): """Test boundary conditions: Tier 1 confidence <= 0.90 does NOT veto CTM.""" monkeypatch.setenv('CTM_ENABLED', 'true') # Tier 1 confidence is 0.89 (<= 0.90 threshold) app.mg_api.detect_morph.return_value = { 'is_morphed': True, 'confidence': 0.89, 'processing_time': 0.05 } with client.session_transaction() as sess: sess['logged_in'] = True data = { 'image': (io.BytesIO(b"boundary image data"), 'boundary_test.jpg'), 'mode': 'deep_forensics', 'detection_type': 'morph' } with patch('src.routes.detection.secure_filename', return_value='boundary_test.jpg'), \ patch('src.routes.detection.os.path.join', return_value='static/uploads/boundary_test.jpg'), \ patch('src.models.ctm_forensic_agent.get_ctm_agent') as mock_get_ctm: mock_ctm_agent = MagicMock() mock_ctm_agent.analyze.return_value = MagicMock( is_morphed=False, confidence=0.70, forensic_steps=16, attention_regions=['nose', 'eyes'], evidence_video_path=None, reasoning_chain=[] ) mock_get_ctm.return_value = mock_ctm_agent response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 200 json_data = response.get_json() # Should respect CTM verdict because Tier 1 confidence (0.89) was not > 0.90 assert json_data['is_morphed'] is False assert json_data['confidence'] == 0.70 assert json_data['tier'] == 'Tier2_Forensic' def test_ctm_not_overruled_when_ctm_confident(client, app, monkeypatch): """Test CTM verdict is respected when CTM confidence is >= 0.85.""" monkeypatch.setenv('CTM_ENABLED', 'true') app.mg_api.detect_morph.return_value = { 'is_morphed': True, 'confidence': 0.92, 'processing_time': 0.05 } with client.session_transaction() as sess: sess['logged_in'] = True data = { 'image': (io.BytesIO(b"dummy image data"), 'confident_ctm.jpg'), 'mode': 'deep_forensics', 'detection_type': 'morph' } with patch('src.routes.detection.secure_filename', return_value='confident_ctm.jpg'), \ patch('src.routes.detection.os.path.join', return_value='static/uploads/confident_ctm.jpg'), \ patch('src.models.ctm_forensic_agent.get_ctm_agent') as mock_get_ctm: mock_ctm_agent = MagicMock() mock_ctm_agent.analyze.return_value = MagicMock( is_morphed=False, confidence=0.91, forensic_steps=16, attention_regions=['forehead'], evidence_video_path=None ) mock_get_ctm.return_value = mock_ctm_agent response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 200 json_data = response.get_json() assert json_data['is_morphed'] is False assert json_data['confidence'] == 0.91 assert json_data['tier'] == 'Tier2_Forensic' assert json_data['forensic_steps'] == 16 assert 'reasoning_chain' in json_data def test_ctm_early_exit_convergence(client, app, monkeypatch): """Test CTM triggers early-exit convergence when confidence > 0.99 at early tick.""" monkeypatch.setenv('CTM_ENABLED', 'true') app.mg_api.detect_morph.return_value = { 'is_morphed': True, 'confidence': 0.60, 'processing_time': 0.04 } with client.session_transaction() as sess: sess['logged_in'] = True data = { 'image': (io.BytesIO(b"early exit image"), 'decisive_portrait.jpg'), 'mode': 'deep_forensics', 'detection_type': 'morph' } with patch('src.routes.detection.secure_filename', return_value='decisive_portrait.jpg'), \ patch('src.routes.detection.os.path.join', return_value='static/uploads/decisive_portrait.jpg'), \ patch('src.models.ctm_forensic_agent.get_ctm_agent') as mock_get_ctm: mock_ctm_agent = MagicMock() mock_ctm_agent.analyze.return_value = MagicMock( is_morphed=True, confidence=0.995, forensic_steps=6, # Converged at tick 6 before reaching 16 attention_regions=['nose_bridge', 'eyes'], evidence_video_path=None, reasoning_chain=[ {'tick': i, 'focus': 'Nose Bridge', 'observation': 'Decisive artifact', 'confidence': 0.995} for i in range(1, 7) ] ) mock_get_ctm.return_value = mock_ctm_agent response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 200 json_data = response.get_json() assert json_data['is_morphed'] is True assert json_data['confidence'] == 0.995 assert json_data['forensic_steps'] == 6 assert len(json_data['reasoning_chain']) == 6 def test_fast_screening_zero_retention(client, app): """Test fast screening with zero-retention purges upload file.""" app.mg_api.detect_morph.return_value = { 'is_morphed': True, 'confidence': 0.95, 'processing_time': 0.045 } with client.session_transaction() as sess: sess['logged_in'] = True sess['role'] = 'admin' data = { 'image': (io.BytesIO(b"portrait binary bytes"), 'passenger_portrait.jpg'), 'mode': 'fast', 'zero_retention': 'true' } with patch('src.routes.detection.os.unlink') as mock_unlink, \ patch('src.routes.detection.os.path.exists', return_value=True): response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 200 json_data = response.get_json() assert json_data['tier'] == 'Tier1_Fast' assert json_data['confidence'] == 0.95 # Verify file unlink was called mock_unlink.assert_called_once() def test_detect_file_too_large(client): """Test /api/detect rejects files exceeding 10MB.""" with client.session_transaction() as sess: sess['logged_in'] = True sess['role'] = 'admin' # Create oversized bytes stream (>10MB) large_data = io.BytesIO(b"0" * (10 * 1024 * 1024 + 1024)) data = { 'image': (large_data, 'large_file.jpg'), 'mode': 'fast' } response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 400 json_data = response.get_json() assert 'exceeds maximum allowed limit' in json_data['error'] assert json_data['code'] == 'FILE_TOO_LARGE' def test_generative_ai_exif_provenance(): """Test deepfake detector identifies generative AI software signatures in EXIF.""" from src.deepfake.deepfake_detector import DeepfakeDetectorAPI detector = DeepfakeDetectorAPI.__wrapped__ if hasattr(DeepfakeDetectorAPI, '__wrapped__') else DeepfakeDetectorAPI() mock_exif = { 305: "Midjourney v6.0 Diffusion" } with patch('PIL.Image.open') as mock_img_open, \ patch('cv2.imread') as mock_imread, \ patch.object(detector, 'preprocess') as mock_prep, \ patch.object(detector, 'detector') as mock_model: mock_pil = MagicMock() mock_pil.getexif.return_value = mock_exif mock_img_open.return_value = mock_pil mock_cv_img = np.zeros((299, 299, 3), dtype=np.uint8) mock_imread.return_value = mock_cv_img import torch mock_prep.return_value = torch.zeros((1, 3, 299, 299)) mock_model.return_value = torch.tensor([[0.2, 0.8]]) mock_model.id2label = {0: 'REAL', 1: 'FAKE'} res = detector.detect("mock_ai_img.jpg") assert res['metadata_verdict'] == "generative_ai_software" assert res['is_deepfake'] is True assert any(step['focus'] == 'Metadata Provenance' for step in res['reasoning_chain']) def test_domain_precedence_both_mode(client, app): """Test in both mode that confirmed deepfake forces is_morphed=False.""" app.mg_api.detect_morph.return_value = { 'is_morphed': True, 'confidence': 0.70, 'processing_time': 0.05 } mock_df_detector = MagicMock() mock_df_detector.detect.return_value = { 'is_deepfake': True, 'confidence': 0.94, 'deepfake_score': 0.94, 'artifacts': {'boundary': 0.8}, 'reasoning_chain': [{'focus': 'Metadata Provenance', 'observation': 'EXIF: generative_ai_software'}] } app.deepfake_detector = mock_df_detector with client.session_transaction() as sess: sess['logged_in'] = True sess['role'] = 'admin' data = { 'image': (io.BytesIO(b"image bytes"), 'ai_face.jpg'), 'detection_type': 'both' } with patch('src.routes.detection.secure_filename', return_value='ai_face.jpg'), \ patch('src.routes.detection.os.path.join', return_value='static/uploads/ai_face.jpg'): response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 200 json_data = response.get_json() assert json_data['is_deepfake'] is True assert json_data['is_morphed'] is False assert json_data['verdict'] == 'AI_GENERATED' assert json_data['classification'] == 'deepfake' def test_telemetry_metric_logging_resilience(client, app): """Test telemetry metric recording succeeds and handles DB failure gracefully.""" app.mg_api.detect_morph.return_value = { 'is_morphed': False, 'confidence': 0.98, 'processing_time': 0.035 } with client.session_transaction() as sess: sess['logged_in'] = True sess['role'] = 'admin' data = { 'image': (io.BytesIO(b"valid portrait test data"), 'clean_portrait.jpg'), 'mode': 'fast' } with patch('src.routes.detection.log_detection_metrics') as mock_log_metrics, \ patch('src.routes.detection.secure_filename', return_value='clean_portrait.jpg'), \ patch('src.routes.detection.os.path.join', return_value='static/uploads/clean_portrait.jpg'): response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 200 json_data = response.get_json() assert json_data['is_morphed'] is False assert 'image_hash' in json_data # Verify log_detection_metrics was called with precomputed image_hash mock_log_metrics.assert_called_once() kwargs = mock_log_metrics.call_args.kwargs assert kwargs['image_hash'] == json_data['image_hash'] assert kwargs['detection_tier'] == 'Tier1_Fast' assert kwargs['processing_time_ms'] == 35.0 # Verify that if log_detection_metrics raises an exception, the HTTP response still returns 200 data2 = { 'image': (io.BytesIO(b"valid portrait test data"), 'clean_portrait.jpg'), 'mode': 'fast' } with patch('src.routes.detection.log_detection_metrics', side_effect=RuntimeError("DB Connection Lost")), \ patch('src.routes.detection.secure_filename', return_value='clean_portrait.jpg'), \ patch('src.routes.detection.os.path.join', return_value='static/uploads/clean_portrait.jpg'): response = client.post('/api/detect', data=data2, content_type='multipart/form-data') assert response.status_code == 200 def test_zero_knowledge_blockchain_logging(client, app): """Test detection with blockchain_logging=true produces tx_id with zero-knowledge privacy.""" app.mg_api.detect_morph.return_value = { 'is_morphed': True, 'confidence': 0.94, 'processing_time': 0.04 } with client.session_transaction() as sess: sess['logged_in'] = True sess['role'] = 'admin' data = { 'image': (io.BytesIO(b"blockchain privacy test image"), 'audit_portrait.jpg'), 'mode': 'fast', 'blockchain_logging': 'true' } mock_logger = MagicMock() mock_logger.log_result.return_value = "0x" + "a" * 64 with patch('src.routes.detection.secure_filename', return_value='audit_portrait.jpg'), \ patch('src.routes.detection.os.path.join', return_value='static/uploads/audit_portrait.jpg'), \ patch('src.verification.blockchain_store.BlockchainLogger', return_value=mock_logger): response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 200 json_data = response.get_json() assert json_data['blockchain_logged'] is True assert json_data['blockchain_tx_id'] == "0x" + "a" * 64 # Verify payload passed to blockchain contains only hashes and metadata, zero raw images/PII mock_logger.log_result.assert_called_once() record = mock_logger.log_result.call_args[0][0] assert 'image_hash' in record assert len(record['image_hash']) == 64 # Full SHA-256 assert 'raw_image' not in record assert 'passenger_name' not in record assert 'name' not in record def test_resilient_local_blockchain_fallback(client, app, tmp_path): """Test blockchain logging seamlessly falls back to local JSON when RPC is unavailable.""" from src.verification.blockchain_store import BlockchainLogger app.mg_api.detect_morph.return_value = { 'is_morphed': True, 'confidence': 0.91, 'processing_time': 0.04 } with client.session_transaction() as sess: sess['logged_in'] = True sess['role'] = 'admin' data = { 'image': (io.BytesIO(b"offline fallback test image"), 'offline_portrait.jpg'), 'mode': 'fast', 'blockchain_logging': 'true' } # Instantiate real BlockchainLogger without active web3 connection logger = BlockchainLogger(blockchain_endpoint="http://localhost:9999") # Unreachable port assert logger.connected is False with patch('src.routes.detection.secure_filename', return_value='offline_portrait.jpg'), \ patch('src.routes.detection.os.path.join', return_value='static/uploads/offline_portrait.jpg'), \ patch('src.verification.blockchain_store.BlockchainLogger', return_value=logger): response = client.post('/api/detect', data=data, content_type='multipart/form-data') assert response.status_code == 200 json_data = response.get_json() assert json_data['blockchain_logged'] is True assert json_data['blockchain_tx_id'].startswith('local_record_')