236 lines
7.7 KiB
Python
236 lines
7.7 KiB
Python
import unittest
|
|
|
|
from couchd.app import App, Config, ConfigError, Response, validate_bind_address
|
|
from couchd.discovery import DiscoveryError
|
|
|
|
|
|
class BindValidationTests(unittest.TestCase):
|
|
def test_rejects_loopback_address(self) -> None:
|
|
with self.assertRaises(ConfigError):
|
|
validate_bind_address("127.0.0.1")
|
|
|
|
def test_rejects_unspecified_address(self) -> None:
|
|
with self.assertRaises(ConfigError):
|
|
validate_bind_address("0.0.0.0")
|
|
|
|
def test_rejects_non_tailscale_address(self) -> None:
|
|
with self.assertRaises(ConfigError):
|
|
validate_bind_address("192.168.1.50")
|
|
|
|
def test_accepts_configured_tailscale_address(self) -> None:
|
|
self.assertEqual(validate_bind_address("100.64.0.15"), "100.64.0.15")
|
|
|
|
|
|
class AuthTests(unittest.TestCase):
|
|
def test_status_requires_bearer_token(self) -> None:
|
|
app = App(Config(bind_address="100.64.0.15", token="secret"))
|
|
response = app.handle_request("GET", "/status", headers={})
|
|
|
|
self.assertEqual(response, Response(401, {"error": "missing or invalid bearer token"}))
|
|
|
|
def test_status_returns_json_payload_when_authenticated(self) -> None:
|
|
app = App(
|
|
Config(bind_address="100.64.0.15", token="secret"),
|
|
status_provider=lambda: {
|
|
"state": "idle",
|
|
"pid": None,
|
|
"display": {"name": "HDMI-1", "mode": "1920x1080@60"},
|
|
"slippi_version": None,
|
|
"controller": {"connected": False, "name": "Xbox Wireless Controller"},
|
|
},
|
|
)
|
|
|
|
response = app.handle_request(
|
|
"GET",
|
|
"/status",
|
|
headers={"Authorization": "Bearer secret"},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.body["state"], "idle")
|
|
self.assertEqual(response.body["display"]["mode"], "1920x1080@60")
|
|
|
|
def test_status_accepts_safe_query_string(self) -> None:
|
|
app = App(
|
|
Config(bind_address="100.64.0.15", token="secret"),
|
|
status_provider=lambda: {"state": "idle", "pid": None},
|
|
)
|
|
|
|
response = app.handle_request(
|
|
"GET",
|
|
"/status?cache_bust=1",
|
|
headers={"Authorization": "Bearer secret"},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.body["state"], "idle")
|
|
|
|
|
|
class RouteTests(unittest.TestCase):
|
|
def test_launch_melee_uses_allowlisted_action(self) -> None:
|
|
calls: list[str] = []
|
|
|
|
class Launcher:
|
|
def launch_melee(self) -> dict[str, object]:
|
|
calls.append("melee")
|
|
return {"state": "launching", "pid": 4321}
|
|
|
|
def open_slippi(self) -> dict[str, object]:
|
|
calls.append("slippi")
|
|
return {"state": "launcher", "pid": 4322}
|
|
|
|
def stop(self) -> dict[str, object]:
|
|
calls.append("stop")
|
|
return {"state": "stopped", "pid": None}
|
|
|
|
app = App(
|
|
Config(bind_address="100.64.0.15", token="secret"),
|
|
launcher=Launcher(),
|
|
)
|
|
|
|
response = app.handle_request(
|
|
"POST",
|
|
"/launch/melee",
|
|
headers={"Authorization": "Bearer secret"},
|
|
)
|
|
|
|
self.assertEqual(calls, ["melee"])
|
|
self.assertEqual(response, Response(202, {"state": "launching", "pid": 4321}))
|
|
|
|
def test_open_slippi_uses_allowlisted_action(self) -> None:
|
|
calls: list[str] = []
|
|
|
|
class Launcher:
|
|
def launch_melee(self) -> dict[str, object]:
|
|
calls.append("melee")
|
|
return {"state": "launching", "pid": 4321}
|
|
|
|
def open_slippi(self) -> dict[str, object]:
|
|
calls.append("slippi")
|
|
return {"state": "launcher", "pid": 4322}
|
|
|
|
def stop(self) -> dict[str, object]:
|
|
calls.append("stop")
|
|
return {"state": "stopped", "pid": None}
|
|
|
|
app = App(
|
|
Config(bind_address="100.64.0.15", token="secret"),
|
|
launcher=Launcher(),
|
|
)
|
|
|
|
response = app.handle_request(
|
|
"POST",
|
|
"/open/slippi",
|
|
headers={"Authorization": "Bearer secret"},
|
|
)
|
|
|
|
self.assertEqual(calls, ["slippi"])
|
|
self.assertEqual(response, Response(202, {"state": "launcher", "pid": 4322}))
|
|
|
|
def test_stop_uses_allowlisted_action(self) -> None:
|
|
calls: list[str] = []
|
|
|
|
class Launcher:
|
|
def launch_melee(self) -> dict[str, object]:
|
|
calls.append("melee")
|
|
return {"state": "launching", "pid": 4321}
|
|
|
|
def open_slippi(self) -> dict[str, object]:
|
|
calls.append("slippi")
|
|
return {"state": "launcher", "pid": 4322}
|
|
|
|
def stop(self) -> dict[str, object]:
|
|
calls.append("stop")
|
|
return {"state": "stopped", "pid": None}
|
|
|
|
app = App(
|
|
Config(bind_address="100.64.0.15", token="secret"),
|
|
launcher=Launcher(),
|
|
)
|
|
|
|
response = app.handle_request(
|
|
"POST",
|
|
"/stop",
|
|
headers={"Authorization": "Bearer secret"},
|
|
)
|
|
|
|
self.assertEqual(calls, ["stop"])
|
|
self.assertEqual(response, Response(202, {"state": "stopped", "pid": None}))
|
|
|
|
def test_rejects_nonempty_body_for_action_route(self) -> None:
|
|
app = App(
|
|
Config(bind_address="100.64.0.15", token="secret"),
|
|
launcher=None,
|
|
)
|
|
|
|
response = app.handle_request(
|
|
"POST",
|
|
"/stop",
|
|
headers={"Authorization": "Bearer secret"},
|
|
body=b"x",
|
|
)
|
|
|
|
self.assertEqual(response, Response(400, {"error": "request body is not allowed"}))
|
|
|
|
def test_rejects_query_variant_of_unknown_route(self) -> None:
|
|
app = App(
|
|
Config(bind_address="100.64.0.15", token="secret"),
|
|
launcher=None,
|
|
)
|
|
|
|
response = app.handle_request(
|
|
"POST",
|
|
"/launch/unknown?x=1",
|
|
headers={"Authorization": "Bearer secret"},
|
|
)
|
|
|
|
self.assertEqual(response, Response(404, {"error": "not found"}))
|
|
|
|
def test_expected_launcher_failures_become_json_conflicts(self) -> None:
|
|
class Launcher:
|
|
def launch_melee(self) -> dict[str, object]:
|
|
raise DiscoveryError("ISO missing")
|
|
|
|
def open_slippi(self) -> dict[str, object]:
|
|
raise AssertionError("not expected")
|
|
|
|
def stop(self) -> dict[str, object]:
|
|
raise AssertionError("not expected")
|
|
|
|
app = App(
|
|
Config(bind_address="100.64.0.15", token="secret"),
|
|
launcher=Launcher(),
|
|
)
|
|
|
|
response = app.handle_request(
|
|
"POST",
|
|
"/launch/melee",
|
|
headers={"Authorization": "Bearer secret"},
|
|
)
|
|
|
|
self.assertEqual(response, Response(409, {"error": "ISO missing"}))
|
|
|
|
def test_unexpected_launcher_failures_become_json_service_errors(self) -> None:
|
|
class Launcher:
|
|
def launch_melee(self) -> dict[str, object]:
|
|
raise RuntimeError("boom")
|
|
|
|
def open_slippi(self) -> dict[str, object]:
|
|
raise AssertionError("not expected")
|
|
|
|
def stop(self) -> dict[str, object]:
|
|
raise AssertionError("not expected")
|
|
|
|
app = App(
|
|
Config(bind_address="100.64.0.15", token="secret"),
|
|
launcher=Launcher(),
|
|
)
|
|
|
|
response = app.handle_request(
|
|
"POST",
|
|
"/launch/melee",
|
|
headers={"Authorization": "Bearer secret"},
|
|
)
|
|
|
|
self.assertEqual(response, Response(503, {"error": "launcher action failed"}))
|