231 lines
8.4 KiB
Python
231 lines
8.4 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from straywild_discovery.registry import (
|
|
LeaseAuthorizationError,
|
|
RoomLimitError,
|
|
RoomNotFoundError,
|
|
RoomRegistry,
|
|
ValidationError,
|
|
)
|
|
|
|
|
|
VALID_ROOM = {
|
|
"room_name": "Pond Friends",
|
|
"port": 7777,
|
|
"current_players": 1,
|
|
"max_players": 8,
|
|
"game_version": "0.20.3-alpha",
|
|
"protocol_version": 12,
|
|
}
|
|
|
|
|
|
class FakeClock:
|
|
def __init__(self) -> None:
|
|
self.now = 1_000.0
|
|
|
|
def __call__(self) -> float:
|
|
return self.now
|
|
|
|
|
|
class RoomRegistryTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.clock = FakeClock()
|
|
self.registry = RoomRegistry(ttl_seconds=30.0, clock=self.clock)
|
|
|
|
def test_create_list_update_and_delete(self) -> None:
|
|
room, token, verification_token = self.registry.create(
|
|
"203.0.113.10", VALID_ROOM
|
|
)
|
|
room = self.registry.verify_endpoint(
|
|
room.room_id, verification_token, "203.0.113.10", 7777
|
|
)
|
|
self.assertEqual(room.address, "203.0.113.10")
|
|
self.assertEqual(self.registry.list_rooms(), [room])
|
|
|
|
self.clock.now += 5.0
|
|
updated_payload = dict(VALID_ROOM, current_players=3)
|
|
updated = self.registry.update(room.room_id, token, room.address, updated_payload)
|
|
self.assertEqual(updated.current_players, 3)
|
|
self.assertEqual(updated.created_at, room.created_at)
|
|
self.assertEqual(updated.expires_at, self.clock.now + 30.0)
|
|
|
|
self.registry.delete(room.room_id, token)
|
|
self.assertEqual(self.registry.list_rooms(), [])
|
|
|
|
def test_public_room_shape_never_discloses_endpoint(self) -> None:
|
|
room, _token, verification_token = self.registry.create(
|
|
"203.0.113.10", VALID_ROOM
|
|
)
|
|
room = self.registry.verify_endpoint(
|
|
room.room_id, verification_token, "203.0.113.10", 49152
|
|
)
|
|
public = room.public_dict()
|
|
self.assertNotIn("address", public)
|
|
self.assertNotIn("port", public)
|
|
self.assertEqual(public["connection_mode"], "direct")
|
|
self.assertEqual(public["ip_privacy"], "peer_visible")
|
|
self.assertEqual(
|
|
room.join_route_dict(),
|
|
{
|
|
"transport": "direct",
|
|
"address": "203.0.113.10",
|
|
"port": 49152,
|
|
"ip_privacy": "peer_visible",
|
|
},
|
|
)
|
|
|
|
def test_relay_room_shape_advertises_privacy_without_endpoint(self) -> None:
|
|
registry = RoomRegistry(
|
|
ttl_seconds=30.0,
|
|
relay_available=True,
|
|
clock=self.clock,
|
|
)
|
|
room, _token, verification_token = registry.create(
|
|
"203.0.113.10", dict(VALID_ROOM, connection_mode="relay")
|
|
)
|
|
room = registry.verify_endpoint(
|
|
room.room_id, verification_token, "203.0.113.10", 49152
|
|
)
|
|
public = room.public_dict()
|
|
self.assertEqual(public["connection_mode"], "relay")
|
|
self.assertEqual(public["ip_privacy"], "relayed")
|
|
self.assertNotIn("address", public)
|
|
self.assertNotIn("port", public)
|
|
|
|
def test_player_can_choose_direct_when_relay_is_available(self) -> None:
|
|
registry = RoomRegistry(
|
|
ttl_seconds=30.0,
|
|
relay_available=True,
|
|
clock=self.clock,
|
|
)
|
|
room, _token, _verification_token = registry.create(
|
|
"203.0.113.10", dict(VALID_ROOM, connection_mode="direct")
|
|
)
|
|
self.assertEqual(room.connection_mode, "direct")
|
|
|
|
def test_dedicated_server_cannot_request_relay(self) -> None:
|
|
registry = RoomRegistry(
|
|
ttl_seconds=30.0,
|
|
relay_available=True,
|
|
clock=self.clock,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
ValidationError,
|
|
"dedicated servers cannot use the privacy relay",
|
|
):
|
|
registry.create(
|
|
"203.0.113.10",
|
|
dict(
|
|
VALID_ROOM,
|
|
host_kind="dedicated",
|
|
connection_mode="relay",
|
|
),
|
|
)
|
|
|
|
def test_relay_request_fails_when_service_has_no_relay(self) -> None:
|
|
with self.assertRaisesRegex(
|
|
ValidationError,
|
|
"privacy relay is not available",
|
|
):
|
|
self.registry.create(
|
|
"203.0.113.10",
|
|
dict(VALID_ROOM, connection_mode="relay"),
|
|
)
|
|
|
|
def test_join_attempts_are_rate_limited_per_requester(self) -> None:
|
|
registry = RoomRegistry(ttl_seconds=300.0, clock=self.clock)
|
|
room, _token, verification_token = registry.create(
|
|
"203.0.113.10", VALID_ROOM
|
|
)
|
|
room = registry.verify_endpoint(
|
|
room.room_id, verification_token, "203.0.113.10", 7777
|
|
)
|
|
for _attempt in range(10):
|
|
token, routed_room = registry.create_join_attempt(
|
|
room.room_id, "198.51.100.20"
|
|
)
|
|
self.assertTrue(token)
|
|
self.assertEqual(routed_room, room)
|
|
with self.assertRaises(RoomLimitError):
|
|
registry.create_join_attempt(room.room_id, "198.51.100.20")
|
|
|
|
def test_empty_dedicated_room_is_valid(self) -> None:
|
|
room, _token, verification_token = self.registry.create(
|
|
"203.0.113.10", dict(VALID_ROOM, current_players=0)
|
|
)
|
|
room = self.registry.verify_endpoint(
|
|
room.room_id, verification_token, "203.0.113.10", 7777
|
|
)
|
|
self.assertEqual(room.current_players, 0)
|
|
|
|
def test_reverification_refreshes_changed_public_endpoint(self) -> None:
|
|
room, _token, verification_token = self.registry.create(
|
|
"203.0.113.10", VALID_ROOM
|
|
)
|
|
first = self.registry.verify_endpoint(
|
|
room.room_id, verification_token, "203.0.113.10", 41000
|
|
)
|
|
self.clock.now += 20.0
|
|
refreshed = self.registry.verify_endpoint(
|
|
room.room_id, verification_token, "198.51.100.25", 51000
|
|
)
|
|
self.assertEqual(refreshed.address, "198.51.100.25")
|
|
self.assertEqual(refreshed.port, 51000)
|
|
self.assertEqual(refreshed.created_at, first.created_at)
|
|
self.assertEqual(refreshed.expires_at, self.clock.now + 30.0)
|
|
|
|
def test_expired_room_is_removed(self) -> None:
|
|
room, token, _verification_token = self.registry.create(
|
|
"203.0.113.10", VALID_ROOM
|
|
)
|
|
self.clock.now += 30.0
|
|
self.assertEqual(self.registry.list_rooms(), [])
|
|
with self.assertRaises(RoomNotFoundError):
|
|
self.registry.update(room.room_id, token, room.address, VALID_ROOM)
|
|
|
|
def test_wrong_token_cannot_mutate_room(self) -> None:
|
|
room, _token, _verification_token = self.registry.create(
|
|
"203.0.113.10", VALID_ROOM
|
|
)
|
|
with self.assertRaises(LeaseAuthorizationError):
|
|
self.registry.delete(room.room_id, "wrong")
|
|
|
|
def test_invalid_room_is_rejected(self) -> None:
|
|
with self.assertRaises(ValidationError):
|
|
self.registry.create("203.0.113.10", dict(VALID_ROOM, current_players=9))
|
|
with self.assertRaises(ValidationError):
|
|
self.registry.create("203.0.113.10", dict(VALID_ROOM, room_name="bad\nname"))
|
|
|
|
def test_filters_use_authored_versions(self) -> None:
|
|
expected, _token, verification_token = self.registry.create(
|
|
"203.0.113.10", VALID_ROOM
|
|
)
|
|
expected = self.registry.verify_endpoint(
|
|
expected.room_id, verification_token, "203.0.113.10", 7777
|
|
)
|
|
self.registry.create(
|
|
"203.0.113.11",
|
|
dict(VALID_ROOM, room_name="Older Room", protocol_version=3),
|
|
)
|
|
self.assertEqual(self.registry.list_rooms(protocol_version=12), [expected])
|
|
self.assertEqual(self.registry.list_rooms(game_version="missing"), [])
|
|
|
|
def test_room_limits_bound_untrusted_advertisements(self) -> None:
|
|
registry = RoomRegistry(
|
|
ttl_seconds=30.0,
|
|
max_rooms=2,
|
|
max_rooms_per_address=1,
|
|
clock=self.clock,
|
|
)
|
|
registry.create("203.0.113.10", VALID_ROOM)
|
|
with self.assertRaises(RoomLimitError):
|
|
registry.create("203.0.113.10", dict(VALID_ROOM, room_name="Duplicate"))
|
|
registry.create("203.0.113.11", dict(VALID_ROOM, room_name="Second"))
|
|
with self.assertRaises(RoomLimitError):
|
|
registry.create("203.0.113.12", dict(VALID_ROOM, room_name="Full"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|