leocrm/tests/test_entity_permissions.py

"""Tests for entity permission service — ACL resolution, ownership, sharing, expiration.

Covers:
- Owner sees own contacts
- Non-owner doesn't see others' contacts
- Shared user sees shared contacts
- Permission expiration
- System admin sees all
- Batch resolution
"""

from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.contact import Contact
from app.models.entity_permission import EntityPermission
from app.models.user import User
from app.services import entity_permission_service as eps
from tests.conftest import seed_tenant_and_users


@pytest.mark.asyncio
class TestEntityPermissions:
    """Tests for entity permission service — ACL resolution, ownership, sharing."""

    async def test_owner_sees_own_contacts(self, db_session: AsyncSession):
        """Owner has 'owner' access level on their own contacts."""
        seed = await seed_tenant_and_users(db_session)
        tenant_id = seed["tenant_a"].id
        user_id = seed["admin_a"].id
        contact_id = seed["company_a"].id

        # Set owner_id on the contact
        contact = await db_session.get(Contact, contact_id)
        contact.owner_id = user_id
        await db_session.commit()

        access = await eps.get_effective_access(
            db_session, tenant_id, user_id, "contact", contact_id
        )
        assert access == "owner", f"Expected 'owner', got '{access}'"

    async def test_non_owner_doesnt_see_others_contacts(self, db_session: AsyncSession):
        """Non-owner without explicit permission gets 'none' access."""
        seed = await seed_tenant_and_users(db_session)
        tenant_id = seed["tenant_a"].id
        owner_id = seed["admin_a"].id
        other_user_id = seed["viewer_a"].id
        contact_id = seed["company_a"].id

        # Set owner_id on the contact
        contact = await db_session.get(Contact, contact_id)
        contact.owner_id = owner_id
        await db_session.commit()

        access = await eps.get_effective_access(
            db_session, tenant_id, other_user_id, "contact", contact_id
        )
        assert access == "none", f"Expected 'none', got '{access}'"

    async def test_shared_user_sees_shared_contacts(self, db_session: AsyncSession):
        """User with explicit 'read' permission sees the contact."""
        seed = await seed_tenant_and_users(db_session)
        tenant_id = seed["tenant_a"].id
        owner_id = seed["admin_a"].id
        shared_user_id = seed["viewer_a"].id
        contact_id = seed["company_a"].id

        # Set owner_id on the contact
        contact = await db_session.get(Contact, contact_id)
        contact.owner_id = owner_id
        await db_session.commit()

        # Grant read permission to viewer
        await eps.create_permission(
            db_session,
            tenant_id=tenant_id,
            entity_type="contact",
            entity_id=str(contact_id),
            principal_type="user",
            principal_id=str(shared_user_id),
            permission_level="read",
            created_by=owner_id,
        )

        access = await eps.get_effective_access(
            db_session, tenant_id, shared_user_id, "contact", contact_id
        )
        assert access == "read", f"Expected 'read', got '{access}'"

    async def test_permission_expiration(self, db_session: AsyncSession):
        """Expired permission returns 'none' access."""
        seed = await seed_tenant_and_users(db_session)
        tenant_id = seed["tenant_a"].id
        owner_id = seed["admin_a"].id
        shared_user_id = seed["viewer_a"].id
        contact_id = seed["company_a"].id

        # Set owner_id on the contact
        contact = await db_session.get(Contact, contact_id)
        contact.owner_id = owner_id
        await db_session.commit()

        # Grant read permission that expired 1 hour ago
        await eps.create_permission(
            db_session,
            tenant_id=tenant_id,
            entity_type="contact",
            entity_id=str(contact_id),
            principal_type="user",
            principal_id=str(shared_user_id),
            permission_level="read",
            expires_at=datetime.now(UTC) - timedelta(hours=1),
            created_by=owner_id,
        )

        access = await eps.get_effective_access(
            db_session, tenant_id, shared_user_id, "contact", contact_id
        )
        assert access == "none", f"Expected 'none' for expired permission, got '{access}'"

    async def test_system_admin_sees_all(self, db_session: AsyncSession):
        """System admin gets 'delete' access on any entity."""
        seed = await seed_tenant_and_users(db_session)
        tenant_id = seed["tenant_a"].id
        contact_id = seed["company_a"].id

        # Create a system admin user
        from app.core.auth import hash_password
        sys_admin = User(
            email="sysadmin@test.com",
            name="System Admin",
            password_hash=hash_password("TestPass123!"),
            is_active=True,
            is_system_admin=True,
            preferences={},
        )
        db_session.add(sys_admin)
        await db_session.commit()

        access = await eps.get_effective_access(
            db_session, tenant_id, sys_admin.id, "contact", contact_id
        )
        assert access == "delete", f"Expected 'delete' for system admin, got '{access}'"

    async def test_batch_resolution(self, db_session: AsyncSession):
        """Batch resolution returns correct access levels for multiple entities."""
        seed = await seed_tenant_and_users(db_session)
        tenant_id = seed["tenant_a"].id
        owner_id = seed["admin_a"].id
        viewer_id = seed["viewer_a"].id
        contact_id = seed["company_a"].id

        # Create a second contact owned by viewer
        contact2 = Contact(
            tenant_id=tenant_id,
            type="company",
            name="Company Viewer",
            displayname="Company Viewer",
            owner_id=viewer_id,
            created_by=viewer_id,
            updated_by=viewer_id,
        )
        db_session.add(contact2)
        await db_session.commit()

        # Set owner on first contact
        contact = await db_session.get(Contact, contact_id)
        contact.owner_id = owner_id
        await db_session.commit()

        # Grant read permission to viewer on first contact
        await eps.create_permission(
            db_session,
            tenant_id=tenant_id,
            entity_type="contact",
            entity_id=str(contact_id),
            principal_type="user",
            principal_id=str(viewer_id),
            permission_level="read",
            created_by=owner_id,
        )

        # Batch resolve for viewer
        result = await eps.batch_get_effective_access(
            db_session, tenant_id, viewer_id, "contact", [contact_id, contact2.id]
        )

        assert result[contact_id] == "read", f"Expected 'read' for shared contact, got '{result[contact_id]}'"
        assert result[contact2.id] == "owner", f"Expected 'owner' for own contact, got '{result[contact2.id]}'"

    async def test_tenant_owned_contact_visible_to_all(self, db_session: AsyncSession):
        """Tenant-owned contact (owner_id IS NULL) is visible as 'read' to all tenant users."""
        seed = await seed_tenant_and_users(db_session)
        tenant_id = seed["tenant_a"].id
        viewer_id = seed["viewer_a"].id
        contact_id = seed["company_a"].id

        # Ensure owner_id is NULL
        contact = await db_session.get(Contact, contact_id)
        contact.owner_id = None
        await db_session.commit()

        access = await eps.get_effective_access(
            db_session, tenant_id, viewer_id, "contact", contact_id
        )
        assert access == "read", f"Expected 'read' for tenant-owned contact, got '{access}'"

    async def test_group_permission_propagation(self, db_session: AsyncSession):
        """User inherits permissions from group membership."""
        seed = await seed_tenant_and_users(db_session)
        tenant_id = seed["tenant_a"].id
        owner_id = seed["admin_a"].id
        viewer_id = seed["viewer_a"].id
        contact_id = seed["company_a"].id

        # Set owner_id on the contact
        contact = await db_session.get(Contact, contact_id)
        contact.owner_id = owner_id
        await db_session.commit()

        # Create a group and add viewer to it
        from app.models.group import Group, UserGroup
        group = Group(tenant_id=tenant_id, name="Viewers")
        db_session.add(group)
        await db_session.flush()

        ug = UserGroup(tenant_id=tenant_id, user_id=viewer_id, group_id=group.id)
        db_session.add(ug)
        await db_session.commit()

        # Grant permission to the group
        await eps.create_permission(
            db_session,
            tenant_id=tenant_id,
            entity_type="contact",
            entity_id=str(contact_id),
            principal_type="group",
            principal_id=str(group.id),
            permission_level="write",
            created_by=owner_id,
        )

        access = await eps.get_effective_access(
            db_session, tenant_id, viewer_id, "contact", contact_id
        )
        assert access == "write", f"Expected 'write' via group, got '{access}'"

    async def test_highest_permission_wins(self, db_session: AsyncSession):
        """When multiple permissions exist, the highest level wins."""
        seed = await seed_tenant_and_users(db_session)
        tenant_id = seed["tenant_a"].id
        owner_id = seed["admin_a"].id
        viewer_id = seed["viewer_a"].id
        contact_id = seed["company_a"].id

        # Set owner_id on the contact
        contact = await db_session.get(Contact, contact_id)
        contact.owner_id = owner_id
        await db_session.commit()

        # Grant read + write permissions (write should win)
        await eps.create_permission(
            db_session,
            tenant_id=tenant_id,
            entity_type="contact",
            entity_id=str(contact_id),
            principal_type="user",
            principal_id=str(viewer_id),
            permission_level="read",
            created_by=owner_id,
        )
        await eps.create_permission(
            db_session,
            tenant_id=tenant_id,
            entity_type="contact",
            entity_id=str(contact_id),
            principal_type="user",
            principal_id=str(viewer_id),
            permission_level="write",
            created_by=owner_id,
        )

        access = await eps.get_effective_access(
            db_session, tenant_id, viewer_id, "contact", contact_id
        )
        assert access == "write", f"Expected 'write' (highest wins), got '{access}'"

    async def test_visible_ids_returns_owned_and_shared(self, db_session: AsyncSession):
        """get_visible_ids returns owned + shared + tenant-owned entities."""
        seed = await seed_tenant_and_users(db_session)
        tenant_id = seed["tenant_a"].id
        owner_id = seed["admin_a"].id
        viewer_id = seed["viewer_a"].id
        contact_id = seed["company_a"].id

        # Create a second contact owned by viewer
        contact2 = Contact(
            tenant_id=tenant_id,
            type="company",
            name="Viewer Owned",
            displayname="Viewer Owned",
            owner_id=viewer_id,
            created_by=viewer_id,
            updated_by=viewer_id,
        )
        db_session.add(contact2)
        await db_session.commit()

        # Set owner on first contact
        contact = await db_session.get(Contact, contact_id)
        contact.owner_id = owner_id
        await db_session.commit()

        # Grant read to viewer on first contact
        await eps.create_permission(
            db_session,
            tenant_id=tenant_id,
            entity_type="contact",
            entity_id=str(contact_id),
            principal_type="user",
            principal_id=str(viewer_id),
            permission_level="read",
            created_by=owner_id,
        )

        visible, access_map = await eps.get_visible_ids(
            db_session, tenant_id, viewer_id, "contact"
        )

        assert contact_id in visible, "Shared contact should be visible"
        assert contact2.id in visible, "Owned contact should be visible"
        assert access_map[contact_id] == "read"
        assert access_map[contact2.id] == "owner"

    async def test_check_entity_access_enforces_required_level(self, db_session: AsyncSession):
        """check_entity_access returns True/False based on required level."""
        seed = await seed_tenant_and_users(db_session)
        tenant_id = seed["tenant_a"].id
        owner_id = seed["admin_a"].id
        viewer_id = seed["viewer_a"].id
        contact_id = seed["company_a"].id

        # Set owner_id on the contact
        contact = await db_session.get(Contact, contact_id)
        contact.owner_id = owner_id
        await db_session.commit()

        # Grant read to viewer
        await eps.create_permission(
            db_session,
            tenant_id=tenant_id,
            entity_type="contact",
            entity_id=str(contact_id),
            principal_type="user",
            principal_id=str(viewer_id),
            permission_level="read",
            created_by=owner_id,
        )

        # Should have read access
        assert await eps.check_entity_access(
            db_session, tenant_id, viewer_id, "contact", contact_id, "read"
        ) is True

        # Should NOT have write access
        assert await eps.check_entity_access(
            db_session, tenant_id, viewer_id, "contact", contact_id, "write"
        ) is False

    async def test_cleanup_expired_permissions(self, db_session: AsyncSession):
        """cleanup_expired_permissions removes expired entries."""
        seed = await seed_tenant_and_users(db_session)
        tenant_id = seed["tenant_a"].id
        owner_id = seed["admin_a"].id
        viewer_id = seed["viewer_a"].id
        contact_id = seed["company_a"].id

        # Create expired permission
        await eps.create_permission(
            db_session,
            tenant_id=tenant_id,
            entity_type="contact",
            entity_id=str(contact_id),
            principal_type="user",
            principal_id=str(viewer_id),
            permission_level="read",
            expires_at=datetime.now(UTC) - timedelta(hours=1),
            created_by=owner_id,
        )

        count = await eps.cleanup_expired_permissions(db_session)
        assert count >= 1, "Expected at least 1 expired permission cleaned up"

        # Verify it's gone
        access = await eps.get_effective_access(
            db_session, tenant_id, viewer_id, "contact", contact_id
        )
        assert access != "write", "Expired write permission should be gone after cleanup"