FHIRFLARE-IG-Toolkit/tests/test_app.py

import unittest
import pytest
import os
import sys
import json
import tarfile
import shutil
from unittest.mock import patch, MagicMock
from flask import Flask, url_for
from flask.testing import FlaskClient

# Add the parent directory (/app) to sys.path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from app import app, db, ProcessedIg
from datetime import datetime

class TestFHIRFlareIGToolkit(unittest.TestCase):
    def setUp(self):
        # Configure the Flask app for testing
        app.config['TESTING'] = True
        app.config['WTF_CSRF_ENABLED'] = False  # Disable CSRF for testing
        app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
        app.config['FHIR_PACKAGES_DIR'] = 'test_packages'
        app.config['SECRET_KEY'] = 'test-secret-key'
        app.config['API_KEY'] = 'test-api-key'

        # Create the test packages directory
        os.makedirs(app.config['FHIR_PACKAGES_DIR'], exist_ok=True)

        # Create the Flask test client
        self.client = app.test_client()

        # Initialize the database
        with app.app_context():
            db.create_all()

    def tearDown(self):
        # Clean up the database and test packages directory
        with app.app_context():
            db.session.remove()
            db.drop_all()
        if os.path.exists(app.config['FHIR_PACKAGES_DIR']):
            shutil.rmtree(app.config['FHIR_PACKAGES_DIR'])

    # Helper method to create a mock .tgz file
    def create_mock_tgz(self, filename, content=None):
        tgz_path = os.path.join(app.config['FHIR_PACKAGES_DIR'], filename)
        with tarfile.open(tgz_path, "w:gz") as tar:
            if content:
                # Create a mock package.json file inside the .tgz
                import io
                package_json = io.BytesIO(json.dumps(content).encode('utf-8'))
                tarinfo = tarfile.TarInfo(name="package/package.json")
                tarinfo.size = len(package_json.getvalue())
                tar.addfile(tarinfo, package_json)
        return tgz_path

    # Test Case 1: Test Homepage Rendering
    def test_homepage(self):
        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'FHIRFLARE IG Toolkit', response.data)

    # Test Case 2: Test Import IG Page Rendering
    def test_import_ig_page(self):
        response = self.client.get('/import-ig')
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'Import IG', response.data)
        self.assertIn(b'Package Name', response.data)
        self.assertIn(b'Package Version', response.data)

    # Test Case 3: Test Import IG Form Submission (Success)
    @patch('services.import_package_and_dependencies')
    def test_import_ig_success(self, mock_import):
        mock_import.return_value = {
            'downloaded': True,
            'errors': [],
            'dependencies': [{'name': 'hl7.fhir.r4.core', 'version': '4.0.1'}]
        }
        response = self.client.post('/import-ig', data={
            'package_name': 'hl7.fhir.us.core',
            'package_version': '3.1.1'
        }, follow_redirects=True)
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'Successfully downloaded hl7.fhir.us.core#3.1.1', response.data)

    # Test Case 4: Test Import IG Form Submission (Failure)
    @patch('services.import_package_and_dependencies')
    def test_import_ig_failure(self, mock_import):
        mock_import.return_value = {
            'downloaded': False,
            'errors': ['Package not found'],
            'dependencies': []
        }
        response = self.client.post('/import-ig', data={
            'package_name': 'invalid.package',
            'package_version': '1.0.0'
        }, follow_redirects=True)
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'Failed to import invalid.package#1.0.0', response.data)

    # Test Case 5: Test Import IG Form Submission (Invalid Input)
    def test_import_ig_invalid_input(self):
        response = self.client.post('/import-ig', data={
            'package_name': 'invalid@package',  # Invalid format
            'package_version': '1.0.0'
        }, follow_redirects=True)
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'Invalid package name format', response.data)

    # Test Case 6: Test View IGs Page Rendering (No Packages)
    def test_view_igs_no_packages(self):
        response = self.client.get('/view-igs')
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'No packages downloaded yet', response.data)

    # Test Case 7: Test View IGs Page Rendering (With Packages)
    def test_view_igs_with_packages(self):
        # Create a mock .tgz file
        self.create_mock_tgz('hl7.fhir.us.core-3.1.1.tgz', {
            'name': 'hl7.fhir.us.core',
            'version': '3.1.1'
        })
        response = self.client.get('/view-igs')
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'hl7.fhir.us.core', response.data)
        self.assertIn(b'3.1.1', response.data)

    # Test Case 8: Test Process IG (Success)
    @patch('services.process_package_file')
    def test_process_ig_success(self, mock_process):
        mock_process.return_value = {
            'resource_types_info': [{'type': 'Patient'}],
            'must_support_elements': {'Patient': ['name']},
            'examples': {'Patient': ['example1.json']}
        }
        self.create_mock_tgz('hl7.fhir.us.core-3.1.1.tgz')
        response = self.client.post('/process-igs', data={
            'filename': 'hl7.fhir.us.core-3.1.1.tgz'
        }, follow_redirects=True)
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'Successfully processed hl7.fhir.us.core#3.1.1', response.data)

    # Test Case 9: Test Process IG (Invalid File)
    def test_process_ig_invalid_file(self):
        response = self.client.post('/process-igs', data={
            'filename': 'invalid-file.txt'
        }, follow_redirects=True)
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'Invalid package file', response.data)

    # Test Case 10: Test Delete IG (Success)
    def test_delete_ig_success(self):
        self.create_mock_tgz('hl7.fhir.us.core-3.1.1.tgz')
        response = self.client.post('/delete-ig', data={
            'filename': 'hl7.fhir.us.core-3.1.1.tgz'
        }, follow_redirects=True)
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'Deleted hl7.fhir.us.core-3.1.1.tgz', response.data)
        self.assertFalse(os.path.exists(os.path.join(app.config['FHIR_PACKAGES_DIR'], 'hl7.fhir.us.core-3.1.1.tgz')))

    # Test Case 11: Test Delete IG (File Not Found)
    def test_delete_ig_file_not_found(self):
        response = self.client.post('/delete-ig', data={
            'filename': 'nonexistent.tgz'
        }, follow_redirects=True)
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'File not found: nonexistent.tgz', response.data)

    # Test Case 12: Test Unload IG (Success)
    def test_unload_ig_success(self):
        with app.app_context():
            processed_ig = ProcessedIg(
                package_name='hl7.fhir.us.core',
                version='3.1.1',
                processed_date=datetime.now(),
                resource_types_info=[{'type': 'Patient'}]
            )
            db.session.add(processed_ig)
            db.session.commit()
            ig_id = processed_ig.id

        response = self.client.post('/unload-ig', data={
            'ig_id': ig_id
        }, follow_redirects=True)
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'Unloaded hl7.fhir.us.core#3.1.1', response.data)

    # Test Case 13: Test Unload IG (Invalid ID)
    def test_unload_ig_invalid_id(self):
        response = self.client.post('/unload-ig', data={
            'ig_id': '999'
        }, follow_redirects=True)
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'Package not found with ID: 999', response.data)

    # Test Case 14: Test View Processed IG Page
    def test_view_processed_ig(self):
        with app.app_context():
            processed_ig = ProcessedIg(
                package_name='hl7.fhir.us.core',
                version='3.1.1',
                processed_date=datetime.now(),
                resource_types_info=[{'type': 'Patient', 'is_profile': False}]
            )
            db.session.add(processed_ig)
            db.session.commit()
            ig_id = processed_ig.id

        response = self.client.get(f'/view-ig/{ig_id}')
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'View hl7.fhir.us.core#3.1.1', response.data)

    # Test Case 15: Test Push IGs Page Rendering
    def test_push_igs_page(self):
        response = self.client.get('/push-igs')
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'Push IGs to FHIR Server', response.data)
        self.assertIn(b'Live Console', response.data)

    # Test Case 16: Test API - Import IG (Success)
    @patch('services.import_package_and_dependencies')
    def test_api_import_ig_success(self, mock_import):
        mock_import.return_value = {
            'downloaded': True,
            'errors': [],
            'dependencies': [{'name': 'hl7.fhir.r4.core', 'version': '4.0.1'}]
        }
        response = self.client.post('/api/import-ig', 
            data=json.dumps({
                'package_name': 'hl7.fhir.us.core',
                'version': '3.1.1',
                'api_key': 'test-api-key'
            }),
            content_type='application/json'
        )
        self.assertEqual(response.status_code, 200)
        data = json.loads(response.data)
        self.assertEqual(data['status'], 'success')
        self.assertEqual(data['package_name'], 'hl7.fhir.us.core')

    # Test Case 17: Test API - Import IG (Invalid API Key)
    def test_api_import_ig_invalid_api_key(self):
        response = self.client.post('/api/import-ig', 
            data=json.dumps({
                'package_name': 'hl7.fhir.us.core',
                'version': '3.1.1',
                'api_key': 'wrong-api-key'
            }),
            content_type='application/json'
        )
        self.assertEqual(response.status_code, 401)
        data = json.loads(response.data)
        self.assertEqual(data['message'], 'Invalid API key')

    # Test Case 18: Test API - Import IG (Missing Parameters)
    def test_api_import_ig_missing_params(self):
        response = self.client.post('/api/import-ig', 
            data=json.dumps({
                'package_name': 'hl7.fhir.us.core',
                'api_key': 'test-api-key'
            }),
            content_type='application/json'
        )
        self.assertEqual(response.status_code, 400)
        data = json.loads(response.data)
        self.assertEqual(data['message'], 'Missing package_name or version')

    # Test Case 19: Test API - Push IG (Success)
    @patch('requests.put')
    def test_api_push_ig_success(self, mock_put):
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.raise_for_status.return_value = None
        mock_put.return_value = mock_response

        self.create_mock_tgz('hl7.fhir.us.core-3.1.1.tgz', {
            'name': 'hl7.fhir.us.core',
            'version': '3.1.1'
        })
        # Add a mock resource file
        with tarfile.open(os.path.join(app.config['FHIR_PACKAGES_DIR'], 'hl7.fhir.us.core-3.1.1.tgz'), "a:gz") as tar:
            resource_data = json.dumps({
                'resourceType': 'Patient',
                'id': 'example'
            }).encode('utf-8')
            import io
            resource_file = io.BytesIO(resource_data)
            tarinfo = tarfile.TarInfo(name="package/Patient-example.json")
            tarinfo.size = len(resource_data)
            tar.addfile(tarinfo, resource_file)

        response = self.client.post('/api/push-ig',
            data=json.dumps({
                'package_name': 'hl7.fhir.us.core',
                'version': '3.1.1',
                'fhir_server_url': 'http://test-server/fhir',
                'include_dependencies': False,
                'api_key': 'test-api-key'
            }),
            content_type='application/json',
            headers={'Accept': 'application/x-ndjson'}
        )
        self.assertEqual(response.status_code, 200)
        response_text = response.data.decode('utf-8')
        self.assertIn('"type": "start"', response_text)
        self.assertIn('"type": "success"', response_text)
        self.assertIn('"status": "success"', response_text)

    # Test Case 20: Test API - Push IG (Invalid API Key)
    def test_api_push_ig_invalid_api_key(self):
        response = self.client.post('/api/push-ig',
            data=json.dumps({
                'package_name': 'hl7.fhir.us.core',
                'version': '3.1.1',
                'fhir_server_url': 'http://test-server/fhir',
                'include_dependencies': False,
                'api_key': 'wrong-api-key'
            }),
            content_type='application/json',
            headers={'Accept': 'application/x-ndjson'}
        )
        self.assertEqual(response.status_code, 401)
        data = json.loads(response.data)
        self.assertEqual(data['message'], 'Invalid API key')

    # Test Case 21: Test API - Push IG (Package Not Found)
    def test_api_push_ig_package_not_found(self):
        response = self.client.post('/api/push-ig',
            data=json.dumps({
                'package_name': 'hl7.fhir.us.core',
                'version': '3.1.1',
                'fhir_server_url': 'http://test-server/fhir',
                'include_dependencies': False,
                'api_key': 'test-api-key'
            }),
            content_type='application/json',
            headers={'Accept': 'application/x-ndjson'}
        )
        self.assertEqual(response.status_code, 404)
        data = json.loads(response.data)
        self.assertEqual(data['message'], 'Package not found: hl7.fhir.us.core#3.1.1')

    # Test Case 22: Test Secret Key - CSRF Protection
    def test_secret_key_csrf(self):
        # Re-enable CSRF for this test
        app.config['WTF_CSRF_ENABLED'] = True
        response = self.client.post('/import-ig', data={
            'package_name': 'hl7.fhir.us.core',
            'package_version': '3.1.1'
        }, follow_redirects=True)
        self.assertEqual(response.status_code, 400)  # CSRF token missing

    # Test Case 23: Test Secret Key - Flash Messages
    def test_secret_key_flash_messages(self):
        # Set a flash message
        with self.client as client:
            with client.session_transaction() as sess:
                sess['_flashes'] = [('success', 'Test message')]
            response = self.client.get('/push-igs')
            self.assertEqual(response.status_code, 200)
            self.assertIn(b'Test message', response.data)

    # Test Case 24: Test Get Structure Definition (Success)
    def test_get_structure_definition_success(self):
        self.create_mock_tgz('hl7.fhir.us.core-3.1.1.tgz')
        with tarfile.open(os.path.join(app.config['FHIR_PACKAGES_DIR'], 'hl7.fhir.us.core-3.1.1.tgz'), "a:gz") as tar:
            sd_data = json.dumps({
                'snapshot': {'element': [{'id': 'Patient.name'}]}
            }).encode('utf-8')
            import io
            sd_file = io.BytesIO(sd_data)
            tarinfo = tarfile.TarInfo(name="package/StructureDefinition-Patient.json")
            tarinfo.size = len(sd_data)
            tar.addfile(tarinfo, sd_file)

        response = self.client.get('/get-structure?package_name=hl7.fhir.us.core&package_version=3.1.1&resource_type=Patient')
        self.assertEqual(response.status_code, 200)
        data = json.loads(response.data)
        self.assertIn('elements', data)
        self.assertEqual(data['elements'][0]['id'], 'Patient.name')

    # Test Case 25: Test Get Structure Definition (Not Found)
    def test_get_structure_definition_not_found(self):
        self.create_mock_tgz('hl7.fhir.us.core-3.1.1.tgz')
        response = self.client.get('/get-structure?package_name=hl7.fhir.us.core&package_version=3.1.1&resource_type=Observation')
        self.assertEqual(response.status_code, 404)
        data = json.loads(response.data)
        self.assertEqual(data['error'], "SD for 'Observation' not found.")

    # Test Case 26: Test Get Example Content (Success)
    def test_get_example_content_success(self):
        self.create_mock_tgz('hl7.fhir.us.core-3.1.1.tgz')
        with tarfile.open(os.path.join(app.config['FHIR_PACKAGES_DIR'], 'hl7.fhir.us.core-3.1.1.tgz'), "a:gz") as tar:
            example_data = json.dumps({
                'resourceType': 'Patient',
                'id': 'example'
            }).encode('utf-8')
            import io
            example_file = io.BytesIO(example_data)
            tarinfo = tarfile.TarInfo(name="package/example-Patient.json")
            tarinfo.size = len(example_data)
            tar.addfile(tarinfo, example_file)

        response = self.client.get('/get-example?package_name=hl7.fhir.us.core&package_version=3.1.1&filename=package/example-Patient.json')
        self.assertEqual(response.status_code, 200)
        data = json.loads(response.data)
        self.assertEqual(data['resourceType'], 'Patient')

    # Test Case 27: Test Get Example Content (Invalid Path)
    def test_get_example_content_invalid_path(self):
        self.create_mock_tgz('hl7.fhir.us.core-3.1.1.tgz')
        response = self.client.get('/get-example?package_name=hl7.fhir.us.core&package_version=3.1.1&filename=invalid/example.json')
        self.assertEqual(response.status_code, 400)
        data = json.loads(response.data)
        self.assertEqual(data['error'], 'Invalid example file path.')

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