"""Unit tests for P1-7 permission system fixes.

Tests:
1. _merge_field_permissions: strictest-wins merge logic
2. invalidate_all_user_permissions: SCAN-based (no KEYS)
3. get_cached_permissions: version validation logic
4. require_write: no broad wildcard *:write
"""

import asyncio
import json
import logging
import uuid

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from app.core.permissions import (
    _FIELD_PERM_SEVERITY,
    _merge_field_permissions,
    _matches_permission,
    _normalize_permissions,
    check_permission,
    CACHE_PREFIX,
)


class TestMergeFieldPermissions:
    """Tests for _merge_field_permissions — strictest-wins merge."""

    def test_empty_incoming_no_change(self):
        existing = {"contacts": {"name": "read"}}
        _merge_field_permissions(existing, {})
        assert existing == {"contacts": {"name": "read"}}

    def test_new_module_added(self):
        existing = {}
        _merge_field_permissions(existing, {"contacts": {"name": "hidden"}})
        assert existing == {"contacts": {"name": "hidden"}}

    def test_strictest_wins_hidden_over_read(self):
        existing = {"contacts": {"name": "read"}}
        _merge_field_permissions(existing, {"contacts": {"name": "hidden"}})
        assert existing["contacts"]["name"] == "hidden"

    def test_strictest_wins_read_does_not_override_hidden(self):
        existing = {"contacts": {"name": "hidden"}}
        _merge_field_permissions(existing, {"contacts": {"name": "read"}})
        assert existing["contacts"]["name"] == "hidden"

    def test_strictest_wins_readonly_over_read(self):
        existing = {"contacts": {"name": "read"}}
        _merge_field_permissions(existing, {"contacts": {"name": "readonly"}})
        assert existing["contacts"]["name"] == "readonly"

    def test_strictest_wins_readonly_does_not_override_hidden(self):
        existing = {"contacts": {"name": "hidden"}}
        _merge_field_permissions(existing, {"contacts": {"name": "readonly"}})
        assert existing["contacts"]["name"] == "hidden"

    def test_multiple_fields_merge_independently(self):
        existing = {"contacts": {"name": "hidden", "email": "read"}}
        _merge_field_permissions(
            existing,
            {"contacts": {"name": "read", "email": "hidden", "phone": "readonly"}},
        )
        assert existing["contacts"]["name"] == "hidden"  # hidden stayed
        assert existing["contacts"]["email"] == "hidden"  # read upgraded to hidden
        assert existing["contacts"]["phone"] == "readonly"  # new field added

    def test_multiple_modules_merge_independently(self):
        existing = {"contacts": {"name": "read"}}
        _merge_field_permissions(
            existing,
            {"users": {"email": "hidden"}, "contacts": {"name": "readonly"}},
        )
        assert existing["contacts"]["name"] == "readonly"
        assert existing["users"]["email"] == "hidden"

    def test_unknown_permission_level_skipped(self, caplog):
        existing = {"contacts": {"name": "read"}}
        with caplog.at_level(logging.WARNING):
            _merge_field_permissions(
                existing,
                {"contacts": {"name": "bogus"}},
            )
        assert existing["contacts"]["name"] == "read"  # unchanged
        assert "Unknown field permission level" in caplog.text

    def test_non_dict_fields_skipped(self):
        existing = {}
        _merge_field_permissions(existing, {"contacts": "not_a_dict"})
        assert existing == {}

    def test_non_string_perm_skipped(self):
        existing = {"contacts": {}}
        _merge_field_permissions(existing, {"contacts": {"name": 123}})
        assert existing["contacts"] == {}

    def test_case_insensitive_perm_level(self):
        existing = {"contacts": {"name": "read"}}
        _merge_field_permissions(existing, {"contacts": {"name": "HIDDEN"}})
        assert existing["contacts"]["name"] == "hidden"

    def test_severity_ordering_constant(self):
        assert _FIELD_PERM_SEVERITY["hidden"] > _FIELD_PERM_SEVERITY["readonly"]
        assert _FIELD_PERM_SEVERITY["readonly"] > _FIELD_PERM_SEVERITY["read"]


class TestInvalidateAllUserPermissions:
    """Tests for invalidate_all_user_permissions — SCAN-based, no KEYS."""

    @pytest.mark.asyncio
    async def test_scan_deletes_all_matching_keys(self):
        from app.core.permissions import invalidate_all_user_permissions

        tenant_id = uuid.uuid4()

        redis_mock = AsyncMock()
        # Simulate SCAN returning keys in two batches then finishing
        call_count = 0

        async def fake_scan(cursor, match, count):
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                return (
                    1,  # non-zero cursor = more to scan
                    [
                        f"{CACHE_PREFIX}:user1:{tenant_id}",
                        f"{CACHE_PREFIX}:user2:{tenant_id}",
                    ],
                )
            else:
                return (
                    0,  # done
                    [f"{CACHE_PREFIX}:user3:{tenant_id}"],
                )

        redis_mock.scan = fake_scan
        redis_mock.delete = AsyncMock()

        await invalidate_all_user_permissions(redis_mock, tenant_id)

        # delete should be called twice — once per batch
        assert redis_mock.delete.call_count == 2
        # First batch: 2 keys
        first_call_args = redis_mock.delete.call_args_list[0]
        assert len(first_call_args[0]) == 2
        # Second batch: 1 key
        second_call_args = redis_mock.delete.call_args_list[1]
        assert len(second_call_args[0]) == 1

    @pytest.mark.asyncio
    async def test_scan_no_keys_no_delete(self):
        from app.core.permissions import invalidate_all_user_permissions

        tenant_id = uuid.uuid4()
        redis_mock = AsyncMock()

        async def fake_scan(cursor, match, count):
            return (0, [])  # no keys found

        redis_mock.scan = fake_scan
        redis_mock.delete = AsyncMock()

        await invalidate_all_user_permissions(redis_mock, tenant_id)

        # delete should not be called when no keys found
        redis_mock.delete.assert_not_called()

    @pytest.mark.asyncio
    async def test_scan_does_not_use_keys_command(self):
        """Ensure invalidate_all_user_permissions uses SCAN, not KEYS."""
        from app.core.permissions import invalidate_all_user_permissions

        tenant_id = uuid.uuid4()
        redis_mock = AsyncMock()

        async def fake_scan(cursor, match, count):
            return (0, [])

        redis_mock.scan = fake_scan
        redis_mock.keys = AsyncMock()

        await invalidate_all_user_permissions(redis_mock, tenant_id)

        # keys() must never be called
        redis_mock.keys.assert_not_called()


class TestGetCachedPermissionsVersionCheck:
    """Tests for get_cached_permissions version validation."""

    @pytest.mark.asyncio
    async def test_cache_hit_version_match_returns_cached(self):
        from app.core.permissions import get_cached_permissions

        user_id = uuid.uuid4()
        tenant_id = uuid.uuid4()
        cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}"

        cached_data = {
            "permissions": ["contacts:read"],
            "denied": [],
            "field_permissions": {},
            "is_system_admin": False,
            "version": 5,
        }

        redis_mock = AsyncMock()
        redis_mock.get = AsyncMock(return_value=json.dumps(cached_data))

        db_mock = AsyncMock()

        # Mock _get_current_permission_version to return matching version
        with patch(
            "app.core.permissions._get_current_permission_version",
            new_callable=AsyncMock,
            return_value=5,
        ):
            result = await get_cached_permissions(db_mock, redis_mock, user_id, tenant_id)

        assert result == cached_data
        redis_mock.setex.assert_not_called()  # no re-caching needed

    @pytest.mark.asyncio
    async def test_cache_version_mismatch_re_resolves(self):
        from app.core.permissions import get_cached_permissions

        user_id = uuid.uuid4()
        tenant_id = uuid.uuid4()
        cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}"

        cached_data = {
            "permissions": ["contacts:read"],
            "denied": [],
            "field_permissions": {},
            "is_system_admin": False,
            "version": 3,  # stale version
        }

        redis_mock = AsyncMock()
        redis_mock.get = AsyncMock(return_value=json.dumps(cached_data))
        redis_mock.delete = AsyncMock()
        redis_mock.setex = AsyncMock()

        db_mock = AsyncMock()

        resolved = {
            "permissions": {"contacts:read", "contacts:write"},
            "denied": set(),
            "field_permissions": {},
            "is_system_admin": False,
            "version": 5,  # new version
        }

        with patch(
            "app.core.permissions._get_current_permission_version",
            new_callable=AsyncMock,
            return_value=5,  # current version differs from cached
        ), patch(
            "app.core.permissions.resolve_permissions",
            new_callable=AsyncMock,
            return_value=resolved,
        ):
            result = await get_cached_permissions(db_mock, redis_mock, user_id, tenant_id)

        # Stale cache should be deleted
        redis_mock.delete.assert_called_once_with(cache_key)
        # New data should be cached
        redis_mock.setex.assert_called_once()
        # Result should have updated permissions
        assert set(result["permissions"]) == {"contacts:read", "contacts:write"}
        assert result["version"] == 5

    @pytest.mark.asyncio
    async def test_cache_miss_resolves_from_db(self):
        from app.core.permissions import get_cached_permissions

        user_id = uuid.uuid4()
        tenant_id = uuid.uuid4()

        redis_mock = AsyncMock()
        redis_mock.get = AsyncMock(return_value=None)  # cache miss
        redis_mock.setex = AsyncMock()

        db_mock = AsyncMock()

        resolved = {
            "permissions": {"contacts:read"},
            "denied": set(),
            "field_permissions": {},
            "is_system_admin": False,
            "version": 1,
        }

        with patch(
            "app.core.permissions.resolve_permissions",
            new_callable=AsyncMock,
            return_value=resolved,
        ):
            result = await get_cached_permissions(db_mock, redis_mock, user_id, tenant_id)

        assert set(result["permissions"]) == {"contacts:read"}
        redis_mock.setex.assert_called_once()


class TestRequireWriteNoWildcard:
    """Tests that require_write does not use broad *:write wildcard."""

    def test_write_permissions_list_has_no_wildcard(self):
        from app.deps import _WRITE_PERMISSIONS

        for perm in _WRITE_PERMISSIONS:
            # No broad wildcards like *:write or *:create
            assert not perm.startswith("*:"), f"Found wildcard permission: {perm}"
            # All permissions should be module:action format
            assert ":" in perm, f"Invalid permission format: {perm}"

    def test_write_permissions_list_includes_contacts_write(self):
        from app.deps import _WRITE_PERMISSIONS

        assert "contacts:write" in _WRITE_PERMISSIONS


class TestCheckPermissionDenyList:
    """Verify deny list still works correctly."""

    def test_deny_overrides_allowed(self):
        resolved = {
            "permissions": {"contacts:read", "contacts:write"},
            "denied": {"contacts:write"},
            "is_system_admin": False,
        }
        assert check_permission(resolved, "contacts:read") is True
        assert check_permission(resolved, "contacts:write") is False

    def test_deny_wildcard_blocks_specific(self):
        resolved = {
            "permissions": {"contacts:read"},
            "denied": {"contacts:*"},
            "is_system_admin": False,
        }
        assert check_permission(resolved, "contacts:read") is False
        assert check_permission(resolved, "contacts:write") is False

    def test_system_admin_ignores_deny(self):
        resolved = {
            "permissions": set(),
            "denied": {"contacts:*"},
            "is_system_admin": True,
        }
        assert check_permission(resolved, "contacts:read") is True