DroidDeck/tools/tests/test_game_exit_diagnostics.py

import contextlib
import io
import os
from pathlib import Path
import runpy
import signal
import subprocess
import sys
import tempfile
from unittest import mock
import unittest


BIN = Path(__file__).resolve().parents[1] / "linuxfs/overlay/usr/local/bin"
GAME_ENV = BIN / "droiddeck-game-env"
MODULE = runpy.run_path(str(GAME_ENV))


class GameExitDiagnosticsTest(unittest.TestCase):
    def run_child(self, body, app_id="42"):
        with contextlib.redirect_stderr(io.StringIO()) as said:
            status = MODULE["start"](
                [sys.executable, "-c", body], os.environ.copy(), None,
                diagnostic_app_id=app_id,
            )
            return status, said.getvalue()

    def test_clean_and_nonzero_exits_are_authoritative(self):
        for child_status in (0, 7):
            with self.subTest(status=child_status):
                status, said = self.run_child("import sys; sys.exit(%d)" % child_status)
                self.assertEqual(status, child_status)
                self.assertIn("event=game_start app_id=42", said)
                self.assertIn(
                    "event=game_exit app_id=42 raw_status=%d normalized_status=%d wrapper_status=%d signal=0" %
                    (child_status, child_status, child_status), said,
                )

    def test_signal_exit_is_normalized_and_records_the_signal(self):
        status, said = self.run_child("import os, signal; os.kill(os.getpid(), signal.SIGTERM)")
        self.assertEqual(status, 128 + signal.SIGTERM)
        self.assertIn(
            "event=game_exit app_id=42 raw_status=-15 normalized_status=143 wrapper_status=143 signal=15", said,
        )

    def test_xbox_exit_is_recorded_without_inventing_a_start(self):
        with contextlib.redirect_stderr(io.StringIO()) as said:
            MODULE["diagnostic_game_exit"]("42", -signal.SIGTERM, "xbox", 241)
        self.assertNotIn("event=game_start", said.getvalue())
        self.assertIn(
            "event=game_exit app_id=42 raw_status=-15 normalized_status=143 wrapper_status=241 signal=15 source=xbox",
            said.getvalue(),
        )

    def test_signal_handler_forwards_to_the_child(self):
        class Process:
            def __init__(self):
                self.forwarded = []

            def send_signal(self, signum):
                self.forwarded.append(signum)

            def wait(self):
                os.kill(os.getpid(), signal.SIGTERM)
                return -signal.SIGTERM

        process = Process()
        self.assertEqual(MODULE["wait_forwarding_signals"](process), 128 + signal.SIGTERM)
        self.assertEqual(process.forwarded, [signal.SIGTERM])

    def test_default_path_still_execs_in_place_without_diagnostic_output(self):
        called = []
        with mock.patch("os.execvpe", side_effect=lambda *args: called.append(args)), \
             contextlib.redirect_stderr(io.StringIO()) as said:
            MODULE["start"](["/game", "arg"], {"A": "B"}, None)
        self.assertEqual(called, [("/game", ["/game", "arg"], {"A": "B"})])
        self.assertNotIn("DROIDDECK_DIAGNOSTICS", said.getvalue())

    def test_failed_selected_launch_does_not_invent_a_start_or_exit(self):
        with mock.patch("subprocess.Popen", side_effect=FileNotFoundError("missing")), \
             contextlib.redirect_stderr(io.StringIO()) as said:
            with self.assertRaises(FileNotFoundError):
                MODULE["start"](["/missing"], {}, None, diagnostic_app_id="42")
        self.assertNotIn("DROIDDECK_DIAGNOSTICS", said.getvalue())

    def test_cleanup_runs_after_a_diagnostic_child(self):
        cleaned = []
        with contextlib.redirect_stderr(io.StringIO()):
            status = MODULE["start"](
                [sys.executable, "-c", "pass"], os.environ.copy(), None,
                after=lambda: cleaned.append(True), diagnostic_app_id="42",
            )
        self.assertEqual(status, 0)
        self.assertEqual(cleaned, [True])

    def test_diagnostic_install_script_keeps_success_recording(self):
        recorded = []
        with mock.patch.dict(MODULE["start"].__globals__, record_install_scripts=lambda digest, env: recorded.append((digest, env))), \
             contextlib.redirect_stderr(io.StringIO()):
            status = MODULE["start"](
                [sys.executable, "-c", "pass"], {"CUSTOM": "kept"}, "digest",
                diagnostic_app_id="42",
            )
        self.assertEqual(status, 0)
        self.assertEqual(recorded, [("digest", {"CUSTOM": "kept"})])

    def test_cleanup_path_keeps_the_original_install_recording_precedence(self):
        recorded = []
        with mock.patch.dict(MODULE["start"].__globals__, record_install_scripts=lambda digest, env: recorded.append((digest, env))):
            status = MODULE["start"](
                [sys.executable, "-c", "pass"], {"CUSTOM": "kept"}, "digest",
                after=lambda: None,
            )
        self.assertEqual(status, 0)
        self.assertEqual(recorded, [])

    def test_main_only_traces_valid_waitforexitandrun_games(self):
        with tempfile.TemporaryDirectory(prefix="game-exit-diagnostics-") as directory:
            root = Path(directory)
            probe = root / "probe"
            probe.write_text("#!" + sys.executable + "\nimport sys\nsys.exit(7)\n")
            probe.chmod(0o755)
            base = dict(os.environ, HOME=directory, PATH=os.defpath,
                        DROIDDECK_ENHANCED_DIAGNOSTICS="1")
            valid = subprocess.run(
                [sys.executable, str(GAME_ENV), str(probe), "waitforexitandrun"],
                env={**base, "STEAM_COMPAT_DATA_PATH": "/compatdata/42"},
                capture_output=True, text=True, timeout=15,
            )
            self.assertEqual(valid.returncode, 7, valid.stderr)
            self.assertIn("event=game_start app_id=42", valid.stderr)
            self.assertIn(
                "event=game_exit app_id=42 raw_status=7 normalized_status=7 wrapper_status=7 signal=0", valid.stderr,
            )

            for verb, prefix in (("run", "/compatdata/42"), ("waitforexitandrun", "/compatdata/not-a-game")):
                result = subprocess.run(
                    [sys.executable, str(GAME_ENV), str(probe), verb],
                    env={**base, "STEAM_COMPAT_DATA_PATH": prefix},
                    capture_output=True, text=True, timeout=15,
                )
                self.assertEqual(result.returncode, 7, result.stderr)
                self.assertNotIn("DROIDDECK_DIAGNOSTICS", result.stderr)

    def test_main_records_xbox_result_without_changing_it(self):
        launch = lambda command, env: 23
        with tempfile.TemporaryDirectory(prefix="xbox-exit-diagnostics-") as directory, \
             mock.patch.dict(os.environ, {
                 "HOME": directory,
                 "PATH": os.defpath,
                 "STEAM_COMPAT_DATA_PATH": "/compatdata/42",
                 "DROIDDECK_ENHANCED_DIAGNOSTICS": "1",
             }, clear=True), \
             mock.patch("sys.argv", ["droiddeck-game-env", "/proton", "waitforexitandrun", "game.exe"]), \
             mock.patch("runpy.run_path", return_value={"launch": launch}), \
             contextlib.redirect_stderr(io.StringIO()) as said:
            status = MODULE["main"]()
        self.assertEqual(status, 23)
        self.assertNotIn("event=game_start", said.getvalue())
        self.assertIn(
            "event=game_exit app_id=42 raw_status=23 normalized_status=23 wrapper_status=23 signal=0 source=xbox",
            said.getvalue(),
        )

    def test_main_preserves_negative_xbox_result_and_reports_wrapper_status(self):
        launch = lambda command, env: -signal.SIGTERM
        with tempfile.TemporaryDirectory(prefix="xbox-signal-diagnostics-") as directory, \
             mock.patch.dict(os.environ, {
                 "HOME": directory,
                 "PATH": os.defpath,
                 "STEAM_COMPAT_DATA_PATH": "/compatdata/42",
                 "DROIDDECK_ENHANCED_DIAGNOSTICS": "1",
             }, clear=True), \
             mock.patch("sys.argv", ["droiddeck-game-env", "/proton", "waitforexitandrun", "game.exe"]), \
             mock.patch("runpy.run_path", return_value={"launch": launch}), \
             contextlib.redirect_stderr(io.StringIO()) as said:
            status = MODULE["main"]()
        self.assertEqual(status, -signal.SIGTERM)
        self.assertIn(
            "event=game_exit app_id=42 raw_status=-15 normalized_status=143 wrapper_status=241 signal=15 source=xbox",
            said.getvalue(),
        )


if __name__ == "__main__":
    unittest.main()