314 lines
12 KiB
Python
314 lines
12 KiB
Python
"""Tests for the admin-UI search views' ``resolved_libraries`` wiring.
|
|
|
|
The two views in ``library/views.py`` that back the Mnemosyne HTML
|
|
search surface (``search_page`` for the global page, ``library_search``
|
|
for the per-library page) are ``@login_required`` debug/admin tools for
|
|
Django-authenticated operators — not MCP endpoints. The MCP auth
|
|
middleware is bypassed, so the view itself is the trusted caller
|
|
responsible for materializing ``resolved_libraries``.
|
|
|
|
Post-Phase-2, ``SearchService`` appends ``_RESOLVED_LIBRARIES_CLAUSE`` to
|
|
every Cypher query. That clause short-circuits when
|
|
``resolved_libraries=None`` but fail-closes on ``[]``. The admin views
|
|
call ``_all_library_uids()`` to build the full UID list and pass it as
|
|
``resolved_libraries=`` — so a session-authenticated operator sees
|
|
every library in Neo4j, including Daedalus workspace-scoped ones.
|
|
|
|
These tests cover:
|
|
|
|
* ``_all_library_uids`` — returns every UID when Neo4j is reachable,
|
|
returns ``[]`` when it isn't, and swallows unexpected model errors
|
|
rather than 500-ing the whole page.
|
|
* Defence in depth: make sure the admin views actually pass that list
|
|
into ``SearchRequest`` so a future refactor doesn't silently
|
|
re-introduce the "zero results for workspace libraries" bug.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase
|
|
from django.urls import reverse
|
|
|
|
from library import views
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _all_library_uids
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class AllLibraryUidsHelperTests(TestCase):
|
|
"""Cover every branch of ``library.views._all_library_uids``."""
|
|
|
|
def test_returns_empty_when_neo4j_unavailable(self):
|
|
"""Helper must not touch ``Library.nodes`` if Neo4j is down."""
|
|
with patch("library.views.neo4j_available", return_value=False):
|
|
self.assertEqual(views._all_library_uids(), [])
|
|
|
|
def test_returns_every_library_uid(self):
|
|
"""All nodes returned from ``Library.nodes.all()`` are enumerated."""
|
|
fake_libs = [
|
|
SimpleNamespace(uid="lib_workspace_1"),
|
|
SimpleNamespace(uid="lib_global_2"),
|
|
SimpleNamespace(uid="lib_workspace_3"),
|
|
]
|
|
fake_nodes = MagicMock()
|
|
fake_nodes.all.return_value = fake_libs
|
|
fake_library_cls = SimpleNamespace(nodes=fake_nodes)
|
|
|
|
with patch("library.views.neo4j_available", return_value=True), \
|
|
patch.dict("sys.modules", {"library.models": SimpleNamespace(Library=fake_library_cls)}):
|
|
result = views._all_library_uids()
|
|
|
|
self.assertEqual(
|
|
sorted(result),
|
|
["lib_global_2", "lib_workspace_1", "lib_workspace_3"],
|
|
)
|
|
|
|
def test_skips_nodes_with_empty_uid(self):
|
|
"""Defensive filter — a node with no uid would break the Cypher IN."""
|
|
fake_libs = [
|
|
SimpleNamespace(uid="lib_ok"),
|
|
SimpleNamespace(uid=""),
|
|
SimpleNamespace(uid=None),
|
|
SimpleNamespace(uid="lib_ok_2"),
|
|
]
|
|
fake_nodes = MagicMock()
|
|
fake_nodes.all.return_value = fake_libs
|
|
fake_library_cls = SimpleNamespace(nodes=fake_nodes)
|
|
|
|
with patch("library.views.neo4j_available", return_value=True), \
|
|
patch.dict("sys.modules", {"library.models": SimpleNamespace(Library=fake_library_cls)}):
|
|
result = views._all_library_uids()
|
|
|
|
self.assertEqual(sorted(result), ["lib_ok", "lib_ok_2"])
|
|
|
|
def test_returns_empty_on_unexpected_exception(self):
|
|
"""Neo4j hiccup must not break the whole search page — log & degrade."""
|
|
fake_nodes = MagicMock()
|
|
fake_nodes.all.side_effect = RuntimeError("neo4j blew up")
|
|
fake_library_cls = SimpleNamespace(nodes=fake_nodes)
|
|
|
|
with patch("library.views.neo4j_available", return_value=True), \
|
|
patch.dict("sys.modules", {"library.models": SimpleNamespace(Library=fake_library_cls)}):
|
|
self.assertEqual(views._all_library_uids(), [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# search_page wiring
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class SearchPageResolvedLibrariesTests(TestCase):
|
|
"""Verify ``search_page`` forwards every library UID as ``resolved_libraries``.
|
|
|
|
Stubs out ``_all_library_uids`` and ``SearchService`` so the test
|
|
asserts on the ``SearchRequest`` that actually reaches the service
|
|
— not on Neo4j behaviour — keeping the test hermetic.
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(
|
|
username="admin", email="a@example.com", password="pw"
|
|
)
|
|
self.client.force_login(self.user)
|
|
|
|
def _patched_search(self, reranker_skip_reason=None):
|
|
"""Return a (request_capture, patch_context) pair.
|
|
|
|
The patch captures the ``SearchRequest`` that ``SearchService.search``
|
|
is called with so assertions can run after the view returns.
|
|
|
|
:param reranker_skip_reason: Value to set on the stub response's
|
|
``reranker_skip_reason`` attribute, for tests that want to
|
|
exercise the "Skipped" badge rendering path.
|
|
"""
|
|
capture: dict = {}
|
|
|
|
def fake_search(self, request):
|
|
capture["request"] = request
|
|
# Return a minimally-shaped response to satisfy the template.
|
|
return SimpleNamespace(
|
|
query=request.query,
|
|
candidates=[],
|
|
images=[],
|
|
total_candidates=0,
|
|
search_time_ms=0.0,
|
|
reranker_used=False,
|
|
reranker_model=None,
|
|
search_types_used=[],
|
|
reranker_skip_reason=reranker_skip_reason,
|
|
)
|
|
|
|
return capture, patch(
|
|
"library.services.search.SearchService.search", fake_search
|
|
)
|
|
|
|
def test_search_page_forwards_resolved_libraries(self):
|
|
capture, patched = self._patched_search()
|
|
with patch(
|
|
"library.views._all_library_uids",
|
|
return_value=["lib_ws_a", "lib_ws_b"],
|
|
), patched:
|
|
response = self.client.post(
|
|
reverse("library:search"),
|
|
{"query": "contract renewal", "rerank": "on"},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
req = capture["request"]
|
|
self.assertEqual(req.query, "contract renewal")
|
|
self.assertEqual(req.resolved_libraries, ["lib_ws_a", "lib_ws_b"])
|
|
|
|
def test_search_page_empty_library_list_is_fail_closed(self):
|
|
"""When Neo4j has zero libraries, the admin page sees nothing.
|
|
|
|
Post-Phase-2, ``SearchRequest.__post_init__`` preserves ``[]`` as
|
|
fail-closed rather than collapsing it to ``None``. That's a
|
|
deliberate trade: the admin page will render zero results if
|
|
Neo4j has no libraries, which is strictly better than silently
|
|
switching to a different auth branch mid-request. The page
|
|
itself is already useless in that state (nothing to search),
|
|
so fail-closed is the safer default.
|
|
"""
|
|
capture, patched = self._patched_search()
|
|
with patch("library.views._all_library_uids", return_value=[]), patched:
|
|
self.client.post(
|
|
reverse("library:search"),
|
|
{"query": "anything", "rerank": "on"},
|
|
)
|
|
|
|
self.assertEqual(capture["request"].resolved_libraries, [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# search_page rerank-status rendering
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class SearchPageRerankBadgeTests(TestCase):
|
|
"""Verify the three-state Re-ranked indicator on the search page.
|
|
|
|
The badge must distinguish:
|
|
|
|
* Success (``reranker_used=True``) — green "Yes"
|
|
* Skipped (``rerank=True`` requested but ``reranker_skip_reason`` set)
|
|
— warning "Skipped" with the reason shown
|
|
* Off (user unchecked the re-rank box) — ghost "Off"
|
|
|
|
This guards the regression that surfaced when Synesis returned 404
|
|
on a mis-constructed rerank URL: the UI said "No" and gave no hint
|
|
the re-ranker had actually failed.
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(
|
|
username="admin", email="a@example.com", password="pw"
|
|
)
|
|
self.client.force_login(self.user)
|
|
|
|
def _run(self, rerank_value, reranker_used, reranker_skip_reason):
|
|
capture: dict = {}
|
|
|
|
def fake_search(self, request):
|
|
capture["request"] = request
|
|
return SimpleNamespace(
|
|
query=request.query,
|
|
candidates=[],
|
|
images=[],
|
|
total_candidates=0,
|
|
search_time_ms=0.0,
|
|
reranker_used=reranker_used,
|
|
reranker_model=None,
|
|
search_types_used=[],
|
|
reranker_skip_reason=reranker_skip_reason,
|
|
)
|
|
|
|
post_data = {"query": "postgresql"}
|
|
if rerank_value is not None:
|
|
post_data["rerank"] = rerank_value
|
|
|
|
with patch("library.views._all_library_uids", return_value=[]), \
|
|
patch("library.services.search.SearchService.search", fake_search):
|
|
response = self.client.post(reverse("library:search"), post_data)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
return response, capture
|
|
|
|
# The badge text ("Yes" / "Skipped" / "Off") sits inside a
|
|
# ``<span class="badge ...">`` element, preceded by template
|
|
# whitespace (indentation + newlines). We therefore match on
|
|
# the ``badge-<state>`` class for the structural assertion and
|
|
# on the unescaped text for the content assertion, which keeps
|
|
# these checks robust against template re-indentation.
|
|
|
|
def test_badge_shows_yes_when_rerank_succeeded(self):
|
|
response, _ = self._run(
|
|
rerank_value="on",
|
|
reranker_used=True,
|
|
reranker_skip_reason=None,
|
|
)
|
|
body = response.content.decode()
|
|
self.assertIn("badge-success", body)
|
|
self.assertIn("Yes", body)
|
|
self.assertNotIn("badge-warning", body)
|
|
self.assertNotIn("badge-ghost", body)
|
|
|
|
def test_badge_shows_skipped_with_reason_on_api_error(self):
|
|
reason = (
|
|
"api_error: 404 Client Error: Not Found for url: "
|
|
"http://pan.helu.ca:8400/v1/v1/rerank"
|
|
)
|
|
response, capture = self._run(
|
|
rerank_value="on",
|
|
reranker_used=False,
|
|
reranker_skip_reason=reason,
|
|
)
|
|
# Sanity: the view actually requested re-ranking.
|
|
self.assertTrue(capture["request"].rerank)
|
|
|
|
body = response.content.decode()
|
|
self.assertIn("badge-warning", body)
|
|
self.assertIn("Skipped", body)
|
|
# Reason shown in-page so the user can debug without grepping logs.
|
|
# Django auto-escapes the colon-space and URL, which is fine.
|
|
self.assertIn("api_error:", body)
|
|
self.assertIn("404", body)
|
|
# Must not claim success.
|
|
self.assertNotIn("badge-success", body)
|
|
|
|
def test_badge_shows_skipped_on_no_system_model(self):
|
|
response, _ = self._run(
|
|
rerank_value="on",
|
|
reranker_used=False,
|
|
reranker_skip_reason="no_system_model",
|
|
)
|
|
body = response.content.decode()
|
|
self.assertIn("badge-warning", body)
|
|
self.assertIn("Skipped", body)
|
|
self.assertIn("no_system_model", body)
|
|
|
|
def test_badge_shows_off_when_rerank_unchecked(self):
|
|
# The view reads ``request.POST.get("rerank", "on") == "on"``, so an
|
|
# unchecked checkbox (key omitted) still yields ``rerank=True`` in the
|
|
# SearchRequest — reflecting the product default that re-ranking is
|
|
# *on* unless the user actively disables it by posting ``rerank=off``.
|
|
response, capture = self._run(
|
|
rerank_value="off",
|
|
reranker_used=False,
|
|
reranker_skip_reason=None,
|
|
)
|
|
self.assertFalse(capture["request"].rerank)
|
|
|
|
body = response.content.decode()
|
|
self.assertIn("badge-ghost", body)
|
|
self.assertIn("Off", body)
|
|
self.assertNotIn("badge-success", body)
|
|
self.assertNotIn("badge-warning", body)
|