import pathlib import re import shlex import subprocess import tempfile import unittest DEPLOY = pathlib.Path(__file__).resolve().parent SCRIPT_PATH = DEPLOY / "w3cs-criu-build-images.sh" SERVICE_PATH = DEPLOY / "w3cs-live@.service" CPU_DEPLOY_PATH = DEPLOY / "deploy-cpu-server.sh" TIMER_PATH = DEPLOY / "w3cs-criu-build-images.timer" SCRIPT = SCRIPT_PATH.read_text(encoding="utf-8") SERVICE = SERVICE_PATH.read_text(encoding="utf-8") CPU_DEPLOY = CPU_DEPLOY_PATH.read_text(encoding="utf-8") TIMER = TIMER_PATH.read_text(encoding="utf-8") def function_source(name: str) -> str: match = re.search( rf"^{re.escape(name)}\(\) \{{\n.*?^\}}\n", SCRIPT, flags=re.MULTILINE | re.DOTALL, ) if not match: raise AssertionError(f"missing shell function: {name}") return match.group(0) class CriuImageReleaseTests(unittest.TestCase): PROXY_ENV = { "W3CS_ADAPTIVE_QUALITY": "1", "W3CS_BLOBREF": "1", "W3CS_COMPACT_DRAW": "1", "W3CS_OCTA": "1", "W3CS_PACKED_POS": "1", "W3CS_QUANT": "1", "W3CS_QUALITY_LEVEL": "8", "W3CS_RAIN_QUAD": "1", "W3CS_TEXDELTA": "1", "W3CS_TYPED_ATTR": "1", } def test_claim_image_and_cold_launch_enable_the_same_proxy_features(self): image_env = function_source("write_game_env") for name, value in self.PROXY_ENV.items(): with self.subTest(name=name): self.assertIn(f"\n{name}={value}\n", image_env) self.assertIn(f"Environment={name}={value}\n", SERVICE) def test_stale_images_are_invalidated_before_the_global_seat_restart(self): command = ( 'W3CS_ENABLE_131="$ENABLE_131" W3CS_VIDEO_SEATS="$VIDEO_SEATS" \\\n' " /usr/local/sbin/w3cs-criu-build-images" ) invocations = [ match.start() for match in re.finditer(re.escape(command), CPU_DEPLOY) ] self.assertEqual(2, len(invocations)) restart = CPU_DEPLOY.index("systemctl restart $SEAT_UNITS") self.assertLess(invocations[0], restart) self.assertLess(restart, invocations[1]) # The first pass must abort deployment instead of reopening the stale # checkpoint race. The second pass preserves first-provision builds and # can safely fall back to cold launch because pre-invalidation ran. first_call = "\n".join(CPU_DEPLOY[invocations[0]:].splitlines()[:2]) self.assertNotIn("||", first_call) self.assertIn("|| \\\n", CPU_DEPLOY[invocations[1]:]) def test_daily_timer_rechecks_system_library_identity(self): fingerprint = function_source("fingerprint") identity = function_source("system_library_identity") self.assertIn('echo "syslibs=$(system_library_identity)"', fingerprint) self.assertIn("/usr/lib/x86_64-linux-gnu", identity) self.assertIn("/usr/lib/i386-linux-gnu", identity) self.assertIn("OnCalendar=*-*-* 07:20:00", TIMER) self.assertIn("Persistent=true", TIMER) self.assertIn('"$APP/deploy/w3cs-criu-build-images.timer"', CPU_DEPLOY) self.assertIn("w3cs-criu-build-images.timer >/dev/null", CPU_DEPLOY) def run_busy_seat(self, fingerprint: str, *, war3_manifest: bool = True): with tempfile.TemporaryDirectory() as directory: root = pathlib.Path(directory) image = root / "images" / "native-1285__tft" / "seat1" (image / "generic").mkdir(parents=True) (image / "generic" / "inventory.img").write_bytes(b"inventory") if war3_manifest: (image / "war3.json").write_text("{}", encoding="utf-8") (image / "meta.json").write_text( '{"fingerprint": "' + fingerprint + '"}', encoding="utf-8" ) lab = root / "lab" lab.mkdir() shell = f""" set -euo pipefail MATRIX=('native-1285|tft|Warcraft III.exe|placeholder|default') BUILD_PROFILE_FILTER= IMAGES={shlex.quote(str(root / 'images'))} LAB={shlex.quote(str(lab))} CRIU_ROOT={shlex.quote(str(root / 'criu'))} fingerprint() {{ printf '%s\\n' current-fingerprint; }} curl() {{ printf '%s\\n' '{{\"busy\":true}}'; }} {function_source('image_is_current')} {function_source('build_seat')} build_seat 1 test -e {shlex.quote(str(image))} """ result = subprocess.run( ["bash"], input=shell, text=True, capture_output=True ) return result def test_busy_seat_drops_a_fingerprint_mismatched_image_before_deferral(self): result = self.run_busy_seat("stale-fingerprint") self.assertNotEqual(0, result.returncode, result.stdout + result.stderr) self.assertIn("DROP native-1285/tft seat1", result.stdout) self.assertIn("DEFER seat1", result.stdout) def test_busy_seat_keeps_a_complete_current_image(self): result = self.run_busy_seat("current-fingerprint") self.assertEqual(0, result.returncode, result.stdout + result.stderr) self.assertNotIn("DROP ", result.stdout) self.assertIn("DEFER seat1", result.stdout) def test_busy_seat_drops_an_incomplete_image_even_when_fingerprint_matches(self): result = self.run_busy_seat("current-fingerprint", war3_manifest=False) self.assertNotEqual(0, result.returncode, result.stdout + result.stderr) self.assertIn("DROP native-1285/tft seat1", result.stdout) def test_invalid_seat_is_rejected_before_a_claim_path_is_touched(self): build_seat = function_source("build_seat") validation = 'if ! [[ "$seat" =~ ^[1-9][0-9]*$ ]]; then' self.assertIn(validation, build_seat) self.assertLess(build_seat.index(validation), build_seat.index("rm -rf -- \"$img\"")) def run_free_seat_build(self, *, fail_after_stop: bool): with tempfile.TemporaryDirectory() as directory: root = pathlib.Path(directory) lab = root / "lab" replay = lab / "replays" / "placeholder.w3g" replay.parent.mkdir(parents=True) replay.write_bytes(b"replay") service_log = root / "systemctl.log" sleep_result = "return 17" if fail_after_stop else "return 0" shell = f""" set -euo pipefail MATRIX=('native-1311-d3d9|tft|Warcraft III.exe|placeholder|131') BUILD_PROFILE_FILTER= IMAGES={shlex.quote(str(root / 'images'))} LAB={shlex.quote(str(lab))} CRIU_ROOT={shlex.quote(str(root / 'criu'))} BUILDER=/unused-builder fingerprint() {{ printf '%s\\n' current-fingerprint; }} curl() {{ printf '%s\\n' '{{\"busy\":false}}'; }} systemctl() {{ printf '%s %s\\n' "$1" "$2" >> {shlex.quote(str(service_log))}; }} sleep() {{ {sleep_result}; }} profile_view() {{ printf '%s\\n' /game; }} profile_prefix() {{ printf '%s\\n' /prefix; }} game_env_file() {{ printf '%s\\n' /env; }} profile_d3d_overrides() {{ printf '%s\\n' overrides; }} layer_base_id() {{ printf '%s\\n' layer; }} ensure_131_prefix() {{ return 0; }} env() {{ return 1; }} {function_source('image_is_current')} {function_source('build_seat')} build_seat 1 """ result = subprocess.run( ["bash"], input=shell, text=True, capture_output=True ) calls = service_log.read_text(encoding="utf-8").splitlines() return result, calls def test_unexpected_failure_after_stop_restarts_once_and_propagates(self): result, calls = self.run_free_seat_build(fail_after_stop=True) self.assertEqual(17, result.returncode, result.stdout + result.stderr) self.assertEqual( ["stop w3cs-live@1", "start w3cs-live@1"], calls ) def test_handled_builder_failure_restarts_once_and_returns_success(self): result, calls = self.run_free_seat_build(fail_after_stop=False) self.assertEqual(0, result.returncode, result.stdout + result.stderr) self.assertEqual( ["stop w3cs-live@1", "start w3cs-live@1"], calls ) if __name__ == "__main__": unittest.main() class VideoLaneImageTests(unittest.TestCase): """A video seat's claim image renders: its baked environment must turn the proxy into clock-only mode and keep rasterization on, while the command seats keep the headless no-raster recorder.""" def test_video_seats_get_a_clock_only_rendering_environment(self): image_env = function_source("write_game_env") video = image_env[image_env.index('if [ "$lane" = video ]'): image_env.index(" return\n")] for line in ("W3_D3D9_CLOCK_ONLY=1", "W3_D3D9_NO_RASTER=0", "W3_D3D9_HEADLESS_FAST=0", "W3_D3D9_CAPTURE_FPS=10"): self.assertIn(f"\n{line}\n", video) for flag in ("W3CS_QUANT", "W3CS_ADAPTIVE_QUALITY", "W3CS_TEXDELTA"): self.assertNotIn(flag, video) command = image_env[image_env.index(" return\n"):] self.assertIn("\nW3_D3D9_NO_RASTER=1\n", command) self.assertIn("\nW3_D3D9_HEADLESS_FAST=1\n", command) self.assertNotIn("CLOCK_ONLY", command) def test_lane_is_part_of_the_fingerprint_and_raster_threads(self): self.assertIn('echo "lane=$(seat_lane "$seat")"', function_source("fingerprint")) build_seat = function_source("build_seat") self.assertIn('[ "$(seat_lane "$seat")" = video ] && raster_threads=2', build_seat) self.assertIn('W3_CLASSIC_LLVMPIPE_THREADS="$raster_threads"', build_seat) def test_video_seat_list_comes_from_the_deploy_or_the_recorded_file(self): self.assertIn('VIDEO_SEATS_FILE=/etc/w3cs/video-seats', SCRIPT) self.assertIn("W3CS_VIDEO_SEATS-$(cat \"$VIDEO_SEATS_FILE\"", SCRIPT) self.assertIn("printf '%s\\n' \"$VIDEO_SEATS\" > /etc/w3cs/video-seats", CPU_DEPLOY) # The seat unit stays one template; the lane is a drop-in. self.assertIn("Environment=W3CS_LANE=video", CPU_DEPLOY) self.assertIn('lane.conf', CPU_DEPLOY) self.assertNotIn("W3CS_LANE", SERVICE) launch = (DEPLOY / "w3cs-seat-launch.sh").read_text(encoding="utf-8") for line in ("export W3_D3D9_NO_RASTER=0", "export W3_D3D9_HEADLESS_FAST=0", "export W3_D3D9_CLOCK_ONLY=1", "export W3_D3D9_CAPTURE_FPS=10"): self.assertIn(line, launch) live = (DEPLOY.parent / "native" / "lab-live.sh").read_text(encoding="utf-8") self.assertIn("--lane video", live) def test_deploy_installs_x264_and_opens_every_seat_ice_range(self): self.assertIn("gstreamer1.0-plugins-ugly", CPU_DEPLOY) self.assertIn("ufw allow 40000:40999/udp", CPU_DEPLOY) self.assertNotIn("ufw allow 40000:40599/udp", CPU_DEPLOY)