From 91f13311a93af88ba92c32298be32c1b193f060b Mon Sep 17 00:00:00 2001 From: Lana Date: Mon, 9 Sep 2024 17:56:13 -0400 Subject: [PATCH] add tests for pagination --- fhirclient/models/bundle.py | 1 - fhirclient/models/fhirsearch.py | 5 +- tests/models/fhirsearch_perform_iter_test.py | 168 ++++++++++--------- 3 files changed, 93 insertions(+), 81 deletions(-) diff --git a/fhirclient/models/bundle.py b/fhirclient/models/bundle.py index a2eaeca0..3922284f 100644 --- a/fhirclient/models/bundle.py +++ b/fhirclient/models/bundle.py @@ -63,7 +63,6 @@ def __next__(self): """ Returns the next BundleEntry in the Bundle's entry list using the internal iterator. """ return next(self._entry_iter) - def elementProperties(self): js = super(Bundle, self).elementProperties() js.extend([ diff --git a/fhirclient/models/fhirsearch.py b/fhirclient/models/fhirsearch.py index 7129bbfc..d1819d46 100644 --- a/fhirclient/models/fhirsearch.py +++ b/fhirclient/models/fhirsearch.py @@ -170,10 +170,7 @@ def perform_resources_iter(self, server) -> Iterator['Resource']: """ Performs the search by calling `perform`, then extracts all Bundle entries and returns an iterator of Resource instances. - # This method is designed to handle cases where the entries in the Bundle - # may be represented as either dictionaries (common in JSON-based data) - # or objects (when the data has been parsed into Python objects). - + :param server: The server against which to perform the search :returns: An iterator of Resource instances """ first_bundle = self.perform(server) diff --git a/tests/models/fhirsearch_perform_iter_test.py b/tests/models/fhirsearch_perform_iter_test.py index 1595c790..bd1239e8 100644 --- a/tests/models/fhirsearch_perform_iter_test.py +++ b/tests/models/fhirsearch_perform_iter_test.py @@ -1,90 +1,93 @@ +import io import json +import os import unittest -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch -from fhirclient.client import FHIRClient +from fhirclient import server + +from fhirclient.models import bundle from fhirclient.models.fhirsearch import FHIRSearch from fhirclient.models.bundle import Bundle class TestFHIRSearchIter(unittest.TestCase): def setUp(self): - # Set up the mock state and client - state = { - "app_id": "AppID", - "app_secret": "AppSecret", - "scope": "user/*.read", - "redirect": "http://test.invalid/redirect", - "patient_id": "PatientID", - "server": { - "base_uri": "http://test.invalid/", - "auth_type": "none", - "auth": { - "app_id": "AppId", - }, - }, - "launch_token": "LaunchToken", - "launch_context": { - "encounter": "EncounterID", - "patient": "PatientID", - }, - "jwt_token": "JwtToken", - } - - self.client = FHIRClient(state=state) - self.server_mock = self.client.server # Use the server from the client setup - - # Load the bundle example from a JSON file - with open('tests/data/examples/bundle-example.json') as f: - self.bundle = json.load(f) - - print(f"Loaded JSON: {self.bundle}") - - # Convert the dictionary to a Bundle object - self.bundle_mock = Bundle(self.bundle, strict=False) - - print(f"Bundle entries after initialization: {self.bundle_mock.entry}") - - # Ensure entries are correctly set - if not self.bundle_mock.entry: - print("Error: Bundle has no entries") - else: - print(f"Bundle entries: {self.bundle_mock.entry}") - - - - # Mock the server's request_json method to return the bundle's JSON - self.server_mock.request_json = MagicMock(return_value=self.bundle) - - # Ensure FHIRSearch.perform uses the mocked server - # FHIRSearch.perform = lambda self, server: Bundle(server.request_json(self.construct())) - FHIRSearch.perform = MagicMock(return_value=self.bundle_mock) - - self.search = FHIRSearch(resource_type='Patient') - - test_bundle_json = { - "resourceType": "Bundle", - "type": "searchset", - "entry": [ - { - "fullUrl": "http://example.com/fhir/Patient/1", - "resource": { - "resourceType": "Patient", - "id": "1", - "name": [{"family": "Doe", "given": ["John"]}] - } - } - ] - } - - self.bundle_mock = Bundle(test_bundle_json, strict=False) - print(f"Test Bundle entries: {self.bundle_mock.entry}") - - def test_perform_iter_single_bundle(self): - result = list(self.search.perform_iter(self.server_mock)) + self.mock_server = MockServer(tmpdir=os.path.join(os.path.dirname(__file__), '..', 'data', 'examples')) + self.search = FHIRSearch(resource_type='Bundle') + self.mock_bundle = self.instantiate_from('bundle-example.json') + + def instantiate_from(self, filename): + datadir = os.path.join(os.path.dirname(__file__), '..', 'data', 'examples') + with io.open(os.path.join(datadir, filename), 'r', encoding='utf-8') as handle: + js = json.load(handle) + self.assertEqual("Bundle", js["resourceType"]) + return bundle.Bundle(js) + + @patch('fhirclient.models.fhirsearch.FHIRSearch.perform') + def test_perform_iter_single_bundle(self, mock_perform): + mock_perform.return_value = self.mock_server.request_json('bundle-example.json') + mock_perform.return_value = self.mock_bundle + + result = list(self.search.perform_iter(self.mock_server)) + + self.assertEqual(len(result), 1) + self.assertIsInstance(result[0], Bundle) self.assertEqual(result[0].resource_type, 'Bundle') - self.assertEqual(result[0].id, self.bundle_mock.id) + self.assertEqual(result[0].id, self.mock_bundle.id) + @patch('fhirclient.models.fhirsearch.FHIRSearch.perform') + def test_perform_iter_no_first_bundle(self, mock_perform): + mock_perform.return_value = None + result = list(self.search.perform_iter(self.mock_server)) + self.assertEqual(result, []) + + @patch('fhirclient.models.fhirsearch.FHIRSearch.perform') + def test_perform_resources_iter_single_page(self, mock_perform): + mock_perform.return_value = self.mock_bundle + result = list(self.search.perform_resources_iter(self.mock_server)) + + # Assert that the iterator yields the correct resources + self.assertEqual(len(result), 2) + self.assertIsInstance(result[0], 'Patient') # Assuming first entry is a Patient (change as needed) + + # Assert that the resource type and ID are correct for the first entry + self.assertEqual(result[0].resource_type, "MedicationRequest") + self.assertEqual(result[0].id, "3123") + + # Assert that the second resource has the correct resource type and ID + self.assertEqual(result[1].resource_type, "Medication") + self.assertEqual(result[1].id, "example") + + # @patch('fhirclient.models.fhirsearch.FHIRSearch.perform') + # def test_perform_resources_iter_multiple_bundles(self, mock_perform): + # # Simulate multiple bundles using iter_pages, with resources in each + # mock_bundle_1 = Bundle(self.bundle_with_resources) + # mock_bundle_2 = Bundle(self.bundle_no_resources) # Second bundle has no resources + # mock_perform.side_effect = [mock_bundle_1, mock_bundle_2, None] + # + # # Test perform_resources_iter() + # result = list(self.search.perform_resources_iter(self.mock_server)) + # + # # Assert that the iterator yields all resources from both bundles + # self.assertEqual(len(result), 3) + # self.assertIsInstance(result[0], Resource) + # self.assertEqual(result[0].resource_type, "Patient") + # self.assertEqual(result[0].id, "1") + # + # @patch('fhirclient.models.fhirsearch.FHIRSearch.perform') + # def test_perform_resources_iter_empty_bundle(self, mock_perform): + # # Simulate perform() returning a bundle with no resources + # mock_perform.return_value = Bundle(self.bundle_no_resources) + # + # # Test perform_resources_iter() + # result = list(self.search.perform_resources_iter(self.mock_server)) + # + # # Assert that the result is an empty list + # self.assertEqual(result, []) + + +""" def test_perform_resources_iter_single_bundle_with_object_entries(self): entry1 = MagicMock(resource='Resource1') entry2 = MagicMock(resource='Resource2') @@ -168,4 +171,17 @@ def iter_pages_mock(bundle, fetch_next_page_func): [entry["resource"] for entry in bundle2_entries] self.assertEqual(result, expected_resources) +""" + +class MockServer(server.FHIRServer): + """ Reads local files. + """ + + def __init__(self, tmpdir: str): + super().__init__(None, base_uri='https://fhir.smarthealthit.org') + self.directory = tmpdir + def request_json(self, path, nosign=False): + assert path + with io.open(os.path.join(self.directory, path), encoding='utf-8') as handle: + return json.load(handle) \ No newline at end of file