96 lines
3.4 KiB
Python
96 lines
3.4 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.4-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 = self.registry.create("203.0.113.10", VALID_ROOM)
|
||
|
|
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_expired_room_is_removed(self) -> None:
|
||
|
|
room, 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 = 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 = self.registry.create("203.0.113.10", VALID_ROOM)
|
||
|
|
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()
|