meldrxtester2.py

import requests
import json
from typing import Optional, Dict, Any
from 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()