distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for LeaseSet2, MetaLease, MetaLeaseSet, and OfflineBlock."""
2
3import os
4import struct
5
6import pytest
7
8from i2p_data.lease_set2 import LeaseSet2, MetaLease, OfflineBlock
9
10
11class TestMetaLease:
12 def _make(self, **kwargs):
13 defaults = {
14 "gateway_hash": os.urandom(32),
15 "flags": 0,
16 "ls_type": 3,
17 "cost": 0,
18 "end_date": 1700000000,
19 }
20 defaults.update(kwargs)
21 return MetaLease(**defaults)
22
23 def test_creation(self):
24 gw = os.urandom(32)
25 ml = MetaLease(gateway_hash=gw, flags=0, ls_type=3, cost=5, end_date=1700000000)
26 assert ml.gateway_hash == gw
27 assert ml.flags == 0
28 assert ml.ls_type == 3
29 assert ml.cost == 5
30 assert ml.end_date == 1700000000
31
32 def test_default_values(self):
33 gw = os.urandom(32)
34 ml = MetaLease(gateway_hash=gw)
35 assert ml.flags == 0
36 assert ml.ls_type == 3
37 assert ml.cost == 0
38 assert ml.end_date == 0
39
40 def test_wrong_hash_length(self):
41 with pytest.raises(ValueError, match="32 bytes"):
42 MetaLease(gateway_hash=b"\x00" * 16)
43
44 def test_roundtrip(self):
45 ml = self._make()
46 data = ml.to_bytes()
47 assert len(data) == MetaLease.SIZE # 40
48 ml2 = MetaLease.from_bytes(data)
49 assert ml == ml2
50
51 def test_from_bytes_too_short(self):
52 with pytest.raises(ValueError, match="40 bytes"):
53 MetaLease.from_bytes(b"\x00" * 10)
54
55 def test_to_bytes_format(self):
56 gw = b"\xaa" * 32
57 ml = MetaLease(gateway_hash=gw, flags=1, ls_type=7, cost=10, end_date=12345)
58 data = ml.to_bytes()
59 assert data[:32] == gw
60 assert struct.unpack("!H", data[32:34])[0] == 1
61 assert data[34] == 7
62 assert data[35] == 10
63 assert struct.unpack("!I", data[36:40])[0] == 12345
64
65 def test_equality(self):
66 gw = os.urandom(32)
67 ml1 = MetaLease(gateway_hash=gw, cost=5, end_date=100)
68 ml2 = MetaLease(gateway_hash=gw, cost=5, end_date=100)
69 assert ml1 == ml2
70
71 def test_inequality(self):
72 ml1 = self._make(cost=1)
73 ml2 = self._make(cost=2)
74 assert ml1 != ml2
75
76 def test_not_equal_to_other_types(self):
77 ml = self._make()
78 assert ml != "not a metalease"
79
80 def test_hash_consistent(self):
81 gw = os.urandom(32)
82 ml1 = MetaLease(gateway_hash=gw, cost=5, end_date=100)
83 ml2 = MetaLease(gateway_hash=gw, cost=5, end_date=100)
84 assert hash(ml1) == hash(ml2)
85
86 def test_repr(self):
87 gw = b"\xab\xcd\xef\x01" + os.urandom(28)
88 ml = MetaLease(gateway_hash=gw, ls_type=7, cost=3)
89 r = repr(ml)
90 assert "MetaLease" in r
91 assert "type=7" in r
92 assert "cost=3" in r
93
94
95class TestLeaseSet2Properties:
96 """Test LeaseSet2 property accessors and flags without serialization."""
97
98 def test_flags_offline(self):
99 assert LeaseSet2.FLAG_OFFLINE_KEYS == 0x0001
100
101 def test_flags_unpublished(self):
102 assert LeaseSet2.FLAG_UNPUBLISHED == 0x0002
103
104 def test_flags_blinded(self):
105 assert LeaseSet2.FLAG_BLINDED == 0x0004
106
107 def test_type(self):
108 assert LeaseSet2.TYPE == 3
109
110 def test_max_leases(self):
111 assert LeaseSet2.MAX_LEASES == 16
112
113 def test_max_enc_keys(self):
114 assert LeaseSet2.MAX_ENC_KEYS == 8
115
116
117class TestOfflineBlock:
118 def test_to_bytes_roundtrip_structure(self):
119 """Test that OfflineBlock.to_bytes() produces correct wire format structure."""
120 from unittest.mock import MagicMock
121 sig_type = MagicMock()
122 sig_type.code = 7
123 ob = OfflineBlock(
124 transient_expires=1700000000,
125 transient_sig_type=sig_type,
126 transient_spk=b"\x11" * 32,
127 offline_signature=b"\x22" * 64,
128 )
129 data = ob.to_bytes()
130 # 4 (expires) + 2 (sig type code) + 32 (spk) + 64 (sig)
131 assert len(data) == 4 + 2 + 32 + 64
132 assert struct.unpack("!I", data[:4])[0] == 1700000000
133 assert struct.unpack("!H", data[4:6])[0] == 7
134 assert data[6:38] == b"\x11" * 32
135 assert data[38:] == b"\x22" * 64