test_app.py

"""Test suite for Flask application endpoints and integration.

This module provides comprehensive tests for the Flask application including
route testing, model integration, caching behavior, and error handling.

Run with: python test_app.py or pytest test_app.py
"""

import json
import os
from pathlib import Path
import tempfile
import unittest
from unittest import mock

# Mock the environment before importing app to avoid initialization errors
with mock.patch.dict(os.environ, {'KEY': 'test_api_key_for_import'}):
    import app as app_module
    from app import Model, app, setup_cache


_SENSITIVE_REPORT_TEXT = 'PRIVATE PATIENT FINDING'


class TestFlaskApplication(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.test_client = app.test_client()
        app.config['TESTING'] = True

    def test_index_route_returns_html(self):
        response = self.test_client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertIn('text/html', response.content_type)

    def test_cache_stats_route(self):
        response = self.test_client.get('/cache/stats')
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.content_type, 'application/json')

        data = json.loads(response.data)
        self.assertIsInstance(data, dict)

    @mock.patch('app.model.predict')
    def test_predict_route_with_valid_data(self, mock_predict):
        mock_predict.return_value = {
            'segments': [{'type': 'body', 'content': 'test'}],
            'text': 'test output',
        }

        response = self.test_client.post('/predict', data='FINDINGS: Normal chest CT')
        self.assertEqual(response.status_code, 200)

        data = json.loads(response.data)
        self.assertIn('segments', data)
        self.assertIn('text', data)

    def test_predict_route_with_empty_data(self):
        response = self.test_client.post('/predict', data='')
        self.assertEqual(response.status_code, 400)

        data = json.loads(response.data)
        self.assertIn('error', data)
        self.assertEqual(data['error'], 'Empty input')
        self.assertIn('message', data)
        self.assertEqual(data['message'], 'Input text is required')
        self.assertIn('max_length', data)

    @mock.patch('app.model.predict')
    def test_predict_with_custom_headers(self, mock_predict):
        mock_predict.return_value = {'segments': [], 'text': 'test'}

        headers = {
            'X-Use-Cache': 'false',
            'X-Sample-ID': 'test_sample',
            'X-Model-ID': 'gemini-2.5-flash',
        }

        response = self.test_client.post(
            '/predict', data='Test report', headers=headers
        )
        self.assertEqual(response.status_code, 200)
        mock_predict.assert_called_once_with(
            'Test report', model_id='gemini-flash-latest'
        )

    @mock.patch('app.cache_manager.get_cached_result')
    def test_predict_with_cache_hit(self, mock_get_cached):
        sample_data = json.loads(Path('static/sample_reports.json').read_text())
        sample = sample_data['samples'][0]
        cached_response = {
            'segments': [{'type': 'body', 'content': 'cached'}],
            'text': 'cached result',
        }
        mock_get_cached.return_value = cached_response

        response = self.test_client.post(
            '/predict',
            data=sample['text'],
            headers={
                'X-Use-Cache': 'true',
                'X-Sample-ID': sample['id'],
            },
        )

        data = json.loads(response.data)
        self.assertTrue(data.get('from_cache'))
        self.assertIn('segments', data)

    def test_matching_catalog_sample_uses_prebuilt_cache(self):
        sample_data = json.loads(Path('static/sample_reports.json').read_text())
        sample = sample_data['samples'][0]
        cached_response = {'segments': [], 'text': 'cached result'}

        with (
            mock.patch.object(
                app_module.cache_manager,
                'get_cached_result',
                autospec=True,
                return_value=cached_response,
            ) as mock_get_cached,
            mock.patch.object(
                app_module.model, 'predict', autospec=True
            ) as mock_predict,
        ):
            response = self.test_client.post(
                '/predict',
                data=sample['text'],
                headers={
                    'X-Use-Cache': 'true',
                    'X-Sample-ID': sample['id'],
                },
            )

        self.assertEqual(response.status_code, 200)
        self.assertTrue(response.get_json()['from_cache'])
        mock_get_cached.assert_called_once()
        mock_predict.assert_not_called()

    def test_mismatched_catalog_sample_bypasses_cache(self):
        cached_response = {'segments': [], 'text': 'wrong cached result'}
        live_response = {'segments': [], 'text': 'live result'}

        with (
            mock.patch.object(
                app_module.cache_manager,
                'get_cached_result',
                autospec=True,
                return_value=cached_response,
            ) as mock_get_cached,
            mock.patch.object(
                app_module.cache_manager, 'cache_result', autospec=True
            ) as mock_cache_result,
            mock.patch.object(
                app_module.model,
                'predict',
                autospec=True,
                return_value=live_response,
            ) as mock_predict,
        ):
            response = self.test_client.post(
                '/predict',
                data='EXAMINATION: A report that is not the chest sample.',
                headers={
                    'X-Use-Cache': 'true',
                    'X-Sample-ID': 'chest_xray',
                },
            )

        self.assertEqual(response.status_code, 200)
        self.assertFalse(response.get_json().get('from_cache', False))
        mock_get_cached.assert_not_called()
        mock_predict.assert_called_once()
        mock_cache_result.assert_not_called()

    def test_unknown_sample_id_is_never_cached(self):
        live_response = {'segments': [], 'text': 'live result'}

        with (
            mock.patch.object(
                app_module.cache_manager,
                'get_cached_result',
                autospec=True,
            ) as mock_get_cached,
            mock.patch.object(
                app_module.cache_manager, 'cache_result', autospec=True
            ) as mock_cache_result,
            mock.patch.object(
                app_module.model,
                'predict',
                autospec=True,
                return_value=live_response,
            ),
        ):
            response = self.test_client.post(
                '/predict',
                data='EXAMINATION: Unknown sample.',
                headers={
                    'X-Use-Cache': 'true',
                    'X-Sample-ID': 'not_in_catalog',
                },
            )

        self.assertEqual(response.status_code, 200)
        mock_get_cached.assert_not_called()
        mock_cache_result.assert_not_called()

    def test_catalog_cache_miss_is_not_written_at_runtime(self):
        sample_data = json.loads(Path('static/sample_reports.json').read_text())
        sample = sample_data['samples'][0]
        live_response = {'segments': [], 'text': 'live result'}

        with (
            mock.patch.object(
                app_module.cache_manager,
                'get_cached_result',
                autospec=True,
                return_value=None,
            ),
            mock.patch.object(
                app_module.cache_manager, 'cache_result', autospec=True
            ) as mock_cache_result,
            mock.patch.object(
                app_module.model,
                'predict',
                autospec=True,
                return_value=live_response,
            ),
        ):
            response = self.test_client.post(
                '/predict',
                data=sample['text'],
                headers={
                    'X-Use-Cache': 'true',
                    'X-Sample-ID': sample['id'],
                },
            )

        self.assertEqual(response.status_code, 200)
        mock_cache_result.assert_not_called()

    def test_unapproved_model_returns_stable_client_error(self):
        with mock.patch.object(
            app_module.model,
            'predict',
            autospec=True,
            return_value={'segments': [], 'text': 'result'},
        ) as mock_predict:
            response = self.test_client.post(
                '/predict',
                data='FINDINGS: Normal chest.',
                headers={
                    'X-Use-Cache': 'false',
                    'X-Model-ID': 'unapproved-model',
                },
            )

        self.assertEqual(response.status_code, 400)
        self.assertEqual(response.get_json()['error'], 'Unsupported model')
        self.assertIn('message', response.get_json())
        mock_predict.assert_not_called()

    @mock.patch.dict(os.environ, {'MODEL_ID': 'configured-current-model'})
    def test_environment_configured_model_is_allowed(self):
        with mock.patch.object(
            app_module.model,
            'predict',
            autospec=True,
            return_value={'segments': [], 'text': 'result'},
        ) as mock_predict:
            response = self.test_client.post(
                '/predict',
                data='FINDINGS: Normal chest.',
                headers={
                    'X-Use-Cache': 'false',
                    'X-Model-ID': 'configured-current-model',
                },
            )

        self.assertEqual(response.status_code, 200)
        mock_predict.assert_called_once_with(
            'FINDINGS: Normal chest.', model_id='configured-current-model'
        )

    def test_legacy_pro_model_resolves_at_http_boundary(self):
        with mock.patch.object(
            app_module.model,
            'predict',
            autospec=True,
            return_value={'segments': [], 'text': 'result'},
        ) as mock_predict:
            response = self.test_client.post(
                '/predict',
                data='FINDINGS: Normal chest.',
                headers={
                    'X-Use-Cache': 'false',
                    'X-Model-ID': 'gemini-2.5-pro',
                },
            )

        self.assertEqual(response.status_code, 200)
        mock_predict.assert_called_once_with(
            'FINDINGS: Normal chest.', model_id='gemini-pro-latest'
        )


class TestModelClass(unittest.TestCase):

    @mock.patch.dict(os.environ, {'KEY': 'test_api_key'})
    def test_model_initialization_with_api_key(self):
        model = Model()
        self.assertEqual(model.gemini_api_key, 'test_api_key')
        self.assertIn('gemini-flash-latest', model._structurers)

    @mock.patch.dict(os.environ, {}, clear=True)
    def test_model_initialization_without_api_key(self):
        with self.assertRaises(ValueError) as context:
            Model()
        self.assertIn('KEY environment variable not set', str(context.exception))

    @mock.patch.dict(os.environ, {'KEY': 'test_key', 'MODEL_ID': 'custom-model'})
    def test_model_initialization_with_custom_model(self):
        model = Model()
        self.assertIn('custom-model', model._structurers)

    @mock.patch.dict(os.environ, {'KEY': 'test_key'})
    @mock.patch('app.RadiologyReportStructurer')
    def test_get_structurer_creates_new_instance(self, mock_structurer_class):
        model = Model()
        model._get_structurer('new-model')

        # Should be called twice: once for default, once for new model
        self.assertEqual(mock_structurer_class.call_count, 2)

    @mock.patch.dict(os.environ, {'KEY': 'test_key'})
    @mock.patch('app.RadiologyReportStructurer')
    def test_predict_calls_structurer(self, mock_structurer_class):
        mock_instance = mock.Mock()
        mock_instance.predict.return_value = {'result': 'test'}
        mock_structurer_class.return_value = mock_instance

        model = Model()
        result = model.predict('test data', 'test-model')

        mock_instance.predict.assert_called_once_with('test data')
        self.assertEqual(result, {'result': 'test'})

    @mock.patch.dict(os.environ, {'KEY': 'test_key'})
    @mock.patch('app.RadiologyReportStructurer')
    def test_predict_does_not_log_report_content(self, mock_structurer_class):
        mock_instance = mock.Mock()
        mock_instance.predict.return_value = {'text': _SENSITIVE_REPORT_TEXT}
        mock_structurer_class.return_value = mock_instance
        model = Model()

        with mock.patch.object(app_module.logger, 'info', autospec=True) as log:
            model.predict(_SENSITIVE_REPORT_TEXT, 'gemini-flash-latest')

        self.assertNotIn(_SENSITIVE_REPORT_TEXT, str(log.call_args_list))


class TestCacheSetup(unittest.TestCase):

    @mock.patch('os.path.exists')
    @mock.patch('shutil.copy2')
    @mock.patch('os.makedirs')
    def test_setup_cache_copies_existing_file(
        self, mock_makedirs, mock_copy, mock_exists
    ):
        mock_exists.side_effect = [True, False]

        with mock.patch.object(os.path, 'getsize', autospec=True, return_value=123):
            cache_dir = setup_cache()

        expected_cache_dir = tempfile.gettempdir() + '/cache'
        mock_makedirs.assert_called_once_with(expected_cache_dir, exist_ok=True)
        mock_copy.assert_called_once_with(
            'cache/sample_cache.json', expected_cache_dir + '/sample_cache.json'
        )
        self.assertEqual(cache_dir, expected_cache_dir)

    @mock.patch('os.path.exists')
    @mock.patch('os.makedirs')
    def test_setup_cache_handles_missing_source(self, mock_makedirs, mock_exists):
        mock_exists.return_value = False

        cache_dir = setup_cache()

        expected_cache_dir = tempfile.gettempdir() + '/cache'
        mock_makedirs.assert_called_once_with(expected_cache_dir, exist_ok=True)
        self.assertEqual(cache_dir, expected_cache_dir)


class TestErrorHandling(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.test_client = app.test_client()
        app.config['TESTING'] = True

    def setUp(self):
        # Suppress all logging during error tests to reduce noise
        import logging

        logging.disable(logging.CRITICAL)

    def tearDown(self):
        # Re-enable logging
        import logging

        logging.disable(logging.NOTSET)

    @mock.patch('app.model.predict')
    @mock.patch('app.logger')
    def test_predict_handles_type_error(self, mock_logger, mock_predict):
        mock_predict.side_effect = TypeError('Invalid type')

        response = self.test_client.post('/predict', data='Test data')
        self.assertEqual(response.status_code, 500)

        data = json.loads(response.data)
        self.assertEqual(data['error'], 'Internal processing error')
        self.assertIn('message', data)

    @mock.patch('app.model.predict')
    @mock.patch('app.logger')
    def test_predict_handles_provider_error(self, mock_logger, mock_predict):
        provider_error = RuntimeError(_SENSITIVE_REPORT_TEXT)
        processing_error = app_module.ReportProcessingError('Report processing failed')
        processing_error.__cause__ = provider_error
        mock_predict.side_effect = processing_error

        response = self.test_client.post('/predict', data='Test data')

        self.assertEqual(response.status_code, 502)
        data = response.get_json()
        self.assertEqual(data['error'], 'Processing unavailable')
        self.assertIn('message', data)
        self.assertNotIn(_SENSITIVE_REPORT_TEXT, response.get_data(as_text=True))
        self.assertNotIn(_SENSITIVE_REPORT_TEXT, str(mock_logger.method_calls))

    @mock.patch('app.model.predict')
    @mock.patch('app.logger')
    def test_predict_handles_general_exception(self, mock_logger, mock_predict):
        mock_predict.side_effect = Exception(_SENSITIVE_REPORT_TEXT)

        response = self.test_client.post('/predict', data='Test data')
        self.assertEqual(response.status_code, 500)

        data = json.loads(response.data)
        self.assertEqual(data['error'], 'Internal processing error')
        self.assertIn('message', data)
        self.assertNotIn(_SENSITIVE_REPORT_TEXT, response.get_data(as_text=True))
        self.assertNotIn(_SENSITIVE_REPORT_TEXT, str(mock_logger.method_calls))


class TestFrontendAndProjectConfiguration(unittest.TestCase):

    def test_demo_studio_contract_is_present_above_supporting_content(self):
        response = app.test_client().get('/')
        page = response.get_data(as_text=True)

        self.assertEqual(response.status_code, 200)
        self.assertIn('id="transformation-studio"', page)
        self.assertIn('id="findings-cards"', page)
        self.assertIn('id="studio-metrics"', page)
        self.assertIn('role="tablist"', page)
        self.assertIn('Research demonstration — not for clinical use', page)
        self.assertLess(
            page.index('id="transformation-studio"'),
            page.index('id="how-it-works"'),
        )

    def test_default_sample_load_is_cached_and_has_no_simulated_delay(self):
        script = Path('static/script.js').read_text()

        self.assertIn("DEFAULT_SAMPLE_ID: 'chest_xray'", script)
        self.assertIn('loadSampleReport(defaultSample, { scroll: false })', script)
        self.assertIn("headers['X-Use-Cache'] = 'true'", script)
        self.assertNotIn('Math.random() * 1000', script)
        self.assertNotIn('window.location.search.length', script)
        self.assertNotIn('window.location.hash.length', script)

    def test_finding_cards_are_lossless_and_keep_alternate_views(self):
        script = Path('static/script.js').read_text()
        template = Path('templates/index.html').read_text()

        self.assertIn('function renderFindingCards(segments)', script)
        self.assertIn('segments.forEach((segment, index)', script)
        self.assertIn('card.dataset.segmentIndex = String(index)', script)
        self.assertIn('No source span', script)
        self.assertIn('data-output-view="report"', template)
        self.assertIn('data-output-view="raw"', template)
        self.assertIn('id="prompt-toggle"', template)

    def test_langextract_links_and_sample_copy_are_professional(self):
        template = Path('templates/index.html').read_text()
        stylesheet = Path('static/style.css').read_text()

        self.assertIn('href="https://github.com/google/langextract"', template)
        self.assertIn('LangExtract on GitHub', template)
        self.assertIn('LangExtract release blog', template)
        self.assertGreaterEqual(template.count('rel="noopener noreferrer"'), 2)
        self.assertIn('<h2 id="sample-heading">Explore sample reports</h2>', template)
        self.assertIn('Choose a study.', template)
        self.assertNotIn('Explore a cached report', template)
        self.assertNotIn('Results appear instantly from the demo cache.', template)
        self.assertIn('.hero-attribution a:focus-visible', stylesheet)

    def test_sample_grid_and_report_default_preserve_responsive_fallbacks(self):
        script = Path('static/script.js').read_text()
        stylesheet = Path('static/style.css').read_text()
        template = Path('templates/index.html').read_text()

        self.assertIn('grid-template-columns: repeat(5, minmax(0, 1fr));', stylesheet)
        mobile_rules = stylesheet[stylesheet.index('@media (max-width: 768px)') :]
        self.assertIn('.sample-buttons {', mobile_rules)
        self.assertIn('display: flex;', mobile_rules)
        self.assertIn('overflow-x: auto;', mobile_rules)
        self.assertLess(
            template.index('id="report-tab"'), template.index('id="findings-tab"')
        )
        self.assertIn("let preferredStructuredView = 'report';", script)
        self.assertIn('selectOutputView(preferredStructuredView);', script)
        self.assertGreaterEqual(script.count('remember: true'), 3)
        self.assertIn(
            "selectOutputView(showRaw ? 'raw' : 'cards', { remember: true });",
            script,
        )
        self.assertIn("tab.getAttribute('aria-selected') === 'true'", script)
        self.assertLess(
            script.index(
                "selectOutputView('cards');",
                script.index('predictButton.addEventListener'),
            ),
            script.index(
                'setOutputViewsAvailable(false);',
                script.index('predictButton.addEventListener'),
            ),
        )
        self.assertNotIn('balanceByColumnCount', script)
        self.assertNotIn('BALANCE_DELAY', script)
        self.assertNotIn('RESIZE_DEBOUNCE', script)

    def test_mobile_and_reduced_motion_studio_rules_are_defined(self):
        script = Path('static/script.js').read_text()
        stylesheet = Path('static/style.css').read_text()

        self.assertIn('@media (max-width: 768px)', stylesheet)
        self.assertIn('.output-wrapper {', stylesheet)
        self.assertIn('order: 1;', stylesheet)
        self.assertIn('.input-wrapper {', stylesheet)
        self.assertIn('order: 2;', stylesheet)
        self.assertIn('@media (prefers-reduced-motion: reduce)', stylesheet)
        self.assertIn('overflow-x: hidden;', stylesheet)
        self.assertIn('function prefersReducedMotion()', script)
        self.assertIn("behavior: prefersReducedMotion() ? 'auto' : 'smooth'", script)

    def test_studio_fallback_and_accessibility_guards_are_present(self):
        script = Path('static/script.js').read_text()
        template = Path('templates/index.html').read_text()

        self.assertIn('function resetStudioMetrics()', script)
        self.assertGreaterEqual(script.count('resetStudioMetrics();'), 4)
        self.assertIn('showGrounding({ focusInput: false })', script)
        self.assertIn('sampleRunQueued = true', script)
        self.assertGreaterEqual(script.count('if (sampleRunQueued) return;'), 2)
        self.assertIn('let sampleLoadTimer = null', script)
        self.assertIn('function setOutputViewsAvailable(available)', script)
        self.assertIn('aria-controls="findings-cards"', template)
        self.assertIn('role="tabpanel"', template)
        self.assertIn('id="raw-toggle" tabindex="-1"', template)
        self.assertIn('aria-disabled="true"', template)

    def test_frontend_sanitizes_prompt_and_raw_fallback(self):
        script = Path('static/script.js').read_text()
        template = Path('templates/index.html').read_text()

        self.assertIn('DOMPurify.sanitize', script)
        self.assertIn('dompurify', template.lower())
        self.assertIn('pre.textContent', script)
        self.assertNotIn("rawOutput.innerHTML = '<pre", script)

    def test_langextract_floor_supports_resolver_params(self):
        project = Path('pyproject.toml').read_text()

        self.assertIn('"langextract>=1.6.0,<2.0.0"', project)

    def test_sample_preprocessing_is_idempotent_and_cache_is_complete(self):
        samples = json.loads(Path('static/sample_reports.json').read_text())['samples']
        cache = json.loads(Path('cache/sample_cache.json').read_text())

        for sample in samples:
            with self.subTest(sample=sample['id']):
                normalized = app_module.preprocess_report(sample['text'])
                self.assertEqual(app_module.preprocess_report(normalized), normalized)
                self.assertIn(f"sample_{sample['id']}", cache)

    def test_pytest_discovers_root_test_files_without_fallback(self):
        project = Path('pyproject.toml').read_text()

        self.assertIn('testpaths = ["test_app.py", "test_validation.py"]', project)


if __name__ == '__main__':
    unittest.main()