Import tests: tests/test_cache
Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.
This commit is contained in:
@@ -0,0 +1,57 @@
|
||||
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:")
|
||||
Reference in New Issue
Block a user