137 lines
5.1 KiB
Python
137 lines
5.1 KiB
Python
"""Unit tests for ``graphCreation.create_graph``.
|
|
|
|
This module turns a JSON bundle into a networkx DiGraph:
|
|
- a dict value becomes a child node ``{parent}.{key}`` (label = key) with an
|
|
edge carrying ``edge_type = key``;
|
|
- a list with no dicts is stored as a plain attribute on the parent;
|
|
- a list containing dicts becomes one child node per dict (indexed when the
|
|
list has more than one element);
|
|
- any other scalar becomes an attribute on the parent.
|
|
|
|
Pure logic, only networkx -- no stubs or mocks needed. These tests import the
|
|
real module directly (conftest prefers the real package when present).
|
|
"""
|
|
|
|
import networkx as nx
|
|
import pytest
|
|
|
|
from graphCreation.create_graph import (
|
|
add_json_to_networkx,
|
|
add_nodes_from_dict,
|
|
process_dictionaries,
|
|
)
|
|
|
|
|
|
def _graph_from(json_data, name="root"):
|
|
g = nx.DiGraph()
|
|
add_json_to_networkx(json_data, name, g)
|
|
return g
|
|
|
|
|
|
# --- entry point / guards --------------------------------------------------
|
|
|
|
def test_rejects_non_digraph():
|
|
with pytest.raises(ValueError):
|
|
add_json_to_networkx({}, "root", nx.Graph()) # undirected, not DiGraph
|
|
|
|
|
|
def test_root_node_created_with_root_label():
|
|
g = _graph_from({})
|
|
assert g.nodes["root_bundle"]["label"] == "root"
|
|
|
|
|
|
def test_bundle_suffix_is_always_appended():
|
|
# SHARP EDGE: the function appends '_bundle' to whatever name it's given.
|
|
# bundles_to_graph already passes ``id + '_bundle'``, so the in-graph root
|
|
# ends up doubly suffixed (e.g. 'patient1_bundle_bundle'). Harmless while
|
|
# consistent, but surprising if anything looks the root up by id.
|
|
g = _graph_from({}, name="patient1_bundle")
|
|
assert "patient1_bundle_bundle" in g.nodes
|
|
|
|
|
|
# --- scalars ---------------------------------------------------------------
|
|
|
|
def test_scalar_values_become_parent_attributes():
|
|
g = _graph_from({"id": "123", "active": True, "count": 5})
|
|
attrs = g.nodes["root_bundle"]
|
|
assert attrs["id"] == "123"
|
|
assert attrs["active"] is True
|
|
assert attrs["count"] == 5
|
|
|
|
|
|
# --- nested dicts ----------------------------------------------------------
|
|
|
|
def test_nested_dict_becomes_child_node_with_edge():
|
|
g = _graph_from({"name": {"family": "Mueller", "given": "Anna"}})
|
|
assert "root_bundle.name" in g.nodes
|
|
assert g.nodes["root_bundle.name"]["label"] == "name"
|
|
assert g.nodes["root_bundle.name"]["family"] == "Mueller"
|
|
assert g.edges["root_bundle", "root_bundle.name"]["edge_type"] == "name"
|
|
|
|
|
|
def test_deeply_nested_dicts_chain_through_nodes():
|
|
g = _graph_from({"a": {"b": {"c": 1}}})
|
|
assert {"root_bundle", "root_bundle.a", "root_bundle.a.b"} <= set(g.nodes)
|
|
assert g.nodes["root_bundle.a.b"]["c"] == 1
|
|
assert g.has_edge("root_bundle.a", "root_bundle.a.b")
|
|
|
|
|
|
# --- lists -----------------------------------------------------------------
|
|
|
|
def test_list_of_scalars_stored_as_attribute():
|
|
g = _graph_from({"tags": ["a", "b", "c"]})
|
|
assert g.nodes["root_bundle"]["tags"] == ["a", "b", "c"]
|
|
assert list(g.nodes) == ["root_bundle"] # no child nodes created
|
|
|
|
|
|
def test_empty_list_stored_as_attribute():
|
|
g = _graph_from({"items": []})
|
|
assert g.nodes["root_bundle"]["items"] == []
|
|
|
|
|
|
def test_single_dict_list_has_no_index_suffix():
|
|
g = _graph_from({"contact": [{"phone": "555"}]})
|
|
assert "root_bundle.contact" in g.nodes
|
|
assert "root_bundle.contact[0]" not in g.nodes
|
|
assert g.nodes["root_bundle.contact"]["phone"] == "555"
|
|
|
|
|
|
def test_multi_dict_list_creates_indexed_nodes():
|
|
g = _graph_from({"entries": [{"x": 1}, {"y": 2}]})
|
|
assert "root_bundle.entries[0]" in g.nodes
|
|
assert "root_bundle.entries[1]" in g.nodes
|
|
assert g.nodes["root_bundle.entries[0]"]["x"] == 1
|
|
assert g.nodes["root_bundle.entries[1]"]["y"] == 2
|
|
|
|
|
|
def test_mixed_list_silently_drops_non_dict_items():
|
|
# FINDING: when a list contains at least one dict, process_dictionaries
|
|
# runs and *skips* every non-dict element -- it is neither stored as an
|
|
# attribute nor turned into a node. The scalar "lost" below disappears.
|
|
# Pinned as current behaviour; flip if such data should be retained.
|
|
g = _graph_from({"mixed": ["lost", {"k": "v"}]})
|
|
assert "root_bundle.mixed[1]" in g.nodes # the dict survives (indexed)
|
|
assert g.nodes["root_bundle.mixed[1]"]["k"] == "v"
|
|
assert "mixed" not in g.nodes["root_bundle"] # not stored as an attribute
|
|
# The scalar appears nowhere in the graph.
|
|
all_attr_values = [v for _, data in g.nodes(data=True) for v in data.values()]
|
|
assert "lost" not in all_attr_values
|
|
|
|
|
|
# --- helpers used directly -------------------------------------------------
|
|
|
|
def test_add_nodes_from_dict_attaches_to_given_parent():
|
|
g = nx.DiGraph()
|
|
g.add_node("p", label="root")
|
|
add_nodes_from_dict(g, "p", {"status": "final"})
|
|
assert g.nodes["p"]["status"] == "final"
|
|
|
|
|
|
def test_process_dictionaries_skips_when_no_dicts_present():
|
|
# Defensive: passing a dict-free list straight to process_dictionaries
|
|
# creates nothing (the inner isinstance check is never satisfied).
|
|
g = nx.DiGraph()
|
|
g.add_node("p", label="root")
|
|
process_dictionaries(["a", "b"], "p", "vals", g)
|
|
assert list(g.nodes) == ["p"]
|