test_cache_service.py•4.32 kB
"""Unit tests for CacheService."""
import sys
import os
import tempfile
import shutil
import time
# Add src to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../src"))
from toggl_mcp.services.cache_service import CacheService
def test_cache_service_set_and_get():
"""Test basic cache set and get."""
with tempfile.TemporaryDirectory() as tmpdir:
cache = CacheService(cache_dir=tmpdir, ttl_hours=24) # Use longer TTL for tests
data = {"users": {"test@example.com": {"matched_entities": []}}}
cache.set("2025-10-06", "2025-10-06", data)
retrieved = cache.get("2025-10-06", "2025-10-06")
assert retrieved == data
def test_cache_service_with_user_id():
"""Test cache with user_id parameter."""
with tempfile.TemporaryDirectory() as tmpdir:
cache = CacheService(cache_dir=tmpdir, ttl_hours=24)
data = {"users": {"test@example.com": {}}}
cache.set("2025-10-06", "2025-10-06", data, user_id="12345")
# Should return None for different key
retrieved_wrong = cache.get("2025-10-06", "2025-10-06")
assert retrieved_wrong is None
# Should return data for same key
retrieved_correct = cache.get("2025-10-06", "2025-10-06", user_id="12345")
assert retrieved_correct == data
def test_cache_service_expiry():
"""Test that cache expires after TTL."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create cache with a very short TTL (1 minute = 1/60 hour)
# We'll manually edit the database to make it expired
cache = CacheService(cache_dir=tmpdir, ttl_hours=24)
data = {"users": {}}
cache.set("2025-10-06", "2025-10-06", data)
# Immediately retrieve - should work
retrieved = cache.get("2025-10-06", "2025-10-06")
assert retrieved is not None
# Manually expire the cache entry by updating its timestamp to an old date
import sqlite3
from datetime import datetime, timedelta
conn = sqlite3.connect(cache.index_db)
cursor = conn.cursor()
old_time = (datetime.now() - timedelta(hours=25)).strftime("%Y-%m-%d %H:%M:%S")
cursor.execute("UPDATE cache SET created_at = ? WHERE id = 1", (old_time,))
conn.commit()
conn.close()
# Should return None now
retrieved_expired = cache.get("2025-10-06", "2025-10-06")
assert retrieved_expired is None
def test_cache_service_update():
"""Test cache update overwrites old data."""
with tempfile.TemporaryDirectory() as tmpdir:
cache = CacheService(cache_dir=tmpdir, ttl_hours=24)
data1 = {"users": {"first@example.com": {}}}
cache.set("2025-10-06", "2025-10-06", data1)
retrieved1 = cache.get("2025-10-06", "2025-10-06")
assert "first@example.com" in retrieved1["users"]
# Update with new data
data2 = {"users": {"second@example.com": {}}}
cache.set("2025-10-06", "2025-10-06", data2)
retrieved2 = cache.get("2025-10-06", "2025-10-06")
assert "second@example.com" in retrieved2["users"]
assert "first@example.com" not in retrieved2["users"]
def test_cache_service_multiple_entries():
"""Test cache with multiple date ranges."""
with tempfile.TemporaryDirectory() as tmpdir:
cache = CacheService(cache_dir=tmpdir, ttl_hours=24)
data1 = {"users": {"first@example.com": {}}}
data2 = {"users": {"second@example.com": {}}}
data3 = {"users": {"third@example.com": {}}}
cache.set("2025-10-06", "2025-10-06", data1)
cache.set("2025-10-07", "2025-10-07", data2)
cache.set("2025-10-08", "2025-10-10", data3)
retrieved1 = cache.get("2025-10-06", "2025-10-06")
retrieved2 = cache.get("2025-10-07", "2025-10-07")
retrieved3 = cache.get("2025-10-08", "2025-10-10")
assert "first@example.com" in retrieved1["users"]
assert "second@example.com" in retrieved2["users"]
assert "third@example.com" in retrieved3["users"]
if __name__ == "__main__":
test_cache_service_set_and_get()
test_cache_service_with_user_id()
test_cache_service_expiry()
test_cache_service_update()
test_cache_service_multiple_entries()
print("All cache tests passed!")