test_validation.py

#!/usr/bin/env python3
"""End-to-end validation tests for radiology report structuring.

This module provides focused validation tests that verify the complete
RadiologyReportStructurer pipeline by comparing actual processing
results against known good cached outputs.

Typical usage example:

  # Run with unittest (built-in)
  python test_validation.py
  python -m unittest test_validation.py -v
  
  # Run with pytest (recommended for CI/CD)
  pytest test_validation.py -v
"""

import json
import os
import sys
import unittest
from typing import Any
from unittest import mock

import langextract as lx

# Add the current directory to path for imports
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from structure_report import RadiologyReportStructurer


_ORIGINAL_RESOLVER_ALIGN = lx.resolver.Resolver.align


class TestRadiologyReportEndToEnd(unittest.TestCase):
    """End-to-end tests for complete RadiologyReportStructurer pipeline."""

    cache_file: str
    sample_data: dict[str, Any]
    structurer: RadiologyReportStructurer

    @classmethod
    def setUpClass(cls):
        cls.cache_file = 'cache/sample_cache.json'
        cls.sample_data = cls._load_sample_cache()
        cls.structurer = RadiologyReportStructurer(
            api_key='test_key', model_id='gemini-2.5-flash'
        )

    @classmethod
    def _load_sample_cache(cls) -> dict[str, Any]:
        if not os.path.exists(cls.cache_file):
            raise FileNotFoundError(f'Sample cache file not found: {cls.cache_file}')

        with open(cls.cache_file, 'r', encoding='utf-8') as f:
            return json.load(f)

    def _validate_response_structure(self, response: dict[str, Any]) -> None:
        self.assertIn('segments', response)
        self.assertIn('text', response)
        self.assertIsInstance(response['segments'], list)
        self.assertIsInstance(response['text'], str)

    def _validate_successful_response(self, response: dict[str, Any]) -> None:
        self._validate_response_structure(response)
        self.assertGreater(len(response['segments']), 0)
        self.assertGreater(len(response['text']), 0)

        for segment in response['segments']:
            self._validate_segment_structure(segment)

    def _validate_segment_structure(self, segment: dict[str, Any]) -> None:
        required_fields = ['type', 'label', 'content', 'intervals']
        for field in required_fields:
            self.assertIn(field, segment)

        valid_types = ['prefix', 'body', 'suffix']
        self.assertIn(segment['type'], valid_types)

        if segment['intervals']:
            for interval in segment['intervals']:
                self.assertIn('startPos', interval)
                self.assertIn('endPos', interval)
                self.assertGreaterEqual(interval['startPos'], 0)
                self.assertGreater(interval['endPos'], interval['startPos'])

    @mock.patch('structure_report.lx.extract')
    def test_end_to_end_processing_pipeline(self, mock_extract):
        mock_result = mock.MagicMock()
        mock_result.extractions = []
        mock_extract.return_value = mock_result

        input_text = 'EXAMINATION: Chest CT\n\nFINDINGS: Normal lungs.\n\nIMPRESSION: No acute findings.'

        response = self.structurer.predict(input_text)

        self._validate_response_structure(response)

        mock_extract.assert_called_once()
        call_args = mock_extract.call_args
        self.assertEqual(call_args[1]['text_or_documents'], input_text)
        self.assertEqual(call_args[1]['model_id'], 'gemini-flash-latest')
        self.assertEqual(
            call_args[1]['resolver_params'],
            {
                'accept_match_lesser': False,
                'fuzzy_alignment_threshold': 0.50,
            },
        )

    def test_legacy_pro_model_resolves_to_rolling_alias(self):
        structurer = RadiologyReportStructurer(
            api_key='test_key', model_id='gemini-2.5-pro'
        )

        self.assertEqual(structurer.model_id, 'gemini-pro-latest')

    def test_current_model_id_is_preserved(self):
        structurer = RadiologyReportStructurer(
            api_key='test_key', model_id='gemini-3.5-flash'
        )

        self.assertEqual(structurer.model_id, 'gemini-3.5-flash')

    def test_all_cached_samples_validation(self):
        self.assertGreater(len(self.sample_data), 0, 'No samples found in cache')

        for sample_key, sample in self.sample_data.items():
            with self.subTest(sample=sample_key):
                self._validate_successful_response(sample)

    def test_error_handling_with_invalid_input(self):
        with self.assertRaises(ValueError) as context:
            self.structurer.predict('')
        self.assertIn('Report text cannot be empty', str(context.exception))

        with self.assertRaises(ValueError):
            self.structurer.predict('   \n\t  ')

    def test_error_handling_with_no_api_key(self):
        error_structurer = RadiologyReportStructurer(api_key=None)

        with mock.patch.object(
            lx,
            'extract',
            autospec=True,
            side_effect=RuntimeError('sensitive provider detail'),
        ):
            with self.assertRaisesRegex(RuntimeError, 'Report processing failed'):
                error_structurer.predict('EXAMINATION: Test')

    def test_predict_does_not_patch_global_resolver(self):
        new_structurer = RadiologyReportStructurer()

        with mock.patch.object(lx, 'extract', autospec=True) as mock_extract:
            mock_result = mock.MagicMock()
            mock_result.extractions = []
            mock_extract.return_value = mock_result
            new_structurer.predict('EXAMINATION: Test')

        self.assertIs(lx.resolver.Resolver.align, _ORIGINAL_RESOLVER_ALIGN)

    def test_resolver_settings_reject_partial_exact_matches(self):
        source_text = 'No acute disease.'
        extraction_text = 'No acute disease absent invented words words words words.'

        accepted = list(
            lx.resolver.Resolver().align(
                [lx.data.Extraction('finding', extraction_text)],
                source_text,
                token_offset=0,
                accept_match_lesser=True,
                fuzzy_alignment_threshold=0.50,
            )
        )
        rejected = list(
            lx.resolver.Resolver().align(
                [lx.data.Extraction('finding', extraction_text)],
                source_text,
                token_offset=0,
                accept_match_lesser=False,
                fuzzy_alignment_threshold=0.50,
            )
        )

        self.assertEqual(
            accepted[0].alignment_status, lx.data.AlignmentStatus.MATCH_LESSER
        )
        self.assertIsNone(rejected[0].alignment_status)
        self.assertIsNone(rejected[0].char_interval)

    def test_section_mapping_core_functionality(self):
        self.assertEqual(
            self.structurer._map_section('findings_prefix'),
            self.structurer._map_section('findings_prefix'),
        )

        self.assertIsNone(self.structurer._map_section('invalid_section'))
        self.assertIsNone(self.structurer._map_section(''))

    def test_exam_prefix_stripping(self):
        self.assertEqual(
            self.structurer._strip_exam_prefix('EXAMINATION: Chest CT'), 'Chest CT'
        )

        self.assertEqual(
            self.structurer._strip_exam_prefix('Normal findings'), 'Normal findings'
        )


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