File size: 6,701 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
import requests
import json
from typing import Optional, Dict, Any
from utils.meldrx import MeldRxAPI  # Assuming meldrx.py contains the updated MeldRxAPI with SMART on FHIR

class MeldRxAPITest:
    """A class to test the functionality of the MeldRxAPI class with SMART on FHIR and Gradio callback."""

    def __init__(self):
        # Placeholder variables (replace with real values for live testing)
        self.client_id = "04bdc9f9a23d488a868b93d594ee5a4a"  # Replace with your actual client ID
        self.redirect_uri = "https://multitransformer-discharge-guard.hf.space/callback"  # Gradio Space callback URL
        self.workspace_id = "09ed4f76-b5ac-42bf-92d5-496933203dbe"  # Replace with your workspace ID
        self.test_patient_id = "05f7c0c2-ebee-44db-ab71-2113e7a70acb"  # Replace with a valid patient ID
        self.client_secret = None  # Optional; set if using a confidential client
        self.api = MeldRxAPI(self.client_id, self.redirect_uri, self.workspace_id, self.client_secret)
        self.auth_code = None
        self.access_token = None

    def _print_section(self, title: str):
        """Helper method to print section headers."""
        print(f"\n{'=' * 10} {title} {'=' * 10}")

    def test_get_auth_url(self):
        """Test generating the SMART on FHIR authorization URL."""
        self._print_section("Testing Get Authorization URL")
        auth_url = self.api.get_authorization_url()
        print(f"Generated Authorization URL:\n{auth_url}")
        print("Action: Open this URL in a browser, log in, and copy the 'code' from the redirect URL.")

    def test_authenticate_with_code(self):
        """Test authentication using an authorization code (simulating Gradio callback)."""
        self._print_section("Testing Authenticate with Code")
        self.auth_code = input("Enter the authorization code from the redirect URL (or press Enter to skip): ").strip()
        if not self.auth_code:
            print("No code provided. Skipping authentication test.")
            return
        result = self.api.authenticate_with_code(self.auth_code)
        print(f"Authentication Result: {result}")
        if result:
            self.access_token = self.api.access_token
            print(f"Access Token: {self.access_token[:10]}... (truncated for brevity)")
        else:
            print("Authentication failed. Subsequent tests may not work as expected.")

    def test_get_patients(self):
        """Test retrieving patients from the FHIR API."""
        self._print_section("Testing Get Patients (FHIR)")
        if not self.access_token:
            print("Not authenticated. Run test_authenticate_with_code first.")
            return
        patients = self.api.get_patients()
        if patients is not None:
            print("Patients Retrieved Successfully:")
            print(json.dumps(patients, indent=2) if patients else "No data returned (expected per docs)")
        else:
            print("Failed to retrieve patients.")

    def test_create_virtual_workspace(self):
        """Test creating a virtual workspace."""
        self._print_section("Testing Create Virtual Workspace")
        if not self.access_token:
            print("Not authenticated. Run test_authenticate_with_code first.")
            return
        success = self.api.create_virtual_workspace(
            snapshot="patient-prefetch",
            patient_id=self.test_patient_id,
            hook="patient-view"
        )
        print(f"Virtual Workspace Creation Result: {success}")
        if success:
            print("Virtual workspace created successfully (no response body expected).")
        else:
            print("Failed to create virtual workspace.")

    def test_get_mips_patients(self):
        """Test retrieving patients from the MIPS API."""
        self._print_section("Testing Get MIPS Patients")
        if not self.access_token:
            print("Not authenticated. Run test_authenticate_with_code first.")
            return
        patients = self.api.get_mips_patients()
        if patients is not None:
            print("MIPS Patients Retrieved Successfully:")
            print(json.dumps(patients, indent=2) if patients else "No data returned")
        else:
            print("Failed to retrieve MIPS patients.")

    def test_get_mips_patient_by_id(self):
        """Test retrieving a specific patient by ID from the MIPS API."""
        self._print_section("Testing Get MIPS Patient by ID")
        if not self.access_token:
            print("Not authenticated. Run test_authenticate_with_code first.")
            return
        patient = self.api.get_mips_patient_by_id(self.test_patient_id)
        if patient is not None:
            print(f"Patient {self.test_patient_id} Retrieved Successfully:")
            print(json.dumps(patient, indent=2) if patient else "No data returned")
        else:
            print(f"Failed to retrieve patient {self.test_patient_id}.")

    def test_get_mips_encounters(self):
        """Test retrieving encounters from the MIPS API (with and without patient filter)."""
        self._print_section("Testing Get MIPS Encounters (All)")
        if not self.access_token:
            print("Not authenticated. Run test_authenticate_with_code first.")
            return
        encounters = self.api.get_mips_encounters()
        if encounters is not None:
            print("All Encounters Retrieved Successfully:")
            print(json.dumps(encounters, indent=2) if encounters else "No data returned")
        else:
            print("Failed to retrieve all encounters.")

        self._print_section("Testing Get MIPS Encounters (Filtered by Patient)")
        encounters_filtered = self.api.get_mips_encounters(self.test_patient_id)
        if encounters_filtered is not None:
            print(f"Encounters for Patient {self.test_patient_id} Retrieved Successfully:")
            print(json.dumps(encounters_filtered, indent=2) if encounters_filtered else "No data returned")
        else:
            print(f"Failed to retrieve encounters for patient {self.test_patient_id}.")

    def run_all_tests(self):
        """Run all test methods in sequence."""
        print("Starting MeldRx API Tests with Gradio Callback Simulation...\n")
        self.test_get_auth_url()
        self.test_authenticate_with_code()
        self.test_get_patients()
        self.test_create_virtual_workspace()
        self.test_get_mips_patients()
        self.test_get_mips_patient_by_id()
        self.test_get_mips_encounters()
        print("\nAll tests completed.")
        self.api.close()


# Run the tests
if __name__ == "__main__":
    tester = MeldRxAPITest()
    tester.run_all_tests()