test_profile_bp.py

"""
US Core Blood Pressure Profile Class API Tests

Mirrors examples/typescript-us-core/profile-bp.test.ts.
"""

import warnings

from fhir_types.hl7_fhir_r4_core.base import CodeableConcept, Coding, Quantity, Reference
from fhir_types.hl7_fhir_r4_core.observation import Observation, ObservationComponent
from fhir_types.hl7_fhir_r4_core.base import Meta
from fhir_types.hl7_fhir_us_core.profiles.observation_uscore_blood_pressure_profile import UscoreBloodPressureProfile

# Pydantic warns when extensions list contains plain dicts instead of Extension
# model instances — this is expected with the current push_extension approach.
warnings.filterwarnings("ignore", category=UserWarning, module="pydantic")


CANONICAL_URL = "http://hl7.org/fhir/us/core/StructureDefinition/us-core-blood-pressure"
VSCAT_CODING = Coding(code="vital-signs", system="http://terminology.hl7.org/CodeSystem/observation-category")


def _make_bp() -> UscoreBloodPressureProfile:
    return UscoreBloodPressureProfile.create(
        status="final",
        subject=Reference(reference="Patient/pt-1"),
    )


# ---------------------------------------------------------------------------
# demo
# ---------------------------------------------------------------------------


def test_import_profiled_observation_from_api_and_read_components() -> None:
    api_response = Observation(
        resourceType="Observation",
        meta=Meta(profile=[CANONICAL_URL]),
        status="final",
        category=[CodeableConcept(coding=[VSCAT_CODING])],
        code=CodeableConcept(coding=[Coding(code="85354-9", system="http://loinc.org", display="Blood pressure panel")]),
        subject=Reference(reference="Patient/pt-1"),
        effectiveDateTime="2024-06-15",
        component=[
            ObservationComponent(
                code=CodeableConcept(coding=[Coding(code="8480-6", system="http://loinc.org")]),
                valueQuantity=Quantity(value=120, unit="mmHg", system="http://unitsofmeasure.org", code="mm[Hg]"),
            ),
            ObservationComponent(
                code=CodeableConcept(coding=[Coding(code="8462-4", system="http://loinc.org")]),
                valueQuantity=Quantity(value=80, unit="mmHg", system="http://unitsofmeasure.org", code="mm[Hg]"),
            ),
        ],
    )

    profile = UscoreBloodPressureProfile.from_resource(api_response)

    assert profile.get_systolic() == {
        "value": 120,
        "unit": "mmHg",
        "system": "http://unitsofmeasure.org",
        "code": "mm[Hg]",
    }
    assert profile.get_diastolic() == {
        "value": 80,
        "unit": "mmHg",
        "system": "http://unitsofmeasure.org",
        "code": "mm[Hg]",
    }
    assert profile.get_effective_date_time() == "2024-06-15"


def test_apply_profile_to_bare_observation_and_populate_it() -> None:
    bare_observation = Observation(resourceType="Observation", status="preliminary", code=CodeableConcept())
    profile = UscoreBloodPressureProfile.apply(bare_observation)

    profile.set_status("final")
    profile.set_code(CodeableConcept(coding=[Coding(code="85354-9", system="http://loinc.org")]))
    profile.set_subject(Reference(reference="Patient/pt-1"))
    profile.set_vscat({})
    profile.set_effective_date_time("2024-06-15")
    profile.set_systolic({"value": 120, "unit": "mmHg"})
    profile.set_diastolic({"value": 80, "unit": "mmHg"})

    assert profile.validate()["errors"] == []
    meta = profile.to_resource().meta
    assert meta is not None
    assert meta.profile is not None
    assert CANONICAL_URL in meta.profile


# ---------------------------------------------------------------------------
# US Core blood pressure profile
# ---------------------------------------------------------------------------


def test_canonical_url_is_exposed() -> None:
    assert UscoreBloodPressureProfile.canonical_url == CANONICAL_URL


def test_create_auto_sets_code_and_meta_profile() -> None:
    profile = _make_bp()
    obs = profile.to_resource()
    assert obs.resourceType == "Observation"
    coding = obs.code.coding
    assert coding is not None
    assert coding[0].code == "85354-9"
    assert coding[0].system == "http://loinc.org"
    meta = obs.meta
    assert meta is not None
    assert meta.profile == [CANONICAL_URL]


def test_freshly_created_profile_is_not_yet_valid_missing_effective() -> None:
    profile = _make_bp()
    errors = profile.validate()["errors"]
    assert errors == [
        "UscoreBloodPressureProfile: at least one of effectiveDateTime, effectivePeriod is required",
    ]


def test_create_auto_populates_component_with_systolic_diastolic_stubs() -> None:
    profile = _make_bp()
    obs = profile.to_resource()
    assert obs.component is not None
    assert len(obs.component) == 2


def test_set_systolic_get_systolic_get_systolic_raw() -> None:
    profile = _make_bp()
    profile.set_systolic({"value": 120, "unit": "mmHg", "system": "http://unitsofmeasure.org", "code": "mm[Hg]"})

    assert profile.get_systolic() == {
        "value": 120,
        "unit": "mmHg",
        "system": "http://unitsofmeasure.org",
        "code": "mm[Hg]",
    }

    raw = profile.get_systolic("raw")
    assert raw is not None
    assert raw.valueQuantity is not None
    assert raw.valueQuantity.value == 120
    coding = raw.code.coding
    assert coding is not None
    assert coding[0].code == "8480-6"


def test_set_diastolic_get_diastolic_get_diastolic_raw() -> None:
    profile = _make_bp()
    profile.set_diastolic({"value": 80, "unit": "mmHg", "system": "http://unitsofmeasure.org", "code": "mm[Hg]"})

    assert profile.get_diastolic() == {
        "value": 80,
        "unit": "mmHg",
        "system": "http://unitsofmeasure.org",
        "code": "mm[Hg]",
    }

    raw = profile.get_diastolic("raw")
    assert raw is not None
    assert raw.valueQuantity is not None
    assert raw.valueQuantity.value == 80
    coding = raw.code.coding
    assert coding is not None
    assert coding[0].code == "8462-4"


def test_both_systolic_and_diastolic_are_in_the_component_array() -> None:
    profile = _make_bp()
    obs = profile.to_resource()
    assert obs.component is not None
    assert len(obs.component) == 2


def test_set_systolic_replaces_an_existing_systolic_component() -> None:
    profile = _make_bp()
    profile.set_systolic({"value": 130, "unit": "mmHg"})
    obs = profile.to_resource()
    assert obs.component is not None
    assert len(obs.component) == 2
    raw = profile.get_systolic("raw")
    assert raw is not None
    assert raw.valueQuantity is not None
    assert raw.valueQuantity.value == 130


def test_set_vscat_adds_category_with_discriminator_values() -> None:
    profile = _make_bp()
    profile.set_vscat({"text": "Vital Signs"})
    flat = profile.get_vscat()
    assert flat is not None
    assert flat["text"] == "Vital Signs"


def test_set_effective_date_time_get_effective_date_time() -> None:
    profile = _make_bp()
    profile.set_effective_date_time("2024-06-15T10:30:00Z")
    assert profile.get_effective_date_time() == "2024-06-15T10:30:00Z"
    assert profile.get_value_quantity() is None


def test_fluent_chaining_across_all_accessor_types() -> None:
    profile = _make_bp()
    result = (
        profile.set_status("final")
        .set_vscat({"text": "Vital Signs"})
        .set_effective_date_time("2024-06-15")
        .set_subject(Reference(reference="Patient/pt-2"))
    )
    assert result is profile
    assert profile.get_status() == "final"
    vscat = profile.get_vscat()
    assert vscat is not None
    assert vscat["text"] == "Vital Signs"
    assert profile.get_effective_date_time() == "2024-06-15"
    subject = profile.get_subject()
    assert subject is not None
    assert subject.reference == "Patient/pt-2"


def test_set_systolic_with_no_args_inserts_discriminator_only_component() -> None:
    profile = _make_bp()
    profile.set_systolic()
    assert profile.get_systolic() is not None


def test_create_with_custom_category_preserves_user_values_and_adds_required_vscat() -> None:
    custom = UscoreBloodPressureProfile.create(
        status="final",
        subject=Reference(reference="Patient/pt-1"),
        category=[CodeableConcept(text="My Category")],
    )
    obs = custom.to_resource()
    assert obs.category is not None
    assert len(obs.category) == 2


def test_create_with_empty_category_still_adds_required_vscat() -> None:
    custom = UscoreBloodPressureProfile.create(
        status="final",
        subject=Reference(reference="Patient/pt-1"),
        category=[],
    )
    obs = custom.to_resource()
    assert obs.category is not None
    assert len(obs.category) == 1


def test_create_with_category_already_containing_vscat_does_not_duplicate_it() -> None:
    custom = UscoreBloodPressureProfile.create(
        status="final",
        subject=Reference(reference="Patient/pt-1"),
        category=[CodeableConcept(coding=[VSCAT_CODING])],
    )
    obs = custom.to_resource()
    assert obs.category is not None
    assert len(obs.category) == 1