mnemosyne/mnemosyne/library/tests/test_managed_by.py

"""Tests for the per-app ``managed_by`` concept.

Covers the pure helpers (``infer_legacy_manager``,
``Library.managed_by_display``), the token-derived stamping and
duplicate-name rejection on the plain create endpoint (Neo4j stubbed
via ``sys.modules``, same style as ``test_views.py``), and the
``WorkspaceStatusSerializer`` surface. Cypher-touching paths are
covered by the manual end-to-end plan, not these unit tests.
"""

from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import patch

from django.contrib.auth import get_user_model
from django.test import TestCase
from rest_framework.test import APIClient

from library.api.serializers import WorkspaceStatusSerializer
from library.models import Library, infer_legacy_manager
from mcp_server.models import UserToken

User = get_user_model()


class InferLegacyManagerTests(TestCase):
    """Truth table for the pre-``managed_by`` inference."""

    def test_null_workspace_is_unmanaged(self):
        self.assertIsNone(infer_legacy_manager(None))
        self.assertIsNone(infer_legacy_manager(""))

    def test_kairos_mail_prefix_is_kairos(self):
        self.assertEqual(
            infer_legacy_manager("kairos-mail-abc123-7"), "Kairos"
        )

    def test_other_workspace_is_daedalus(self):
        self.assertEqual(
            infer_legacy_manager("2f9c4a1e-uuid-ish"), "Daedalus"
        )


class ManagedByDisplayTests(TestCase):
    """``managed_by_display`` on in-memory (unsaved) Library nodes."""

    def test_stamped_label_wins(self):
        lib = Library(name="x", managed_by="Spelunker")
        self.assertEqual(lib.managed_by_display, "Spelunker")

    def test_stamped_label_wins_over_inference(self):
        lib = Library(
            name="x", managed_by="My Token", workspace_id="kairos-mail-a-1"
        )
        self.assertEqual(lib.managed_by_display, "My Token")

    def test_legacy_workspace_falls_back_to_inference(self):
        lib = Library(name="x", workspace_id="ws-uuid")
        self.assertEqual(lib.managed_by_display, "Daedalus")

    def test_unmanaged_is_empty_string(self):
        lib = Library(name="x")
        self.assertEqual(lib.managed_by_display, "")


class _FakeLibrary:
    """Stand-in for the neomodel Library on the plain create endpoint."""

    class DoesNotExist(Exception):
        pass

    existing = None          # what find_library_by_name_ci returns
    instances = []           # constructor kwargs, in order
    save_raises = None       # exception instance save() should raise, if any

    def __init__(self, **kwargs):
        type(self).instances.append(kwargs)
        self.__dict__.update(kwargs)
        self.uid = "lib-new"
        self.workspace_id = None
        self.created_at = None

    def save(self):
        if type(self).save_raises is not None:
            raise type(self).save_raises
        return self


def _fake_find_ci(name):
    _FakeLibrary.ci_queries.append(name)
    return _FakeLibrary.existing


def _fake_models_module():
    return SimpleNamespace(
        Library=_FakeLibrary, find_library_by_name_ci=_fake_find_ci
    )


class LibraryCreateStampingTests(TestCase):
    """POST /library/api/libraries/ stamps ``managed_by`` and rejects dupes."""

    def setUp(self):
        self.user = User.objects.create_user(username="op", password="pw")
        self.client = APIClient()
        _FakeLibrary.existing = None
        _FakeLibrary.instances = []
        _FakeLibrary.save_raises = None
        _FakeLibrary.ci_queries = []

    def _post(self, token=None, name="Docs"):
        self.client.force_authenticate(user=self.user, token=token)
        with patch.dict("sys.modules", {"library.models": _fake_models_module()}):
            return self.client.post(
                "/library/api/libraries/",
                {"name": name, "library_type": "technical"},
                format="json",
            )

    def test_token_create_stamps_token_name(self):
        response = self._post(token=UserToken(name="Spelunker"))

        self.assertEqual(response.status_code, 201)
        self.assertEqual(_FakeLibrary.instances[0]["managed_by"], "Spelunker")
        self.assertEqual(response.json()["managed_by"], "Spelunker")

    def test_session_create_leaves_managed_by_null(self):
        response = self._post(token=None)

        self.assertEqual(response.status_code, 201)
        self.assertIsNone(_FakeLibrary.instances[0]["managed_by"])

    def test_duplicate_name_returns_409_name_conflict(self):
        _FakeLibrary.existing = SimpleNamespace(
            uid="lib-old", name="Docs", managed_by_display="Daedalus"
        )
        response = self._post(token=UserToken(name="Spelunker"))

        self.assertEqual(response.status_code, 409)
        body = response.json()
        self.assertEqual(body["code"], "name_conflict")
        self.assertIn("Docs", body["detail"])
        self.assertEqual(body["uid"], "lib-old")
        self.assertEqual(body["managed_by"], "Daedalus")
        self.assertEqual(_FakeLibrary.instances, [])

    def test_duplicate_check_is_case_insensitive(self):
        """A case-variant name 409s and reports the existing spelling."""
        _FakeLibrary.existing = SimpleNamespace(
            uid="lib-old", name="Amazon Connect", managed_by_display="Spelunker"
        )
        response = self._post(token=None, name="amazon connect")

        self.assertEqual(response.status_code, 409)
        # The lookup received the posted name (case-folding happens in
        # Cypher), and the response names the existing spelling.
        self.assertEqual(_FakeLibrary.ci_queries, ["amazon connect"])
        self.assertIn("Amazon Connect", response.json()["detail"])
        self.assertEqual(_FakeLibrary.instances, [])

    def test_duplicate_of_unmanaged_reports_null_manager(self):
        _FakeLibrary.existing = SimpleNamespace(
            uid="lib-old", name="Docs", managed_by_display=""
        )
        response = self._post(token=None)

        self.assertEqual(response.status_code, 409)
        self.assertIsNone(response.json()["managed_by"])

    def test_save_race_returns_409_not_500(self):
        """An exact-name twin landing between pre-check and save 409s."""
        from neomodel.exceptions import UniqueProperty

        _FakeLibrary.save_raises = UniqueProperty("name")
        response = self._post(token=None)

        self.assertEqual(response.status_code, 409)
        self.assertEqual(response.json()["code"], "name_conflict")


class WorkspaceStatusSerializerManagedByTests(TestCase):
    """The workspace status payload carries ``managed_by`` (nullable)."""

    BASE = {
        "workspace_id": "ws_a",
        "library_uid": "lib_1",
        "name": "W",
        "library_type": "technical",
        "description": "",
        "item_count": 0,
        "chunk_count": 0,
        "created_at": "2026-01-01T00:00:00Z",
    }

    def test_managed_by_value_round_trips(self):
        s = WorkspaceStatusSerializer(data={**self.BASE, "managed_by": "Daedalus"})
        self.assertTrue(s.is_valid(), s.errors)
        self.assertEqual(s.validated_data["managed_by"], "Daedalus")

    def test_managed_by_null_accepted(self):
        s = WorkspaceStatusSerializer(data={**self.BASE, "managed_by": None})
        self.assertTrue(s.is_valid(), s.errors)

    def test_managed_by_absent_accepted(self):
        s = WorkspaceStatusSerializer(data=self.BASE)
        self.assertTrue(s.is_valid(), s.errors)