from __future__ import annotations from http.client import HTTPConnection import json import socketserver import threading import unittest from netfishing_discovery.registry import RoomRegistry from netfishing_discovery.server import ( DiscoveryHTTPServer, ServerConfig, TraversalUDPServer, ) VALID_ROOM = { "room_name": "Pond Friends", "port": 7777, "current_players": 1, "max_players": 8, "game_version": "0.6.7-alpha", "protocol_version": 3, } class DiscoveryHTTPTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: config = ServerConfig( bind_host="127.0.0.1", bind_port=0, room_ttl_seconds=30.0, build_revision="test-build", ) cls.server = DiscoveryHTTPServer(config, RoomRegistry(ttl_seconds=30.0)) cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True) cls.thread.start() cls.port = cls.server.server_address[1] @classmethod def tearDownClass(cls) -> None: cls.server.shutdown() cls.server.server_close() cls.thread.join(timeout=2.0) def request( self, method: str, path: str, payload: dict[str, object] | None = None, token: str | None = None, ) -> tuple[int, dict[str, object] | None]: connection = HTTPConnection("127.0.0.1", self.port, timeout=2.0) headers: dict[str, str] = {} body = None if payload is not None: body = json.dumps(payload) headers["Content-Type"] = "application/json" if token is not None: headers["Authorization"] = f"Bearer {token}" connection.request(method, path, body=body, headers=headers) response = connection.getresponse() response_body = response.read() connection.close() decoded = json.loads(response_body) if response_body else None return response.status, decoded def test_room_lease_lifecycle(self) -> None: status, created = self.request("POST", "/v1/rooms", VALID_ROOM) self.assertEqual(status, 201) assert created is not None room = created["room"] assert isinstance(room, dict) room_id = room["room_id"] token = created["lease_token"] traversal = created["traversal"] assert isinstance(traversal, dict) self.server.registry.verify_endpoint( str(room_id), str(traversal["verification_token"]), "127.0.0.1", 7777, ) self.assertIsInstance(room_id, str) self.assertIsInstance(token, str) status, listing = self.request("GET", "/v1/rooms?protocol_version=3") self.assertEqual(status, 200) assert listing is not None listed_rooms = listing["rooms"] self.assertTrue(any(candidate["room_id"] == room_id for candidate in listed_rooms)) status, updated = self.request( "PUT", f"/v1/rooms/{room_id}", dict(VALID_ROOM, game_version="0.6.6-alpha"), str(token), ) self.assertEqual(status, 409) assert updated is not None self.assertEqual(updated["error"]["code"], "game_version_mismatch") self.assertEqual( updated["error"]["required_game_version"], "0.6.7-alpha" ) status, updated = self.request( "PUT", f"/v1/rooms/{room_id}", dict(VALID_ROOM, current_players=2), str(token), ) self.assertEqual(status, 200) assert updated is not None self.assertEqual(updated["room"]["current_players"], 2) status, _body = self.request("DELETE", f"/v1/rooms/{room_id}", token=str(token)) self.assertEqual(status, 204) def test_invalid_payload_returns_structured_error(self) -> None: status, body = self.request("POST", "/v1/rooms", dict(VALID_ROOM, port=70_000)) self.assertEqual(status, 400) assert body is not None self.assertEqual(body["error"]["code"], "invalid_room") def test_outdated_game_cannot_list_a_room(self) -> None: status, body = self.request( "POST", "/v1/rooms", dict(VALID_ROOM, game_version="0.6.6-alpha"), ) self.assertEqual(status, 409) assert body is not None error = body["error"] assert isinstance(error, dict) self.assertEqual(error["code"], "game_version_mismatch") self.assertEqual(error["required_game_version"], "0.6.7-alpha") self.assertIn("will not be listed", str(error["message"])) def test_empty_dedicated_room_can_be_listed(self) -> None: status, created = self.request( "POST", "/v1/rooms", dict(VALID_ROOM, current_players=0) ) self.assertEqual(status, 201) assert created is not None room = created["room"] assert isinstance(room, dict) traversal = created["traversal"] assert isinstance(traversal, dict) self.server.registry.verify_endpoint( str(room["room_id"]), str(traversal["verification_token"]), "127.0.0.1", 7777, ) self.assertEqual(room["current_players"], 0) self.request( "DELETE", f"/v1/rooms/{room['room_id']}", token=str(created["lease_token"]), ) def test_join_attempt_reports_observed_endpoint_to_host(self) -> None: status, created = self.request("POST", "/v1/rooms", VALID_ROOM) self.assertEqual(status, 201) assert created is not None room = created["room"] traversal = created["traversal"] assert isinstance(room, dict) assert isinstance(traversal, dict) room_id = str(room["room_id"]) self.server.registry.verify_endpoint( room_id, str(traversal["verification_token"]), "127.0.0.1", 7777, ) status, join = self.request( "POST", f"/v1/rooms/{room_id}/join-attempts" ) self.assertEqual(status, 201) assert join is not None self.server.registry.register_join_endpoint( str(join["join_token"]), "198.51.100.7", 49152 ) status, pending = self.request( "GET", f"/v1/rooms/{room_id}/join-attempts", token=str(created["lease_token"]), ) self.assertEqual(status, 200) assert pending is not None self.assertEqual( pending["endpoints"], [{"address": "198.51.100.7", "port": 49152}], ) def test_health_endpoint(self) -> None: status, body = self.request("GET", "/health") self.assertEqual(status, 200) assert body is not None self.assertEqual(body["status"], "ok") self.assertEqual(body["version"], "0.6.7-alpha") self.assertEqual(body["build_revision"], "test-build") def test_udp_rendezvous_does_not_spawn_per_packet_threads(self) -> None: self.assertFalse(issubclass(TraversalUDPServer, socketserver.ThreadingMixIn)) if __name__ == "__main__": unittest.main()