From 6324b2d7078da9d57f4f709a65935d40c0c17a52 Mon Sep 17 00:00:00 2001
From: Wang Jianan <wangjianan.wjn@gmail.com>
Date: Thu, 3 Sep 2026 19:50:11 -0700
Subject: [PATCH] test: cover get_dd_changelog and signal_analytics MCP tools

Both tools had no test references. Add mock-backed unit tests that
pin the Cypher parameter contract, query routing, result shaping,
error handling, and markdown formatting. No source changes.

- tests/tools/test_dd_changelog.py: VersionTool.get_dd_changelog and
  format_dd_changelog_report (15 tests)
- tests/llm/test_signal_analytics.py: _signal_analytics, its two query
  shapes, and _format_analytics (33 tests)

The unknown-filter-key test pins the current behaviour (keys are
silently dropped, unlike group_by which is validated) so that a
deliberate change shows up in review.
---
 tests/llm/test_signal_analytics.py | 265 +++++++++++++++++++++++++++++
 tests/tools/test_dd_changelog.py   | 265 +++++++++++++++++++++++++++++
 2 files changed, 530 insertions(+)
 create mode 100644 tests/llm/test_signal_analytics.py
 create mode 100644 tests/tools/test_dd_changelog.py

diff --git a/tests/llm/test_signal_analytics.py b/tests/llm/test_signal_analytics.py
new file mode 100644
index 000000000..b79b385a2
--- /dev/null
+++ b/tests/llm/test_signal_analytics.py
@@ -0,0 +1,265 @@
+"""Tests for the ``signal_analytics`` MCP tool implementation.
+
+``_signal_analytics`` turns a facility + group_by + filters request into one
+of two Cypher shapes (with or without the ``CHECKED_WITH`` join) and renders
+the counts as a markdown table. These tests pin input validation, query
+routing, parameterisation, error handling, and the formatter — all against a
+mocked GraphClient, no live Neo4j.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+from neo4j.exceptions import ServiceUnavailable
+
+from imas_codex.llm.search_tools import (
+    _ALLOWED_GROUP_BY,
+    NEO4J_NOT_RUNNING_MSG,
+    _format_analytics,
+    _signal_analytics,
+)
+
+
+@pytest.fixture
+def gc() -> MagicMock:
+    client = MagicMock()
+    client.query.return_value = []
+    return client
+
+
+def _issued(gc: MagicMock) -> tuple[str, dict]:
+    """Return (cypher, params) of the single query the tool issued."""
+    gc.query.assert_called_once()
+    (cypher,), params = gc.query.call_args
+    return cypher, params
+
+
+# ============================================================================
+# group_by validation
+# ============================================================================
+
+
+class TestGroupByValidation:
+    def test_default_group_by_is_status(self, gc):
+        _signal_analytics("tcv", gc=gc)
+
+        cypher, params = _issued(gc)
+        assert "s.status AS status" in cypher
+        assert params == {"facility": "tcv"}
+
+    def test_invalid_dimension_rejected_before_querying(self, gc):
+        result = _signal_analytics("tcv", group_by=["status", "bogus"], gc=gc)
+
+        assert result.startswith("Invalid group_by dimensions")
+        assert "bogus" in result
+        assert "status" in result  # allowed list is echoed back
+        gc.query.assert_not_called()
+
+    @pytest.mark.parametrize("dim", sorted(_ALLOWED_GROUP_BY))
+    def test_every_allowed_dimension_is_accepted(self, gc, dim):
+        result = _signal_analytics("tcv", group_by=[dim], gc=gc)
+
+        gc.query.assert_called_once()
+        assert not result.startswith("Invalid")
+
+
+# ============================================================================
+# Query routing: simple vs CHECKED_WITH join
+# ============================================================================
+
+
+class TestQueryRouting:
+    def test_plain_dimensions_use_simple_query(self, gc):
+        _signal_analytics("tcv", group_by=["physics_domain", "diagnostic"], gc=gc)
+
+        cypher, _ = _issued(gc)
+        assert "CHECKED_WITH" not in cypher
+        assert (
+            "s.physics_domain AS physics_domain, s.diagnostic AS diagnostic" in cypher
+        )
+        assert "count(s) AS count" in cypher
+        assert "ORDER BY count DESC" in cypher
+
+    @pytest.mark.parametrize("dim", ["check_status", "error_type"])
+    def test_check_dimensions_join_checked_with(self, gc, dim):
+        _signal_analytics("tcv", group_by=[dim], gc=gc)
+
+        cypher, _ = _issued(gc)
+        assert "OPTIONAL MATCH (s)-[c:CHECKED_WITH]->()" in cypher
+        assert "count(DISTINCT s) AS count" in cypher
+
+    def test_check_status_is_derived_from_success_flag(self, gc):
+        _signal_analytics("tcv", group_by=["check_status"], gc=gc)
+
+        cypher, _ = _issued(gc)
+        assert "WHEN c IS NULL THEN 'unchecked'" in cypher
+        assert "WHEN c.success = true THEN 'passed'" in cypher
+        assert "ELSE 'failed' END AS check_status" in cypher
+
+    def test_error_type_dimension_reads_relationship_property(self, gc):
+        _signal_analytics("tcv", group_by=["error_type"], gc=gc)
+
+        cypher, _ = _issued(gc)
+        assert "c.error_type AS error_type" in cypher
+
+    def test_check_filter_alone_routes_to_join_query(self, gc):
+        _signal_analytics(
+            "tcv", group_by=["status"], filters={"check_status": "failed"}, gc=gc
+        )
+
+        cypher, _ = _issued(gc)
+        assert "CHECKED_WITH" in cypher
+        assert "c.success = false" in cypher
+
+    @pytest.mark.parametrize(
+        ("value", "clause"),
+        [
+            ("passed", "c.success = true"),
+            ("failed", "c.success = false"),
+            ("unchecked", "c IS NULL"),
+        ],
+    )
+    def test_check_status_filter_values(self, gc, value, clause):
+        _signal_analytics("tcv", filters={"check_status": value}, gc=gc)
+
+        cypher, _ = _issued(gc)
+        assert clause in cypher
+
+    def test_unknown_check_status_value_adds_no_clause(self, gc):
+        _signal_analytics("tcv", filters={"check_status": "maybe"}, gc=gc)
+
+        cypher, _ = _issued(gc)
+        assert "CHECKED_WITH" in cypher
+        assert "c.success" not in cypher
+        assert "c IS NULL" not in cypher
+
+    def test_error_type_filter_is_parameterised(self, gc):
+        _signal_analytics("tcv", filters={"error_type": "timeout"}, gc=gc)
+
+        cypher, params = _issued(gc)
+        assert "c.error_type = $f_error_type" in cypher
+        assert params["f_error_type"] == "timeout"
+        assert "timeout" not in cypher
+
+
+# ============================================================================
+# Node-property filters
+# ============================================================================
+
+
+class TestFilters:
+    def test_allowed_filter_becomes_where_clause_and_parameter(self, gc):
+        _signal_analytics("tcv", filters={"physics_domain": "magnetics"}, gc=gc)
+
+        cypher, params = _issued(gc)
+        assert "s.physics_domain = $f_physics_domain" in cypher
+        assert params == {"facility": "tcv", "f_physics_domain": "magnetics"}
+
+    def test_filter_values_are_never_interpolated(self, gc):
+        hostile = "x' OR 1=1 //"
+        _signal_analytics("tcv", filters={"status": hostile}, gc=gc)
+
+        cypher, params = _issued(gc)
+        assert hostile not in cypher
+        assert params["f_status"] == hostile
+
+    @pytest.mark.parametrize("group_by", [["status"], ["check_status"]])
+    def test_unknown_filter_key_is_silently_dropped(self, gc, group_by):
+        """Pins current behaviour: unknown filter keys are ignored, not rejected.
+
+        This is asymmetric with ``group_by`` (which is validated). If that is
+        changed to an error, this test should flip to assert the message and
+        ``gc.query.assert_not_called()``.
+        """
+        _signal_analytics("tcv", group_by=group_by, filters={"bogus": "x"}, gc=gc)
+
+        cypher, params = _issued(gc)
+        assert "bogus" not in cypher
+        assert "f_bogus" not in params
+
+    def test_node_filters_also_apply_in_join_query(self, gc):
+        _signal_analytics(
+            "tcv",
+            group_by=["check_status"],
+            filters={"diagnostic": "magnetics", "check_status": "passed"},
+            gc=gc,
+        )
+
+        cypher, params = _issued(gc)
+        assert "s.diagnostic = $f_diagnostic" in cypher
+        assert params["f_diagnostic"] == "magnetics"
+        assert "c.success = true" in cypher
+
+
+# ============================================================================
+# Error handling
+# ============================================================================
+
+
+class TestErrorHandling:
+    def test_service_unavailable_returns_setup_hint(self, gc):
+        gc.query.side_effect = ServiceUnavailable("connection refused")
+
+        assert _signal_analytics("tcv", gc=gc) == NEO4J_NOT_RUNNING_MSG
+
+    def test_other_errors_are_reported_not_raised(self, gc):
+        gc.query.side_effect = RuntimeError("kaboom")
+
+        result = _signal_analytics("tcv", gc=gc)
+
+        assert result.startswith("Analytics error:")
+
+
+# ============================================================================
+# Formatting
+# ============================================================================
+
+
+class TestFormatAnalytics:
+    def test_empty_results_message(self):
+        assert (
+            _format_analytics(["status"], [], "tcv")
+            == "No signals found for facility 'tcv'."
+        )
+
+    def test_table_with_total_and_percentages(self):
+        results = [
+            {"status": "checked", "count": 3},
+            {"status": "discovered", "count": 1},
+        ]
+
+        report = _format_analytics(["status"], results, "tcv")
+
+        assert report.startswith("## Signal Analytics for tcv")
+        assert "Total: 4 signals" in report
+        assert "| status | count | % |" in report
+        assert "| checked | 3 | 75.0 |" in report
+        assert "| discovered | 1 | 25.0 |" in report
+
+    def test_cross_tabulation_columns_follow_group_by_order(self):
+        results = [{"physics_domain": "magnetics", "status": "checked", "count": 2}]
+
+        report = _format_analytics(["physics_domain", "status"], results, "tcv")
+
+        assert "| physics_domain | status | count | % |" in report
+        assert "| magnetics | checked | 2 | 100.0 |" in report
+
+    def test_missing_dimension_value_rendered_as_dash(self):
+        report = _format_analytics(["physics_domain"], [{"count": 2}], "tcv")
+
+        assert "| — | 2 | 100.0 |" in report
+
+    def test_total_uses_thousands_separator(self):
+        report = _format_analytics(["status"], [{"status": "a", "count": 12345}], "tcv")
+
+        assert "Total: 12,345 signals" in report
+
+    def test_end_to_end_through_tool(self, gc):
+        gc.query.return_value = [{"status": "checked", "count": 2}]
+
+        report = _signal_analytics("tcv", gc=gc)
+
+        assert "## Signal Analytics for tcv" in report
+        assert "| checked | 2 | 100.0 |" in report
diff --git a/tests/tools/test_dd_changelog.py b/tests/tools/test_dd_changelog.py
new file mode 100644
index 000000000..481216eda
--- /dev/null
+++ b/tests/tools/test_dd_changelog.py
@@ -0,0 +1,265 @@
+"""Tests for the DD changelog tool and its report formatter.
+
+``VersionTool.get_dd_changelog`` ranks IMASNode paths by how volatile they
+have been across DD versions. These tests pin the query parameter contract,
+result shaping, and error handling against a mocked GraphClient, and the
+markdown formatter's header, row, and truncation-hint behaviour.
+
+No live Neo4j required — all graph access goes through ``MagicMock``.
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from imas_codex.llm.search_formatters import format_dd_changelog_report
+from imas_codex.tools.version_tool import VersionTool
+
+
+def _row(
+    path: str = "equilibrium/time_slice/profiles_1d/psi",
+    ids: str = "equilibrium",
+    **overrides,
+) -> dict:
+    """Build one changelog row in the shape returned by the Cypher query."""
+    row = {
+        "path": path,
+        "ids": ids,
+        "lifecycle_status": "active",
+        "change_count": 3,
+        "type_variety": 2,
+        "change_types": ["units_changed", "documentation_changed"],
+        "was_renamed": 0,
+        "volatility_score": 7,
+    }
+    row.update(overrides)
+    return row
+
+
+# ============================================================================
+# VersionTool.get_dd_changelog
+# ============================================================================
+
+
+class TestGetDDChangelog:
+    """Query contract and result shaping of the changelog tool."""
+
+    @pytest.mark.asyncio
+    async def test_default_call_passes_null_filters_and_limit_50(self):
+        gc = MagicMock()
+        gc.query.return_value = []
+
+        result = await VersionTool(gc).get_dd_changelog()
+
+        gc.query.assert_called_once()
+        _, kwargs = gc.query.call_args
+        assert kwargs == {
+            "ids_filter": None,
+            "from_version": None,
+            "to_version": None,
+            "limit": 50,
+        }
+        assert result == {
+            "results": [],
+            "total": 0,
+            "ids_filter": None,
+            "version_range": None,
+            "limit": 50,
+        }
+
+    @pytest.mark.asyncio
+    async def test_filters_are_forwarded_as_query_parameters(self):
+        gc = MagicMock()
+        gc.query.return_value = []
+
+        result = await VersionTool(gc).get_dd_changelog(
+            ids_filter="equilibrium",
+            from_version="3.30.0",
+            to_version="3.39.0",
+            limit=10,
+        )
+
+        _, kwargs = gc.query.call_args
+        assert kwargs == {
+            "ids_filter": "equilibrium",
+            "from_version": "3.30.0",
+            "to_version": "3.39.0",
+            "limit": 10,
+        }
+        assert result["ids_filter"] == "equilibrium"
+        assert result["limit"] == 10
+        assert result["version_range"] == {"from": "3.30.0", "to": "3.39.0"}
+
+    @pytest.mark.asyncio
+    async def test_open_ended_version_range_uses_empty_string(self):
+        gc = MagicMock()
+        gc.query.return_value = []
+
+        lower_only = await VersionTool(gc).get_dd_changelog(from_version="3.30.0")
+        upper_only = await VersionTool(gc).get_dd_changelog(to_version="4.0.0")
+
+        assert lower_only["version_range"] == {"from": "3.30.0", "to": ""}
+        assert upper_only["version_range"] == {"from": "", "to": "4.0.0"}
+
+    @pytest.mark.asyncio
+    async def test_query_is_parameterised_not_interpolated(self):
+        gc = MagicMock()
+        gc.query.return_value = []
+
+        await VersionTool(gc).get_dd_changelog(ids_filter="equilibrium", limit=5)
+
+        (cypher,), _ = gc.query.call_args
+        for placeholder in ("$ids_filter", "$from_version", "$to_version", "$limit"):
+            assert placeholder in cypher
+        assert "equilibrium" not in cypher
+        assert "IMASNodeChange" in cypher
+        assert "ORDER BY volatility_score DESC" in cypher
+
+    @pytest.mark.asyncio
+    async def test_rows_are_copied_into_results_in_order(self):
+        gc = MagicMock()
+        rows = [
+            _row(volatility_score=9),
+            _row(path="equilibrium/time", volatility_score=4),
+        ]
+        gc.query.return_value = rows
+
+        result = await VersionTool(gc).get_dd_changelog()
+
+        assert result["results"] == rows
+        assert result["total"] == 2
+        # Results are fresh dicts, not the driver's record objects.
+        assert result["results"][0] is not rows[0]
+
+    @pytest.mark.asyncio
+    async def test_none_rows_yield_empty_results(self):
+        gc = MagicMock()
+        gc.query.return_value = None
+
+        result = await VersionTool(gc).get_dd_changelog()
+
+        assert result["results"] == []
+        assert result["total"] == 0
+
+    @pytest.mark.asyncio
+    async def test_query_failure_is_returned_not_raised(self):
+        gc = MagicMock()
+        gc.query.side_effect = RuntimeError("bolt connection refused")
+
+        result = await VersionTool(gc).get_dd_changelog()
+
+        assert result == {"error": "Failed to query changelog: bolt connection refused"}
+
+
+# ============================================================================
+# format_dd_changelog_report
+# ============================================================================
+
+
+class TestFormatDDChangelogReport:
+    """Markdown rendering of the changelog result."""
+
+    def test_error_result_renders_error_line(self):
+        assert format_dd_changelog_report({"error": "boom"}) == "Error: boom"
+
+    def test_header_without_filters(self):
+        report = format_dd_changelog_report(
+            {
+                "results": [_row()],
+                "total": 1,
+                "ids_filter": None,
+                "version_range": None,
+                "limit": 50,
+            }
+        )
+
+        assert report.splitlines()[0] == "## DD Changelog — 1 most volatile paths"
+        assert "(IDS:" not in report
+        assert "Version range" not in report
+
+    def test_header_with_ids_and_closed_version_range(self):
+        report = format_dd_changelog_report(
+            {
+                "results": [],
+                "total": 0,
+                "ids_filter": "equilibrium",
+                "version_range": {"from": "3.30.0", "to": "3.39.0"},
+                "limit": 50,
+            }
+        )
+
+        assert "## DD Changelog — 0 most volatile paths (IDS: equilibrium)" in report
+        assert "Version range: 3.30.0 → 3.39.0" in report
+
+    def test_open_ended_range_labels(self):
+        def render(version_range: dict) -> str:
+            return format_dd_changelog_report(
+                {
+                    "results": [],
+                    "total": 0,
+                    "version_range": version_range,
+                    "limit": 50,
+                }
+            )
+
+        assert "Version range: 3.30.0 → latest" in render({"from": "3.30.0", "to": ""})
+        assert "Version range: earliest → 4.0.0" in render({"from": "", "to": "4.0.0"})
+        assert "Version range" not in render({"from": "", "to": ""})
+
+    def test_empty_results_still_render_table_header(self):
+        report = format_dd_changelog_report({"results": [], "total": 0, "limit": 50})
+
+        assert (
+            "| Rank | Path | IDS | Lifecycle | Changes | Types | Renamed | Score |"
+            in report
+        )
+        table_lines = [ln for ln in report.splitlines() if ln.startswith("|")]
+        assert len(table_lines) == 2  # header + separator only
+
+    def test_row_rendering(self):
+        renamed = _row(
+            path="core_profiles/time",
+            ids="core_profiles",
+            was_renamed=1,
+            lifecycle_status="obsolescent",
+            change_types=["renamed"],
+            change_count=1,
+            volatility_score=6,
+        )
+        report = format_dd_changelog_report(
+            {"results": [_row(), renamed], "total": 2, "limit": 50}
+        )
+
+        assert (
+            "| 1 | `equilibrium/time_slice/profiles_1d/psi` | equilibrium |  | 3 "
+            "| units_changed, documentation_changed |  | 7 |"
+        ) in report
+        assert (
+            "| 2 | `core_profiles/time` | core_profiles | obsolescent | 1 "
+            "| renamed | ✓ | 6 |"
+        ) in report
+
+    def test_missing_lifecycle_and_types_are_tolerated(self):
+        report = format_dd_changelog_report(
+            {
+                "results": [_row(lifecycle_status=None, change_types=None)],
+                "total": 1,
+                "limit": 50,
+            }
+        )
+
+        assert (
+            "| 1 | `equilibrium/time_slice/profiles_1d/psi` | equilibrium |  | 3 "
+            "|  |  | 7 |"
+        ) in report
+
+    def test_truncation_hint_only_when_limit_reached(self):
+        at_limit = format_dd_changelog_report(
+            {"results": [_row(), _row(path="x")], "total": 2, "limit": 2}
+        )
+        under_limit = format_dd_changelog_report(
+            {"results": [_row()], "total": 1, "limit": 50}
+        )
+
+        assert "*Showing top 2 — use `limit` to see more.*" in at_limit
+        assert "Showing top" not in under_limit