distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for SqliteNetDB — persistent RouterInfo storage."""
2
3import os
4import tempfile
5import time
6
7import pytest
8
9from i2p_netdb.sqlite_netdb import SqliteNetDB
10
11
12@pytest.fixture
13def db(tmp_path):
14 """Create a SqliteNetDB in a temp directory."""
15 ndb = SqliteNetDB(str(tmp_path))
16 yield ndb
17 ndb.close()
18
19
20class TestSqliteNetDBStore:
21 def test_store_and_count(self, db):
22 db.store(b"\x01" * 32, b"routerinfo1")
23 assert db.count() == 1
24
25 def test_store_multiple(self, db):
26 for i in range(5):
27 db.store(bytes([i]) * 32, f"data{i}".encode())
28 assert db.count() == 5
29
30 def test_store_replace(self, db):
31 key = b"\xaa" * 32
32 db.store(key, b"old")
33 db.store(key, b"new")
34 assert db.count() == 1
35 rows = db.load_all()
36 assert rows[0][1] == b"new"
37
38
39class TestSqliteNetDBLoad:
40 def test_load_all_empty(self, db):
41 assert db.load_all() == []
42
43 def test_load_all_returns_non_expired(self, db):
44 key = b"\x01" * 32
45 db.store(key, b"data", ttl_ms=86_400_000) # 24h TTL
46 rows = db.load_all()
47 assert len(rows) == 1
48 assert rows[0][0] == key
49 assert rows[0][1] == b"data"
50
51 def test_load_excludes_expired(self, db):
52 key = b"\x02" * 32
53 db.store(key, b"expired", ttl_ms=0) # already expired
54 # wait a tiny bit so expires_at < now
55 time.sleep(0.01)
56 rows = db.load_all()
57 assert len(rows) == 0
58
59
60class TestSqliteNetDBEvict:
61 def test_evict_expired(self, db):
62 db.store(b"\x01" * 32, b"expired", ttl_ms=0)
63 time.sleep(0.01)
64 db.store(b"\x02" * 32, b"fresh", ttl_ms=86_400_000)
65 evicted = db.evict_expired()
66 assert evicted == 1
67 assert db.count() == 1
68
69 def test_evict_nothing(self, db):
70 db.store(b"\x01" * 32, b"fresh", ttl_ms=86_400_000)
71 assert db.evict_expired() == 0
72
73 def test_evict_all(self, db):
74 for i in range(3):
75 db.store(bytes([i]) * 32, b"x", ttl_ms=0)
76 time.sleep(0.01)
77 evicted = db.evict_expired()
78 assert evicted == 3
79 assert db.count() == 0
80
81
82class TestSqliteNetDBLifecycle:
83 def test_creates_db_file(self, tmp_path):
84 db = SqliteNetDB(str(tmp_path))
85 db.store(b"\x01" * 32, b"data")
86 db.close()
87 assert (tmp_path / "netdb.sqlite3").exists()
88
89 def test_creates_parent_dirs(self, tmp_path):
90 nested = tmp_path / "a" / "b" / "c"
91 db = SqliteNetDB(str(nested))
92 db.store(b"\x01" * 32, b"data")
93 db.close()
94 assert (nested / "netdb.sqlite3").exists()
95
96 def test_persists_across_open_close(self, tmp_path):
97 db1 = SqliteNetDB(str(tmp_path))
98 db1.store(b"\x01" * 32, b"persist")
99 db1.close()
100
101 db2 = SqliteNetDB(str(tmp_path))
102 rows = db2.load_all()
103 assert len(rows) == 1
104 assert rows[0][1] == b"persist"
105 db2.close()