114 lines
3.7 KiB
Python
114 lines
3.7 KiB
Python
"""Unit tests for ``node_generator`` and ``edge_generator``.
|
|
|
|
Both take a populated ``networkx.DiGraph`` and yield the tuples BioCypher
|
|
expects. The label-normalisation logic (capitalize, ``resource`` ->
|
|
``resourceType``, skipping dummy/search/meta/link) is the interesting part.
|
|
"""
|
|
|
|
import networkx as nx
|
|
import pytest
|
|
|
|
import import_fhir_to_nx_diGraph as pipeline
|
|
|
|
|
|
def _emit(generator):
|
|
return list(generator)
|
|
|
|
|
|
# --- node_generator --------------------------------------------------------
|
|
|
|
def test_resource_node_uses_resource_type_as_label():
|
|
g = nx.DiGraph()
|
|
g.add_node("p1", label="resource", resourceType="Patient", unique_id="P-1")
|
|
(node_id, label, props), = _emit(pipeline.node_generator(g))
|
|
assert node_id == "P-1"
|
|
assert label == "Patient"
|
|
|
|
|
|
def test_label_is_capitalized():
|
|
g = nx.DiGraph()
|
|
g.add_node("o1", label="observation")
|
|
(_, label, _), = _emit(pipeline.node_generator(g))
|
|
assert label == "Observation"
|
|
|
|
|
|
def test_falls_back_to_node_key_without_unique_id():
|
|
g = nx.DiGraph()
|
|
g.add_node("node-key", label="observation")
|
|
(node_id, _, _), = _emit(pipeline.node_generator(g))
|
|
assert node_id == "node-key"
|
|
|
|
|
|
@pytest.mark.parametrize("label", ["dummy", "Dummy"])
|
|
def test_dummy_nodes_are_skipped(label):
|
|
g = nx.DiGraph()
|
|
g.add_node("d", label=label)
|
|
assert _emit(pipeline.node_generator(g)) == []
|
|
|
|
|
|
@pytest.mark.parametrize("label", ["search", "meta", "link"])
|
|
def test_metadata_nodes_are_skipped(label):
|
|
g = nx.DiGraph()
|
|
g.add_node("m", label=label)
|
|
assert _emit(pipeline.node_generator(g)) == []
|
|
|
|
|
|
def test_generator_mutates_graph_label_in_place():
|
|
# Documented side effect: the generator rewrites each node's 'label' to
|
|
# its normalised form. Worth pinning because anything that iterates the
|
|
# graph afterwards sees the mutated value.
|
|
g = nx.DiGraph()
|
|
g.add_node("o1", label="observation")
|
|
_emit(pipeline.node_generator(g))
|
|
assert g.nodes["o1"]["label"] == "Observation"
|
|
|
|
|
|
def test_node_without_label_raises():
|
|
# KNOWN SHARP EDGE: a node missing both 'label' and 'resourceType' makes
|
|
# label None, and None.capitalize() raises. Pinning it as expected
|
|
# behaviour; flip this test if you decide such nodes should be skipped.
|
|
g = nx.DiGraph()
|
|
g.add_node("orphan")
|
|
with pytest.raises(AttributeError):
|
|
_emit(pipeline.node_generator(g))
|
|
|
|
|
|
# --- edge_generator --------------------------------------------------------
|
|
|
|
def test_edge_label_combines_endpoint_labels():
|
|
g = nx.DiGraph()
|
|
g.add_node("a", label="patient")
|
|
g.add_node("b", label="observation")
|
|
g.add_edge("a", "b", id="e1")
|
|
(edge_id, source, target, label, _), = _emit(pipeline.edge_generator(g))
|
|
assert (edge_id, source, target, label) == (
|
|
"e1", "a", "b", "Patient_to_Observation",
|
|
)
|
|
|
|
|
|
def test_edge_resource_endpoints_use_resource_type():
|
|
g = nx.DiGraph()
|
|
g.add_node("a", label="resource", resourceType="Patient")
|
|
g.add_node("b", label="resource", resourceType="Encounter")
|
|
g.add_edge("a", "b", id="e1")
|
|
(_, _, _, label, _), = _emit(pipeline.edge_generator(g))
|
|
assert label == "Patient_to_Encounter"
|
|
|
|
|
|
def test_edge_without_id_gets_generated_uuid():
|
|
g = nx.DiGraph()
|
|
g.add_node("a", label="patient")
|
|
g.add_node("b", label="observation")
|
|
g.add_edge("a", "b")
|
|
(edge_id, *_), = _emit(pipeline.edge_generator(g))
|
|
assert isinstance(edge_id, str) and len(edge_id) == 36 # uuid4 string
|
|
|
|
|
|
def test_edge_uses_unique_id_for_endpoints_when_present():
|
|
g = nx.DiGraph()
|
|
g.add_node("a", label="patient", unique_id="P-1")
|
|
g.add_node("b", label="observation", unique_id="O-1")
|
|
g.add_edge("a", "b", id="e1")
|
|
(_, source, target, _, _), = _emit(pipeline.edge_generator(g))
|
|
assert (source, target) == ("P-1", "O-1")
|