arsam
d71cea666a
Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.
58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
import asyncio
|
|
import pytest
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inmemory_cache_miss():
|
|
from axon.services.cache import InMemoryCache
|
|
cache = InMemoryCache()
|
|
assert await cache.get("missing") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inmemory_cache_set_and_get():
|
|
from axon.services.cache import InMemoryCache
|
|
cache = InMemoryCache()
|
|
await cache.set("k", "v", ttl=60)
|
|
assert await cache.get("k") == "v"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inmemory_cache_expiry():
|
|
from axon.services.cache import InMemoryCache
|
|
cache = InMemoryCache()
|
|
await cache.set("k", "v", ttl=0)
|
|
# Force expiry by backdating
|
|
key = list(cache._store.keys())[0]
|
|
value, expires_at = cache._store[key]
|
|
from datetime import UTC, datetime, timedelta
|
|
cache._store[key] = (value, datetime.now(UTC) - timedelta(seconds=1))
|
|
assert await cache.get("k") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inmemory_cache_ping():
|
|
from axon.services.cache import InMemoryCache
|
|
cache = InMemoryCache()
|
|
assert await cache.ping() is True
|
|
|
|
|
|
def test_make_cache_key_is_deterministic():
|
|
from axon.services.cache import make_cache_key
|
|
k1 = make_cache_key("search", query="hello", max_results=10)
|
|
k2 = make_cache_key("search", query="hello", max_results=10)
|
|
assert k1 == k2
|
|
|
|
|
|
def test_make_cache_key_differs_by_params():
|
|
from axon.services.cache import make_cache_key
|
|
k1 = make_cache_key("search", query="hello")
|
|
k2 = make_cache_key("search", query="world")
|
|
assert k1 != k2
|
|
|
|
|
|
def test_make_cache_key_has_prefix():
|
|
from axon.services.cache import make_cache_key
|
|
k = make_cache_key("extract", url="https://x.com")
|
|
assert k.startswith("extract:")
|