add tests for pagination

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([

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)

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)