Spaces:

juanquy
/

MorphGuard

Running

MorphGuard / tests /test_detection_api.py

feat(web3): implement resilient zero-loss local blockchain fallback logging (Story 3.3)

7cfebff 2 months ago

21.1 kB

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_')

	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_')