netfishing-discovery-server/tests/test_registry.py

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.8.1-alpha",
    "protocol_version": 5,
}


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=3),
        )
        self.assertEqual(self.registry.list_rooms(protocol_version=5), [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()