"""Unit tests for ``generate_neo4j_import_script``. It scans a directory of BioCypher CSV output, classifies each ``*-header.csv`` as a node or relationship, and emits a ``neo4j-admin import`` shell script. We drive it with a temporary directory so no real ``/neo4j_import`` is needed. """ import os import stat import import_fhir_to_nx_diGraph as pipeline def _touch(directory, name): (directory / name).write_text("") def _generate(tmp_path): ret = pipeline.generate_neo4j_import_script( directory_path=str(tmp_path) + os.sep, output_file="neo4j-admin-import-call.sh", ) out = tmp_path / "neo4j-admin-import-call.sh" return out.read_text(), out, ret def test_node_with_parts_is_included_as_nodes(tmp_path): _touch(tmp_path, "Patient-header.csv") _touch(tmp_path, "Patient-part0.csv") script, _, _ = _generate(tmp_path) assert '--nodes="/neo4j_import/Patient-header.csv,/neo4j_import/Patient-part.*"' in script def test_association_is_classified_as_relationship(tmp_path): _touch(tmp_path, "PatientToObservationAssociation-header.csv") _touch(tmp_path, "PatientToObservationAssociation-part0.csv") script, _, _ = _generate(tmp_path) assert "--relationships=" in script assert "PatientToObservationAssociation" in script def test_has_pattern_is_classified_as_relationship(tmp_path): _touch(tmp_path, "Encounter_has_Diagnosis-header.csv") _touch(tmp_path, "Encounter_has_Diagnosis-part0.csv") script, _, _ = _generate(tmp_path) assert "--relationships=" in script assert "--nodes=" not in script def test_header_without_part_files_is_dropped(tmp_path): # SHARP EDGE: entities are only emitted when a matching *-part* file # exists. A lone header silently produces no import command. Pinning so # the behaviour is intentional, not a surprise empty graph. _touch(tmp_path, "Patient-header.csv") # no part file script, _, _ = _generate(tmp_path) assert "--nodes=" not in script assert "--relationships=" not in script def test_script_is_written_and_executable(tmp_path): _touch(tmp_path, "Patient-header.csv") _touch(tmp_path, "Patient-part0.csv") _, out, ret = _generate(tmp_path) assert out.exists() mode = os.stat(out).st_mode assert mode & stat.S_IXUSR # owner-executable # SHARP EDGE: the docstring promises it returns the script path, but the # function has no `return`, so callers actually get None. Pinning the # real behaviour; fix the function (or the docstring) and flip this. assert ret is None def test_script_contains_version_branching(tmp_path): _touch(tmp_path, "Patient-header.csv") _touch(tmp_path, "Patient-part0.csv") script, _, _ = _generate(tmp_path) assert "neo4j-admin database import full" in script # v5+ branch assert "neo4j-admin import --database=neo4j" in script # legacy branch