Andros Fenollosa
bfd8b510b6
Clean architecture (core use cases + infra gateways), X25519/AES-GCM E2E encryption, tkinter GUI, and a pytest suite. 45 tests passing.
43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
"""Tests for the prune_stale_peers use case."""
|
|
|
|
from neotalk.core.entities.peer import Peer
|
|
from neotalk.core.use_cases.roster.prune_stale_peers import (
|
|
prune_stale_peers_use_case,
|
|
)
|
|
|
|
|
|
def _peer(peer_id: str, last_seen: float) -> Peer:
|
|
return Peer(peer_id, peer_id, "host", 1, b"k", last_seen)
|
|
|
|
|
|
def test_prune_keeps_recent_peers_and_drops_old_ones():
|
|
# Given a roster with one fresh and one stale peer
|
|
roster = {"fresh": _peer("fresh", 95.0), "stale": _peer("stale", 80.0)}
|
|
|
|
# When pruning at t=100 with a 10s timeout
|
|
response = prune_stale_peers_use_case(roster, now=100.0, timeout=10.0)
|
|
|
|
# Then only the fresh peer survives
|
|
assert set(response["data"]["roster"]) == {"fresh"}
|
|
assert response["data"]["removed"] == ["stale"]
|
|
|
|
|
|
def test_prune_keeps_a_peer_exactly_on_the_boundary():
|
|
# Given a peer seen exactly timeout seconds ago
|
|
roster = {"edge": _peer("edge", 90.0)}
|
|
|
|
# When pruning at the boundary
|
|
response = prune_stale_peers_use_case(roster, now=100.0, timeout=10.0)
|
|
|
|
# Then it is kept
|
|
assert set(response["data"]["roster"]) == {"edge"}
|
|
|
|
|
|
def test_prune_on_empty_roster_returns_empty():
|
|
# Given no peers
|
|
# When pruning
|
|
response = prune_stale_peers_use_case({}, now=1.0, timeout=1.0)
|
|
|
|
# Then nothing is removed
|
|
assert response["data"]["roster"] == {}
|
|
assert response["data"]["removed"] == []
|