FabledScribe/tests/test_shared_provenance.py

"""Shared records must announce themselves.

A record another user owns is that person's suggestion, not the operator's
settled practice — and the two are indistinguishable unless every surface that
hands one over says whose it is. These tests hold that line at the labelling
helpers and at the process→skill manifest, which is the most consequential
passive surface Scribe has (each entry becomes an auto-surfacing skill file).
"""
from unittest.mock import AsyncMock, MagicMock, patch

import pytest


def _session_returning_rows(rows):
    result = MagicMock()
    result.all.return_value = rows
    session = AsyncMock()
    session.__aenter__ = AsyncMock(return_value=session)
    session.__aexit__ = AsyncMock(return_value=False)
    session.execute = AsyncMock(return_value=result)
    return session


@pytest.mark.asyncio
async def test_label_marks_only_foreign_rows():
    from scribe.services.access import label_shared_items
    items = [
        {"id": 1, "user_id": 7},   # mine
        {"id": 2, "user_id": 9},   # someone else's
    ]
    with patch("scribe.services.access.async_session") as cls:
        cls.return_value = _session_returning_rows([(9, "alex")])
        out = await label_shared_items(user_id=7, items=items)

    mine, theirs = out
    # An unmarked row must be unambiguously the caller's own.
    assert "shared" not in mine and "owner" not in mine
    assert theirs["shared"] is True
    assert theirs["owner"] == "alex"


@pytest.mark.asyncio
async def test_label_skips_the_query_when_everything_is_mine():
    """No foreign rows means no user lookup at all — labelling shouldn't cost a
    query on the common single-owner path."""
    from scribe.services.access import label_shared_items
    with patch("scribe.services.access.async_session") as cls:
        out = await label_shared_items(user_id=7, items=[{"id": 1, "user_id": 7}])
        cls.assert_not_called()
    assert out == [{"id": 1, "user_id": 7}]


@pytest.mark.asyncio
async def test_provenance_is_empty_for_your_own_record():
    from scribe.services.access import describe_provenance
    note = MagicMock(id=1, user_id=7)
    assert await describe_provenance(user_id=7, note=note) == {}


@pytest.mark.asyncio
async def test_provenance_names_the_owner_and_permission():
    from scribe.services.access import describe_provenance
    note = MagicMock(id=1, user_id=9)
    owner = MagicMock(username="alex")
    session = AsyncMock()
    session.__aenter__ = AsyncMock(return_value=session)
    session.__aexit__ = AsyncMock(return_value=False)
    session.get = AsyncMock(return_value=owner)
    with patch("scribe.services.access.async_session") as cls, \
         patch("scribe.services.access.get_note_permission",
               AsyncMock(return_value="viewer")):
        cls.return_value = session
        got = await describe_provenance(user_id=7, note=note)
    assert got == {"shared": True, "owner": "alex", "permission": "viewer"}


@pytest.mark.asyncio
async def test_shared_process_skill_stub_never_claims_to_be_the_operators():
    """The stub description IS the skill's auto-surface trigger and the only
    provenance a reader sees. For a shared Process it must name the author and
    require a go-ahead — never read as "the operator's saved process"."""
    from scribe.services import plugin_context

    items = [
        {"id": 1, "title": "Drift Audit", "snippet": "Sweep for drift.", "user_id": 7},
        {"id": 2, "title": "Deploy Dance", "snippet": "Ship it.", "user_id": 9},
    ]
    with patch.object(plugin_context.knowledge_svc, "query_knowledge",
                      AsyncMock(return_value=(items, 2))), \
         patch.object(plugin_context, "label_shared_items",
                      AsyncMock(side_effect=lambda uid, its: [
                          it if it["user_id"] == uid
                          else {**it, "shared": True, "owner": "alex"}
                          for it in its
                      ])):
        out = await plugin_context.build_process_manifest(user_id=7)

    own, shared = out["processes"]

    assert "operator's saved Scribe process" in own["description"]
    assert "shared" not in own

    assert shared["shared"] is True
    assert shared["owner"] == "alex"
    assert "operator's saved Scribe process" not in shared["description"]
    assert "alex" in shared["description"]
    assert "go-ahead" in shared["description"]