import asyncio import pytest from fastapi.testclient import TestClient from httpx import AsyncClient from app.main import app @pytest.mark.asyncio async def test_add_vendor_success(): payload = { "vendor": "test_vendor", "doctors": ["doc1", "doc2"], "concurrent_doctors": 2, "immediate_unblock": False, "rx_max_hold_time": 60, "rx_max_block_time": 30, "rx_notification_interval": 10 } async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.post("/api/1rx/v1/scheduler/add_vendor", json=payload) assert response.status_code == 200 assert "already exists" in response.json()["data"]["message"] or "added to the pool" in response.json()["data"]["message"] @pytest.mark.asyncio async def test_add_vendor_duplicate(): payload = { "vendor": "test_vendor_dup", "doctors": ["doc3"], "concurrent_doctors": 1, "immediate_unblock": False, "rx_max_hold_time": 60, "rx_max_block_time": 30, "rx_notification_interval": 10 } async with AsyncClient(app=app, base_url="http://test") as ac: await ac.post("/api/1rx/v1/scheduler/add_vendor", json=payload) response = await ac.post("/api/1rx/v1/scheduler/add_vendor", json=payload) assert response.status_code == 200 assert "already exists" in response.json()["data"]["message"] @pytest.mark.asyncio async def test_add_doctor_success(): # Add vendor first vendor_payload = { "vendor": "adddoc_vendor", "doctors": [], "concurrent_doctors": 1, "immediate_unblock": False, "rx_max_hold_time": 60, "rx_max_block_time": 30, "rx_notification_interval": 10 } doctor_payload = {"vendor_id": "adddoc_vendor", "id": "docX"} async with AsyncClient(app=app, base_url="http://test") as ac: await ac.post("/api/1rx/v1/scheduler/add_vendor", json=vendor_payload) response = await ac.post("/api/1rx/v1/scheduler/add_doctor", json=doctor_payload) assert response.status_code == 200 assert "added to vendor" in response.json()["data"]["message"] @pytest.mark.asyncio async def test_add_doctor_duplicate(): vendor_payload = { "vendor": "dupdoc_vendor", "doctors": [], "concurrent_doctors": 1, "immediate_unblock": False, "rx_max_hold_time": 60, "rx_max_block_time": 30, "rx_notification_interval": 10 } doctor_payload = {"vendor_id": "dupdoc_vendor", "id": "docY"} async with AsyncClient(app=app, base_url="http://test") as ac: await ac.post("/api/1rx/v1/scheduler/add_vendor", json=vendor_payload) await ac.post("/api/1rx/v1/scheduler/add_doctor", json=doctor_payload) response = await ac.post("/api/1rx/v1/scheduler/add_doctor", json=doctor_payload) assert response.status_code == 200 assert "already exists" in response.json()["data"]["message"] @pytest.mark.asyncio async def test_remove_doctor_not_found(): payload = {"vendor_id": "nonexistent_vendor", "id": "docZ"} async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.post("/api/1rx/v1/scheduler/remove_doctor", json=payload) assert response.status_code == 200 assert "not found" in response.json()["data"]["message"] @pytest.mark.asyncio async def test_assign_doctor_success_and_failures(): vendor_payload = { "vendor": "assigndoc_vendor", "doctors": ["docA"], "concurrent_doctors": 1, "immediate_unblock": False, "rx_max_hold_time": 60, "rx_max_block_time": 30, "rx_notification_interval": 10 } assign_payload = {"vendor_id": "assigndoc_vendor", "prescription_id": "rx1"} async with AsyncClient(app=app, base_url="http://test") as ac: await ac.post("/api/1rx/v1/scheduler/add_vendor", json=vendor_payload) response = await ac.post("/api/1rx/v1/scheduler/assign_doctor", json=assign_payload) assert response.status_code == 200 assert response.json()["data"]["doctor_ids"] == ["docA"] # Try assigning again (should fail) response2 = await ac.post("/api/1rx/v1/scheduler/assign_doctor", json=assign_payload) assert response2.status_code == 400 @pytest.mark.asyncio async def test_reassign_doctor_no_doctors(): vendor_payload = { "vendor": "reassigndoc_vendor", "doctors": [], "concurrent_doctors": 1, "immediate_unblock": False, "rx_max_hold_time": 60, "rx_max_block_time": 30, "rx_notification_interval": 10 } assign_payload = {"vendor_id": "reassigndoc_vendor", "prescription_id": "rx2"} async with AsyncClient(app=app, base_url="http://test") as ac: await ac.post("/api/1rx/v1/scheduler/add_vendor", json=vendor_payload) response = await ac.post("/api/1rx/v1/scheduler/reassign_doctor", json=assign_payload) assert response.status_code == 404 @pytest.mark.asyncio async def test_block_doctor_not_found(): payload = {"vendor_id": "blockdoc_vendor", "block_doctor_id": "docB", "prescription_id": "rx3"} async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.post("/api/1rx/v1/scheduler/block_doctor", json=payload) assert response.status_code == 404 @pytest.mark.asyncio async def test_release_doctor_not_found(): payload = {"vendor_id": "releasedoc_vendor", "doctor_id": "docC", "prescription_id": "rx4"} async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.post("/api/1rx/v1/scheduler/release_doctor", json=payload) assert response.status_code == 404 @pytest.mark.asyncio async def test_hold_prescription_not_found(): payload = {"vendor_id": "holddoc_vendor", "prescription_id": "rx5"} async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.post("/api/1rx/v1/scheduler/hold", json=payload) assert response.status_code == 404 @pytest.mark.asyncio async def test_list_all_cached_doctors_empty(): async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.get("/api/1rx/v1/scheduler/list_all_cached_doctors?vendor_id=empty_vendor") assert response.status_code == 200 assert response.json()["data"]["doctor_ids"] == [] @pytest.mark.asyncio async def test_get_prescriptions_in_queue_empty(): async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.get("/api/1rx/v1/scheduler/get_prescriptions_in_queue?vendor_id=empty_vendor") assert response.status_code == 404 @pytest.mark.asyncio async def test_assigned_doctors_empty(): async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.get("/api/1rx/v1/scheduler/assigned_doctors") assert response.status_code == 404 @pytest.mark.asyncio async def test_shutdown(): # Just check the endpoint returns 200 async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.get("/api/1rx/v1/scheduler/shutdown") assert response.status_code == 200 assert "Server shutting down" in response.text @pytest.mark.asyncio async def test_get_all_available_doctors_sample(): """Test the structure and content of get_all_available_doctors response.""" async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.get("/api/1rx/v1/scheduler/get-all-available-doctors") assert response.status_code == 200 data = response.json()["data"] assert isinstance(data, list) # Check for known vendors and structure vendors = {v["vendor"] for v in data} assert "wellness_forever" in vendors assert "zepto" in vendors # Check for null concurrent_doctors and empty doctors for v in data: assert "vendor" in v assert "doctors" in v assert "concurrent_doctors" in v assert "immediate_unblock" in v assert "rx_max_hold_time" in v assert "rx_max_block_time" in v assert "rx_notification_interval" in v # concurrent_doctors can be None assert v["concurrent_doctors"] is None or isinstance(v["concurrent_doctors"], int) # doctors is always a list assert isinstance(v["doctors"], list) @pytest.mark.asyncio async def test_assign_doctor_null_concurrent_and_empty_doctors(): """Assign doctor to vendor with null concurrent_doctors and empty doctors list (should fail gracefully).""" assign_payload = {"vendor_id": "wellness_forever", "prescription_id": "rx_null1"} async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.post("/api/1rx/v1/scheduler/assign_doctor", json=assign_payload) assert response.status_code in (200, 400, 404) # Should not assign any doctors if response.status_code == 200: assert response.json()["data"]["doctor_ids"] == [] @pytest.mark.asyncio async def test_assign_doctor_single_doctor(): """Assign doctor to vendor with a single doctor and null concurrent_doctors (should assign that doctor).""" assign_payload = {"vendor_id": "zepto", "prescription_id": "rx_single1"} async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.post("/api/1rx/v1/scheduler/assign_doctor", json=assign_payload) assert response.status_code == 200 # Should assign the only doctor assert response.json()["data"]["doctor_ids"] == ["1484ec9c-0baa-45fd-828b-d890b28b2468"] @pytest.mark.asyncio async def test_assign_doctor_multiple_doctors_null_concurrent(): """Assign doctor to vendor with multiple doctors and null concurrent_doctors (should assign all).""" # Add a vendor with multiple doctors and null concurrent_doctors vendor_payload = { "vendor": "platinumrx", "doctors": ["a88f4861-6a5d-493f-9060-244c40a9ace6", "doc_extra"], "concurrent_doctors": None, "immediate_unblock": False, "rx_max_hold_time": 900, "rx_max_block_time": 600, "rx_notification_interval": 25 } assign_payload = {"vendor_id": "platinumrx", "prescription_id": "rx_multi1"} async with AsyncClient(app=app, base_url="http://test") as ac: await ac.post("/api/1rx/v1/scheduler/add_vendor", json=vendor_payload) response = await ac.post("/api/1rx/v1/scheduler/assign_doctor", json=assign_payload) assert response.status_code == 200 # Should assign all doctors in the list assert set(response.json()["data"]["doctor_ids"]) == set(["a88f4861-6a5d-493f-9060-244c40a9ace6", "doc_extra"])