"""Tests for DMS Phase 7 — SEARCH & INDEXING.

Covers:
  * index_document — creates entry, updates entry (index_version increments),
    populates search_text, extracts tags
  * deindex_document — removes index entry
  * reindex_tenant — indexes all non-deleted tenant documents
  * search_documents — q filter, status filter, document_type filter,
    repository filter, folder filter, content_type filter, tags filter,
    indexed_after/indexed_before, sort+order
  * create_saved_search, update_saved_search, delete_saved_search
  * execute_saved_search — increments use_count, updates last_used_at
  * Queries: get_search_index_for_document, get_saved_search (own/shared/
    not-accessible), list_saved_searches
  * Signals: document save triggers index, version save triggers re-index,
    soft-deleted document gets deindexed
  * HTTP API: GET /dms/search/, saved search CRUD, execute saved search
  * Permission gates: 401 unauthenticated, 403 no permission
"""

from __future__ import annotations

import uuid

import pytest
from django.utils import timezone

from simorgh.apps.dms.documents.services import add_version, publish_version
from simorgh.apps.dms.repositories.models import Repository
from simorgh.apps.dms.search import queries, services
from simorgh.apps.dms.search.iam_permissions import PERM_SEARCH_QUERY, PERM_SEARCH_SAVE
from simorgh.apps.dms.search.models import DocumentSearchIndex, SavedSearch
from simorgh.apps.dms.search.services import SearchParams
from simorgh.core.context import RequestContext, use_request_context

ALL_PERMS = frozenset({PERM_SEARCH_QUERY, PERM_SEARCH_SAVE})


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

@pytest.fixture
def alice_membership(alice, tenant_acme, acme_tree, role_admin):
    from simorgh.apps.memberships.models import Membership

    m = Membership.objects.create(
        tenant=tenant_acme,
        organization_node=acme_tree["root"],
        role=role_admin,
    )
    m.users.add(alice)
    return m


@pytest.fixture
def bob(db):
    from django.contrib.auth import get_user_model

    User = get_user_model()
    return User.objects.create_user(
        "+989000000099", password="x", email="bob@example.com",
        is_superuser=True,
    )


@pytest.fixture
def srch_ctx(alice, tenant_acme, acme_tree):
    return RequestContext(
        actor=alice,
        tenant=tenant_acme,
        org_node_ids=frozenset({acme_tree["root"].pk}),
        permissions=ALL_PERMS,
    )


@pytest.fixture
def repo(tenant_acme, acme_tree):
    return Repository.objects.create(
        name="Search Test Repo",
        slug="search-test-repo",
        tenant=tenant_acme,
        organization_node=acme_tree["root"],
    )


@pytest.fixture
def doc(repo, tenant_acme, acme_tree):
    from simorgh.apps.dms.documents.models import Document

    return Document.objects.create(
        title="Search Test Document",
        repository=repo,
        tenant=tenant_acme,
        organization_node=acme_tree["root"],
    )


@pytest.fixture
def doc2(repo, tenant_acme, acme_tree):
    """A second document in the same tenant for multi-doc search tests."""
    from simorgh.apps.dms.documents.models import Document

    return Document.objects.create(
        title="Another Document",
        repository=repo,
        tenant=tenant_acme,
        organization_node=acme_tree["root"],
    )


@pytest.fixture
def file_asset(tenant_acme, acme_tree):
    from simorgh.apps.dms.assets.constants import APP_CONTEXT
    from simorgh.apps.storage.models import FileMetadata, FileUploadStatus

    return FileMetadata.objects.create(
        filename="report.pdf",
        content_type="application/pdf",
        size_bytes=4096,
        path=f"test/phase7/{uuid.uuid4()}.pdf",
        tenant=tenant_acme,
        organization_node=acme_tree["root"],
        upload_status=FileUploadStatus.READY,
        app_context=APP_CONTEXT,
    )


@pytest.fixture
def version(doc, file_asset, tenant_acme, acme_tree):
    v = add_version(
        document=doc,
        file_asset=file_asset,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    publish_version(v)
    return v


@pytest.fixture
def indexed_doc(doc, version, tenant_acme):
    """Document that has been indexed (with a version)."""
    # Re-index explicitly to ensure a fresh index entry
    doc.refresh_from_db()
    return services.index_document(doc)


@pytest.fixture
def saved_search(alice, tenant_acme, acme_tree):
    return services.create_saved_search(
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        owner=alice,
        name="My Saved Search",
        query_params={"q": "report"},
    )


# ---------------------------------------------------------------------------
# Service — index_document
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_index_document_creates_entry(doc, tenant_acme):
    idx = services.index_document(doc)
    assert idx.pk is not None
    assert idx.tenant_id == tenant_acme.pk
    assert idx.title == doc.title


@pytest.mark.django_db
def test_index_document_idempotent_increments_version(doc):
    idx1 = services.index_document(doc)
    first_version = idx1.index_version
    idx2 = services.index_document(doc)
    assert idx1.pk == idx2.pk
    assert idx2.index_version == first_version + 1


@pytest.mark.django_db
def test_index_document_with_version_populates_content_type(doc, version, tenant_acme):
    idx = services.index_document(doc)
    assert idx.content_type == "application/pdf"


@pytest.mark.django_db
def test_index_document_with_version_populates_file_name(doc, version, tenant_acme):
    idx = services.index_document(doc)
    assert idx.file_name == "report.pdf"


@pytest.mark.django_db
def test_index_document_builds_search_text(doc, version, tenant_acme):
    idx = services.index_document(doc)
    assert doc.title in idx.search_text


@pytest.mark.django_db
def test_index_document_extracts_tags(doc, tenant_acme):
    doc.extra = {"tags": ["finance", "Q4"]}
    doc.save()
    idx = services.index_document(doc)
    assert "finance" in idx.tags
    assert "q4" in idx.tags  # lowercase


@pytest.mark.django_db
def test_index_document_no_version_still_works(doc, tenant_acme):
    """A document without any version can still be indexed."""
    idx = services.index_document(doc)
    assert idx.pk is not None
    assert idx.content_type == ""
    assert idx.version_label == ""


# ---------------------------------------------------------------------------
# Service — deindex_document
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_deindex_document_removes_entry(doc, tenant_acme):
    services.index_document(doc)
    assert DocumentSearchIndex.objects.filter(document=doc).exists()
    services.deindex_document(doc.pk)
    assert not DocumentSearchIndex.objects.filter(document=doc).exists()


@pytest.mark.django_db
def test_deindex_document_noop_when_not_indexed(doc):
    # Must not raise
    services.deindex_document(doc.pk)


# ---------------------------------------------------------------------------
# Service — reindex_tenant
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_reindex_tenant_indexes_all_documents(doc, doc2, tenant_acme):
    count = services.reindex_tenant(tenant_acme.pk)
    assert count >= 2
    assert DocumentSearchIndex.objects.filter(tenant=tenant_acme, document=doc).exists()
    assert DocumentSearchIndex.objects.filter(tenant=tenant_acme, document=doc2).exists()


@pytest.mark.django_db
def test_reindex_tenant_skips_deleted_documents(doc, tenant_acme):
    doc.is_deleted = True
    doc.save()
    count = services.reindex_tenant(tenant_acme.pk)
    # Deleted documents are excluded
    assert not DocumentSearchIndex.objects.filter(document=doc).exists()


# ---------------------------------------------------------------------------
# Service — search_documents
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_search_documents_returns_results_for_tenant(indexed_doc, tenant_acme):
    results = list(services.search_documents(tenant_acme.pk, SearchParams()))
    assert len(results) >= 1


@pytest.mark.django_db
def test_search_documents_q_filter(doc, version, tenant_acme):
    idx = services.index_document(doc)
    results = list(services.search_documents(
        tenant_acme.pk, SearchParams(q="Search Test Document")
    ))
    pks = [r.pk for r in results]
    assert idx.pk in pks


@pytest.mark.django_db
def test_search_documents_q_no_match(doc, tenant_acme):
    services.index_document(doc)
    results = list(services.search_documents(
        tenant_acme.pk, SearchParams(q="xyznonexistenttokenxyz")
    ))
    assert len(results) == 0


@pytest.mark.django_db
def test_search_documents_status_filter(doc, tenant_acme):
    idx = services.index_document(doc)
    results = list(services.search_documents(
        tenant_acme.pk, SearchParams(status=doc.status)
    ))
    pks = [r.pk for r in results]
    assert idx.pk in pks


@pytest.mark.django_db
def test_search_documents_content_type_filter(doc, version, tenant_acme):
    idx = services.index_document(doc)
    results = list(services.search_documents(
        tenant_acme.pk, SearchParams(content_type="application/pdf")
    ))
    pks = [r.pk for r in results]
    assert idx.pk in pks


@pytest.mark.django_db
def test_search_documents_content_type_no_match(doc, version, tenant_acme):
    services.index_document(doc)
    results = list(services.search_documents(
        tenant_acme.pk, SearchParams(content_type="video/mp4")
    ))
    assert len(results) == 0


@pytest.mark.django_db
def test_search_documents_repository_filter(doc, repo, tenant_acme):
    idx = services.index_document(doc)
    results = list(services.search_documents(
        tenant_acme.pk, SearchParams(repository_id=str(repo.public_id))
    ))
    pks = [r.pk for r in results]
    assert idx.pk in pks


@pytest.mark.django_db
def test_search_documents_tags_any_match(doc, tenant_acme):
    doc.extra = {"tags": ["hr", "policy"]}
    doc.save()
    idx = services.index_document(doc)
    # Match on ONE of two tags
    results = list(services.search_documents(
        tenant_acme.pk, SearchParams(tags=["hr", "finance"])
    ))
    pks = [r.pk for r in results]
    assert idx.pk in pks


@pytest.mark.django_db
def test_search_documents_tags_no_match(doc, tenant_acme):
    doc.extra = {"tags": ["hr"]}
    doc.save()
    services.index_document(doc)
    results = list(services.search_documents(
        tenant_acme.pk, SearchParams(tags=["finance"])
    ))
    assert len(results) == 0


@pytest.mark.django_db
def test_search_documents_indexed_before_filter(doc, tenant_acme):
    services.index_document(doc)
    results_past = list(services.search_documents(
        tenant_acme.pk,
        SearchParams(indexed_before=timezone.now() - timezone.timedelta(days=1)),
    ))
    assert len(results_past) == 0


@pytest.mark.django_db
def test_search_documents_indexed_after_filter(doc, tenant_acme):
    idx = services.index_document(doc)
    results = list(services.search_documents(
        tenant_acme.pk,
        SearchParams(indexed_after=timezone.now() - timezone.timedelta(hours=1)),
    ))
    pks = [r.pk for r in results]
    assert idx.pk in pks


@pytest.mark.django_db
def test_search_documents_sort_asc(doc, doc2, tenant_acme):
    services.index_document(doc)
    services.index_document(doc2)
    results = list(services.search_documents(
        tenant_acme.pk, SearchParams(sort="title", order="asc")
    ))
    titles = [r.title for r in results]
    assert titles == sorted(titles)


# ---------------------------------------------------------------------------
# Service — saved search CRUD
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_create_saved_search(alice, tenant_acme, acme_tree):
    ss = services.create_saved_search(
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        owner=alice,
        name="Finance Reports",
        query_params={"q": "finance", "tags": ["finance"]},
        is_shared=True,
    )
    assert ss.pk is not None
    assert ss.name == "Finance Reports"
    assert ss.is_shared is True
    assert ss.owner_id == alice.pk


@pytest.mark.django_db
def test_update_saved_search_name(saved_search):
    updated = services.update_saved_search(saved_search, name="Updated Name")
    assert updated.name == "Updated Name"


@pytest.mark.django_db
def test_update_saved_search_query_params(saved_search):
    updated = services.update_saved_search(
        saved_search, query_params={"q": "updated", "status": "published"}
    )
    assert updated.query_params["q"] == "updated"


@pytest.mark.django_db
def test_update_saved_search_is_shared(saved_search):
    updated = services.update_saved_search(saved_search, is_shared=True)
    assert updated.is_shared is True


@pytest.mark.django_db
def test_delete_saved_search(saved_search):
    services.delete_saved_search(saved_search)
    saved_search.refresh_from_db()
    assert saved_search.is_deleted is True


@pytest.mark.django_db
def test_execute_saved_search_increments_use_count(
    saved_search, doc, indexed_doc, tenant_acme
):
    assert saved_search.use_count == 0
    services.execute_saved_search(saved_search, tenant_acme.pk)
    saved_search.refresh_from_db()
    assert saved_search.use_count == 1


@pytest.mark.django_db
def test_execute_saved_search_updates_last_used_at(
    saved_search, doc, indexed_doc, tenant_acme
):
    assert saved_search.last_used_at is None
    services.execute_saved_search(saved_search, tenant_acme.pk)
    saved_search.refresh_from_db()
    assert saved_search.last_used_at is not None


@pytest.mark.django_db
def test_execute_saved_search_returns_matching_results(
    saved_search, indexed_doc, tenant_acme
):
    """Saved search with q='report' should match the indexed doc (title contains
    'Search Test Document' and file_name 'report.pdf')."""
    results = list(services.execute_saved_search(saved_search, tenant_acme.pk))
    pks = [r.pk for r in results]
    assert indexed_doc.pk in pks


# ---------------------------------------------------------------------------
# Queries
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_get_search_index_for_document_not_indexed(doc, tenant_acme):
    # The post_save signal auto-indexes on creation; manually deindex first
    services.deindex_document(doc.pk)
    result = queries.get_search_index_for_document(doc)
    assert result is None


@pytest.mark.django_db
def test_get_search_index_for_document_returns_entry(doc, tenant_acme):
    idx = services.index_document(doc)
    result = queries.get_search_index_for_document(doc)
    assert result is not None
    assert result.pk == idx.pk


@pytest.mark.django_db
def test_get_saved_search_own(alice, saved_search, tenant_acme):
    result = queries.get_saved_search(tenant_acme.pk, str(saved_search.public_id), user=alice)
    assert result.pk == saved_search.pk


@pytest.mark.django_db
def test_get_saved_search_shared(bob, alice, tenant_acme, acme_tree):
    ss = services.create_saved_search(
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        owner=alice,
        name="Shared Search",
        is_shared=True,
    )
    # bob can access because is_shared=True
    result = queries.get_saved_search(tenant_acme.pk, str(ss.public_id), user=bob)
    assert result.pk == ss.pk


@pytest.mark.django_db
def test_get_saved_search_private_not_accessible(db, alice, tenant_acme, acme_tree):
    from django.contrib.auth import get_user_model
    from simorgh.apps.dms.common.exceptions import AssetNotFound

    User = get_user_model()
    charlie = User.objects.create_user(
        "+989000000077", password="x", email="charlie@example.com",
        is_superuser=False,
    )
    ss = services.create_saved_search(
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        owner=alice,
        name="Private Search",
        is_shared=False,
    )
    with pytest.raises(AssetNotFound):
        queries.get_saved_search(tenant_acme.pk, str(ss.public_id), user=charlie)


@pytest.mark.django_db
def test_get_saved_search_not_found_raises(alice, tenant_acme):
    from simorgh.apps.dms.common.exceptions import AssetNotFound

    with pytest.raises(AssetNotFound):
        queries.get_saved_search(tenant_acme.pk, str(uuid.uuid4()), user=alice)


@pytest.mark.django_db
def test_list_saved_searches_own(alice, saved_search, tenant_acme):
    results = list(queries.list_saved_searches(tenant_acme.pk, alice))
    pks = [r.pk for r in results]
    assert saved_search.pk in pks


@pytest.mark.django_db
def test_list_saved_searches_includes_shared(bob, alice, tenant_acme, acme_tree):
    ss = services.create_saved_search(
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        owner=alice,
        name="Shared For Bob",
        is_shared=True,
    )
    results = list(queries.list_saved_searches(tenant_acme.pk, bob))
    pks = [r.pk for r in results]
    assert ss.pk in pks


@pytest.mark.django_db
def test_list_saved_searches_excludes_others_private(bob, alice, tenant_acme, acme_tree):
    services.create_saved_search(
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        owner=alice,
        name="Alice Only",
        is_shared=False,
    )
    results = list(queries.list_saved_searches(tenant_acme.pk, bob))
    # Bob should not see alice's private search
    for r in results:
        assert not (r.owner_id == alice.pk and r.is_shared is False)


# ---------------------------------------------------------------------------
# Signals
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_signal_indexes_on_document_save(doc, tenant_acme):
    """Saving a document automatically creates/updates its search index entry."""
    # Disconnect to get a fresh state, then trigger save manually
    doc.title = "Signal Test Save"
    doc.save()
    assert DocumentSearchIndex.objects.filter(document=doc).exists()


@pytest.mark.django_db
def test_signal_deindexes_soft_deleted_document(doc, tenant_acme):
    services.index_document(doc)
    assert DocumentSearchIndex.objects.filter(document=doc).exists()

    doc.is_deleted = True
    doc.save()
    assert not DocumentSearchIndex.objects.filter(document=doc).exists()


@pytest.mark.django_db
def test_signal_reindexes_on_version_save(doc, file_asset, tenant_acme, acme_tree):
    """Adding a version re-indexes the parent document."""
    services.index_document(doc)
    initial_version = DocumentSearchIndex.objects.get(document=doc).index_version

    v = add_version(
        document=doc,
        file_asset=file_asset,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    publish_version(v)

    idx = DocumentSearchIndex.objects.get(document=doc)
    assert idx.index_version > initial_version


# ---------------------------------------------------------------------------
# HTTP API — document search
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_search_returns_results(
    client, alice, tenant_acme, alice_membership, indexed_doc, srch_ctx
):
    client.force_login(alice)
    with use_request_context(srch_ctx):
        resp = client.get(
            "/api/v1/dms/search/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    data = resp.json()
    assert isinstance(data, list)
    pks = [r["public_id"] for r in data]
    assert str(indexed_doc.public_id) in pks


@pytest.mark.django_db
def test_api_search_with_q_filter(
    client, alice, tenant_acme, alice_membership, indexed_doc, srch_ctx
):
    client.force_login(alice)
    with use_request_context(srch_ctx):
        resp = client.get(
            "/api/v1/dms/search/?q=Search+Test+Document",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    pks = [r["public_id"] for r in resp.json()]
    assert str(indexed_doc.public_id) in pks


@pytest.mark.django_db
def test_api_search_requires_auth(client, tenant_acme):
    resp = client.get("/api/v1/dms/search/", HTTP_X_TENANT=tenant_acme.slug)
    assert resp.status_code == 401


@pytest.mark.django_db
def test_api_search_requires_perm(client, alice, tenant_acme):
    # alice has no membership → middleware gives empty permissions → 403
    client.force_login(alice)
    resp = client.get("/api/v1/dms/search/", HTTP_X_TENANT=tenant_acme.slug)
    assert resp.status_code == 403


# ---------------------------------------------------------------------------
# HTTP API — saved search list/create
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_list_saved_searches(
    client, alice, tenant_acme, alice_membership, saved_search, srch_ctx
):
    client.force_login(alice)
    with use_request_context(srch_ctx):
        resp = client.get(
            "/api/v1/dms/search/saved/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    pks = [r["public_id"] for r in resp.json()]
    assert str(saved_search.public_id) in pks


@pytest.mark.django_db
def test_api_create_saved_search(
    client, alice, tenant_acme, alice_membership, srch_ctx
):
    client.force_login(alice)
    with use_request_context(srch_ctx):
        resp = client.post(
            "/api/v1/dms/search/saved/",
            data={
                "name": "API Created Search",
                "query_params": {"q": "contract"},
                "is_shared": True,
            },
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 201
    data = resp.json()
    assert data["name"] == "API Created Search"
    assert data["is_shared"] is True


@pytest.mark.django_db
def test_api_create_saved_search_missing_name_returns_400(
    client, alice, tenant_acme, alice_membership, srch_ctx
):
    client.force_login(alice)
    with use_request_context(srch_ctx):
        resp = client.post(
            "/api/v1/dms/search/saved/",
            data={"query_params": {}},
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 400


@pytest.mark.django_db
def test_api_saved_search_list_requires_auth(client, tenant_acme):
    resp = client.get("/api/v1/dms/search/saved/", HTTP_X_TENANT=tenant_acme.slug)
    assert resp.status_code == 401


@pytest.mark.django_db
def test_api_saved_search_list_requires_perm(client, alice, tenant_acme):
    client.force_login(alice)
    resp = client.get("/api/v1/dms/search/saved/", HTTP_X_TENANT=tenant_acme.slug)
    assert resp.status_code == 403


# ---------------------------------------------------------------------------
# HTTP API — saved search detail/update/delete
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_saved_search_detail(
    client, alice, tenant_acme, alice_membership, saved_search, srch_ctx
):
    client.force_login(alice)
    with use_request_context(srch_ctx):
        resp = client.get(
            f"/api/v1/dms/search/saved/{saved_search.public_id}/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert resp.json()["public_id"] == str(saved_search.public_id)


@pytest.mark.django_db
def test_api_saved_search_update(
    client, alice, tenant_acme, alice_membership, saved_search, srch_ctx
):
    client.force_login(alice)
    with use_request_context(srch_ctx):
        resp = client.put(
            f"/api/v1/dms/search/saved/{saved_search.public_id}/",
            data={"name": "Updated Via API"},
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert resp.json()["name"] == "Updated Via API"


@pytest.mark.django_db
def test_api_saved_search_delete(
    client, alice, tenant_acme, alice_membership, saved_search, srch_ctx
):
    client.force_login(alice)
    with use_request_context(srch_ctx):
        resp = client.delete(
            f"/api/v1/dms/search/saved/{saved_search.public_id}/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 204
    saved_search.refresh_from_db()
    assert saved_search.is_deleted is True


@pytest.mark.django_db
def test_api_saved_search_not_found_returns_404(
    client, alice, tenant_acme, alice_membership, srch_ctx
):
    client.force_login(alice)
    with use_request_context(srch_ctx):
        resp = client.get(
            f"/api/v1/dms/search/saved/{uuid.uuid4()}/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 404


@pytest.mark.django_db
def test_api_saved_search_update_by_non_owner_returns_403(
    client, alice, tenant_acme, alice_membership, acme_tree
):
    """A regular non-owner non-superuser must get 403 when updating a shared (but
    not owned) saved search."""
    from django.contrib.auth import get_user_model

    User = get_user_model()
    charlie = User.objects.create_user(
        "+989000000088", password="x", email="charlie@example.com",
        is_superuser=False,
    )
    # Add charlie to the existing membership so org_node_id resolves
    alice_membership.users.add(charlie)

    # Create a shared search owned by alice — charlie can SEE it but not MODIFY it
    shared_ss = services.create_saved_search(
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        owner=alice,
        name="Shared By Alice",
        is_shared=True,
    )

    charlie_ctx = RequestContext(
        actor=charlie,
        tenant=tenant_acme,
        org_node_ids=frozenset({acme_tree["root"].pk}),
        permissions=ALL_PERMS,
    )
    client.force_login(charlie)
    with use_request_context(charlie_ctx):
        resp = client.put(
            f"/api/v1/dms/search/saved/{shared_ss.public_id}/",
            data={"name": "Charlie Override"},
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 403


# ---------------------------------------------------------------------------
# HTTP API — execute saved search
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_execute_saved_search(
    client, alice, tenant_acme, alice_membership, saved_search, indexed_doc, srch_ctx
):
    client.force_login(alice)
    with use_request_context(srch_ctx):
        resp = client.post(
            f"/api/v1/dms/search/saved/{saved_search.public_id}/execute/",
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert isinstance(resp.json(), list)


@pytest.mark.django_db
def test_api_execute_saved_search_increments_use_count(
    client, alice, tenant_acme, alice_membership, saved_search, indexed_doc, srch_ctx
):
    client.force_login(alice)
    with use_request_context(srch_ctx):
        client.post(
            f"/api/v1/dms/search/saved/{saved_search.public_id}/execute/",
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    saved_search.refresh_from_db()
    assert saved_search.use_count == 1


@pytest.mark.django_db
def test_api_execute_saved_search_not_found_returns_404(
    client, alice, tenant_acme, alice_membership, srch_ctx
):
    client.force_login(alice)
    with use_request_context(srch_ctx):
        resp = client.post(
            f"/api/v1/dms/search/saved/{uuid.uuid4()}/execute/",
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 404


@pytest.mark.django_db
def test_api_execute_saved_search_requires_auth(client, tenant_acme, saved_search):
    resp = client.post(
        f"/api/v1/dms/search/saved/{saved_search.public_id}/execute/",
        HTTP_X_TENANT=tenant_acme.slug,
    )
    assert resp.status_code == 401


@pytest.mark.django_db
def test_api_execute_saved_search_requires_perm(
    client, alice, tenant_acme, saved_search
):
    # alice has no membership → middleware gives empty permissions → 403
    client.force_login(alice)
    resp = client.post(
        f"/api/v1/dms/search/saved/{saved_search.public_id}/execute/",
        HTTP_X_TENANT=tenant_acme.slug,
    )
    assert resp.status_code == 403