Merge pull request #38 from VirtualFlyBrain/connectivity-testing

test connectivity queries and build term hierarchies

Author: Clare72

.github/workflows/performance-test.yml

@@ -225,17 +225,11 @@ jobs:
           echo "" >> $GITHUB_STEP_SUMMARY
           cat performance.md >> $GITHUB_STEP_SUMMARY
           
-      - name: Commit Performance Report
-        if: always()
+      - name: Commit and Push Performance Report
+        if: always() && github.ref == 'refs/heads/main'
         run: |
           git config --local user.email "action@github.com"
           git config --local user.name "GitHub Action"
           git add performance.md
           git diff --staged --quiet || git commit -m "Update performance test results [skip ci]"
-          
-      - name: Push Performance Report
-        if: always()
-        uses: ad-m/github-push-action@master
-        with:
-          github_token: ${{ secrets.GITHUB_TOKEN }}
-          branch: ${{ github.ref }}
+          git push origin HEAD:main

.gitignore

@@ -14,3 +14,4 @@ test_results.py
 .vscode/settings.json
 temp_examples_output.txt
 json_block_*.json
+.idea/

src/test/term_info_queries_test.py

@@ -299,7 +299,7 @@ def test_term_info_serialization_neuron_class2(self):
         self.assertFalse("thumbnail" in serialized)
 
         self.assertTrue("references" in serialized)
-        self.assertEqual(7, len(serialized["references"]))
+        self.assertEqual(9, len(serialized["references"]))
 
         self.assertTrue("targetingSplits" in serialized)
         self.assertEqual(6, len(serialized["targetingSplits"]))

src/test/test_downstream_class_connectivity.py

@@ -0,0 +1,120 @@
+"""Tests for DownstreamClassConnectivity query.
+
+Tests the query that finds downstream partner neuron classes for a given
+neuron class, using the pre-indexed downstream_connectivity_query Solr field.
+"""
+
+import pytest
+import pandas as pd
+
+from vfbquery.vfb_queries import (
+    get_downstream_class_connectivity,
+    DownstreamClassConnectivity_to_schema,
+)
+
+# FBbt_00001482 = lineage NB3-2 primary interneuron — known to have
+# downstream_connectivity_query data in the vfb_json Solr core.
+TEST_CLASS = "FBbt_00001482"
+# A class that is unlikely to have downstream connectivity data.
+EMPTY_CLASS = "FBbt_00000001"
+
+
+class TestDownstreamClassConnectivityDict:
+    """Tests using return_dataframe=False (dict output)."""
+
+    @pytest.mark.integration
+    def test_returns_results(self):
+        result = get_downstream_class_connectivity(
+            TEST_CLASS, return_dataframe=False, force_refresh=True
+        )
+        assert isinstance(result, dict)
+        assert result["count"] > 0
+        assert len(result["rows"]) > 0
+
+    @pytest.mark.integration
+    def test_row_has_expected_keys(self):
+        result = get_downstream_class_connectivity(
+            TEST_CLASS, return_dataframe=False, limit=1, force_refresh=True
+        )
+        assert result["rows"], "Expected at least one row"
+        row = result["rows"][0]
+        expected_keys = {
+            "id", "downstream_class", "total_n", "connected_n",
+            "percent_connected", "pairwise_connections", "total_weight", "avg_weight",
+        }
+        assert expected_keys.issubset(row.keys())
+
+    @pytest.mark.integration
+    def test_headers_present(self):
+        result = get_downstream_class_connectivity(
+            TEST_CLASS, return_dataframe=False, limit=1, force_refresh=True
+        )
+        assert "headers" in result
+        assert "downstream_class" in result["headers"]
+
+    @pytest.mark.integration
+    def test_limit_respected(self):
+        result = get_downstream_class_connectivity(
+            TEST_CLASS, return_dataframe=False, limit=3, force_refresh=True
+        )
+        assert len(result["rows"]) <= 3
+        # count should reflect total, not the limited set
+        assert result["count"] >= len(result["rows"])
+
+    @pytest.mark.integration
+    def test_empty_class_returns_zero(self):
+        result = get_downstream_class_connectivity(
+            EMPTY_CLASS, return_dataframe=False, force_refresh=True
+        )
+        assert result["count"] == 0
+        assert result["rows"] == []
+
+
+class TestDownstreamClassConnectivityDataFrame:
+    """Tests using return_dataframe=True (DataFrame output)."""
+
+    @pytest.mark.integration
+    def test_returns_dataframe(self):
+        df = get_downstream_class_connectivity(
+            TEST_CLASS, return_dataframe=True, force_refresh=True
+        )
+        assert isinstance(df, pd.DataFrame)
+        assert not df.empty
+
+    @pytest.mark.integration
+    def test_dataframe_has_expected_columns(self):
+        df = get_downstream_class_connectivity(
+            TEST_CLASS, return_dataframe=True, limit=1, force_refresh=True
+        )
+        expected_cols = {
+            "id", "downstream_class", "total_n", "connected_n",
+            "percent_connected", "pairwise_connections", "total_weight", "avg_weight",
+        }
+        assert expected_cols.issubset(set(df.columns))
+
+    @pytest.mark.integration
+    def test_limit_respected(self):
+        df = get_downstream_class_connectivity(
+            TEST_CLASS, return_dataframe=True, limit=5, force_refresh=True
+        )
+        assert len(df) <= 5
+
+    @pytest.mark.integration
+    def test_empty_class_returns_empty_dataframe(self):
+        df = get_downstream_class_connectivity(
+            EMPTY_CLASS, return_dataframe=True, force_refresh=True
+        )
+        assert isinstance(df, pd.DataFrame)
+        assert df.empty
+
+
+class TestDownstreamClassConnectivitySchema:
+    def test_schema_generation(self):
+        schema = DownstreamClassConnectivity_to_schema(
+            "test neuron class", {"short_form": TEST_CLASS}
+        )
+        assert schema.query == "DownstreamClassConnectivity"
+        assert schema.function == "get_downstream_class_connectivity"
+        assert schema.preview == 5
+        assert "downstream_class" in schema.preview_columns
+        assert "percent_connected" in schema.preview_columns

src/test/test_hierarchy.py

@@ -0,0 +1,166 @@
+"""Tests for get_hierarchy function.
+
+Tests the hierarchy tree builder for both part_of (brain region structure)
+and subclass_of (cell type hierarchies), in both ancestor and descendant
+directions.
+"""
+
+import pytest
+
+from vfbquery.vfb_queries import get_hierarchy
+
+
+# Known test terms
+MUSHROOM_BODY = "FBbt_00005801"
+KENYON_CELL = "FBbt_00003686"
+
+
+class TestHierarchyValidation:
+    def test_invalid_relationship_raises(self):
+        with pytest.raises(ValueError, match="relationship"):
+            get_hierarchy(KENYON_CELL, relationship="invalid")
+
+    def test_invalid_direction_raises(self):
+        with pytest.raises(ValueError, match="direction"):
+            get_hierarchy(KENYON_CELL, direction="invalid")
+
+
+class TestSubclassOfDescendants:
+    @pytest.mark.integration
+    def test_returns_descendants(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'descendants', max_depth=1)
+        assert result['id'] == KENYON_CELL
+        assert result['label'] == 'Kenyon cell'
+        assert result['relationship'] == 'subclass_of'
+        assert 'descendants' in result
+        assert len(result['descendants']) > 0
+
+    @pytest.mark.integration
+    def test_descendants_have_id_and_label(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'descendants', max_depth=1)
+        for child in result['descendants']:
+            assert 'id' in child
+            assert 'label' in child
+            assert child['id'].startswith('FBbt_')
+            assert child['label'] != child['id']  # label should be resolved
+
+    @pytest.mark.integration
+    def test_depth_1_has_no_grandchildren(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'descendants', max_depth=1)
+        for child in result['descendants']:
+            assert 'descendants' not in child
+
+    @pytest.mark.integration
+    def test_depth_2_has_nested_children(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'descendants', max_depth=2)
+        has_grandchildren = any('descendants' in child for child in result['descendants'])
+        assert has_grandchildren, "At least one direct subclass should have its own subclasses"
+
+
+class TestSubclassOfAncestors:
+    @pytest.mark.integration
+    def test_returns_ancestors(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'ancestors', max_depth=1)
+        assert 'ancestors' in result
+        assert len(result['ancestors']) > 0
+
+    @pytest.mark.integration
+    def test_ancestors_have_id_and_label(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'ancestors', max_depth=1)
+        for anc in result['ancestors']:
+            assert 'id' in anc
+            assert 'label' in anc
+
+    @pytest.mark.integration
+    def test_kenyon_cell_ancestor_is_mb_intrinsic_neuron(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'ancestors', max_depth=1)
+        ancestor_ids = [a['id'] for a in result['ancestors']]
+        assert 'FBbt_00007484' in ancestor_ids  # mushroom body intrinsic neuron
+
+    @pytest.mark.integration
+    def test_depth_2_has_nested_ancestors(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'ancestors', max_depth=2)
+        has_grandparent = any('ancestors' in anc for anc in result['ancestors'])
+        assert has_grandparent
+
+
+class TestPartOfDescendants:
+    @pytest.mark.integration
+    def test_returns_parts(self):
+        result = get_hierarchy(MUSHROOM_BODY, 'part_of', 'descendants', max_depth=1)
+        assert result['id'] == MUSHROOM_BODY
+        assert result['label'] == 'mushroom body'
+        assert 'descendants' in result
+        assert len(result['descendants']) > 0
+
+    @pytest.mark.integration
+    def test_parts_have_id_and_label(self):
+        result = get_hierarchy(MUSHROOM_BODY, 'part_of', 'descendants', max_depth=1)
+        for part in result['descendants']:
+            assert 'id' in part
+            assert 'label' in part
+            assert part['id'].startswith('FBbt_')
+
+
+class TestPartOfAncestors:
+    @pytest.mark.integration
+    def test_mushroom_body_part_of_protocerebrum(self):
+        result = get_hierarchy(MUSHROOM_BODY, 'part_of', 'ancestors', max_depth=1)
+        assert 'ancestors' in result
+        ancestor_ids = [a['id'] for a in result['ancestors']]
+        assert 'FBbt_00003627' in ancestor_ids  # protocerebrum
+
+
+class TestDisplayOutput:
+    @pytest.mark.integration
+    def test_display_field_present(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'both', max_depth=1)
+        assert 'display' in result
+        assert isinstance(result['display'], str)
+        assert 'Kenyon cell' in result['display']
+
+    @pytest.mark.integration
+    def test_display_shows_ancestors(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'both', max_depth=1)
+        assert 'ancestors' in result['display'].lower()
+        assert 'mushroom body intrinsic neuron' in result['display']
+
+    @pytest.mark.integration
+    def test_display_shows_tree_connectors(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'descendants', max_depth=1)
+        assert '├──' in result['display'] or '└──' in result['display']
+
+    @pytest.mark.integration
+    def test_html_field_present(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'both', max_depth=1)
+        assert 'html' in result
+        assert '<!DOCTYPE html>' in result['html']
+        assert 'Kenyon cell' in result['html']
+
+    @pytest.mark.integration
+    def test_html_contains_vfb_links(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'descendants', max_depth=1)
+        assert 'virtualflybrain.org' in result['html']
+        assert KENYON_CELL in result['html']
+
+
+class TestBothDirections:
+    @pytest.mark.integration
+    def test_both_returns_ancestors_and_descendants(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'both', max_depth=1)
+        assert 'ancestors' in result
+        assert 'descendants' in result
+        assert len(result['ancestors']) > 0
+        assert len(result['descendants']) > 0
+
+    @pytest.mark.integration
+    def test_descendants_only_has_no_ancestors(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'descendants', max_depth=1)
+        assert 'descendants' in result
+        assert 'ancestors' not in result
+
+    @pytest.mark.integration
+    def test_ancestors_only_has_no_descendants(self):
+        result = get_hierarchy(KENYON_CELL, 'subclass_of', 'ancestors', max_depth=1)
+        assert 'ancestors' in result
+        assert 'descendants' not in result