"""Tests for TunnelId, HopConfig, and TunnelInfo. TDD: written before implementation. """ import io import os import time import pytest from i2p_data.tunnel import TunnelId, HopConfig, TunnelInfo # --------------------------------------------------------------------------- # TunnelId # --------------------------------------------------------------------------- class TestTunnelId: """TunnelId: 4-byte unsigned integer identifier for tunnels.""" def test_construction_valid(self): tid = TunnelId(42) assert int(tid) == 42 def test_construction_zero(self): tid = TunnelId(0) assert int(tid) == 0 def test_construction_max(self): tid = TunnelId(2**32 - 1) assert int(tid) == 2**32 - 1 def test_construction_negative_raises(self): with pytest.raises(ValueError): TunnelId(-1) def test_construction_too_large_raises(self): with pytest.raises(ValueError): TunnelId(2**32) def test_to_bytes(self): tid = TunnelId(0x01020304) assert tid.to_bytes() == b"\x01\x02\x03\x04" def test_to_bytes_zero(self): tid = TunnelId(0) assert tid.to_bytes() == b"\x00\x00\x00\x00" def test_from_bytes_roundtrip(self): original = TunnelId(12345) restored = TunnelId.from_bytes(original.to_bytes()) assert original == restored def test_from_bytes_specific(self): tid = TunnelId.from_bytes(b"\x00\x00\x00\x2a") assert int(tid) == 42 def test_from_stream(self): stream = io.BytesIO(b"\x00\x00\x01\x00extra") tid = TunnelId.from_stream(stream) assert int(tid) == 256 # Stream should have advanced past the 4 bytes assert stream.read() == b"extra" def test_equality(self): assert TunnelId(100) == TunnelId(100) def test_inequality(self): assert TunnelId(100) != TunnelId(200) def test_equality_different_type(self): assert TunnelId(100) != 100 def test_hash_same(self): assert hash(TunnelId(100)) == hash(TunnelId(100)) def test_hash_usable_in_set(self): s = {TunnelId(1), TunnelId(2), TunnelId(1)} assert len(s) == 2 def test_is_zero_true(self): assert TunnelId(0).is_zero() is True def test_is_zero_false(self): assert TunnelId(1).is_zero() is False def test_repr(self): r = repr(TunnelId(42)) assert "42" in r def test_int(self): tid = TunnelId(999) assert int(tid) == 999 # --------------------------------------------------------------------------- # HopConfig # --------------------------------------------------------------------------- class TestHopConfig: """HopConfig: configuration for one hop in a tunnel.""" def _make_hop_config(self): return HopConfig( receive_tunnel_id=TunnelId(1), send_tunnel_id=TunnelId(2), receive_key=os.urandom(32), send_key=os.urandom(32), iv_key=os.urandom(32), reply_key=os.urandom(32), reply_iv=os.urandom(16), layer_key=os.urandom(32), ) def test_construction(self): hc = self._make_hop_config() assert isinstance(hc.receive_tunnel_id, TunnelId) assert isinstance(hc.send_tunnel_id, TunnelId) def test_field_access(self): rk = os.urandom(32) sk = os.urandom(32) ivk = os.urandom(32) rpk = os.urandom(32) rpiv = os.urandom(16) lk = os.urandom(32) hc = HopConfig( receive_tunnel_id=TunnelId(10), send_tunnel_id=TunnelId(20), receive_key=rk, send_key=sk, iv_key=ivk, reply_key=rpk, reply_iv=rpiv, layer_key=lk, ) assert int(hc.receive_tunnel_id) == 10 assert int(hc.send_tunnel_id) == 20 assert hc.receive_key == rk assert hc.send_key == sk assert hc.iv_key == ivk assert hc.reply_key == rpk assert hc.reply_iv == rpiv assert hc.layer_key == lk # --------------------------------------------------------------------------- # TunnelInfo # --------------------------------------------------------------------------- class TestTunnelInfo: """TunnelInfo: metadata about a built tunnel.""" def _make_tunnel_info(self, **overrides): defaults = dict( tunnel_id=TunnelId(500), gateway=os.urandom(32), length=3, creation_time=1_000_000, expiration=2_000_000, ) defaults.update(overrides) return TunnelInfo(**defaults) def test_construction(self): ti = self._make_tunnel_info() assert int(ti.tunnel_id) == 500 assert ti.length == 3 assert ti.creation_time == 1_000_000 assert ti.expiration == 2_000_000 def test_gateway_size(self): gw = os.urandom(32) ti = self._make_tunnel_info(gateway=gw) assert ti.gateway == gw assert len(ti.gateway) == 32 def test_gateway_wrong_size_raises(self): with pytest.raises(ValueError): self._make_tunnel_info(gateway=b"\x00" * 16) def test_is_expired_past(self): ti = self._make_tunnel_info(expiration=1_000) assert ti.is_expired(now_ms=2_000) is True def test_is_expired_future(self): ti = self._make_tunnel_info(expiration=5_000) assert ti.is_expired(now_ms=1_000) is False def test_is_expired_exact(self): ti = self._make_tunnel_info(expiration=1_000) # At exact expiration time, should be expired (<=) assert ti.is_expired(now_ms=1_000) is True def test_is_expired_default_now(self): # Far-future expiration should not be expired ti = self._make_tunnel_info(expiration=int(time.time() * 1000) + 600_000) assert ti.is_expired() is False def test_equality(self): gw = os.urandom(32) a = self._make_tunnel_info(tunnel_id=TunnelId(1), gateway=gw) b = self._make_tunnel_info(tunnel_id=TunnelId(1), gateway=gw) assert a == b def test_inequality(self): gw = os.urandom(32) a = self._make_tunnel_info(tunnel_id=TunnelId(1), gateway=gw) b = self._make_tunnel_info(tunnel_id=TunnelId(2), gateway=gw) assert a != b def test_equality_different_type(self): ti = self._make_tunnel_info() assert ti != "not a tunnel" def test_repr(self): r = repr(self._make_tunnel_info()) assert "500" in r