DEV Community

Cover image for Testing HL7-to-FHIR Pipelines Without a Hospital: Mocking HAPI FHIR with respx
Budi Widhiyanto
Budi Widhiyanto

Posted on

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
Enter fullscreen mode Exit fullscreen mode

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"
Enter fullscreen mode Exit fullscreen mode

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"]
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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"
Enter fullscreen mode Exit fullscreen mode

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}"
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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.AsyncClient with FastAPI's app directly (no server startup needed) and respx to mock HAPI. The pattern uses a context manager for the mock and FastAPI's lifespan context:

import pytest
import respx
from httpx import ASGITransport, AsyncClient, Response
from app.main import app

def _hapi_mock():
    mock = respx.mock(base_url="http://localhost:8080/fhir", assert_all_called=False)
    # Mock StructureDefinition PUTs (FastAPI registers profiles on startup)
    mock.put("/StructureDefinition/au-patient").mock(
        return_value=Response(200, json={"resourceType": "StructureDefinition", "id": "au-patient"})
    )
    # ... other StructureDefinition mocks ...
    mock.put("/Patient").mock(
        return_value=Response(201, json={"resourceType": "Patient", "id": "pat-1"},
                              headers={"Location": "/Patient/pat-1/_history/1"})
    )
    mock.post("/Condition").mock(
        return_value=Response(201, json={"resourceType": "Condition", "id": "cond-1"})
    )
    return mock

class TestPatientEndpoint:
    @pytest.mark.asyncio
    async def test_with_diagnosis_returns_condition_ids(self):
        """POST /fhir/Patient returns patientId, conditionIds, and correlationId."""
        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": "int-test-002",
                            "mrn": "1234567",
                            "name": {"family": "Smith", "given": "Jane"},
                            "birthDate": "19920315",
                            "gender": "F",
                            "address": {"line": "1 Test St", "city": "Sydney",
                                        "state": "NSW", "postalCode": "2000"},
                            "diagnoses": [{"code": "O80", "display": "Normal delivery"}],
                        },
                    )
        assert response.status_code == 200
        body = response.json()
        assert body["conditionIds"] == ["cond-1"]
Enter fullscreen mode Exit fullscreen mode

This test verifies the entire chain: the endpoint accepts the payload, the transformer builds valid FHIR resources, the HAPI client sends them with the right HTTP methods, and the response is formatted correctly. All without Docker. The lifespan_context ensures FastAPI's startup hooks run (including profile registration with HAPI), which is why the StructureDefinition mocks are needed.

Testing Error Paths

The error tests are where integration tests become important. I want to verify that invalid payloads produce RFC 7807 problem+json responses:

@pytest.mark.asyncio
async def test_empty_mrn_returns_422(self):
    """Empty MRN rejected with problem+json response."""
    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": "int-test-003",
                    "mrn": "",
                    "name": {"family": "Smith", "given": "Jane"},
                    "birthDate": "19920315",
                    "gender": "F",
                    "address": {"line": "1 Test St", "city": "Sydney",
                                "state": "NSW", "postalCode": "2000"},
                },
            )
    assert response.status_code == 422
    assert response.headers["content-type"] == "application/problem+json"
Enter fullscreen mode Exit fullscreen mode

The unit test layer also exercises the error handling directly:

from app.errors import problem_response

def test_problem_response_status():
    resp = problem_response(422, "Validation Error", "Missing field")
    assert resp.status_code == 422

def test_problem_response_content_type():
    resp = problem_response(422, "Validation Error", "Missing field")
    assert resp.media_type == "application/problem+json"
Enter fullscreen mode Exit fullscreen mode

Without respx, testing HAPI failure paths would require either making HAPI actually reject a resource (which means running HAPI) or patching internals so deep that the test tells you nothing about real behavior.


The CI Pipeline

All of this runs in GitHub Actions on every push:

name: CI
on:
  push:
    branches: [main, master]
  pull_request:
    branches: [main, master]

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    strategy:
      fail-fast: false
      matrix:
        profile_region: [au, eu]
    env:
      PROFILE_REGION: ${{ matrix.profile_region }}
      PROFILE_COUNTRY: ${{ matrix.profile_region == 'eu' && 'uk' || '' }}

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install dependencies
        working-directory: ./fastapi
        run: pip install ".[dev]"

      - name: Lint app
        working-directory: ./fastapi
        run: ruff check app/

      - name: Lint tests
        run: ruff check tests/

      - name: Type check
        working-directory: ./fastapi
        run: mypy app/ --ignore-missing-imports

      - name: Run tests with coverage
        run: |
          python -m pytest tests/ -v \
            --cov=fastapi/app \
            --cov-report=term-missing \
            --cov-fail-under=80
Enter fullscreen mode Exit fullscreen mode

No Docker in CI. No Mirth. No HAPI. The entire test suite runs in under 10 seconds. The matrix strategy runs the full suite twice - once with PROFILE_REGION=au and once with PROFILE_REGION=eu - so both AU and EU profile paths are exercised in CI. The coverage threshold is set at 80% (actual coverage is 90%), so if someone adds a new transformer without tests, CI catches it.

The lint steps run ruff check on both application code and tests (separately, since they live in different directories). The type check step (mypy) catches type errors at the function boundary level. Between the three checks, most bugs are caught before the code ever runs.


What I Don't Test (and Why)

Being honest about what you don't test matters as much as coverage numbers.

Mirth Connect's JavaScript runtime is not tested in CI. However, the contract between Mirth and FastAPI is tested: 11 unit tests in test_mirth_channel_contract.py parse the actual sample HL7 messages using the same field positions as the Mirth JavaScript transformer, build the flat JSON payload, and validate it against the real FastAPI Pydantic models. This covers AU and EU messages, HL7 escape sequences (\T\&), and invalid inputs (missing MRN → ValidationError). What's not tested is Mirth's E4X runtime semantics - its toString() behavior, subcomponent drilling, and actual MLLP framing -
which would require running Mirth in Docker. The live MLLP smoke test covers that gap manually.

HAPI FHIR profile validation is not fully tested. The validate_before_persist setting exists in the app config but the unit tests don't exercise HAPI's $validate endpoint against real profiles. The mock returns whatever I tell it to. This means I'm testing that my code sends a valid resource, not that HAPI accepts it. The end-to-end test suite (which runs against a live HAPI instance in Docker) does validate this path, but those tests are skipped in CI.

MLLP protocol handling is not tested. That's Mirth Connect's job, and Mirth has its own test suite. I test the boundary I own: the HTTP interface between Mirth and FastAPI.

Each of these gaps maps to a deliberate architectural decision. The pipeline's layers have clear boundaries specifically so each layer's tests don't need to boot the other layers.


Patterns Worth Stealing

If you're building a healthcare integration and want to set up a similar test suite, here's the minimal recipe:

Use respx for any httpx-based FHIR client. It mocks at the transport level, so your tests exercise the actual URL construction, header setting, and body serialization code. unittest.mock.patch skips all of that.

Make transformers pure functions. If build_patient(payload, profile) returns a FHIR resource without side effects, it's testable without any mocking at all. Push all IO (HTTP calls, file writes, logging) to the edges.

Parametrize valueset mappings. Gender codes, observation status codes, encounter class mappings, all of these have a finite set of valid inputs. A @pytest.mark.parametrize decorator turns the entire mapping table into test cases in one block.

Test the failure modes, not just the happy path. Healthcare data is messy. Missing fields, empty arrays, unexpected codes. Each one deserves a test that documents what the pipeline does rather than crashing with an unhandled KeyError.

Set a coverage floor, not a ceiling. My threshold is 80%. Actual coverage is 90%. The floor catches regressions without encouraging people to write meaningless tests just to hit a number.


Try It

git clone https://github.com/budityw23/maternity-hl7-to-fhir-pipeline.git
cd maternity-hl7-to-fhir-pipeline

# Install dev dependencies
cd fastapi && pip install ".[dev]" && cd ..

# Run AU profile tests
python -m pytest tests/ -v --cov=fastapi/app --cov-report=term-missing

# Run EU profile tests
PROFILE_REGION=eu PROFILE_COUNTRY=uk python -m pytest tests/ -v
Enter fullscreen mode Exit fullscreen mode

No Docker required. The unit + integration suite runs in seconds. E2E tests (tests/e2e/) need the Docker stack running - see docker compose up.

Source code: github.com/budityw23/maternity-hl7-to-fhir-pipeline


This is the fourth article in my series on the Maternity HL7-to-FHIR Pipeline. The first article covers the full architecture, the second explains the Mirth + FastAPI split, and the third shows how we added EU FHIR profile support to the same codebase.

Top comments (0)