118 lines
4 KiB
Python
118 lines
4 KiB
Python
from __future__ import annotations
|
|
|
|
from http.client import HTTPConnection
|
|
import json
|
|
import threading
|
|
import unittest
|
|
|
|
from netfishing_discovery.registry import RoomRegistry
|
|
from netfishing_discovery.server import DiscoveryHTTPServer, ServerConfig
|
|
|
|
|
|
VALID_ROOM = {
|
|
"room_name": "Pond Friends",
|
|
"port": 7777,
|
|
"current_players": 1,
|
|
"max_players": 8,
|
|
"game_version": "0.6.4-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)
|
|
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"]
|
|
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, 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_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)
|
|
self.assertEqual(room["current_players"], 0)
|
|
self.request(
|
|
"DELETE",
|
|
f"/v1/rooms/{room['room_id']}",
|
|
token=str(created["lease_token"]),
|
|
)
|
|
|
|
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")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|