Testing HL7-to-FHIR Pipelines Without a Hospital: Mocking HAPI FHIR with respx

You can't spin up a maternity ward in CI. But you can get 274 tests and 90% coverage without one. Some Background I've been building an open-source Maternity HL7-to-FHIR Pipeline that converts legacy HL7 v2.5 messages into FHIR R4 resources with Australian and European FHIR profiles. The pipeline has three layers: Mirth Connect for HL7 ingestion, FastAPI for FHIR transformation, and HAPI FHIR Server for persistence. If you want the full architecture story, the first article covers the end-to-end design, and the second article explains why I split responsibilities between Mirth and FastAPI. This article is about the part that made me confident enough to publish the whole thing: the tests. Specifically, how to build a reliable test suite for a healthcare integration pipeline when you don't have access to a hospital system, a live HAPI FHIR server, or an MLLP connection. The Testing Problem in Healthcare Integration Healthcare integration pipelines are awkward to test. The input is a decades-old wire protocol (MLLP) carrying pipe-delimited messages with positional field semantics. The output is validated FHIR resources persisted to a server that enforces its own schema rules. Between input and output, there's transformation logic full of edge cases: merging blood pressure readings, mapping HL7 gender codes to FHIR valuesets, handling missing fields gracefully. If you test the whole thing end-to-end, you need Mirth Connect, a HAPI FHIR server, and something pretending to be a hospital. That's a docker compose up before every test run. CI becomes slow. Debugging becomes painful. You get flaky tests because you're now dependent on container startup timing and network behavior. The alternative is to separate what you're actually testing from what you're testing through . My FastAPI layer doesn't care about MLLP or Mirth. It receives flat JSON payloads, transforms them into FHIR resources, and sends them to HAPI over HTTP. Every one of those steps is testable in isolation if you mock the HTTP boundary to HAPI. Test Architecture: Two Layers, Clear Boundaries The test suite has 222 unit tests, 31 integration tests, and 21 end-to-end tests, organized like this: tests/ |-- unit/ # 222 tests, no network, no Docker | |-- test_patient_transformer.py | |-- test_condition_transformer.py | |-- test_encounter_transformer.py | |-- test_observation_transformer.py # 54 tests incl. BP panel merging | |-- test_ips_composition.py | |-- test_consent.py | |-- test_eu_transformers.py | |-- test_au_profiles.py | |-- test_profile_registry.py | |-- test_profile_contamination.py | |-- test_mirth_channel_contract.py | |-- test_validate_resource.py | |-- test_errors.py | |-- test_logging.py | |-- test_middleware.py | -- test_validation.py | |-- integration/ # 31 tests, full HTTP round-trip | |-- test_patient_endpoint.py | |-- test_encounter_endpoint.py | |-- test_observation_endpoint.py | |-- test_health_endpoint.py | |-- test_consent_endpoint.py | |-- test_ips_endpoint.py | |-- test_validate_endpoint.py | -- test_eu_pipeline.py | -- e2e/ # 21 tests, live Docker stack required |-- test_au_pipeline.py -- test_eu_pipeline.py Unit tests exercise the transformation logic directly. They call transformer functions with Pydantic input models and assert on the FHIR resource output. No HTTP, no server, no Docker. They run in under 2 seconds. Integration tests hit the FastAPI endpoints through httpx.AsyncClient with HAPI responses mocked by respx . They verify the full request lifecycle: payload validation, transformation, HAPI client calls, response formatting, and error handling. Still no Docker needed. End-to-end tests run against a live Docker stack (Mirth + FastAPI + HAPI FHIR) and are skipped in CI. They verify the full pipeline from MLLP message to persisted FHIR resource. The key insight: unit and integration layers use respx to mock HAPI FHIR, but they test different things. Why respx (Not unittest.mock) The FastAPI layer talks to HAPI through httpx , an async HTTP client. That means the natural mocking tool is respx , which intercepts httpx requests at the transport level. Why not unittest.mock.patch ? Because patching httpx.AsyncClient.put gives you a dumb mock that returns whatever you tell it to. You lose the ability to assert on: The exact URL the client constructed (did it include the right resource type and identifier?) The request body (is the FHIR resource actually valid JSON with the right structure?) The HTTP method (was it a PUT for upserts vs POST for creates?) Whether conditional headers were set correctly With respx , you define route patterns and response fixtures. The mock behaves like a real HTTP endpoint. If your code sends a request to an unexpected URL or with an unexpected method, the test fails with a clear message about what was actually called. Here's the pattern I use in the integration tests. HAPI routes are mocked with a helper that sets up all the StructureDefinition and resource endpoints: import respx from httpx import ASGITransport , AsyncClient , Response def _hapi_mock (): mock = respx . mock ( base_url = " http://localhost:8080/fhir " , assert_all_called = False ) # Mock StructureDefinition endpoints (profile registration on startup) mock . put ( " /StructureDefinition/au-patient " ). mock ( return_value = Response ( 200 , json = { " resourceType " : " StructureDefinition " , " id " : " au-patient " }) ) # ... other StructureDefinition mocks ... # Mock the HAPI FHIR conditional PUT for Patient mock . put ( " /Patient " ). mock ( return_value = Response ( 201 , json = { " resourceType " : " Patient " , " id " : " pat-1 " }, headers = { " Location " : " /Patient/pat-1/_history/1 " }, ) ) return mock async def test_patient_upsert_sends_conditional_put (): """ Verify that patient creation uses conditional PUT with identifier query. """ with _hapi_mock (): async with app . router . lifespan_context ( app ): async with AsyncClient ( transport = ASGITransport ( app = app ), base_url = " http://testserver " ) as client : response = await client . post ( " /fhir/Patient " , json = { " correlationId " : " test-001 " , " mrn " : " 1234567 " , " name " : { " family " : " TEST " , " given " : " PATIENT " }, " birthDate " : " 19920315 " , " gender " : " F " , " address " : { " line " : " 14 SAMPLE ST " , " city " : " SYDNEY " , " state " : " NSW " , " postalCode " : " 2000 " }, }, ) assert response . status_code == 200 body = response . json () assert body [ " patientId " ] == " pat-1 " The HapiClient.upsert_resource method constructs a conditional PUT with If-None-Exist headers internally. By mocking at the respx transport level, the test verifies the full chain: payload parsing, FHIR resource construction, HTTP method selection, and response formatting. If the transformer produces a malformed resource, the test catches it at the HTTP boundary, not through a vague assertion error. Testing Transformers: Input Model In, FHIR Resource Out Each transformer function has the same shape: take a Pydantic input model and a ProfileConfig , return a FHIR resource. This makes them pure functions (aside from configuration), which makes them easy to test. Here's a simplified example of testing the patient transformer: from app.models.adt_payload import AdtPayload , NamePayload , AddressPayload from app.profiles.au_profile import AU_PROFILE from app.transformers.patient import build_patient def test_patient_basic_fields (): """ ADT payload maps correctly to FHIR Patient resource. """ payload = AdtPayload ( correlationId = " test-001 " , messageType = " ADT^A01 " , mrn = " 1234567 " , ihi = " 8003608166690503 " , name = NamePayload ( family = " TEST " , given = " PATIENT " , middle = " MARY " , prefix = " MS " ), birthDate = " 19920315 " , gender = " F " , address = AddressPayload ( line = " 14 SAMPLE ST " , city = " SYDNEY " , state = " NSW " , postalCode = " 2000 " , country = " AU " , ), phone = " 0412345678 " , ) patient = build_patient ( payload , AU_PROFILE ) # Check identifier mapping mrn_id = patient . identifier [ 0 ] assert mrn_id . value == " 1234567 " assert mrn_id . system == " http://hospital.local/mrn " # Check HL7 gender code mapped to FHIR valueset assert patient . gender == " female " # HL7 "F" -> FHIR "female" # Check name structure assert patient . name [ 0 ]. family == " TEST " assert patient . name [ 0 ]. given == [ " PATIENT " , " MARY " ] assert patient . name [ 0 ]. prefix == [ " MS " ] Notice what's not here: no HTTP mocking, no server setup, no async/await. The transformer is a function. The test calls the function. The assertion checks the output. Each test runs in microseconds. The ProfileConfig parameter is what makes the same transformer work for both AU and EU profiles - it carries profile URLs, terminology systems, and timezone offsets. Tests pass AU_PROFILE or EU_PROFILE directly. This is where having fhir.resources as a dependency pays off. The build_patient function returns a Patient Pydantic model, not a raw dict. If the function accidentally sets gender to "F" instead of "female" , the Pydantic model raises a validation error inside the transformer , before the test even gets to the assertions. Testing Edge Cases: Where Healthcare Gets Interesting General-purpose APIs have edge cases. Healthcare APIs have clinically significant edge cases. Here are the ones that taught me the most. Blood Pressure Panel Merging In HL7, blood pressure comes as two separate OBX segments: one for systolic (LOINC 8480-6 ) and one for diastolic (LOINC 8462-4 ). In FHIR, they should be a single Observation with panel code 85354-9 and two component[] entries. The tricky part: they're only a panel if they appear as consecutive OBX segments. A systolic reading followed by a body weight followed by a diastolic reading is two separate observations, not a panel. from app.models.oru_payload import ObservationPayload , OruPayload from app.profiles.au_profile import AU_PROFILE from app.transformers.observation import build_observations def _obs ( ** overrides ) -> ObservationPayload : defaults = { " setId " : 1 , " code " : " 29463-7 " , " display " : " Body weight " , " value " : 68.5 , " unitCode " : " kg " , " status " : " F " } defaults . update ( overrides ) return ObservationPayload ( ** defaults ) def _payload ( observations ): return OruPayload ( correlationId = " test-004 " , mrn = " 1234567 " , observations = observations ) def test_bp_codes_merged_into_panel (): """ Systolic + diastolic OBX segments merge into one BP panel. """ payload = _payload ([ _obs ( code = " 8480-6 " , display = " Systolic BP " , value = 120 , unitCode = " mm[Hg] " ), _obs ( code = " 8462-4 " , display = " Diastolic BP " , value = 80 , unitCode = " mm[Hg] " ), ]) results = build_observations ( payload , " Patient/2 " , None , AU_PROFILE ) assert len ( results ) == 1 # One panel, not two observations assert results [ 0 ]. code . coding [ 0 ]. code == " 85354-9 " # BP panel code def test_mixed_bp_and_simple (): """ BP pair merges; non-BP observations remain individual. """ payload = _payload ([ _obs ( code = " 8480-6 " , display = " Systolic BP " , value = 120 , unitCode = " mm[Hg] " ), _obs ( code = " 8462-4 " , display = " Diastolic BP " , value = 80 , unitCode = " mm[Hg] " ), _obs ( code = " 29463-7 " , display = " Body weight " , value = 68.5 , unitCode = " kg " ), _obs ( code = " 55283-6 " , display = " Fetal heart rate " , value = 145 , unitCode = " /min " ), ]) results = build_observations ( payload , " Patient/2 " , " Encounter/4 " , AU_PROFILE ) assert len ( results ) == 3 codes = [ r . code . coding [ 0 ]. code for r in results ] assert " 85354-9 " in codes # BP panel assert " 29463-7 " in codes # Body weight assert " 55283-6 " in codes # Fetal heart rate def test_orphan_systolic_built_individually (): """ Systolic without diastolic -> individual observation. """ payload = _payload ([ _obs ( code = " 8480-6 " , display = " Systolic BP " , value = 120 , unitCode = " mm[Hg] " ), ]) results = build_observations ( payload , " Patient/2 " , None , AU_PROFILE ) assert len ( results ) == 1 assert results [ 0 ]. code . coding [ 0 ]. code == " 8480-6 " # Not wrapped in panel Three tests, three different behaviors from the same function. The build_observations function takes a full OruPayload (not a raw list), a patient reference, an optional encounter reference, and a ProfileConfig . The BP merging logic scans all ObservationPayload items for matching systolic + diastolic LOINC codes and merges them into a panel. Orphan readings stay standalone. The test suite covers additional scenarios: orphan diastolic, worst-case status interpretation (one preliminary + one final = panel is preliminary), and abnormal flag propagation. Mirth Contract Tests: Catching Drift Without Running Mirth The Mirth JavaScript transformer parses HL7 messages by field position and produces flat JSON for FastAPI. If the field positions in the Mirth transformer and the Pydantic models in FastAPI ever drift apart, the pipeline silently produces wrong FHIR resources. I test the contract without running Mirth. The test file contains a minimal HL7 v2 parser (about 60 lines of Python) that extracts fields by the same positional rules as the Mirth JavaScript. It reads the actual synthetic .hl7 sample files from disk, builds the flat JSON payload, and validates it against the real Pydantic models: from app.models.adt_payload import AdtPayload def test_adt_maps_to_valid_patient_payload () -> None : segments = _load ( " adt_a01_normal_delivery.hl7 " ) assert _message_type ( segments ) == " ADT " payload = build_patient_payload ( segments , " test-adt " ) model = AdtPayload . model_validate ( payload ) # raises if the contract is broken assert model . mrn == " 1234567 " assert model . ihi == " 8003608166690503 " assert model . name . family == " TEST " assert model . gender == " F " assert len ( model . diagnoses ) == 1 assert model . diagnoses [ 0 ]. code == " O80 " The build_patient_payload function mirrors Mirth's field extraction: PID-3.1 for MRN, PID-3.5 for identifier type, PID-5 for name components. If someone changes the Pydantic model to rename a field or make a previously optional field required, this test fails immediately - even though Mirth isn't running. The contract tests cover all three message types (ADT, ORM, ORU) in both AU and EU formats, HL7 escape sequence decoding ( \T\ → & , \S\ → ^ ), and invalid inputs (missing MRN triggers ValidationError ). That's 11 tests total, all running without Docker. What they don't test is Mirth's E4X runtime - its toString() behavior and subcomponent drilling - which requires a live MLLP smoke test. Profile Cross-Contamination: Zero-Leakage Tests When the same transformer serves both AU and EU profiles, there's a risk that AU profile URLs leak into EU output, or EU terminology systems appear in AU resources. Per-profile tests won't catch this - each profile passes its own assertions. But the wrong profile URL breaks FHIR validation in the other jurisdiction. The contamination tests run the same payload through both profiles and assert on the serialized JSON: AU_MARKERS = [ " au-patient " , " au-condition " , " au-vitalsigns-bloodpressure " , " hl7.org.au/fhir/CodeSystem/icd-10-am " , " ns.electronichealth.net.au " , ] EU_MARKERS = [ " patient-eu " , " condition-eu-core " , " fhir/sid/icd-10 " , " fhir.nhs.uk/Id/nhs-number " , ] def test_au_output_contains_only_au_values () -> None : out = _all_output ( AU_PROFILE ) for marker in AU_MARKERS : assert marker in out , f " expected AU marker missing: { marker } " for marker in EU_MARKERS : assert marker not in out , f " AU output leaked EU marker: { marker } " def test_eu_output_contains_only_eu_values () -> None : out = _all_output ( build_eu_profile ( " uk " )) for marker in EU_MARKERS : assert marker in out , f " expected EU marker missing: { marker } " for marker in [ " hl7.org.au " , " icd-10-am " , " ns.electronichealth.net.au " ]: assert marker not in out , f " EU output leaked AU marker: { marker } " The _all_output helper runs every transformer (patient, conditions, observations) and concatenates the serialized FHIR JSON into one string. Then it checks for marker strings from the wrong region. If a refactoring accidentally hardcodes an AU profile URL instead of reading from ProfileConfig , this test catches it. There's also a parametrized test for EU national identifier systems. UK uses NHS numbers, Netherlands uses BSN, Germany uses KVID-10, Ireland uses PPSN. Each country code must produce the correct identifier system in the output. Gender Code Mapping HL7 v2 uses single-character gender codes. FHIR uses full words. The mapping isn't one-to-one: from app.valuesets.hl7_to_fhir_gender import map_gender @pytest.mark.parametrize ( " hl7_code,fhir_code " , [ ( " F " , " female " ), ( " M " , " male " ), ( " O " , " other " ), ( " U " , " unknown " ), ( " A " , " other " ), # Ambiguous -> other ( " N " , " unknown " ), # Not applicable -> unknown ( "" , " unknown " ), # Empty -> unknown ]) def test_gender_mapping ( hl7_code , fhir_code ): """ HL7 gender codes map correctly to FHIR Administrative Gender. """ assert map_gender ( hl7_code ) == fhir_code Parametrized tests are perfect for valueset mappings. Seven test cases, one function, zero ambiguity about what the transformer does with each input. If a hospital sends gender code "A" (Ambiguous), we know it maps to FHIR "other" , not that it crashes or silently drops the field. The actual codebase tests gender mapping through individual unit tests ( test_gender_female , test_gender_male , test_gender_unknown ), but @pytest.mark.parametrize would be equally effective for this finite mapping table. Missing and Empty Fields Hospital systems are inconsistent about what they include. Some always send a middle name. Some never do. Some send an empty string where others send nothing at all. from app.models.adt_payload import AdtPayload , NamePayload , AddressPayload from app.profiles.au_profile import AU_PROFILE from app.transformers.patient import build_patient from app.transformers.condition import build_conditions def _sample_payload ( ** overrides ) -> AdtPayload : defaults = { " correlationId " : " test-uuid-001 " , " mrn " : " 1234567 " , " name " : NamePayload ( family = " TEST " , given = " PATIENT " ), " birthDate " : " 19920315 " , " gender " : " F " , " address " : AddressPayload ( line = " 14 SAMPLE ST " , city = " SYDNEY " , state = " NSW " , postalCode = " 2000 " ), } defaults . update ( overrides ) return AdtPayload ( ** defaults ) def test_patient_without_middle_name (): """ Patient resource builds correctly when middle name is absent. """ patient = build_patient ( _sample_payload (), AU_PROFILE ) assert patient . name [ 0 ]. given == [ " PATIENT " ] # List with one entry, not crash def test_patient_with_empty_diagnosis_list (): """ Patient with no diagnoses produces no Conditions. """ payload = _sample_payload () conditions = build_conditions ( payload , " Patient/1 " , AU_PROFILE ) assert conditions == [] def test_patient_with_no_phone (): """ Missing phone number doesn ' t set telecom at all. """ patient = build_patient ( _sample_payload ( phone = "" ), AU_PROFILE ) assert patient . telecom is None # Not an empty list, not a placeholder These tests are not clever. They are simple and boring, and that is what I want. The build_patient and build_conditions are separate functions - the endpoint orchestrates both, but tests exercise them independently. Every "what if this field is empty" question has a documented, tested answer. Integration Tests: The Full HTTP Round-Trip Unit tests verify the transformation logic. Integration tests verify that the FastAPI endpoints wire everything together correctly: payload validation, transformer calls, HAPI client calls, response formatting. The integration tests use httpx.A