134 lines
4.9 KiB
Python
134 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from netfishing_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.6.8-alpha",
|
|
"protocol_version": 3,
|
|
}
|
|
|
|
|
|
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_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=2),
|
|
)
|
|
self.assertEqual(self.registry.list_rooms(protocol_version=3), [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()
|