Skip to content

Commit

Permalink
Mock the Census API for tests so that Actions doesn't actually hit th…
Browse files Browse the repository at this point in the history
…e API when running tests (#864)

* first pass at conversion function; will add tests next

* added function to init

* tests for the datetime_to_str added

* forgot to add to __all__

* added a param to include time in the output with tests

* separated the line that was too long

* removed functions from phdi.linkage

* add function to container utils

* added mocking of census api with pytest monkeypatch

* added geocoded_response as fixture

* finished mocking the fhir census geocoder

* forgot to run black

* removed duplicate mocking

* removed extra line

* removed extra comment
  • Loading branch information
robertmitchellv authored Oct 17, 2023
1 parent 3d4b5ea commit 77942ac
Show file tree
Hide file tree
Showing 2 changed files with 180 additions and 242 deletions.
235 changes: 96 additions & 139 deletions tests/fhir/geospatial/test_fhir_census_client.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,30 @@
from unittest import mock
import copy
import json
import pathlib
import copy
import pytest

from phdi.fhir.geospatial.census import CensusFhirGeocodeClient
from phdi.geospatial.core import GeocodeResult


def test_geocode_resource_census():
census_client = CensusFhirGeocodeClient()
assert census_client is not None
FHIR_BUNDLE_PATH = pathlib.Path(__file__).parent.parent.parent / "assets" / "general"


@pytest.fixture
def patient_bundle_census():
with open(FHIR_BUNDLE_PATH / "patient_bundle_census.json") as file:
return json.load(file)


@pytest.fixture
def patient_bundle_census_extension():
with open(FHIR_BUNDLE_PATH / "patient_bundle_census_extension.json") as file:
return json.load(file)


geocoded_response = GeocodeResult(
@pytest.fixture
def geocoded_response():
return GeocodeResult(
line=["239 Greene St", "Apt 4L"],
city="NEW YORK",
state="NY",
Expand All @@ -23,17 +36,28 @@ def test_geocode_resource_census():
census_tract="59",
)

bundle = json.load(
open(
pathlib.Path(__file__).parent.parent.parent
/ "assets"
/ "general"
/ "patient_bundle_census.json"
)

@pytest.fixture
def census_client(monkeypatch, geocoded_response):
client = CensusFhirGeocodeClient()

def mock_geocode_from_dict(*args, **kwargs):
return geocoded_response

monkeypatch.setattr(
client._CensusFhirGeocodeClient__client,
"geocode_from_dict",
mock_geocode_from_dict,
)

patient = bundle["entry"][1]["resource"]
standardized_patient = copy.deepcopy(patient)
return client


def _get_patient_from_bundle(bundle):
return bundle["entry"][1]["resource"]


def _extract_address(standardized_patient, geocoded_response, extension=False):
address = standardized_patient["address"][0]
address["city"] = geocoded_response.city
address["state"] = geocoded_response.state
Expand All @@ -48,31 +72,61 @@ def test_geocode_resource_census():
],
}
)
address["_line"] = []
address["_line"].append(
{
"extension": [
{
"url": "http://hl7.org/fhir/StructureDefinition/"
+ "iso21090-ADXP-censusTract",
"valueString": "59",
}
]
},
)
address["_line"].append(
{
if extension:
address["_line"][0]["extension"].append(
{
"url": "http://hl7.org/fhir/StructureDefinition/"
+ "iso21090-ADXP-censusTract",
"valueString": "59",
}
)
address["_line"][1] = {
"extension": [
{
"url": "http://hl7.org/fhir/StructureDefinition/"
+ "iso21090-ADXP-censusTract",
"valueString": "59",
}
]
},
)
census_client.geocode_from_str = mock.Mock()
census_client.geocode_from_str.return_value = geocoded_response
}
else:
address["_line"] = []
address["_line"].append(
{
"extension": [
{
"url": "http://hl7.org/fhir/StructureDefinition/"
+ "iso21090-ADXP-censusTract",
"valueString": "59",
}
]
},
)
address["_line"].append(
{
"extension": [
{
"url": "http://hl7.org/fhir/StructureDefinition/"
+ "iso21090-ADXP-censusTract",
"valueString": "59",
}
]
},
)
return address


def test_geocode_resource_census(
patient_bundle_census,
patient_bundle_census_extension,
geocoded_response,
census_client,
):
assert census_client is not None

patient = _get_patient_from_bundle(patient_bundle_census)
standardized_patient = copy.deepcopy(patient)
_extract_address(standardized_patient, geocoded_response)

# Case 1: Overwrite = False
returned_patient = census_client.geocode_resource(patient, overwrite=False)
Expand All @@ -89,48 +143,9 @@ def test_geocode_resource_census():
assert returned_patient == patient

# Case 3: Patient already has an extension on line, and it's preserved.
bundle = json.load(
open(
pathlib.Path(__file__).parent.parent.parent
/ "assets"
/ "general"
/ "patient_bundle_census_extension.json"
)
)

patient = bundle["entry"][1]["resource"]
patient = _get_patient_from_bundle(patient_bundle_census_extension)
standardized_patient = copy.deepcopy(patient)

address = standardized_patient["address"][0]
address["city"] = geocoded_response.city
address["state"] = geocoded_response.state
address["postalCode"] = geocoded_response.postal_code
address["extension"] = []
address["extension"].append(
{
"url": "http://hl7.org/fhir/StructureDefinition/geolocation",
"extension": [
{"url": "latitude", "valueDecimal": geocoded_response.lat},
{"url": "longitude", "valueDecimal": geocoded_response.lng},
],
}
)
address["_line"][0]["extension"].append(
{
"url": "http://hl7.org/fhir/StructureDefinition/"
+ "iso21090-ADXP-censusTract",
"valueString": "59",
}
)
address["_line"][1] = {
"extension": [
{
"url": "http://hl7.org/fhir/StructureDefinition/"
+ "iso21090-ADXP-censusTract",
"valueString": "59",
}
]
}
_extract_address(standardized_patient, geocoded_response, extension=True)

# Case 3: Patient already has 1 extension on _line, and it's preserved.
returned_patient = census_client.geocode_resource(patient, overwrite=True)
Expand All @@ -145,73 +160,15 @@ def test_geocode_resource_census():
)


def test_geocode_bundle_census():
census_client = CensusFhirGeocodeClient()
def test_geocode_bundle_census(patient_bundle_census, geocoded_response, census_client):
assert census_client is not None

geocoded_response = GeocodeResult(
line=["239 Greene St", "Apt 4L"],
city="NEW YORK",
state="NY",
lat=40.729656537689166,
lng=-73.99550002689155,
county_fips="36061",
county_name="New York",
postal_code="10003",
census_tract="59",
)
standardized_bundle = copy.deepcopy(patient_bundle_census)
patient = _get_patient_from_bundle(standardized_bundle)
_extract_address(patient, geocoded_response)

bundle = json.load(
open(
pathlib.Path(__file__).parent.parent.parent
/ "assets"
/ "general"
/ "patient_bundle_census.json"
)
returned_bundle = census_client.geocode_bundle(
patient_bundle_census, overwrite=False
)
standardized_bundle = copy.deepcopy(bundle)
patient = standardized_bundle["entry"][1]["resource"]
address = patient["address"][0]
address["city"] = geocoded_response.city
address["state"] = geocoded_response.state
address["postalCode"] = geocoded_response.postal_code
address["extension"] = []
address["extension"].append(
{
"url": "http://hl7.org/fhir/StructureDefinition/geolocation",
"extension": [
{"url": "latitude", "valueDecimal": geocoded_response.lat},
{"url": "longitude", "valueDecimal": geocoded_response.lng},
],
}
)
address["_line"] = []
address["_line"].append(
{
"extension": [
{
"url": "http://hl7.org/fhir/StructureDefinition/"
+ "iso21090-ADXP-censusTract",
"valueString": "59",
}
]
},
)
address["_line"].append(
{
"extension": [
{
"url": "http://hl7.org/fhir/StructureDefinition/"
+ "iso21090-ADXP-censusTract",
"valueString": "59",
}
]
},
)

census_client.geocode_from_str = mock.Mock()
census_client.geocode_from_str.return_value = geocoded_response

returned_bundle = census_client.geocode_bundle(bundle, overwrite=False)
assert standardized_bundle == returned_bundle
assert bundle != standardized_bundle
assert patient_bundle_census != standardized_bundle
Loading

0 comments on commit 77942ac

Please sign in to comment.