"""Tests for peer history tracking.""" import time import pytest class TestDBHistory: def test_record_and_success_rate(self): from i2p_peer.history import DBHistory h = DBHistory() for _ in range(7): h.record_store(success=True, latency_ms=100.0) for _ in range(3): h.record_store(success=False, latency_ms=200.0) assert abs(h.success_rate - 0.7) < 0.001 def test_max_entries_eviction(self): from i2p_peer.history import DBHistory h = DBHistory(max_entries=10) for i in range(20): h.record_store(success=True, latency_ms=float(i)) # Should have only 10 entries assert len(h.recent_entries) == 10 # Oldest should be evicted, so min latency should be 10.0 latencies = [e.latency_ms for e in h.recent_entries] assert min(latencies) == 10.0 def test_average_latency(self): from i2p_peer.history import DBHistory h = DBHistory() h.record_store(success=True, latency_ms=100.0) h.record_store(success=True, latency_ms=200.0) h.record_lookup(success=True, latency_ms=300.0) assert abs(h.average_latency - 200.0) < 0.001 def test_average_latency_empty(self): from i2p_peer.history import DBHistory h = DBHistory() assert h.average_latency == 0.0 def test_success_rate_empty(self): from i2p_peer.history import DBHistory h = DBHistory() assert h.success_rate == 0.0 def test_record_lookup(self): from i2p_peer.history import DBHistory h = DBHistory() h.record_lookup(success=True, latency_ms=50.0) h.record_lookup(success=False, latency_ms=100.0) assert abs(h.success_rate - 0.5) < 0.001 def test_recent_entries_returns_list(self): from i2p_peer.history import DBHistory h = DBHistory() h.record_store(success=True) entries = h.recent_entries assert len(entries) == 1 assert entries[0].success is True class TestTunnelHistory: def test_build_success_rate(self): from i2p_peer.history import TunnelHistory h = TunnelHistory() for _ in range(8): h.record_build(success=True) for _ in range(2): h.record_build(success=False) assert abs(h.build_success_rate - 0.8) < 0.001 def test_participation_rate(self): from i2p_peer.history import TunnelHistory h = TunnelHistory() for _ in range(6): h.record_participation(success=True, data_transferred=1024) for _ in range(4): h.record_participation(success=False) assert abs(h.participation_rate - 0.6) < 0.001 def test_tunnel_history_empty(self): from i2p_peer.history import TunnelHistory h = TunnelHistory() assert h.build_success_rate == 0.0 assert h.participation_rate == 0.0 def test_tunnel_history_max_entries(self): from i2p_peer.history import TunnelHistory h = TunnelHistory(max_entries=5) for _ in range(10): h.record_build(success=True) # Internal entries should be capped assert len(h._entries) == 5