MaverickMCP

"""
Mock cache manager implementation for testing.
"""

import time
from typing import Any


class MockCacheManager:
    """
    Mock implementation of ICacheManager for testing.

    This implementation uses in-memory storage and provides predictable
    behavior for testing cache-dependent functionality.
    """

    def __init__(self):
        """Initialize the mock cache manager."""
        self._data: dict[str, dict[str, Any]] = {}
        self._call_log: list[dict[str, Any]] = []

    async def get(self, key: str) -> Any:
        """Get data from mock cache."""
        self._log_call("get", {"key": key})

        if key not in self._data:
            return None

        entry = self._data[key]

        # Check if expired
        if "expires_at" in entry and entry["expires_at"] < time.time():
            del self._data[key]
            return None

        return entry["value"]

    async def set(self, key: str, value: Any, ttl: int | None = None) -> bool:
        """Store data in mock cache."""
        self._log_call("set", {"key": key, "value": value, "ttl": ttl})

        entry = {"value": value}

        if ttl is not None:
            entry["expires_at"] = time.time() + ttl

        self._data[key] = entry
        return True

    async def delete(self, key: str) -> bool:
        """Delete a key from mock cache."""
        self._log_call("delete", {"key": key})

        if key in self._data:
            del self._data[key]
            return True

        return False

    async def exists(self, key: str) -> bool:
        """Check if a key exists in mock cache."""
        self._log_call("exists", {"key": key})

        if key not in self._data:
            return False

        entry = self._data[key]

        # Check if expired
        if "expires_at" in entry and entry["expires_at"] < time.time():
            del self._data[key]
            return False

        return True

    async def clear(self, pattern: str | None = None) -> int:
        """Clear cache entries."""
        self._log_call("clear", {"pattern": pattern})

        if pattern is None:
            count = len(self._data)
            self._data.clear()
            return count

        # Simple pattern matching (only supports prefix*)
        if pattern.endswith("*"):
            prefix = pattern[:-1]
            keys_to_delete = [k for k in self._data.keys() if k.startswith(prefix)]
        else:
            keys_to_delete = [k for k in self._data.keys() if k == pattern]

        for key in keys_to_delete:
            del self._data[key]

        return len(keys_to_delete)

    async def get_many(self, keys: list[str]) -> dict[str, Any]:
        """Get multiple values at once."""
        self._log_call("get_many", {"keys": keys})

        results = {}
        for key in keys:
            value = await self.get(key)
            if value is not None:
                results[key] = value

        return results

    async def set_many(self, items: list[tuple[str, Any, int | None]]) -> int:
        """Set multiple values at once."""
        self._log_call("set_many", {"items_count": len(items)})

        success_count = 0
        for key, value, ttl in items:
            if await self.set(key, value, ttl):
                success_count += 1

        return success_count

    async def delete_many(self, keys: list[str]) -> int:
        """Delete multiple keys."""
        self._log_call("delete_many", {"keys": keys})

        deleted_count = 0
        for key in keys:
            if await self.delete(key):
                deleted_count += 1

        return deleted_count

    async def exists_many(self, keys: list[str]) -> dict[str, bool]:
        """Check existence of multiple keys."""
        self._log_call("exists_many", {"keys": keys})

        results = {}
        for key in keys:
            results[key] = await self.exists(key)

        return results

    async def count_keys(self, pattern: str) -> int:
        """Count keys matching a pattern."""
        self._log_call("count_keys", {"pattern": pattern})

        if pattern.endswith("*"):
            prefix = pattern[:-1]
            return len([k for k in self._data.keys() if k.startswith(prefix)])
        else:
            return 1 if pattern in self._data else 0

    async def get_or_set(
        self, key: str, default_value: Any, ttl: int | None = None
    ) -> Any:
        """Get value from cache, setting it if it doesn't exist."""
        self._log_call(
            "get_or_set", {"key": key, "default_value": default_value, "ttl": ttl}
        )

        value = await self.get(key)
        if value is not None:
            return value

        await self.set(key, default_value, ttl)
        return default_value

    async def increment(self, key: str, amount: int = 1) -> int:
        """Increment a numeric value in cache."""
        self._log_call("increment", {"key": key, "amount": amount})

        current = await self.get(key)

        if current is None:
            new_value = amount
        else:
            try:
                current_int = int(current)
                new_value = current_int + amount
            except (ValueError, TypeError):
                raise ValueError(f"Key {key} contains non-numeric value: {current}")

        await self.set(key, new_value)
        return new_value

    async def set_if_not_exists(
        self, key: str, value: Any, ttl: int | None = None
    ) -> bool:
        """Set a value only if the key doesn't already exist."""
        self._log_call("set_if_not_exists", {"key": key, "value": value, "ttl": ttl})

        if await self.exists(key):
            return False

        return await self.set(key, value, ttl)

    async def get_ttl(self, key: str) -> int | None:
        """Get the remaining time-to-live for a key."""
        self._log_call("get_ttl", {"key": key})

        if key not in self._data:
            return None

        entry = self._data[key]

        if "expires_at" not in entry:
            return None

        remaining = int(entry["expires_at"] - time.time())
        return max(0, remaining)

    async def expire(self, key: str, ttl: int) -> bool:
        """Set expiration time for an existing key."""
        self._log_call("expire", {"key": key, "ttl": ttl})

        if key not in self._data:
            return False

        self._data[key]["expires_at"] = time.time() + ttl
        return True

    # Testing utilities

    def _log_call(self, method: str, args: dict[str, Any]) -> None:
        """Log method calls for testing verification."""
        self._call_log.append(
            {
                "method": method,
                "args": args,
                "timestamp": time.time(),
            }
        )

    def get_call_log(self) -> list[dict[str, Any]]:
        """Get the log of method calls for testing verification."""
        return self._call_log.copy()

    def clear_call_log(self) -> None:
        """Clear the method call log."""
        self._call_log.clear()

    def get_cache_contents(self) -> dict[str, Any]:
        """Get all cache contents for testing verification."""
        return {k: v["value"] for k, v in self._data.items()}

    def set_cache_contents(self, contents: dict[str, Any]) -> None:
        """Set cache contents directly for testing setup."""
        self._data.clear()
        for key, value in contents.items():
            self._data[key] = {"value": value}

    def simulate_cache_expiry(self, key: str) -> None:
        """Simulate cache expiry for testing."""
        if key in self._data:
            self._data[key]["expires_at"] = time.time() - 1