File size: 7,377 Bytes
5372c12 09daffd 5372c12 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 |
import unittest
from unittest.mock import patch, Mock
import json
from io import StringIO
from contextlib import redirect_stdout
from utils.meldrx import MeldRxAPI # Import the class from meldrx.py
class TestMeldRxAPI(unittest.TestCase):
def setUp(self):
"""Set up test fixtures before each test."""
self.client_id = "test_client_id"
self.client_secret = "test_client_secret"
self.workspace_id = "test_workspace_id"
self.patient_id = "test_patient_id"
self.api = MeldRxAPI(self.client_id, self.client_secret, self.workspace_id)
def tearDown(self):
"""Clean up after each test."""
self.api.close()
@patch('meldrx.requests.Session.post')
def test_authenticate_success(self, mock_post):
"""Test successful authentication."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {"access_token": "test_token"}
mock_post.return_value = mock_response
result = self.api.authenticate()
self.assertTrue(result)
self.assertEqual(self.api.access_token, "test_token")
@patch('meldrx.requests.Session.post')
def test_authenticate_failure(self, mock_post):
"""Test authentication failure."""
mock_response = Mock()
mock_response.status_code = 401
mock_response.raise_for_status.side_effect = requests.exceptions.HTTPError("Unauthorized")
mock_post.return_value = mock_response
with redirect_stdout(StringIO()) as output:
result = self.api.authenticate()
self.assertFalse(result)
self.assertIsNone(self.api.access_token)
self.assertIn("Authentication failed: Unauthorized", output.getvalue())
@patch('meldrx.requests.Session.get')
@patch('meldrx.MeldRxAPI.authenticate', return_value=True)
def test_get_patients_success(self, mock_auth, mock_get):
"""Test successful retrieval of patients from FHIR API."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.text = json.dumps({"resourceType": "Bundle", "entry": [{"resource": {"id": self.patient_id}}]})
mock_response.json.return_value = json.loads(mock_response.text)
mock_get.return_value = mock_response
result = self.api.get_patients()
self.assertIsNotNone(result)
self.assertEqual(result["entry"][0]["resource"]["id"], self.patient_id)
@patch('meldrx.requests.Session.get')
@patch('meldrx.MeldRxAPI.authenticate', return_value=False)
def test_get_patients_auth_failure(self, mock_auth, mock_get):
"""Test patient retrieval fails due to authentication."""
with redirect_stdout(StringIO()) as output:
result = self.api.get_patients()
self.assertIsNone(result)
mock_get.assert_not_called()
self.assertIn("Cannot proceed without authentication", output.getvalue())
@patch('meldrx.requests.Session.post')
@patch('meldrx.MeldRxAPI.authenticate', return_value=True)
def test_create_virtual_workspace_success(self, mock_auth, mock_post):
"""Test successful creation of a virtual workspace."""
mock_response = Mock()
mock_response.status_code = 201
mock_post.return_value = mock_response
result = self.api.create_virtual_workspace()
self.assertTrue(result)
@patch('meldrx.requests.Session.get')
@patch('meldrx.MeldRxAPI.authenticate', return_value=True)
def test_get_mips_patients_success(self, mock_auth, mock_get):
"""Test successful retrieval of patients from MIPS API."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.text = json.dumps({"resourceType": "Bundle", "entry": [{"resource": {"id": self.patient_id}}]})
mock_response.json.return_value = json.loads(mock_response.text)
mock_get.return_value = mock_response
result = self.api.get_mips_patients()
self.assertIsNotNone(result)
self.assertEqual(result["entry"][0]["resource"]["id"], self.patient_id)
@patch('meldrx.requests.Session.get')
@patch('meldrx.MeldRxAPI.authenticate', return_value=True)
def test_get_mips_patient_by_id_success(self, mock_auth, mock_get):
"""Test successful retrieval of a patient by ID from MIPS API."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.text = json.dumps({"resourceType": "Patient", "id": self.patient_id})
mock_response.json.return_value = json.loads(mock_response.text)
mock_get.return_value = mock_response
result = self.api.get_mips_patient_by_id(self.patient_id)
self.assertIsNotNone(result)
self.assertEqual(result["id"], self.patient_id)
@patch('meldrx.requests.Session.get')
@patch('meldrx.MeldRxAPI.authenticate', return_value=True)
def test_get_mips_encounters_success(self, mock_auth, mock_get):
"""Test successful retrieval of encounters from MIPS API."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.text = json.dumps({"resourceType": "Bundle", "entry": [{"resource": {"id": "enc1"}}]})
mock_response.json.return_value = json.loads(mock_response.text)
mock_get.return_value = mock_response
result = self.api.get_mips_encounters(self.patient_id)
self.assertIsNotNone(result)
self.assertEqual(result["entry"][0]["resource"]["id"], "enc1")
@patch('meldrx.requests.Session.put')
@patch('meldrx.MeldRxAPI.authenticate', return_value=True)
def test_update_fhir_patient_success(self, mock_auth, mock_put):
"""Test successful update of patient data in FHIR API."""
mock_response = Mock()
mock_response.status_code = 200
mock_put.return_value = mock_response
patient_data = {"resourceType": "Patient", "id": self.patient_id, "name": [{"family": "Doe"}]}
result = self.api.update_fhir_patient(self.patient_id, patient_data)
self.assertTrue(result)
@patch('meldrx.requests.Session.put')
@patch('meldrx.MeldRxAPI.authenticate', return_value=True)
def test_update_mips_patient_success(self, mock_auth, mock_put):
"""Test successful update of patient data in MIPS API."""
mock_response = Mock()
mock_response.status_code = 200
mock_put.return_value = mock_response
patient_data = {"resourceType": "Patient", "id": self.patient_id, "name": [{"family": "Doe"}]}
result = self.api.update_mips_patient(self.patient_id, patient_data)
self.assertTrue(result)
@patch('meldrx.requests.Session.put')
@patch('meldrx.MeldRxAPI.authenticate', return_value=False)
def test_update_fhir_patient_auth_failure(self, mock_auth, mock_put):
"""Test patient update fails due to authentication in FHIR API."""
patient_data = {"resourceType": "Patient", "id": self.patient_id}
with redirect_stdout(StringIO()) as output:
result = self.api.update_fhir_patient(self.patient_id, patient_data)
self.assertFalse(result)
mock_put.assert_not_called()
self.assertIn("Cannot proceed without authentication", output.getvalue())
if __name__ == "__main__":
unittest.main(verbosity=2) # Increased verbosity for detailed output |