from __future__ import annotations from contextlib import redirect_stderr, redirect_stdout import io from pathlib import Path import tempfile import unittest from unittest.mock import MagicMock from ptlinux.application.commands.play import ( game_exit_metadata, game_failure_reason, game_readiness_signal, play_locked, print_game_outcome, runtime_issue_summary, ) from ptlinux.command import CommandOutcome from ptlinux.depot import DepotStartOption from ptlinux.paths import AppPaths from ptlinux.runtime import RuntimeSelection from ptlinux.settings import Settings def missing_status_message(status: int) -> str: return ( f"Gamescope exited without a recorded game status (compositor status {status}); " "the game's exit result is unknown.\n" ) class GamescopeSessionOutcomeTests(unittest.TestCase): def test_missing_status_is_not_attributed_to_the_game_or_umu(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) paths = AppPaths(*(root / name for name in ("data", "config", "state", "cache"))) paths.runtime_view.mkdir(parents=True) api = MagicMock() api.game_import_requires_verification.return_value = False api._load_or_default_settings.return_value = Settings(runtime="ge-proton") api.is_running_inside_steam_runtime.return_value = False api.read_cached_depot_launch_catalog.return_value = None api.read_current_depot_file_catalog.return_value = None api._current_game_log.return_value = None api._current_game_crash.return_value = None api.select_runtime.return_value = RuntimeSelection("ge-proton", "umu-run", None, None) api.select_depot_start_option.return_value = DepotStartOption("Babel", "Game.exe", "PTEU_REBORN") api._runtime_issue_summary.side_effect = runtime_issue_summary api._game_exit_metadata.side_effect = game_exit_metadata for ready in (True, False): for compositor_status, wrapper_status in ((0, 1), (-11, 139), (247, 247), (0, 0)): with self.subTest(ready=ready, compositor=compositor_status, wrapper=wrapper_status): api.current_telemetry.return_value.reset_mock() api._game_readiness_signal.return_value = "window_ready" if ready else None api.run_game.return_value = CommandOutcome( [], wrapper_status, "", missing_status_message(compositor_status), 120.0, False, False, ) self.assertEqual(play_locked(api, paths, None, False, False, False, True, reuse_prepared=True), 1) report = api._emit.call_args.args[0] self.assertFalse(report["ok"]) self.assertEqual(report["launch_succeeded"], ready) event_name = "game_exited" if ready else "game_launch_failed" event = next(call for call in api.current_telemetry.return_value.send_event.call_args_list if call.args[0] == event_name) metadata = event.kwargs["metadata"] self.assertEqual(metadata["failure_reason"], "gamescope_game_status_missing") self.assertFalse(metadata["game_exit_status_available"]) self.assertEqual(metadata["exit_status_source"], "gamescope_supervisor") self.assertEqual(metadata["gamescope_exit_code"], compositor_status) self.assertEqual(metadata["exit_code"], wrapper_status) self.assertNotIn("signal_inferred", metadata) self.assertNotIn("signal", metadata) self.assertIn("before the game reported its exit status", event.kwargs["message"]) self.assertNotIn("Game client exited with status", event.kwargs["message"]) def test_truncated_output_retains_observed_shutdown_status(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) paths = AppPaths(*(root / name for name in ("data", "config", "state", "cache"))) paths.runtime_view.mkdir(parents=True) api = MagicMock() api.game_import_requires_verification.return_value = False api._load_or_default_settings.return_value = Settings(runtime="ge-proton") api.is_running_inside_steam_runtime.return_value = False api.read_cached_depot_launch_catalog.return_value = None api.read_current_depot_file_catalog.return_value = None api._current_game_log.return_value = None api._current_game_crash.return_value = None api.select_runtime.return_value = RuntimeSelection("ge-proton", "umu-run", None, None) api.select_depot_start_option.return_value = DepotStartOption("Babel", "Game.exe", "PTEU_REBORN") api._runtime_issue_summary.side_effect = runtime_issue_summary api._game_readiness_signal.side_effect = game_readiness_signal api._gamescope_presentation_failure.return_value = None api.parse_window_diagnostic.return_value = None api._game_exit_metadata.side_effect = game_exit_metadata def run(*args, **kwargs): observe = kwargs["output_line_filter"] observe("Game window ready: PristonTale EU\n") observe(missing_status_message(247)) return CommandOutcome([], 247, "", "final child cleanup output\n", 120.0, False, False) api.run_game.side_effect = run self.assertEqual(play_locked(api, paths, None, False, False, False, True, reuse_prepared=True), 1) event = next(call for call in api.current_telemetry.return_value.send_event.call_args_list if call.args[0] == "game_exited") metadata = event.kwargs["metadata"] self.assertEqual(metadata["failure_reason"], "gamescope_game_status_missing") self.assertEqual(metadata["gamescope_exit_code"], 247) self.assertNotIn("signal_inferred", metadata) self.assertTrue(metadata["game_window_ready"]) def test_unobserved_window_is_not_readiness_evidence(self) -> None: for message in ( "Game window ready: unmanaged PristonTale EU\n", "Game window monitoring unavailable; process started: PristonTale EU\n", ): with self.subTest(message=message): self.assertIsNone(game_readiness_signal(message)) self.assertFalse(runtime_issue_summary(message)["game_window_ready"]) self.assertEqual(game_readiness_signal(message + "Game window ready: PristonTale EU\n"), "window_ready") def test_human_output_distinguishes_unknown_status_in_old_and_new_logs(self) -> None: messages = ( missing_status_message(0), missing_status_message(-11), "Gamescope exited without a recorded game status; the game session was interrupted.\n", ) for message in messages: with self.subTest(message=message): outcome = CommandOutcome([], 1, "", message, 120.0, False, False) issues = runtime_issue_summary(message) output = io.StringIO() with redirect_stdout(output), redirect_stderr(output): print_game_outcome(outcome, None, issues) self.assertIn("before the game reported its exit status", output.getvalue()) self.assertNotIn("Game client exited", output.getvalue()) self.assertNotIn("crashed", output.getvalue()) def test_missing_status_preserves_stronger_failure_evidence(self) -> None: cases = ( ({"runtime_out_of_memory": True}, None, False, "direct3d_resource_exhaustion"), (None, {"path": "/tmp/crash_game.dat"}, False, "client_crash_dump"), (None, None, True, "session_timeout"), ) for native_log, crash, timed_out, reason in cases: with self.subTest(reason=reason): outcome = CommandOutcome([], 1, "", missing_status_message(0), 120.0, timed_out, False) self.assertEqual(game_failure_reason(outcome, native_log, crash, {}), reason) def test_human_output_keeps_crash_and_timeout_evidence(self) -> None: for timed_out, crash, expected in ( (False, {"path": "/tmp/crash_game.dat"}, "crashed and wrote a dump"), (True, None, "timed out and was stopped"), ): with self.subTest(expected=expected): outcome = CommandOutcome([], 1, "", missing_status_message(0), 120.0, timed_out, False) output = io.StringIO() with redirect_stderr(output): print_game_outcome(outcome, None, runtime_issue_summary(outcome.stderr), crash) self.assertIn(expected, output.getvalue()) self.assertNotIn("before the game reported", output.getvalue()) if __name__ == "__main__": unittest.main()