# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. ========= # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # ========= Copyright 2026 @ Strukto.AI All Rights Reserved. ========= import os import time from datetime import datetime, timedelta, timezone import pytest import pytest_asyncio from mirage.cache.index import IndexEntry, LookupStatus from mirage.cache.index.redis import RedisIndexCacheStore REDIS_URL = os.environ.get("REDIS_URL", "") pytestmark = pytest.mark.skipif(not REDIS_URL, reason="REDIS_URL not set") @pytest_asyncio.fixture() async def store(): s = RedisIndexCacheStore(ttl=60, url=REDIS_URL, key_prefix="test:") await s.clear() yield s await s.clear() await s.close() @pytest_asyncio.fixture() def entry(): return IndexEntry( id="file1", name="Test File", resource_type="text/plain", remote_time="2026-04-01T00:00:00.000Z", vfs_name="test_file.txt", ) @pytest.mark.asyncio async def test_put_and_get(store, entry): await store.put("/folder/test_file.txt", entry) result = await store.get("/folder/test_file.txt") assert result.entry is not None assert result.entry.id == "file1" assert result.entry.name == "Test File" @pytest.mark.asyncio async def test_put_sets_index_time(store, entry): await store.put("/folder/test_file.txt", entry) result = await store.get("/folder/test_file.txt") assert result.entry is not None assert result.entry.index_time != "" @pytest.mark.asyncio async def test_put_preserves_existing_index_time(store): entry = IndexEntry(id="b", name="b.txt", resource_type="file", index_time="2026-01-01T00:00:00") await store.put("/data/b.txt", entry) result = await store.get("/data/b.txt") assert result.entry.index_time == "2026-01-01T00:00:00" @pytest.mark.asyncio async def test_get_not_found(store): result = await store.get("/nonexistent/file.txt") assert result.status == LookupStatus.NOT_FOUND assert result.entry is None @pytest.mark.asyncio async def test_list_dir_not_found(store): result = await store.list_dir("/some_dir") assert result.status == LookupStatus.NOT_FOUND @pytest.mark.asyncio async def test_set_dir_and_list(store): entries = [ ("a.txt", IndexEntry(id="a", name="A", resource_type="text/plain", vfs_name="a.txt")), ("b.txt", IndexEntry(id="b", name="B", resource_type="text/plain", vfs_name="b.txt")), ] await store.set_dir("/mydir", entries) result = await store.list_dir("/mydir") assert result.status is None assert sorted(result.entries) == ["/mydir/a.txt", "/mydir/b.txt"] @pytest.mark.asyncio async def test_set_dir_root(store): entries = [ ("top.txt", IndexEntry(id="t", name="T", resource_type="text/plain", vfs_name="top.txt")), ] await store.set_dir("/", entries) result = await store.list_dir("/") assert result.status is None assert result.entries == ["/top.txt"] @pytest.mark.asyncio async def test_set_dir_entries_are_retrievable(store): entries = [("x.csv", IndexEntry(id="x1", name="x.csv", resource_type="file"))] await store.set_dir("/root", entries) result = await store.get("/root/x.csv") assert result.entry is not None assert result.entry.id == "x1" @pytest.mark.asyncio async def test_list_dir_expired(store): entries = [ ("file.txt", IndexEntry(id="f", name="F", resource_type="text/plain", vfs_name="file.txt")), ] past = datetime.now(timezone.utc) + timedelta(seconds=1) await store.set_dir("/dir", entries, expired_at=past) time.sleep(1.5) result = await store.list_dir("/dir") assert result.status in (LookupStatus.NOT_FOUND, LookupStatus.EXPIRED) @pytest.mark.asyncio async def test_list_dir_fresh(store): entries = [ ("file.txt", IndexEntry(id="f", name="F", resource_type="text/plain", vfs_name="file.txt")), ] future = datetime.now(timezone.utc) + timedelta(seconds=3600) await store.set_dir("/dir", entries, expired_at=future) result = await store.list_dir("/dir") assert result.status is None assert result.entries == ["/dir/file.txt"] @pytest.mark.asyncio async def test_set_dir_overwrites_previous(store): await store.set_dir("/d", [ ("old.txt", IndexEntry(id="o", name="old.txt", resource_type="file")) ]) await store.set_dir("/d", [ ("new.txt", IndexEntry(id="n", name="new.txt", resource_type="file")) ]) result = await store.list_dir("/d") assert result.entries is not None assert len(result.entries) == 1 assert "new.txt" in result.entries[0] @pytest.mark.asyncio async def test_invalidate_dir_evicts_child_entries(store, entry): await store.set_dir("/folder", [("test.txt", entry)]) assert (await store.get("/folder/test.txt")).entry is not None await store.invalidate_dir("/folder") assert (await store.get("/folder/test.txt")).status == LookupStatus.NOT_FOUND assert (await store.list_dir("/folder")).status == LookupStatus.NOT_FOUND @pytest.mark.asyncio async def test_clear(store, entry): await store.put("/folder/test.txt", entry) await store.set_dir("/folder", [("test.txt", entry)]) await store.clear() result = await store.get("/folder/test.txt") assert result.status == LookupStatus.NOT_FOUND result = await store.list_dir("/folder") assert result.status == LookupStatus.NOT_FOUND @pytest.mark.asyncio async def test_key_prefix_isolation(): s1 = RedisIndexCacheStore(ttl=60, url=REDIS_URL, key_prefix="ns1:") s2 = RedisIndexCacheStore(ttl=60, url=REDIS_URL, key_prefix="ns2:") await s1.clear() await s2.clear() await s1.put("/shared", IndexEntry(id="a", name="a", resource_type="file")) assert (await s2.get("/shared")).status == LookupStatus.NOT_FOUND assert (await s1.get("/shared")).entry is not None await s1.clear() await s2.clear() await s1.close() await s2.close()