tests for connectivity queries - always expect data

Author: Clare72

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_neuron_neuron_connectivity.py

@@ -1,89 +1,118 @@
-#!/usr/bin/env python3
-"""
-Test suite for NeuronNeuronConnectivityQuery.
+"""Tests for NeuronNeuronConnectivityQuery.
 
 Tests the query that finds neurons connected to a given neuron.
 This implements the neuron_neuron_connectivity_query from the VFB XMI specification.
-
-Test cases:
-1. Query execution with known neuron
-2. Schema generation and validation
-3. Term info integration (if applicable)
-4. Preview results validation
 """
 
-import unittest
-import sys
-import os
-
-# Add the src directory to the path
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+import pytest
+import pandas as pd
 
 from vfbquery.vfb_queries import (
     get_neuron_neuron_connectivity,
     NeuronNeuronConnectivityQuery_to_schema,
-    get_term_info
 )
 
-class NeuronNeuronConnectivityTest(unittest.TestCase):
-    """Test suite for neuron_neuron_connectivity_query"""
-
-    def setUp(self):
-        """Set up test fixtures"""
-        # Test neuron: LPC1 (FlyEM-HB:1775513344) [VFB_jrchk00s]
-        self.test_neuron = "VFB_jrchk00s"
-
-    def test_query_execution(self):
-        """Test that the query executes successfully"""
-        print(f"\n=== Testing neuron_neuron_connectivity_query execution ===")
-        result = get_neuron_neuron_connectivity(self.test_neuron, return_dataframe=False, limit=5)
-        self.assertIsNotNone(result, "Query should return a result")
-        self.assertIsInstance(result, dict, "Result should be a dictionary")
-        print(f"Query returned {result.get('count', 0)} results")
-        if 'data' in result and len(result['data']) > 0:
-            first_result = result['data'][0]
-            self.assertIn('id', first_result, "Result should contain 'id' field")
-            self.assertIn('label', first_result, "Result should contain 'label' field")
-            print(f"First result: {first_result.get('label', 'N/A')} ({first_result.get('id', 'N/A')})")
-        else:
-            print("No connected neurons found (this is OK if none exist)")
+# VFB_jrchk00s = LPC1 (FlyEM-HB:1775513344) — known to have connectivity data.
+TEST_NEURON = "VFB_jrchk00s"
+
+
+class TestNeuronNeuronConnectivityDict:
+    """Tests using return_dataframe=False (dict output)."""
+
+    @pytest.mark.integration
+    def test_returns_results(self):
+        result = get_neuron_neuron_connectivity(
+            TEST_NEURON, return_dataframe=False
+        )
+        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_neuron_neuron_connectivity(
+            TEST_NEURON, return_dataframe=False, limit=1
+        )
+        assert result["rows"], "Expected at least one row"
+        row = result["rows"][0]
+        expected_keys = {"id", "label", "outputs", "inputs", "tags"}
+        assert expected_keys.issubset(row.keys())
+
+    @pytest.mark.integration
+    def test_headers_present(self):
+        result = get_neuron_neuron_connectivity(
+            TEST_NEURON, return_dataframe=False, limit=1
+        )
+        assert "headers" in result
+        assert "label" in result["headers"]
+        assert "outputs" in result["headers"]
+        assert "inputs" in result["headers"]
+
+    @pytest.mark.integration
+    def test_limit_respected(self):
+        result = get_neuron_neuron_connectivity(
+            TEST_NEURON, return_dataframe=False, limit=3
+        )
+        assert len(result["rows"]) <= 3
+        assert result["count"] >= len(result["rows"])
+
+    @pytest.mark.integration
+    def test_direction_upstream(self):
+        all_result = get_neuron_neuron_connectivity(
+            TEST_NEURON, return_dataframe=False
+        )
+        up_result = get_neuron_neuron_connectivity(
+            TEST_NEURON, return_dataframe=False, direction='upstream'
+        )
+        assert up_result["count"] > 0
+        assert up_result["count"] <= all_result["count"]
+
+    @pytest.mark.integration
+    def test_direction_downstream(self):
+        all_result = get_neuron_neuron_connectivity(
+            TEST_NEURON, return_dataframe=False
+        )
+        down_result = get_neuron_neuron_connectivity(
+            TEST_NEURON, return_dataframe=False, direction='downstream'
+        )
+        assert down_result["count"] > 0
+        assert down_result["count"] <= all_result["count"]
+
+
+class TestNeuronNeuronConnectivityDataFrame:
+    """Tests using return_dataframe=True (DataFrame output)."""
+
+    @pytest.mark.integration
+    def test_returns_dataframe(self):
+        df = get_neuron_neuron_connectivity(
+            TEST_NEURON, return_dataframe=True
+        )
+        assert isinstance(df, pd.DataFrame)
+        assert not df.empty
+
+    @pytest.mark.integration
+    def test_dataframe_has_expected_columns(self):
+        df = get_neuron_neuron_connectivity(
+            TEST_NEURON, return_dataframe=True, limit=1
+        )
+        expected_cols = {"id", "label", "outputs", "inputs", "tags"}
+        assert expected_cols.issubset(set(df.columns))
+
+    @pytest.mark.integration
+    def test_limit_respected(self):
+        df = get_neuron_neuron_connectivity(
+            TEST_NEURON, return_dataframe=True, limit=5
+        )
+        assert len(df) <= 5
+
 
+class TestNeuronNeuronConnectivitySchema:
     def test_schema_generation(self):
-        """Test schema function generates correct structure"""
-        print(f"\n=== Testing neuron_neuron_connectivity_query schema generation ===")
-        test_name = "LPC1"
-        test_takes = {"short_form": self.test_neuron}
-        schema = NeuronNeuronConnectivityQuery_to_schema(test_name, test_takes)
-        self.assertIsNotNone(schema, "Schema should not be None")
-        self.assertEqual(schema.query, "NeuronNeuronConnectivityQuery", "Query name should match")
-        self.assertEqual(schema.label, f"Neurons connected to {test_name}", "Label should be formatted correctly")
-        self.assertEqual(schema.function, "get_neuron_neuron_connectivity", "Function name should match")
-        self.assertEqual(schema.preview, 5, "Preview should be 5")
-        expected_columns = ["id", "label", "outputs", "inputs", "tags"]
-        self.assertEqual(schema.preview_columns, expected_columns, f"Preview columns should be {expected_columns}")
-        print(f"Schema generated successfully: {schema.label}")
-
-    def test_preview_results(self):
-        """Test that preview results are properly formatted"""
-        print(f"\n=== Testing preview results ===")
-        result = get_neuron_neuron_connectivity(self.test_neuron, return_dataframe=False, limit=3)
-        self.assertIsNotNone(result, "Query should return a result")
-        if 'data' in result and len(result['data']) > 0:
-            first_result = result['data'][0]
-            self.assertIn('id', first_result, "Preview result should have 'id'")
-            self.assertIn('label', first_result, "Preview result should have 'label'")
-            print(f"First preview result: {first_result.get('label', 'N/A')}")
-        else:
-            print("No preview results available (this is OK if no connected neurons exist)")
-
-
-def run_tests():
-    """Run the test suite"""
-    suite = unittest.TestLoader().loadTestsFromTestCase(NeuronNeuronConnectivityTest)
-    runner = unittest.TextTestRunner(verbosity=2)
-    result = runner.run(suite)
-    return result.wasSuccessful()
-
-if __name__ == '__main__':
-    success = run_tests()
-    sys.exit(0 if success else 1)
+        schema = NeuronNeuronConnectivityQuery_to_schema(
+            "LPC1", {"short_form": TEST_NEURON}
+        )
+        assert schema.query == "NeuronNeuronConnectivityQuery"
+        assert schema.function == "get_neuron_neuron_connectivity"
+        assert schema.label == "Neurons connected to LPC1"
+        assert schema.preview == 5
+        assert schema.preview_columns == ["id", "label", "outputs", "inputs", "tags"]