mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-11 00:14:00 +00:00
Josh Creek
64 lines
2.8 KiB
Python
64 lines
2.8 KiB
Python
"""Dependency-free structural checks for the versioned control-plane contract."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).parent
|
|
|
|
|
|
class ContractTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.openapi = json.loads((ROOT / "openapi.json").read_text())
|
|
cls.events = json.loads((ROOT / "websocket-events.json").read_text())
|
|
|
|
def test_openapi_is_versioned_and_has_core_surfaces(self):
|
|
self.assertEqual(self.openapi["openapi"], "3.1.0")
|
|
operations = {
|
|
operation["operationId"]
|
|
for path in self.openapi["paths"].values()
|
|
for operation in path.values()
|
|
if isinstance(operation, dict) and "operationId" in operation
|
|
}
|
|
self.assertTrue({
|
|
"createSteamSession", "getProfile", "createQueueTicket",
|
|
"heartbeatQueueTicket", "cancelQueueTicket", "acceptProposal",
|
|
"declineProposal", "getAssignment", "registerServer",
|
|
"submitMatchResult",
|
|
} <= operations)
|
|
|
|
def test_mutations_require_idempotency_and_revision(self):
|
|
parameters = self.openapi["components"]["parameters"]
|
|
self.assertEqual(parameters["IdempotencyKey"]["name"], "Idempotency-Key")
|
|
self.assertTrue(parameters["IdempotencyKey"]["required"])
|
|
self.assertEqual(parameters["ExpectedRevision"]["name"], "If-Match-Revision")
|
|
for path, methods in self.openapi["paths"].items():
|
|
for method, operation in methods.items():
|
|
if method not in {"post", "delete", "put", "patch"} or "operationId" not in operation:
|
|
continue
|
|
if operation["operationId"] == "createSteamSession":
|
|
continue
|
|
refs = {item.get("$ref") for item in operation.get("parameters", [])}
|
|
self.assertIn("#/components/parameters/IdempotencyKey", refs, path)
|
|
|
|
def test_state_vocabulary_is_shared(self):
|
|
queue_states = self.openapi["components"]["schemas"]["QueueState"]["enum"]
|
|
websocket_states = self.events["$defs"]["stateChanged"]["allOf"][1]["properties"]["state"]["enum"]
|
|
self.assertEqual(queue_states, websocket_states)
|
|
self.assertIn("ASSIGNMENT_READY", queue_states)
|
|
self.assertIn("RESULT_PENDING", queue_states)
|
|
|
|
def test_events_have_revisioned_envelopes_and_no_credentials(self):
|
|
envelope = self.events["$defs"]["envelope"]
|
|
self.assertEqual(envelope["required"], ["event", "revision", "resource_id", "occurred_at"])
|
|
serialized = json.dumps(self.events).lower()
|
|
self.assertNotIn("access_token", serialized)
|
|
self.assertNotIn("web_api_ticket", serialized)
|
|
self.assertNotIn("relay_ticket", serialized)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|