daniel156161
a2414f0b3d
Build and Push Docker Container / build-and-push (push) Successful in 2m18s
- Register Zenith 1.1 with balanced duel verdicts and forced-outcome overrides. - Prioritize permanent-trail space in Constrictor and health-aware hazard routes. - Deduct executor queue delays and return fallback directions on move timeouts. - Correlate response headers, replay timings, and sampled structured trace logs. - Move shared metrics publication off the response path without losing local counts. - Allow configurable tournament opponents and add a psql helper recipe. - Preserve empty latency values through compact replay encoding and migration. - Update Arena game links and use default assets for missing customizations. - Cover strategy, HTTP lifecycle, tracing, replay codecs, and dashboard behavior. - Validate with 260 passing Python tests, 4 skips, and 2 passing JavaScript tests.
126 lines
5.2 KiB
Python
126 lines
5.2 KiB
Python
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from server.database.compact.codec import (
|
|
decode_body, decode_latency, decode_move, decode_points, encode_body,
|
|
encode_latency, encode_move, encode_points, from_epoch_us, to_epoch_us,
|
|
)
|
|
from server.database.compact.records import PointsState, SnakeState, TurnRecord, compact_turn
|
|
from server.database.compact.replay_blob import pack_replay, unpack_replay
|
|
from server.database.compact.thinking import derived_values, restore_thinking, slim_thinking
|
|
|
|
def _body(*points):
|
|
return [{"x": x, "y": y} for x, y in points]
|
|
|
|
class TestCodec(unittest.TestCase):
|
|
def test_body_roundtrip_including_stacked_tail(self):
|
|
body = _body((5, 5), (5, 6), (4, 6), (4, 6))
|
|
encoded = encode_body(body)
|
|
self.assertEqual(decode_body(encoded), body)
|
|
# 1 format byte + head pair + 3 step codes
|
|
self.assertEqual(len(encoded), 6)
|
|
|
|
def test_body_roundtrip_with_wrapped_jump(self):
|
|
body = _body((0, 5), (10, 5), (9, 5))
|
|
self.assertEqual(decode_body(encode_body(body)), body)
|
|
|
|
def test_body_out_of_byte_range_falls_back_to_json(self):
|
|
body = _body((-1, 3), (300, 3))
|
|
self.assertEqual(decode_body(encode_body(body)), body)
|
|
|
|
def test_empty_body(self):
|
|
self.assertEqual(decode_body(encode_body([])), [])
|
|
|
|
def test_points_keep_none_as_unchanged(self):
|
|
self.assertIsNone(encode_points(None))
|
|
self.assertIsNone(decode_points(None))
|
|
self.assertEqual(decode_points(encode_points([])), [])
|
|
food = _body((1, 2), (10, 0))
|
|
self.assertEqual(decode_points(encode_points(food)), food)
|
|
|
|
def test_moves_and_latency(self):
|
|
for move in ("up", "down", "left", "right", None):
|
|
self.assertEqual(decode_move(encode_move(move)), move)
|
|
self.assertEqual(decode_latency(encode_latency("77")), "77")
|
|
self.assertEqual(decode_latency(encode_latency("")), "")
|
|
self.assertIsNone(decode_latency(encode_latency(None)))
|
|
self.assertIsNone(encode_latency("timeout"))
|
|
|
|
def test_epoch_microseconds_roundtrip(self):
|
|
value = datetime(2026, 11, 4, 21, 41, 48, 123456, tzinfo=timezone.utc)
|
|
self.assertEqual(from_epoch_us(to_epoch_us(value)), value)
|
|
self.assertEqual(from_epoch_us(to_epoch_us(value.isoformat())), value)
|
|
|
|
class TestThinking(unittest.TestCase):
|
|
def setUp(self):
|
|
self.derived = derived_values(
|
|
turn=12, move="up", you={"head": {"x": 1, "y": 2}, "health": 90, "length": 4},
|
|
food=[{"x": 0, "y": 0}], width=11, height=11, snake_count=2,
|
|
)
|
|
|
|
def test_strips_and_restores_derived_fields(self):
|
|
thinking = {
|
|
"turn": 12, "move": "up", "head": {"x": 1, "y": 2}, "health": 90,
|
|
"length": 4, "food_count": 1, "board_size": "11x11", "snakes": 2,
|
|
"scores": {"up": 1.5}, "mode": "duel",
|
|
}
|
|
slim = slim_thinking(thinking, self.derived)
|
|
self.assertEqual(set(slim), {"scores", "mode", "_derived"})
|
|
self.assertEqual(restore_thinking(slim, self.derived), thinking)
|
|
|
|
def test_keeps_values_that_differ_from_derived(self):
|
|
thinking = {"turn": 11, "health": 89, "snakes": True}
|
|
slim = slim_thinking(thinking, self.derived)
|
|
self.assertEqual(slim, thinking)
|
|
self.assertEqual(restore_thinking(slim, self.derived), thinking)
|
|
|
|
def test_payload_using_marker_key_is_stored_verbatim(self):
|
|
thinking = {"_derived": "mine", "turn": 12}
|
|
self.assertEqual(restore_thinking(slim_thinking(thinking, self.derived), self.derived), thinking)
|
|
|
|
def test_non_dict_payload_passes_through(self):
|
|
self.assertIsNone(slim_thinking(None, self.derived))
|
|
self.assertEqual(slim_thinking(["a"], self.derived), ["a"])
|
|
|
|
class TestCompactTurn(unittest.TestCase):
|
|
def _compact(self, turn, food, previous):
|
|
states = [SnakeState("me", turn, 90, 5, _body((1, 1)))]
|
|
return compact_turn(turn, None, "up", None, food, [], states, "me", 11, 11, previous)
|
|
|
|
def test_unchanged_food_is_stored_as_none_only_after_consecutive_turn(self):
|
|
food = _body((3, 3))
|
|
first, points = self._compact(1, food, None)
|
|
second, points = self._compact(2, list(food), points)
|
|
after_gap, _ = self._compact(4, list(food), points)
|
|
self.assertEqual(first.food, food)
|
|
self.assertIsNone(second.food)
|
|
self.assertEqual(second.hazards, None)
|
|
self.assertEqual(after_gap.food, food)
|
|
|
|
def test_previous_state_from_other_turn_is_not_used(self):
|
|
record, _ = self._compact(5, [], PointsState(5, [], []))
|
|
self.assertEqual(record.food, [])
|
|
|
|
class TestReplayBlob(unittest.TestCase):
|
|
def test_pack_roundtrip(self):
|
|
observed = datetime(2026, 11, 4, 12, 0, 0, 5, tzinfo=timezone.utc)
|
|
turns = [
|
|
TurnRecord(0, observed, None, None, _body((1, 1)), []),
|
|
TurnRecord(1, observed, "left", {"scores": {"left": 1.0}, "_derived": 3}, None, None),
|
|
]
|
|
states = [
|
|
SnakeState("me", 0, 100, None, _body((5, 5), (5, 5), (5, 5))),
|
|
SnakeState("me", 1, 99, 42, _body((4, 5), (5, 5), (5, 5))),
|
|
SnakeState("enemy", 1, 99, 7, _body((-1, 2))),
|
|
]
|
|
unpacked_turns, unpacked_states = unpack_replay(pack_replay(turns, states))
|
|
self.assertEqual(unpacked_turns, turns)
|
|
self.assertEqual(sorted(unpacked_states, key=lambda s: (s.snake_id, s.turn)), sorted(states, key=lambda s: (s.snake_id, s.turn)))
|
|
|
|
def test_unknown_format_is_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
unpack_replay(b"", 99)
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|