"""Tests for PeerSelector (tunnel hop selection) and TunnelHopConfig. TDD: tests written before implementation. """ import os import struct import pytest from i2p_peer.hop_config import TunnelHopConfig from i2p_peer.organizer import PeerTier, ProfileOrganizer from i2p_peer.profile import PeerProfile from i2p_peer.selector import PeerSelector # --- Helpers --- def _make_hash(i: int) -> bytes: """Create a unique 32-byte peer hash from an integer.""" return i.to_bytes(4, "big") + b"\x00" * 28 def _make_profile(i: int, capacity: float = 0.5, speed: float = 0.5, tunnel_builds: int = 20) -> PeerProfile: """Create a PeerProfile with given scores.""" p = PeerProfile(_make_hash(i)) p.capacity = capacity p.speed = speed p.tunnel_builds_succeeded = tunnel_builds p.tunnel_builds_failed = 0 p.record_latency(200.0) # give it some latency data p.heard_from() return p def _organizer_with_profiles(profiles: list[PeerProfile]) -> ProfileOrganizer: org = ProfileOrganizer() for p in profiles: org.add_profile(p) return org # --- TunnelHopConfig tests --- class TestTunnelHopConfig: def test_fields_present(self): cfg = TunnelHopConfig( peer_hash=os.urandom(32), receive_tunnel_id=12345, layer_key=os.urandom(32), iv_key=os.urandom(32), reply_key=os.urandom(32), reply_iv=os.urandom(16), is_gateway=True, is_endpoint=False, ) assert len(cfg.peer_hash) == 32 assert len(cfg.layer_key) == 32 assert len(cfg.iv_key) == 32 assert len(cfg.reply_key) == 32 assert len(cfg.reply_iv) == 16 assert cfg.receive_tunnel_id == 12345 assert cfg.is_gateway is True assert cfg.is_endpoint is False def test_defaults(self): cfg = TunnelHopConfig( peer_hash=os.urandom(32), receive_tunnel_id=1, layer_key=os.urandom(32), iv_key=os.urandom(32), reply_key=os.urandom(32), reply_iv=os.urandom(16), ) assert cfg.is_gateway is False assert cfg.is_endpoint is False # --- PeerSelector.select_peers tests --- class TestSelectPeers: def test_basic_selection(self): """Select N peers from profiles.""" profiles = [_make_profile(i, capacity=0.5) for i in range(10)] org = _organizer_with_profiles(profiles) sel = PeerSelector(org) result = sel.select_peers(3) assert len(result) == 3 # All returned items should be 32-byte hashes for h in result: assert isinstance(h, bytes) assert len(h) == 32 def test_exclude_list_respected(self): """Excluded peers must not appear in results.""" profiles = [_make_profile(i, capacity=0.5) for i in range(5)] org = _organizer_with_profiles(profiles) sel = PeerSelector(org) excluded = {_make_hash(0), _make_hash(1), _make_hash(2)} result = sel.select_peers(5, exclude=excluded) for h in result: assert h not in excluded # Only 2 non-excluded peers exist assert len(result) == 2 def test_higher_scoring_peers_preferred(self): """Peers with higher capacity/speed should be selected first.""" # Create peers with distinct tiers fast_peer = _make_profile(1, capacity=0.9, speed=0.8, tunnel_builds=20) high_cap_peer = _make_profile(2, capacity=0.7, speed=0.3, tunnel_builds=5) standard_peer = _make_profile(3, capacity=0.4, speed=0.2, tunnel_builds=5) failing_peer = _make_profile(4, capacity=0.1, speed=0.1, tunnel_builds=5) org = _organizer_with_profiles([fast_peer, high_cap_peer, standard_peer, failing_peer]) sel = PeerSelector(org) # Request 2 peers — should get the best tiers result = sel.select_peers(2) assert len(result) == 2 # The fast and high-cap peers should be selected assert _make_hash(1) in result assert _make_hash(2) in result def test_empty_peer_set(self): """Empty organizer returns empty list.""" org = ProfileOrganizer() sel = PeerSelector(org) result = sel.select_peers(3) assert result == [] def test_fewer_peers_than_requested(self): """If fewer peers exist than requested, return all available.""" profiles = [_make_profile(i, capacity=0.5) for i in range(2)] org = _organizer_with_profiles(profiles) sel = PeerSelector(org) result = sel.select_peers(10) assert len(result) == 2 # --- PeerSelector.select_hops tests --- class TestSelectHops: def test_returns_correct_number(self): """select_hops returns the requested number of TunnelHopConfig objects.""" profiles = [_make_profile(i, capacity=0.5) for i in range(5)] org = _organizer_with_profiles(profiles) sel = PeerSelector(org) hops = sel.select_hops(3) assert len(hops) == 3 for hop in hops: assert isinstance(hop, TunnelHopConfig) def test_hop_key_sizes(self): """Each hop has correct key sizes.""" profiles = [_make_profile(i, capacity=0.5) for i in range(5)] org = _organizer_with_profiles(profiles) sel = PeerSelector(org) hops = sel.select_hops(3) for hop in hops: assert len(hop.peer_hash) == 32 assert len(hop.layer_key) == 32 assert len(hop.iv_key) == 32 assert len(hop.reply_key) == 32 assert len(hop.reply_iv) == 16 assert isinstance(hop.receive_tunnel_id, int) assert hop.receive_tunnel_id > 0 def test_gateway_and_endpoint_flags(self): """First hop is gateway, last hop is endpoint.""" profiles = [_make_profile(i, capacity=0.5) for i in range(5)] org = _organizer_with_profiles(profiles) sel = PeerSelector(org) hops = sel.select_hops(3) assert hops[0].is_gateway is True assert hops[0].is_endpoint is False assert hops[1].is_gateway is False assert hops[1].is_endpoint is False assert hops[2].is_gateway is False assert hops[2].is_endpoint is True def test_single_hop_is_both_gateway_and_endpoint(self): """A single-hop tunnel is both gateway and endpoint.""" profiles = [_make_profile(0, capacity=0.5)] org = _organizer_with_profiles(profiles) sel = PeerSelector(org) hops = sel.select_hops(1) assert len(hops) == 1 assert hops[0].is_gateway is True assert hops[0].is_endpoint is True def test_keys_are_unique_per_hop(self): """Each hop should have distinct random keys.""" profiles = [_make_profile(i, capacity=0.5) for i in range(5)] org = _organizer_with_profiles(profiles) sel = PeerSelector(org) hops = sel.select_hops(3) layer_keys = [hop.layer_key for hop in hops] assert len(set(layer_keys)) == 3 # all unique def test_exclude_in_select_hops(self): """Exclude set is passed through to peer selection.""" profiles = [_make_profile(i, capacity=0.5) for i in range(5)] org = _organizer_with_profiles(profiles) sel = PeerSelector(org) excluded = {_make_hash(0), _make_hash(1)} hops = sel.select_hops(3, exclude=excluded) hop_hashes = {hop.peer_hash for hop in hops} for h in excluded: assert h not in hop_hashes def test_empty_peers_returns_empty_hops(self): """Empty organizer returns empty hop list.""" org = ProfileOrganizer() sel = PeerSelector(org) hops = sel.select_hops(3) assert hops == [] def test_fewer_peers_than_hops(self): """If fewer peers than requested hops, return what's available.""" profiles = [_make_profile(i, capacity=0.5) for i in range(2)] org = _organizer_with_profiles(profiles) sel = PeerSelector(org) hops = sel.select_hops(5) assert len(hops) == 2 # First is gateway, last is endpoint assert hops[0].is_gateway is True assert hops[-1].is_endpoint is True