1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
|
"""Tests for the (path, size, mtime) -> hash cache (indexer/cache.py)."""
import pytest
from meshbay_node.indexer.cache import IndexCache
@pytest.fixture
async def cache(tmp_path):
c = IndexCache(db_path=tmp_path / "index_cache.db")
await c.open()
yield c
await c.close()
@pytest.mark.asyncio
async def test_put_then_lookup_hits(cache):
await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123",
type="video", added_at=42)
hit = await cache.lookup("/lib/a.mkv", size=1000, mtime=111.0)
assert hit is not None
assert hit.hash == "abc123"
assert hit.type == "video"
assert hit.added_at == 42
@pytest.mark.asyncio
async def test_lookup_misses_on_unknown_path(cache):
assert await cache.lookup("/lib/never-seen.mkv", size=1, mtime=1.0) is None
@pytest.mark.asyncio
async def test_lookup_misses_on_different_mtime(cache):
await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123",
type="video", added_at=42)
assert await cache.lookup("/lib/a.mkv", size=1000, mtime=222.0) is None
@pytest.mark.asyncio
async def test_lookup_misses_on_different_size(cache):
await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123",
type="video", added_at=42)
assert await cache.lookup("/lib/a.mkv", size=2000, mtime=111.0) is None
@pytest.mark.asyncio
async def test_put_overwrites_previous_row_for_same_path(cache):
await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="old",
type="video", added_at=1)
await cache.put("/lib/a.mkv", size=2000, mtime=222.0, hash="new",
type="video", added_at=2)
assert await cache.lookup("/lib/a.mkv", size=1000, mtime=111.0) is None
hit = await cache.lookup("/lib/a.mkv", size=2000, mtime=222.0)
assert hit.hash == "new"
@pytest.mark.asyncio
async def test_cache_survives_reopen(tmp_path):
db_path = tmp_path / "index_cache.db"
c1 = IndexCache(db_path=db_path)
await c1.open()
await c1.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123",
type="video", added_at=42)
await c1.close()
c2 = IndexCache(db_path=db_path)
await c2.open()
hit = await c2.lookup("/lib/a.mkv", size=1000, mtime=111.0)
await c2.close()
assert hit is not None
assert hit.hash == "abc123"
|