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()