netfishing-discovery-server/tests/test_http_api.py

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