"""Tests for DMS Phase 5 — ACL & SECURITY.

Covers:
  * ACLPolicy CRUD service operations
  * ACLRule add / remove / duplicate guard
  * DocumentPermission grant / revoke / duplicate guard
  * ShareLink create / use / revoke / exhaustion / expiry / password
  * resolve_document_permissions — ALLOW, DENY, deny-wins, EVERYONE
  * Full HTTP API: policies, rules, doc permissions, share links, public use
  * Permission gates (401, 403)
"""

from __future__ import annotations

import uuid
from datetime import timedelta

import pytest
from django.utils import timezone

from simorgh.apps.dms.documents.services import add_version
from simorgh.apps.dms.permissions import queries, services
from simorgh.apps.dms.permissions.iam_permissions import (
    PERM_ACL_MANAGE,
    PERM_ACL_VIEW,
    PERM_SHARE_MANAGE,
    PERM_SHARE_VIEW,
)
from simorgh.apps.dms.permissions.models import (
    ACLAction,
    ACLEffect,
    ACLPolicy,
    ACLPrincipalType,
    ACLSubjectType,
    DocumentPermission,
    ShareLink,
)
from simorgh.apps.dms.permissions.services import ACLError, ShareLinkError
from simorgh.apps.dms.repositories.models import Repository
from simorgh.core.context import RequestContext, use_request_context


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

ALL_PERMS = frozenset({PERM_ACL_VIEW, PERM_ACL_MANAGE, PERM_SHARE_VIEW, PERM_SHARE_MANAGE})


@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 acl_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="Main",
        slug="main",
        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="Policy Doc",
        repository=repo,
        tenant=tenant_acme,
        organization_node=acme_tree["root"],
    )


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

    return FileMetadata.objects.create(
        filename="policy.pdf",
        content_type="application/pdf",
        size_bytes=2048,
        path=f"test/phase5/{_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):
    return add_version(
        document=doc,
        file_asset=file_asset,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )


@pytest.fixture
def policy(doc, tenant_acme, acme_tree):
    return services.create_policy(
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        subject_type=ACLSubjectType.DOCUMENT,
        subject_id=str(doc.public_id),
        name="Test Policy",
        inheritable=False,
    )


# ---------------------------------------------------------------------------
# Service — ACLPolicy
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_create_policy(doc, tenant_acme, acme_tree):
    p = services.create_policy(
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        subject_type=ACLSubjectType.DOCUMENT,
        subject_id=str(doc.public_id),
        name="My Policy",
        inheritable=True,
    )
    assert p.pk is not None
    assert p.name == "My Policy"
    assert p.subject_type == ACLSubjectType.DOCUMENT
    assert p.inheritable is True


@pytest.mark.django_db
def test_create_policy_invalid_subject_type_raises(tenant_acme, acme_tree):
    with pytest.raises(ACLError, match="Invalid subject type"):
        services.create_policy(
            tenant_id=tenant_acme.pk,
            organization_node_id=acme_tree["root"].pk,
            subject_type="galaxy",
            subject_id="abc",
            name="Bad",
        )


@pytest.mark.django_db
def test_update_policy(policy):
    policy = services.update_policy(policy, name="Renamed", is_active=False)
    assert policy.name == "Renamed"
    assert policy.is_active is False


@pytest.mark.django_db
def test_delete_policy_soft_deletes_rules(policy, tenant_acme, acme_tree):
    rule = services.add_rule(
        policy,
        principal_type=ACLPrincipalType.EVERYONE,
        action=ACLAction.VIEW,
    )
    services.delete_policy(policy)
    policy.refresh_from_db()
    rule.refresh_from_db()
    assert policy.is_deleted is True
    assert rule.is_deleted is True


# ---------------------------------------------------------------------------
# Service — ACLRule
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_add_rule(policy):
    rule = services.add_rule(
        policy,
        principal_type=ACLPrincipalType.USER,
        principal_id="42",
        action=ACLAction.VIEW,
        effect=ACLEffect.ALLOW,
        priority=10,
    )
    assert rule.pk is not None
    assert rule.action == ACLAction.VIEW
    assert rule.effect == ACLEffect.ALLOW
    assert rule.priority == 10


@pytest.mark.django_db
def test_add_rule_invalid_action_raises(policy):
    with pytest.raises(ACLError, match="Invalid action"):
        services.add_rule(
            policy,
            principal_type=ACLPrincipalType.EVERYONE,
            action="fly",
        )


@pytest.mark.django_db
def test_add_rule_duplicate_raises(policy):
    services.add_rule(
        policy,
        principal_type=ACLPrincipalType.EVERYONE,
        action=ACLAction.VIEW,
    )
    with pytest.raises(ACLError, match="already exists"):
        services.add_rule(
            policy,
            principal_type=ACLPrincipalType.EVERYONE,
            action=ACLAction.VIEW,
        )


@pytest.mark.django_db
def test_everyone_principal_id_cleared(policy):
    rule = services.add_rule(
        policy,
        principal_type=ACLPrincipalType.EVERYONE,
        action=ACLAction.VIEW,
        principal_id="should-be-ignored",
    )
    assert rule.principal_id == ""


@pytest.mark.django_db
def test_remove_rule(policy):
    rule = services.add_rule(
        policy,
        principal_type=ACLPrincipalType.EVERYONE,
        action=ACLAction.DOWNLOAD,
    )
    services.remove_rule(rule)
    rule.refresh_from_db()
    assert rule.is_deleted is True


# ---------------------------------------------------------------------------
# Service — DocumentPermission
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_grant_document_permission(doc, tenant_acme, acme_tree, alice):
    perm = services.grant_document_permission(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        principal_type=ACLPrincipalType.USER,
        principal_id=str(alice.pk),
        action=ACLAction.VIEW,
        effect=ACLEffect.ALLOW,
        granted_by=alice,
    )
    assert perm.pk is not None
    assert perm.action == ACLAction.VIEW
    assert perm.effect == ACLEffect.ALLOW


@pytest.mark.django_db
def test_grant_duplicate_raises(doc, tenant_acme, acme_tree, alice):
    services.grant_document_permission(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        principal_type=ACLPrincipalType.USER,
        principal_id=str(alice.pk),
        action=ACLAction.VIEW,
    )
    with pytest.raises(ACLError, match="already exists"):
        services.grant_document_permission(
            document=doc,
            tenant_id=tenant_acme.pk,
            organization_node_id=acme_tree["root"].pk,
            principal_type=ACLPrincipalType.USER,
            principal_id=str(alice.pk),
            action=ACLAction.VIEW,
        )


@pytest.mark.django_db
def test_revoke_document_permission(doc, tenant_acme, acme_tree, alice):
    perm = services.grant_document_permission(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        principal_type=ACLPrincipalType.USER,
        principal_id=str(alice.pk),
        action=ACLAction.DOWNLOAD,
    )
    services.revoke_document_permission(perm)
    perm.refresh_from_db()
    assert perm.is_deleted is True


# ---------------------------------------------------------------------------
# Service — ShareLink
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_create_share_link(doc, tenant_acme, acme_tree, alice):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        created_by=alice,
        label="For client",
        allow_download=True,
        allow_preview=True,
    )
    assert link.pk is not None
    assert link.token is not None
    assert link.is_active is True
    assert link.password_hash == ""


@pytest.mark.django_db
def test_create_share_link_with_password(doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        password="s3cr3t",
    )
    assert link.password_hash != ""
    assert link.password_hash != "s3cr3t"


@pytest.mark.django_db
def test_use_share_link(doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        max_uses=3,
    )
    link = services.use_share_link(link)
    assert link.use_count == 1
    link = services.use_share_link(link)
    assert link.use_count == 2


@pytest.mark.django_db
def test_use_share_link_exhausted_raises(doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        max_uses=1,
    )
    services.use_share_link(link)
    with pytest.raises(ShareLinkError, match="maximum use count"):
        services.use_share_link(link)


@pytest.mark.django_db
def test_use_share_link_expired_raises(doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        expires_at=timezone.now() - timedelta(hours=1),
    )
    with pytest.raises(ShareLinkError, match="expired"):
        services.use_share_link(link)


@pytest.mark.django_db
def test_use_share_link_wrong_password_raises(doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        password="correct",
    )
    with pytest.raises(ShareLinkError, match="Incorrect password"):
        services.use_share_link(link, password="wrong")


@pytest.mark.django_db
def test_use_share_link_correct_password(doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        password="correct",
    )
    link = services.use_share_link(link, password="correct")
    assert link.use_count == 1


@pytest.mark.django_db
def test_use_revoked_share_link_raises(doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    services.revoke_share_link(link)
    with pytest.raises(ShareLinkError, match="revoked"):
        services.use_share_link(link)


@pytest.mark.django_db
def test_revoke_share_link(doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    link = services.revoke_share_link(link)
    assert link.is_active is False


# ---------------------------------------------------------------------------
# resolve_document_permissions
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_resolve_allow_via_doc_permission(doc, tenant_acme, acme_tree, alice):
    services.grant_document_permission(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        principal_type=ACLPrincipalType.USER,
        principal_id=str(alice.pk),
        action=ACLAction.VIEW,
    )
    allowed = services.resolve_document_permissions(
        doc, user_id=str(alice.pk)
    )
    assert ACLAction.VIEW in allowed


@pytest.mark.django_db
def test_resolve_deny_overrides_allow(doc, tenant_acme, acme_tree, alice):
    services.grant_document_permission(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        principal_type=ACLPrincipalType.USER,
        principal_id=str(alice.pk),
        action=ACLAction.VIEW,
        effect=ACLEffect.ALLOW,
    )
    services.grant_document_permission(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        principal_type=ACLPrincipalType.EVERYONE,
        action=ACLAction.VIEW,
        effect=ACLEffect.DENY,
    )
    allowed = services.resolve_document_permissions(
        doc, user_id=str(alice.pk)
    )
    # DENY wins over ALLOW for VIEW
    assert ACLAction.VIEW not in allowed


@pytest.mark.django_db
def test_resolve_everyone_principal(doc, tenant_acme, acme_tree):
    services.grant_document_permission(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        principal_type=ACLPrincipalType.EVERYONE,
        action=ACLAction.VIEW,
    )
    # Any user id should match EVERYONE
    allowed = services.resolve_document_permissions(
        doc, user_id="9999"
    )
    assert ACLAction.VIEW in allowed


@pytest.mark.django_db
def test_resolve_via_policy_rule(doc, tenant_acme, acme_tree, policy):
    services.add_rule(
        policy,
        principal_type=ACLPrincipalType.ROLE,
        principal_id="admin",
        action=ACLAction.MANAGE,
    )
    allowed = services.resolve_document_permissions(
        doc, user_id="99", role_slugs=["admin"]
    )
    assert ACLAction.MANAGE in allowed


@pytest.mark.django_db
def test_resolve_expired_doc_permission_ignored(doc, tenant_acme, acme_tree, alice):
    services.grant_document_permission(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        principal_type=ACLPrincipalType.USER,
        principal_id=str(alice.pk),
        action=ACLAction.VIEW,
        expires_at=timezone.now() - timedelta(seconds=1),
    )
    allowed = services.resolve_document_permissions(
        doc, user_id=str(alice.pk)
    )
    assert ACLAction.VIEW not in allowed


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

@pytest.mark.django_db
def test_list_policies_for_subject(doc, tenant_acme, acme_tree, policy):
    qs = queries.list_policies_for_subject(
        tenant_acme.pk, ACLSubjectType.DOCUMENT, str(doc.public_id)
    )
    assert qs.filter(pk=policy.pk).exists()


@pytest.mark.django_db
def test_get_share_link_by_token(doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    found = queries.get_share_link_by_token(str(link.token))
    assert found is not None
    assert found.pk == link.pk

    not_found = queries.get_share_link_by_token(str(uuid.uuid4()))
    assert not_found is None


# ---------------------------------------------------------------------------
# HTTP API — ACL Policies
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_create_policy(client, alice, tenant_acme, acme_tree, alice_membership,
                            doc, acl_ctx):
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.post(
            "/api/v1/dms/acl/policies/",
            data={
                "subject_type": "document",
                "subject_id": str(doc.public_id),
                "name": "API Policy",
                "inheritable": False,
            },
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 201
    data = resp.json()
    assert data["name"] == "API Policy"
    assert data["subject_type"] == "document"


@pytest.mark.django_db
def test_api_list_policies(client, alice, tenant_acme, acme_tree, alice_membership,
                            doc, policy, acl_ctx):
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.get(
            "/api/v1/dms/acl/policies/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    ids = [p["public_id"] for p in resp.json()]
    assert str(policy.public_id) in ids


@pytest.mark.django_db
def test_api_update_policy(client, alice, tenant_acme, acme_tree, alice_membership,
                            policy, acl_ctx):
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.patch(
            f"/api/v1/dms/acl/policies/{policy.public_id}/",
            data={"name": "Renamed Policy"},
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert resp.json()["name"] == "Renamed Policy"


@pytest.mark.django_db
def test_api_delete_policy(client, alice, tenant_acme, acme_tree, alice_membership,
                            policy, acl_ctx):
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.delete(
            f"/api/v1/dms/acl/policies/{policy.public_id}/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 204
    policy.refresh_from_db()
    assert policy.is_deleted is True


# ---------------------------------------------------------------------------
# HTTP API — ACL Rules
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_add_rule(client, alice, tenant_acme, acme_tree, alice_membership,
                       policy, acl_ctx):
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.post(
            f"/api/v1/dms/acl/policies/{policy.public_id}/rules/",
            data={
                "principal_type": "everyone",
                "action": "view",
                "effect": "allow",
            },
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 201
    assert resp.json()["action"] == "view"


@pytest.mark.django_db
def test_api_list_rules(client, alice, tenant_acme, acme_tree, alice_membership,
                         policy, acl_ctx):
    services.add_rule(policy, principal_type=ACLPrincipalType.EVERYONE, action=ACLAction.VIEW)
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.get(
            f"/api/v1/dms/acl/policies/{policy.public_id}/rules/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert len(resp.json()) == 1


@pytest.mark.django_db
def test_api_remove_rule(client, alice, tenant_acme, acme_tree, alice_membership,
                          policy, acl_ctx):
    rule = services.add_rule(
        policy, principal_type=ACLPrincipalType.EVERYONE, action=ACLAction.DOWNLOAD
    )
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.delete(
            f"/api/v1/dms/acl/policies/{policy.public_id}/rules/{rule.public_id}/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 204


# ---------------------------------------------------------------------------
# HTTP API — Document Permissions
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_grant_doc_permission(client, alice, tenant_acme, acme_tree, alice_membership,
                                   doc, acl_ctx):
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.post(
            f"/api/v1/dms/documents/{doc.public_id}/acl/",
            data={
                "principal_type": "user",
                "principal_id": "99",
                "action": "view",
                "effect": "allow",
            },
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 201
    data = resp.json()
    assert data["action"] == "view"
    assert data["principal_id"] == "99"


@pytest.mark.django_db
def test_api_revoke_doc_permission(client, alice, tenant_acme, acme_tree, alice_membership,
                                    doc, acl_ctx):
    perm = services.grant_document_permission(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        principal_type=ACLPrincipalType.USER,
        principal_id="55",
        action=ACLAction.VIEW,
    )
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.delete(
            f"/api/v1/dms/documents/{doc.public_id}/acl/{perm.public_id}/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 204
    perm.refresh_from_db()
    assert perm.is_deleted is True


@pytest.mark.django_db
def test_api_resolve_permissions(client, alice, tenant_acme, acme_tree, alice_membership,
                                  doc, acl_ctx):
    services.grant_document_permission(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        principal_type=ACLPrincipalType.USER,
        principal_id=str(alice.pk),
        action=ACLAction.VIEW,
    )
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.get(
            f"/api/v1/dms/documents/{doc.public_id}/acl/resolve/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert "view" in resp.json()["allowed_actions"]


# ---------------------------------------------------------------------------
# HTTP API — Share Links
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_create_share_link(client, alice, tenant_acme, acme_tree, alice_membership,
                                doc, acl_ctx):
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.post(
            f"/api/v1/dms/documents/{doc.public_id}/share-links/",
            data={"label": "Review link", "allow_download": True},
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 201
    data = resp.json()
    assert data["label"] == "Review link"
    assert "token" in data


@pytest.mark.django_db
def test_api_list_share_links(client, alice, tenant_acme, acme_tree, alice_membership,
                               doc, acl_ctx):
    services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.get(
            f"/api/v1/dms/documents/{doc.public_id}/share-links/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert len(resp.json()) == 1


@pytest.mark.django_db
def test_api_revoke_share_link(client, alice, tenant_acme, acme_tree, alice_membership,
                                doc, acl_ctx):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.post(
            f"/api/v1/dms/documents/{doc.public_id}/share-links/{link.public_id}/revoke/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert resp.json()["is_active"] is False


@pytest.mark.django_db
def test_api_delete_share_link(client, alice, tenant_acme, acme_tree, alice_membership,
                                doc, acl_ctx):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    client.force_login(alice)
    with use_request_context(acl_ctx):
        resp = client.delete(
            f"/api/v1/dms/documents/{doc.public_id}/share-links/{link.public_id}/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 204
    link.refresh_from_db()
    assert link.is_deleted is True


# ---------------------------------------------------------------------------
# Public share link use endpoint
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_use_share_link(client, doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        allow_download=True,
        allow_preview=True,
    )
    resp = client.post(f"/api/v1/dms/share/{link.token}/use/")
    assert resp.status_code == 200
    data = resp.json()
    assert data["allow_download"] is True
    assert data["use_count"] == 1


@pytest.mark.django_db
def test_api_use_share_link_wrong_token(client):
    resp = client.post(f"/api/v1/dms/share/{uuid.uuid4()}/use/")
    assert resp.status_code == 404


@pytest.mark.django_db
def test_api_use_share_link_expired(client, doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        expires_at=timezone.now() - timedelta(hours=1),
    )
    resp = client.post(f"/api/v1/dms/share/{link.token}/use/")
    assert resp.status_code == 403


@pytest.mark.django_db
def test_api_use_share_link_password_required(client, doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        password="s3cr3t",
    )
    resp = client.post(f"/api/v1/dms/share/{link.token}/use/",
                       data={}, content_type="application/json")
    assert resp.status_code == 403


@pytest.mark.django_db
def test_api_use_share_link_with_password(client, doc, tenant_acme, acme_tree):
    link = services.create_share_link(
        document=doc,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        password="s3cr3t",
    )
    resp = client.post(
        f"/api/v1/dms/share/{link.token}/use/",
        data={"password": "s3cr3t"},
        content_type="application/json",
    )
    assert resp.status_code == 200


# ---------------------------------------------------------------------------
# Permission gate tests
# ---------------------------------------------------------------------------

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


@pytest.mark.django_db
def test_api_policy_requires_perm(client, alice, tenant_acme, acme_tree, doc):
    ctx_no_perm = RequestContext(
        actor=alice,
        tenant=tenant_acme,
        org_node_ids=frozenset({acme_tree["root"].pk}),
        permissions=frozenset(),
    )
    client.force_login(alice)
    with use_request_context(ctx_no_perm):
        resp = client.get(
            "/api/v1/dms/acl/policies/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 403