diff --git a/src/main.py b/src/main.py index c8d8751..6cfb8b0 100644 --- a/src/main.py +++ b/src/main.py @@ -51,7 +51,7 @@ app.add_middleware( from src.routers import ( wiki, tools, graph, vector, hybrid_rag, consolidation, ingestion, entity_linking, webhooks, rag_search, content, - maintenance + maintenance, volatile ) app.include_router(wiki.router) @@ -66,6 +66,7 @@ app.include_router(webhooks.router) app.include_router(rag_search.router) app.include_router(content.router) app.include_router(maintenance.router) +app.include_router(volatile.router) # Mount static files directory for Wiki.js integration scripts static_dir = Path(__file__).parent.parent / "static" diff --git a/src/models/volatile.py b/src/models/volatile.py new file mode 100644 index 0000000..0ecdf43 --- /dev/null +++ b/src/models/volatile.py @@ -0,0 +1,130 @@ +""" +Volatile memory models for Library Desk. + +Provides models for ephemeral cached data with TTL - weather, news, financial data, +transit schedules, and other time-sensitive external information. +""" + +from pydantic import BaseModel, Field +from typing import Dict, Any, Optional, List +from datetime import datetime +from enum import Enum + + +class VolatileNamespace(str, Enum): + """ + Predefined namespaces for volatile data. + + Each namespace can have different default TTLs and refresh schedules. + """ + # Real-time external data + WEATHER = "weather" # Current conditions, forecasts + NEWS = "news" # Headlines, breaking news + FINANCIAL = "financial" # Stock prices, exchange rates, crypto + TRANSIT = "transit" # Train/bus schedules, delays, disruptions + TRAFFIC = "traffic" # Commute times, road conditions + AIR_QUALITY = "air_quality" # Pollution levels, pollen counts + SPORTS = "sports" # Live scores, upcoming matches + + # System/integration data + SOCIAL = "social" # Social media mentions, notifications + SYSTEM = "system" # Service health, infrastructure status + + # Ephemeral context + CONTEXT = "context" # Conversation context, session state + CUSTOM = "custom" # User-defined volatile data + + +# Default TTLs per namespace (in seconds) +NAMESPACE_DEFAULT_TTL: Dict[str, int] = { + VolatileNamespace.WEATHER: 1800, # 30 min - weather changes slowly + VolatileNamespace.NEWS: 3600, # 1 hour - news cycles + VolatileNamespace.FINANCIAL: 300, # 5 min - markets move fast + VolatileNamespace.TRANSIT: 300, # 5 min - schedules update frequently + VolatileNamespace.TRAFFIC: 600, # 10 min - traffic patterns + VolatileNamespace.AIR_QUALITY: 3600, # 1 hour - air quality stable + VolatileNamespace.SPORTS: 60, # 1 min - live scores + VolatileNamespace.SOCIAL: 600, # 10 min - social notifications + VolatileNamespace.SYSTEM: 60, # 1 min - system health + VolatileNamespace.CONTEXT: 3600, # 1 hour - session context + VolatileNamespace.CUSTOM: 3600, # 1 hour - default for custom +} + + +class VolatileRecord(BaseModel): + """ + A volatile cache record with TTL. + + Volatile records are ephemeral data stored in Redis with automatic expiration. + Used for weather, news, financial data, and other time-sensitive information. + """ + key: str = Field(..., description="Record key (e.g., 'rotterdam', 'nos-headlines')") + namespace: str = Field(..., description="Namespace (e.g., 'weather', 'news', 'financial')") + data: Dict[str, Any] = Field(..., description="Actual content/payload") + source: Optional[str] = Field(None, description="Origin API/service (e.g., 'openweathermap', 'nos.nl')") + created_at: datetime = Field(default_factory=datetime.utcnow, description="When record was created") + updated_at: datetime = Field(default_factory=datetime.utcnow, description="When record was last updated") + ttl: int = Field(..., ge=60, le=604800, description="Time-to-live in seconds (max 7 days)") + refresh_schedule: Optional[str] = Field(None, description="Cron expression for scheduled refresh") + user: str = Field(..., description="User identifier for multi-tenancy") + + +class VolatileRecordCreate(BaseModel): + """Request model for creating/updating a volatile record.""" + data: Dict[str, Any] = Field(..., description="Content to store") + source: Optional[str] = Field(None, description="Origin API/service") + ttl: Optional[int] = Field(None, ge=60, le=604800, description="TTL in seconds (uses namespace default if not set)") + refresh_schedule: Optional[str] = Field(None, description="Cron expression for scheduled refresh") + + +class VolatileRecordResponse(BaseModel): + """Response model for a volatile record.""" + key: str = Field(..., description="Record key") + namespace: str = Field(..., description="Namespace") + data: Dict[str, Any] = Field(..., description="Stored content") + source: Optional[str] = Field(None, description="Origin API/service") + created_at: datetime = Field(..., description="Creation timestamp") + updated_at: datetime = Field(..., description="Last update timestamp") + ttl: int = Field(..., description="TTL in seconds") + ttl_remaining: int = Field(..., description="Seconds until expiration") + refresh_schedule: Optional[str] = Field(None, description="Cron expression if scheduled") + user: str = Field(..., description="User identifier") + + +class VolatileListResponse(BaseModel): + """Response model for listing volatile records.""" + namespace: str = Field(..., description="Namespace queried") + keys: List[str] = Field(..., description="List of keys in namespace") + count: int = Field(..., description="Number of keys") + user: str = Field(..., description="User identifier") + + +class VolatileScheduledResponse(BaseModel): + """Response model for records needing refresh.""" + records: List[VolatileRecordResponse] = Field(..., description="Records with refresh schedules") + count: int = Field(..., description="Number of scheduled records") + user: str = Field(..., description="User identifier") + + +class VolatileStatsResponse(BaseModel): + """Response model for volatile cache statistics.""" + total_records: int = Field(..., description="Total volatile records for user") + by_namespace: Dict[str, int] = Field(..., description="Record count per namespace") + scheduled_count: int = Field(..., description="Records with refresh schedules") + total_memory_bytes: Optional[int] = Field(None, description="Approximate memory usage") + user: str = Field(..., description="User identifier") + + +class VolatileDeleteResponse(BaseModel): + """Response model for delete operation.""" + key: str = Field(..., description="Deleted key") + namespace: str = Field(..., description="Namespace") + deleted: bool = Field(..., description="Whether record was found and deleted") + user: str = Field(..., description="User identifier") + + +class VolatileBulkDeleteResponse(BaseModel): + """Response model for bulk delete operations.""" + namespace: Optional[str] = Field(None, description="Namespace if namespace-wide delete") + deleted_count: int = Field(..., description="Number of records deleted") + user: str = Field(..., description="User identifier") diff --git a/src/routers/volatile.py b/src/routers/volatile.py new file mode 100644 index 0000000..eb3845c --- /dev/null +++ b/src/routers/volatile.py @@ -0,0 +1,271 @@ +""" +Volatile cache router for Library Desk API. + +Endpoints for ephemeral cached data with TTL - weather, news, financial, etc. +""" + +from fastapi import APIRouter, HTTPException, Depends, Query +import logging + +from src.models.volatile import ( + VolatileRecordCreate, + VolatileRecordResponse, + VolatileListResponse, + VolatileScheduledResponse, + VolatileStatsResponse, + VolatileDeleteResponse, + VolatileBulkDeleteResponse, + VolatileNamespace, +) +from src.services.volatile_service import VolatileCacheService +from src.core.dependencies import verify_api_key, RedisDep +from src.core.multi_tenancy import DEFAULT_USER +from src.config import get_settings + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/volatile", tags=["Volatile Cache"]) + + +def get_volatile_service(redis: RedisDep) -> VolatileCacheService: + """Get volatile cache service instance.""" + settings = get_settings() + return VolatileCacheService(redis_client=redis, settings=settings) + + +@router.get("/stats", response_model=VolatileStatsResponse) +async def get_stats( + user: str = Query(default=DEFAULT_USER, description="User identifier"), + redis: RedisDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Get volatile cache statistics. + + Returns counts of records by namespace and scheduled refresh info. + """ + service = get_volatile_service(redis) + stats = await service.get_stats(user) + + return VolatileStatsResponse( + total_records=stats["total_records"], + by_namespace=stats["by_namespace"], + scheduled_count=stats["scheduled_count"], + total_memory_bytes=stats.get("total_memory_bytes"), + user=user, + ) + + +@router.get("/scheduled", response_model=VolatileScheduledResponse) +async def get_scheduled( + user: str = Query(default=DEFAULT_USER, description="User identifier"), + redis: RedisDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Get records with refresh schedules. + + Used by scheduler to determine what volatile data needs refreshing. + Returns all records that have a refresh_schedule cron expression set. + """ + service = get_volatile_service(redis) + records = await service.get_scheduled(user) + + return VolatileScheduledResponse( + records=records, + count=len(records), + user=user, + ) + + +@router.get("/namespaces") +async def list_namespaces( + api_key: str = Depends(verify_api_key) +): + """ + List available namespaces and their default TTLs. + + Returns predefined namespaces with their default TTL values. + Custom namespaces can also be used with the default TTL. + """ + from src.models.volatile import NAMESPACE_DEFAULT_TTL + + return { + "namespaces": [ + { + "name": ns.value, + "default_ttl": NAMESPACE_DEFAULT_TTL.get(ns, 3600), + "description": _get_namespace_description(ns), + } + for ns in VolatileNamespace + ] + } + + +def _get_namespace_description(ns: VolatileNamespace) -> str: + """Get human-readable description for namespace.""" + descriptions = { + VolatileNamespace.WEATHER: "Weather conditions and forecasts", + VolatileNamespace.NEWS: "Headlines and breaking news", + VolatileNamespace.FINANCIAL: "Stock prices, exchange rates, crypto", + VolatileNamespace.TRANSIT: "Train/bus schedules, delays", + VolatileNamespace.TRAFFIC: "Commute times, road conditions", + VolatileNamespace.AIR_QUALITY: "Pollution levels, pollen counts", + VolatileNamespace.SPORTS: "Live scores, upcoming matches", + VolatileNamespace.SOCIAL: "Social media mentions, notifications", + VolatileNamespace.SYSTEM: "Service health, infrastructure status", + VolatileNamespace.CONTEXT: "Conversation context, session state", + VolatileNamespace.CUSTOM: "User-defined volatile data", + } + return descriptions.get(ns, "Custom namespace") + + +@router.get("/{namespace}", response_model=VolatileListResponse) +async def list_keys( + namespace: str, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + redis: RedisDep = None, + api_key: str = Depends(verify_api_key) +): + """ + List all keys in a namespace. + + Returns the list of keys stored in the specified namespace. + """ + service = get_volatile_service(redis) + keys = await service.list_namespace(user, namespace) + + return VolatileListResponse( + namespace=namespace, + keys=keys, + count=len(keys), + user=user, + ) + + +@router.delete("/{namespace}", response_model=VolatileBulkDeleteResponse) +async def delete_namespace( + namespace: str, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + redis: RedisDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Delete all records in a namespace. + + Removes all volatile data for the specified namespace. + """ + service = get_volatile_service(redis) + deleted = await service.delete_namespace(user, namespace) + + return VolatileBulkDeleteResponse( + namespace=namespace, + deleted_count=deleted, + user=user, + ) + + +@router.get("/{namespace}/{key}", response_model=VolatileRecordResponse) +async def get_record( + namespace: str, + key: str, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + redis: RedisDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Get a volatile record. + + Returns the record if it exists and has not expired. + + **Example:** + ``` + GET /volatile/weather/rotterdam?user=jpmschweitzer + ``` + """ + service = get_volatile_service(redis) + record = await service.get(user, namespace, key) + + if not record: + raise HTTPException( + status_code=404, + detail=f"Record '{key}' not found in namespace '{namespace}'" + ) + + return record + + +@router.post("/{namespace}/{key}", response_model=VolatileRecordResponse) +async def set_record( + namespace: str, + key: str, + request: VolatileRecordCreate, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + redis: RedisDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Store or update a volatile record. + + Creates or updates a record with the specified TTL. + If TTL is not provided, the namespace default is used. + + **Example Request:** + ```json + { + "data": { + "temperature": 18, + "conditions": "Partly cloudy", + "humidity": 65 + }, + "source": "openweathermap", + "ttl": 1800, + "refresh_schedule": "0 * * * *" + } + ``` + + **Refresh Schedule:** + Optional cron expression for automatic refresh. The scheduler + will query `/volatile/scheduled` and trigger refreshes. + """ + service = get_volatile_service(redis) + + try: + record = await service.set( + user=user, + namespace=namespace, + key=key, + data=request.data, + source=request.source, + ttl=request.ttl, + refresh_schedule=request.refresh_schedule, + ) + return record + + except Exception as e: + logger.error(f"Failed to store volatile record: {e}") + raise HTTPException(status_code=500, detail="Failed to store record") + + +@router.delete("/{namespace}/{key}", response_model=VolatileDeleteResponse) +async def delete_record( + namespace: str, + key: str, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + redis: RedisDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Delete a volatile record. + + Removes the record from the cache. + """ + service = get_volatile_service(redis) + deleted = await service.delete(user, namespace, key) + + return VolatileDeleteResponse( + key=key, + namespace=namespace, + deleted=deleted, + user=user, + ) diff --git a/src/services/volatile_service.py b/src/services/volatile_service.py new file mode 100644 index 0000000..9391bd7 --- /dev/null +++ b/src/services/volatile_service.py @@ -0,0 +1,433 @@ +""" +Volatile Cache service for Library Desk. + +Provides ephemeral data storage with TTL for time-sensitive information: +- Weather, news, financial data +- Transit schedules, traffic conditions +- System status, social notifications +""" + +import json +import logging +import hashlib +from datetime import datetime +from typing import List, Optional, Dict, Any + +import redis.asyncio as aioredis + +from src.config import Settings +from src.models.volatile import ( + VolatileRecord, + VolatileRecordResponse, + VolatileNamespace, + NAMESPACE_DEFAULT_TTL, +) + +logger = logging.getLogger(__name__) + + +class VolatileCacheService: + """ + Service for volatile data with TTL. + + Stores ephemeral data in Redis with automatic expiration. + Supports multiple namespaces with configurable TTLs. + """ + + # Redis key prefix for volatile data + KEY_PREFIX = "volatile" + + def __init__( + self, + redis_client: aioredis.Redis, + settings: Settings + ): + """ + Initialize volatile cache service. + + Args: + redis_client: Async Redis client + settings: Application settings + """ + self.redis = redis_client + self.settings = settings + + logger.info("Initialized VolatileCacheService") + + def _build_key(self, user: str, namespace: str, key: str) -> str: + """ + Build Redis key for volatile record. + + Pattern: {user}:volatile:{namespace}:{key_hash} + Uses hash to ensure safe key characters and consistent length. + """ + key_hash = hashlib.md5(key.encode()).hexdigest()[:12] + return f"{user}:{self.KEY_PREFIX}:{namespace}:{key_hash}" + + def _build_pattern(self, user: str, namespace: Optional[str] = None) -> str: + """Build pattern for key scanning.""" + if namespace: + return f"{user}:{self.KEY_PREFIX}:{namespace}:*" + return f"{user}:{self.KEY_PREFIX}:*" + + def _get_default_ttl(self, namespace: str) -> int: + """Get default TTL for a namespace.""" + try: + ns = VolatileNamespace(namespace) + return NAMESPACE_DEFAULT_TTL.get(ns, self.settings.volatile_default_ttl) + except ValueError: + return self.settings.volatile_default_ttl + + def _serialize_record(self, record: VolatileRecord) -> str: + """Serialize record to JSON for storage.""" + return json.dumps({ + "key": record.key, + "namespace": record.namespace, + "data": record.data, + "source": record.source, + "created_at": record.created_at.isoformat(), + "updated_at": record.updated_at.isoformat(), + "ttl": record.ttl, + "refresh_schedule": record.refresh_schedule, + "user": record.user, + }) + + def _deserialize_record(self, data: str) -> VolatileRecord: + """Deserialize record from JSON.""" + obj = json.loads(data) + return VolatileRecord( + key=obj["key"], + namespace=obj["namespace"], + data=obj["data"], + source=obj.get("source"), + created_at=datetime.fromisoformat(obj["created_at"]), + updated_at=datetime.fromisoformat(obj["updated_at"]), + ttl=obj["ttl"], + refresh_schedule=obj.get("refresh_schedule"), + user=obj["user"], + ) + + async def get( + self, + user: str, + namespace: str, + key: str + ) -> Optional[VolatileRecordResponse]: + """ + Get a volatile record. + + Args: + user: User identifier + namespace: Data namespace + key: Record key + + Returns: + Record if found and not expired, None otherwise + """ + redis_key = self._build_key(user, namespace, key) + + try: + data = await self.redis.get(redis_key) + if not data: + return None + + record = self._deserialize_record(data) + + # Get TTL remaining + ttl_remaining = await self.redis.ttl(redis_key) + if ttl_remaining < 0: + return None + + return VolatileRecordResponse( + key=record.key, + namespace=record.namespace, + data=record.data, + source=record.source, + created_at=record.created_at, + updated_at=record.updated_at, + ttl=record.ttl, + ttl_remaining=max(0, ttl_remaining), + refresh_schedule=record.refresh_schedule, + user=record.user, + ) + + except Exception as e: + logger.error(f"Failed to get volatile record {redis_key}: {e}") + return None + + async def set( + self, + user: str, + namespace: str, + key: str, + data: Dict[str, Any], + source: Optional[str] = None, + ttl: Optional[int] = None, + refresh_schedule: Optional[str] = None + ) -> VolatileRecordResponse: + """ + Store or update a volatile record. + + Args: + user: User identifier + namespace: Data namespace + key: Record key + data: Content to store + source: Origin API/service + ttl: TTL in seconds (uses namespace default if not set) + refresh_schedule: Optional cron expression for refresh + + Returns: + The stored record + """ + redis_key = self._build_key(user, namespace, key) + + # Use provided TTL or namespace default + effective_ttl = ttl if ttl is not None else self._get_default_ttl(namespace) + + # Check if record exists (for created_at) + existing = await self.get(user, namespace, key) + now = datetime.utcnow() + + record = VolatileRecord( + key=key, + namespace=namespace, + data=data, + source=source, + created_at=existing.created_at if existing else now, + updated_at=now, + ttl=effective_ttl, + refresh_schedule=refresh_schedule, + user=user, + ) + + try: + serialized = self._serialize_record(record) + await self.redis.setex(redis_key, effective_ttl, serialized) + + logger.debug(f"Stored volatile record {redis_key} with TTL {effective_ttl}s") + + return VolatileRecordResponse( + key=record.key, + namespace=record.namespace, + data=record.data, + source=record.source, + created_at=record.created_at, + updated_at=record.updated_at, + ttl=record.ttl, + ttl_remaining=effective_ttl, + refresh_schedule=record.refresh_schedule, + user=record.user, + ) + + except Exception as e: + logger.error(f"Failed to store volatile record {redis_key}: {e}") + raise + + async def delete( + self, + user: str, + namespace: str, + key: str + ) -> bool: + """ + Delete a volatile record. + + Args: + user: User identifier + namespace: Data namespace + key: Record key + + Returns: + True if record was deleted, False if not found + """ + redis_key = self._build_key(user, namespace, key) + + try: + deleted = await self.redis.delete(redis_key) + if deleted: + logger.debug(f"Deleted volatile record {redis_key}") + return deleted > 0 + + except Exception as e: + logger.error(f"Failed to delete volatile record {redis_key}: {e}") + return False + + async def list_namespace( + self, + user: str, + namespace: str + ) -> List[str]: + """ + List all keys in a namespace. + + Args: + user: User identifier + namespace: Data namespace + + Returns: + List of keys (original keys, not Redis keys) + """ + pattern = self._build_pattern(user, namespace) + + try: + keys = [] + async for redis_key in self.redis.scan_iter(match=pattern): + # Get the record to retrieve original key + data = await self.redis.get(redis_key) + if data: + record = self._deserialize_record(data) + keys.append(record.key) + + return keys + + except Exception as e: + logger.error(f"Failed to list namespace {namespace}: {e}") + return [] + + async def get_scheduled( + self, + user: str + ) -> List[VolatileRecordResponse]: + """ + Get all records with refresh schedules. + + Used by scheduler to determine what needs refreshing. + + Args: + user: User identifier + + Returns: + List of records with refresh_schedule set + """ + pattern = self._build_pattern(user) + + try: + scheduled = [] + async for redis_key in self.redis.scan_iter(match=pattern): + data = await self.redis.get(redis_key) + if data: + record = self._deserialize_record(data) + if record.refresh_schedule: + ttl_remaining = await self.redis.ttl(redis_key) + scheduled.append(VolatileRecordResponse( + key=record.key, + namespace=record.namespace, + data=record.data, + source=record.source, + created_at=record.created_at, + updated_at=record.updated_at, + ttl=record.ttl, + ttl_remaining=max(0, ttl_remaining), + refresh_schedule=record.refresh_schedule, + user=record.user, + )) + + return scheduled + + except Exception as e: + logger.error(f"Failed to get scheduled records: {e}") + return [] + + async def get_stats( + self, + user: str + ) -> Dict[str, Any]: + """ + Get cache statistics for user. + + Args: + user: User identifier + + Returns: + Statistics dict + """ + pattern = self._build_pattern(user) + + try: + by_namespace: Dict[str, int] = {} + total = 0 + scheduled = 0 + + async for redis_key in self.redis.scan_iter(match=pattern): + data = await self.redis.get(redis_key) + if data: + record = self._deserialize_record(data) + total += 1 + by_namespace[record.namespace] = by_namespace.get(record.namespace, 0) + 1 + if record.refresh_schedule: + scheduled += 1 + + return { + "total_records": total, + "by_namespace": by_namespace, + "scheduled_count": scheduled, + "total_memory_bytes": None, # Could implement with DEBUG MEMORY + } + + except Exception as e: + logger.error(f"Failed to get stats: {e}") + return { + "total_records": 0, + "by_namespace": {}, + "scheduled_count": 0, + "total_memory_bytes": None, + } + + async def delete_namespace( + self, + user: str, + namespace: str + ) -> int: + """ + Delete all records in a namespace. + + Args: + user: User identifier + namespace: Data namespace + + Returns: + Number of records deleted + """ + pattern = self._build_pattern(user, namespace) + + try: + deleted = 0 + async for redis_key in self.redis.scan_iter(match=pattern): + await self.redis.delete(redis_key) + deleted += 1 + + logger.info(f"Deleted {deleted} records from namespace {namespace}") + return deleted + + except Exception as e: + logger.error(f"Failed to delete namespace {namespace}: {e}") + return 0 + + async def delete_all( + self, + user: str + ) -> int: + """ + Delete all volatile records for user. + + Args: + user: User identifier + + Returns: + Number of records deleted + """ + pattern = self._build_pattern(user) + + try: + deleted = 0 + async for redis_key in self.redis.scan_iter(match=pattern): + await self.redis.delete(redis_key) + deleted += 1 + + logger.info(f"Deleted all {deleted} volatile records for user {user}") + return deleted + + except Exception as e: + logger.error(f"Failed to delete all records: {e}") + return 0 diff --git a/tests/test_volatile.py b/tests/test_volatile.py new file mode 100644 index 0000000..2e64562 --- /dev/null +++ b/tests/test_volatile.py @@ -0,0 +1,287 @@ +""" +Tests for volatile cache router and service. + +Tests: +- Volatile record CRUD operations +- Namespace listing and management +- Scheduled record retrieval +- TTL behavior +""" + +import pytest +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock, patch + +from src.models.volatile import ( + VolatileRecord, + VolatileRecordCreate, + VolatileRecordResponse, + VolatileListResponse, + VolatileScheduledResponse, + VolatileStatsResponse, + VolatileDeleteResponse, + VolatileBulkDeleteResponse, + VolatileNamespace, + NAMESPACE_DEFAULT_TTL, +) + + +class TestVolatileModels: + """Test volatile data models.""" + + def test_volatile_record_creation(self): + """Test VolatileRecord model creation.""" + record = VolatileRecord( + key="rotterdam", + namespace="weather", + data={"temperature": 18, "conditions": "Cloudy"}, + source="openweathermap", + ttl=1800, + user="jpmschweitzer", + ) + assert record.key == "rotterdam" + assert record.namespace == "weather" + assert record.data["temperature"] == 18 + assert record.ttl == 1800 + assert record.refresh_schedule is None + + def test_volatile_record_with_schedule(self): + """Test VolatileRecord with refresh schedule.""" + record = VolatileRecord( + key="nos-headlines", + namespace="news", + data={"headlines": ["Test headline"]}, + source="nos.nl", + ttl=3600, + refresh_schedule="0 * * * *", + user="jpmschweitzer", + ) + assert record.refresh_schedule == "0 * * * *" + + def test_volatile_record_create(self): + """Test VolatileRecordCreate model.""" + create = VolatileRecordCreate( + data={"price": 150.50, "change": 2.3}, + source="alpha_vantage", + ttl=300, + ) + assert create.data["price"] == 150.50 + assert create.ttl == 300 + + def test_volatile_record_response(self): + """Test VolatileRecordResponse model.""" + response = VolatileRecordResponse( + key="rotterdam", + namespace="weather", + data={"temperature": 18}, + source="openweathermap", + created_at=datetime.utcnow(), + updated_at=datetime.utcnow(), + ttl=1800, + ttl_remaining=1500, + user="jpmschweitzer", + ) + assert response.ttl_remaining == 1500 + assert response.ttl == 1800 + + +class TestVolatileNamespaces: + """Test volatile namespaces and defaults.""" + + def test_all_namespaces_have_default_ttl(self): + """Verify all namespaces have default TTLs defined.""" + for ns in VolatileNamespace: + assert ns in NAMESPACE_DEFAULT_TTL, f"Missing TTL for {ns}" + assert NAMESPACE_DEFAULT_TTL[ns] > 0 + + def test_weather_default_ttl(self): + """Test weather namespace default TTL.""" + assert NAMESPACE_DEFAULT_TTL[VolatileNamespace.WEATHER] == 1800 # 30 min + + def test_financial_default_ttl(self): + """Test financial namespace default TTL.""" + assert NAMESPACE_DEFAULT_TTL[VolatileNamespace.FINANCIAL] == 300 # 5 min + + def test_sports_default_ttl(self): + """Test sports namespace default TTL (fast updates).""" + assert NAMESPACE_DEFAULT_TTL[VolatileNamespace.SPORTS] == 60 # 1 min + + +class TestVolatileListResponse: + """Test list response models.""" + + def test_list_response(self): + """Test VolatileListResponse model.""" + response = VolatileListResponse( + namespace="weather", + keys=["rotterdam", "amsterdam", "utrecht"], + count=3, + user="jpmschweitzer", + ) + assert response.count == 3 + assert "rotterdam" in response.keys + + +class TestVolatileScheduledResponse: + """Test scheduled records response.""" + + def test_scheduled_response_empty(self): + """Test empty scheduled response.""" + response = VolatileScheduledResponse( + records=[], + count=0, + user="jpmschweitzer", + ) + assert response.count == 0 + assert response.records == [] + + def test_scheduled_response_with_records(self): + """Test scheduled response with records.""" + record = VolatileRecordResponse( + key="nos-headlines", + namespace="news", + data={"headlines": []}, + source="nos.nl", + created_at=datetime.utcnow(), + updated_at=datetime.utcnow(), + ttl=3600, + ttl_remaining=3000, + refresh_schedule="0 */6 * * *", + user="jpmschweitzer", + ) + response = VolatileScheduledResponse( + records=[record], + count=1, + user="jpmschweitzer", + ) + assert response.count == 1 + assert response.records[0].refresh_schedule == "0 */6 * * *" + + +class TestVolatileStatsResponse: + """Test stats response model.""" + + def test_stats_response(self): + """Test VolatileStatsResponse model.""" + response = VolatileStatsResponse( + total_records=15, + by_namespace={"weather": 3, "news": 5, "financial": 7}, + scheduled_count=2, + total_memory_bytes=None, + user="jpmschweitzer", + ) + assert response.total_records == 15 + assert response.by_namespace["weather"] == 3 + assert response.scheduled_count == 2 + + +class TestVolatileDeleteResponses: + """Test delete response models.""" + + def test_delete_response(self): + """Test VolatileDeleteResponse model.""" + response = VolatileDeleteResponse( + key="rotterdam", + namespace="weather", + deleted=True, + user="jpmschweitzer", + ) + assert response.deleted is True + + def test_delete_not_found(self): + """Test delete response when record not found.""" + response = VolatileDeleteResponse( + key="nonexistent", + namespace="weather", + deleted=False, + user="jpmschweitzer", + ) + assert response.deleted is False + + def test_bulk_delete_response(self): + """Test VolatileBulkDeleteResponse model.""" + response = VolatileBulkDeleteResponse( + namespace="weather", + deleted_count=5, + user="jpmschweitzer", + ) + assert response.deleted_count == 5 + assert response.namespace == "weather" + + +class TestVolatileService: + """Test VolatileCacheService functionality.""" + + @pytest.fixture + def mock_redis(self): + """Create mock Redis client.""" + redis = AsyncMock() + redis.get = AsyncMock(return_value=None) + redis.setex = AsyncMock() + redis.delete = AsyncMock(return_value=1) + redis.ttl = AsyncMock(return_value=1500) + redis.scan_iter = MagicMock(return_value=iter([])) + return redis + + @pytest.fixture + def mock_settings(self): + """Create mock settings.""" + settings = MagicMock() + settings.volatile_default_ttl = 3600 + return settings + + @pytest.fixture + def volatile_service(self, mock_redis, mock_settings): + """Create VolatileCacheService with mocks.""" + from src.services.volatile_service import VolatileCacheService + return VolatileCacheService( + redis_client=mock_redis, + settings=mock_settings + ) + + def test_build_key(self, volatile_service): + """Test Redis key building.""" + key = volatile_service._build_key("jpmschweitzer", "weather", "rotterdam") + assert key.startswith("jpmschweitzer:volatile:weather:") + assert len(key) > 30 # Has hash suffix + + def test_build_pattern(self, volatile_service): + """Test pattern building.""" + pattern = volatile_service._build_pattern("jpmschweitzer", "weather") + assert pattern == "jpmschweitzer:volatile:weather:*" + + def test_build_pattern_all(self, volatile_service): + """Test pattern building for all namespaces.""" + pattern = volatile_service._build_pattern("jpmschweitzer") + assert pattern == "jpmschweitzer:volatile:*" + + def test_get_default_ttl_known_namespace(self, volatile_service): + """Test default TTL for known namespace.""" + ttl = volatile_service._get_default_ttl("weather") + assert ttl == 1800 # Weather namespace default + + def test_get_default_ttl_unknown_namespace(self, volatile_service): + """Test default TTL for unknown namespace.""" + ttl = volatile_service._get_default_ttl("unknown_namespace") + assert ttl == 3600 # Falls back to settings default + + @pytest.mark.asyncio + async def test_get_not_found(self, volatile_service, mock_redis): + """Test get when record not found.""" + mock_redis.get.return_value = None + result = await volatile_service.get("jpmschweitzer", "weather", "rotterdam") + assert result is None + + @pytest.mark.asyncio + async def test_delete_success(self, volatile_service, mock_redis): + """Test successful delete.""" + mock_redis.delete.return_value = 1 + result = await volatile_service.delete("jpmschweitzer", "weather", "rotterdam") + assert result is True + + @pytest.mark.asyncio + async def test_delete_not_found(self, volatile_service, mock_redis): + """Test delete when record not found.""" + mock_redis.delete.return_value = 0 + result = await volatile_service.delete("jpmschweitzer", "weather", "nonexistent") + assert result is False