Adding CLI tests

.gitignore

@@ -1,3 +1,7 @@
+.idea/
+*.duckdb
+tests/output/
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

closurizer/closurizer.py

@@ -145,13 +145,14 @@ def add_closure(kg_archive: str,
 
     print(edges_query)
 
+    additional_node_constraints = f"where {additional_node_constraints}" if additional_node_constraints else ""
     nodes_query = f"""        
     create or replace table denormalized_nodes as
     select nodes.*, 
         {"".join([node_columns(node_field) for node_field in node_fields])}
     from nodes
         {node_joins('has_phenotype')}
-    where {additional_node_constraints}
+    {additional_node_constraints}
     group by nodes.*
     """
     print(nodes_query)

pyproject.toml

@@ -15,6 +15,9 @@ duckdb = "^0.9.1"
 [tool.poetry.scripts]
 closurizer = "closurizer.cli:main"
 
+[tool.poetry.group.dev.dependencies]
+pytest = "^8.1.1"
+
 [build-system]
 requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"

tests/__init__.py

@@ -0,0 +1,6 @@
+"""Tests for bibliomancer."""
+
+from pathlib import Path
+
+INPUT_DIR = Path(__file__).parent / "input"
+OUTPUT_DIR = Path(__file__).parent / "output"

tests/conftest.py

@@ -0,0 +1,8 @@
+import pytest
+from click.testing import CliRunner
+
+
+@pytest.fixture
+def runner() -> CliRunner:
+    runner = CliRunner()
+    return runner

tests/input/rg.tsv

@@ -0,0 +1,10 @@
+subject	predicate	object
+X:4	rdfs:subClassOf	X:4
+X:4	rdfs:subClassOf	X:3
+X:4	rdfs:subClassOf	X:2
+X:4	rdfs:subClassOf	X:1
+X:3	rdfs:subClassOf	X:3
+X:3	rdfs:subClassOf	X:2
+X:3	rdfs:subClassOf	X:1
+X:2	rdfs:subClassOf	X:2
+X:2	rdfs:subClassOf	X:1

tests/input/test_edges.tsv

@@ -0,0 +1,5 @@
+subject	predicate	object	has_evidence	publications	negated
+X:1	biolink:related_to	Y:1	ECO:1	PMID:1	False
+X:2	biolink:related_to	Y:2	ECO:1	PMID:1	False
+X:3	biolink:related_to	Y:3	ECO:1	PMID:1	False
+X:4	biolink:related_to	Y:4	ECO:1	PMID:1	True

tests/input/test_nodes.tsv

@@ -0,0 +1,9 @@
+id	name	category	in_taxon	in_taxon_label
+X:1	x1	Gene	NCBITaxon:9606	human
+X:2	x2	Gene	NCBITaxon:9606	human
+X:3	x3	Gene	NCBITaxon:9606	human
+X:4	x4	Gene	NCBITaxon:9606	human
+Y:1	y1	Gene	NCBITaxon:9606	human
+Y:2	y2	Gene	NCBITaxon:9606	human
+Y:3	y3	Gene	NCBITaxon:9606	human
+Y:4	y4	Gene	NCBITaxon:9606	human

tests/test_cli.py

@@ -0,0 +1,50 @@
+import csv
+
+import pytest
+
+from closurizer.cli import main
+from tests import INPUT_DIR, OUTPUT_DIR
+
+
+def test_help(runner):
+    """
+    Tests help message
+
+    :param runner:
+    :return:
+    """
+    result = runner.invoke(main, ["--help"])
+    assert result.exit_code == 0
+    assert "edges" in result.output
+
+
+def test_cli_run(runner):
+    """
+    Tests closurize command
+
+    :param runner:
+    :return:
+    """
+    kg_file = INPUT_DIR / "bundle.tar.gz"
+    rg_file = INPUT_DIR / "rg.tsv"
+    output_node_file = OUTPUT_DIR / "nodes.csv"
+    output_edges_file = OUTPUT_DIR / "edges-denorm.csv"
+    OUTPUT_DIR.mkdir(exist_ok=True, parents=True)
+    result = runner.invoke(main, ["--kg", kg_file, "--closure", rg_file, "--nodes-output", output_node_file, "--edges-output", output_edges_file])
+    if result.exit_code != 0:
+        print(result.output)
+    assert result.exit_code == 0
+    assert output_node_file.exists()
+    found = False
+    with open(output_edges_file, "r") as f:
+        reader = csv.DictReader(f, delimiter="\t")
+        rows = list(reader)
+        for row in rows:
+            print(row)
+            if row["subject"] == "X:4" and row["object"] == "Y:4":
+                found = True
+                assert set(row["subject_closure"].split("|")) == {"X:4", "X:3", "X:2", "X:1"}
+    assert found
+
+
+