distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for NetDB data store."""
2
3import os
4import time
5
6import pytest
7
8
9class TestNetDBEntry:
10 def test_construct_router_info(self):
11 from i2p_netdb.datastore import NetDBEntry, EntryType
12 data = os.urandom(500)
13 entry = NetDBEntry(key=os.urandom(32), entry_type=EntryType.ROUTER_INFO,
14 data=data, received_ms=1000)
15 assert entry.entry_type == EntryType.ROUTER_INFO
16 assert entry.data == data
17
18 def test_construct_lease_set(self):
19 from i2p_netdb.datastore import NetDBEntry, EntryType
20 entry = NetDBEntry(key=os.urandom(32), entry_type=EntryType.LEASE_SET,
21 data=os.urandom(200), received_ms=1000)
22 assert entry.entry_type == EntryType.LEASE_SET
23
24 def test_is_expired(self):
25 from i2p_netdb.datastore import NetDBEntry, EntryType
26 entry = NetDBEntry(key=os.urandom(32), entry_type=EntryType.ROUTER_INFO,
27 data=b"", received_ms=1000, expiration_ms=5000)
28 assert entry.is_expired(now_ms=6000)
29 assert not entry.is_expired(now_ms=4000)
30
31
32class TestDataStore:
33 def test_put_and_get(self):
34 from i2p_netdb.datastore import DataStore, NetDBEntry, EntryType
35 store = DataStore()
36 key = os.urandom(32)
37 entry = NetDBEntry(key=key, entry_type=EntryType.ROUTER_INFO,
38 data=b"router data", received_ms=1000)
39 store.put(entry)
40 result = store.get(key)
41 assert result is not None
42 assert result.data == b"router data"
43
44 def test_get_missing(self):
45 from i2p_netdb.datastore import DataStore
46 store = DataStore()
47 assert store.get(os.urandom(32)) is None
48
49 def test_remove(self):
50 from i2p_netdb.datastore import DataStore, NetDBEntry, EntryType
51 store = DataStore()
52 key = os.urandom(32)
53 store.put(NetDBEntry(key, EntryType.ROUTER_INFO, b"data", 1000))
54 store.remove(key)
55 assert store.get(key) is None
56
57 def test_count(self):
58 from i2p_netdb.datastore import DataStore, NetDBEntry, EntryType
59 store = DataStore()
60 for i in range(5):
61 store.put(NetDBEntry(os.urandom(32), EntryType.ROUTER_INFO, b"", 1000))
62 assert store.count() == 5
63
64 def test_count_by_type(self):
65 from i2p_netdb.datastore import DataStore, NetDBEntry, EntryType
66 store = DataStore()
67 for _ in range(3):
68 store.put(NetDBEntry(os.urandom(32), EntryType.ROUTER_INFO, b"", 1000))
69 for _ in range(2):
70 store.put(NetDBEntry(os.urandom(32), EntryType.LEASE_SET, b"", 1000))
71 assert store.count_by_type(EntryType.ROUTER_INFO) == 3
72 assert store.count_by_type(EntryType.LEASE_SET) == 2
73
74 def test_remove_expired(self):
75 from i2p_netdb.datastore import DataStore, NetDBEntry, EntryType
76 store = DataStore()
77 store.put(NetDBEntry(os.urandom(32), EntryType.ROUTER_INFO, b"", 1000, 500))
78 store.put(NetDBEntry(os.urandom(32), EntryType.ROUTER_INFO, b"", 1000, 9999999999))
79 store.remove_expired(now_ms=1000)
80 assert store.count() == 1
81
82 def test_get_all_keys(self):
83 from i2p_netdb.datastore import DataStore, NetDBEntry, EntryType
84 store = DataStore()
85 keys = [os.urandom(32) for _ in range(3)]
86 for k in keys:
87 store.put(NetDBEntry(k, EntryType.ROUTER_INFO, b"", 1000))
88 all_keys = store.get_all_keys()
89 assert len(all_keys) == 3
90 for k in keys:
91 assert k in all_keys
92
93 def test_closest_keys(self):
94 from i2p_netdb.datastore import DataStore, NetDBEntry, EntryType
95 store = DataStore()
96 for _ in range(10):
97 store.put(NetDBEntry(os.urandom(32), EntryType.ROUTER_INFO, b"", 1000))
98 target = os.urandom(32)
99 closest = store.closest_keys(target, n=3)
100 assert len(closest) == 3