chore: merge the round 5 installer fixes

tests/test_install_experience.py

@@ -565,6 +565,35 @@ class AFolderThisUserCannotEnter(unittest.TestCase):
                 os.chmod(locked, 0o755)
         self.assertEqual(state, "wrong_hash")
 
+    @unittest.skipIf(os.name == "nt" or os.geteuid() == 0,
+                     "a folder without search permission is POSIX")
+    def test_a_bios_folder_under_it_is_refused_in_words(self):
+        """--dest /root/.config/retroarch/system without sudo: _missing_dest
+        ended --check and the install on a PermissionError traceback."""
+        with FakeRepository() as repo:
+            lock = repo.root / "lock"
+            lock.mkdir()
+            repo.dest = lock / "system"
+            os.chmod(lock, 0o000)
+            try:
+                runs = {"check": repo.run("--check"), "install": repo.run(),
+                        "events": repo.run("--events")}
+            finally:
+                os.chmod(lock, 0o755)
+            self.assertFalse(repo.dest.exists())
+        for name, proc in runs.items():
+            with self.subTest(run=name):
+                self.assertNotIn("Traceback", proc.stderr)
+                self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
+                self.assertIn(f"this user cannot open {lock} (Permission denied)",
+                              proc.stderr)
+                self.assertIn("choose another folder", proc.stderr)
+        events = [json.loads(line) for line in runs["events"].stdout.splitlines()]
+        errors = [e for e in events if e["event"] == "error"]
+        self.assertEqual([(e["stage"], e["reason"]) for e in errors],
+                         [("dest", "unreadable")])
+        self.assertIn(str(lock), errors[0]["message"])
+
 
 class OutputToAFileAsksNothing(unittest.TestCase):
     """`install.py > log.txt` at a terminal: the menu went into the file
@@ -589,6 +618,40 @@ class OutputToAFileAsksNothing(unittest.TestCase):
         self.assertNotIn("Enter) install everything", proc.stdout)
         self.assertEqual(len(installed), len(FakeRepository.PAYLOADS))
 
+    @unittest.skipIf(os.name == "nt", "a pseudo-terminal for stdin")
+    def test_a_pipe_is_named_a_pipe(self):
+        """`install.py | tee log.txt` was told its output went to a file."""
+        import pty
+
+        master, slave = pty.openpty()
+        self.addCleanup(os.close, master)
+        self.addCleanup(os.close, slave)
+        with FakeRepository() as repo:
+            piped = subprocess.run(
+                [*repo.command(), "--dry-run"], env=repo.env(), stdin=slave,
+                capture_output=True, text=True, timeout=60, check=False)
+            log = repo.root / "log.txt"
+            with open(log, "w", encoding="utf-8") as handle:
+                filed = subprocess.run(
+                    [*repo.command(), "--dry-run"], env=repo.env(), stdin=slave,
+                    stdout=handle, stderr=subprocess.PIPE, text=True, timeout=60,
+                    check=False)
+        self.assertIn("Output goes to a pipe, so nothing is asked", piped.stderr)
+        self.assertIn("run it without the pipe to answer", piped.stderr)
+        self.assertIn("Output goes to a file, so nothing is asked", filed.stderr)
+        self.assertIn("run it without the redirection to answer", filed.stderr)
+        for proc in (piped, filed):
+            self.assertIn("Add --no-input to say so", proc.stderr)
+
+    def test_the_kind_of_output(self):
+        self.assertEqual(install._output_kind(io.StringIO()), "")
+        read, write = os.pipe()
+        with open(read, "rb") as reader, open(write, "w") as writer:
+            self.assertEqual(install._output_kind(writer), "pipe")
+            del reader
+        with tempfile.TemporaryFile("w", dir=REPO_ROOT / "tmp") as handle:
+            self.assertEqual(install._output_kind(handle), "file")
+
 
 class _Tty(io.StringIO):
     def isatty(self) -> bool:

tests/test_install_tui.py

@@ -29,6 +29,7 @@ import tempfile
 import termios
 import threading
 import time
+import types
 import unittest
 import warnings
 from pathlib import Path, PureWindowsPath
@@ -2594,6 +2595,60 @@ class IdleExit(unittest.TestCase):
         self.assertEqual(install.tui_farewell(state),
                          ["RetroBIOS: No button pressed, closing."])
 
+    def test_an_error_screen_closes_with_its_code(self):
+        """Closed idle on "No internet: offline mode", the launcher read 0
+        and showed nothing."""
+        state = fresh_state()
+        state.stack = ["setup", "checking"]
+        install._fail(state, ["Give the file list with --manifest FILE."], 1)
+        state.error_title = "No internet: offline mode"
+        self.assertEqual(self.idle_close(state), 1)
+        self.assertEqual(install.tui_farewell(state), [
+            "RetroBIOS: No button pressed, closing.",
+            "RetroBIOS: No internet: offline mode",
+            "Give the file list with --manifest FILE."])
+
+    def test_a_result_closes_with_the_install_code(self):
+        state = fresh_state()
+        state.stack = ["setup", "result"]
+        state.code = 2
+        state.summary = {"event": "summary", "mode": "install", "installed": 1,
+                         "up_to_date": 0, "pending": 0,
+                         "failed": [{"dest": "a.bin", "reason": "disk full"}],
+                         "skipped": [], "left_out": [], "kept": [], "interrupted": False}
+        state.dest = SYSTEM_DIR
+        self.assertEqual(self.idle_close(state), 2)
+        farewell = install.tui_farewell(state)
+        self.assertEqual(farewell[:2], ["RetroBIOS: No button pressed, closing.",
+                                        "RetroBIOS: Finished with errors"])
+
+    def test_any_other_screen_closes_with_0(self):
+        state = fresh_state()
+        state.stack = ["setup"]
+        self.assertEqual(self.idle_close(state), 0)
+
+    def idle_close(self, state: install.TuiState) -> int:
+        """run_tui on state, no key ever pressed, closed by the idle timer."""
+        class Quiet(Crash.FakeTerminal):
+            def keys(self, timeout):
+                time.sleep(0.01)
+                return []
+
+        out = io.StringIO()
+        work = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp"))
+        self.addCleanup(shutil.rmtree, work, True)
+        with mock.patch.object(install, "_TuiTerminal", Quiet), \
+                mock.patch.object(install.tempfile, "mkdtemp", return_value=str(work)), \
+                mock.patch.object(install, "tui_start", return_value=(state, [])), \
+                mock.patch.object(install, "_open_local_sources"), \
+                mock.patch.object(install, "detect_os", return_value="linux"), \
+                mock.patch.dict(os.environ, {install.TUI_IDLE_ENV: "0.05"}), \
+                mock.patch.object(sys, "stdout", out), mock.patch.object(sys, "stderr", out):
+            code = install.run_tui(_args())
+        self.assertTrue(state.idle_closed)
+        self.assertIn("No button pressed, closing.", out.getvalue())
+        return code
+
 
 class AReadErrorIsNotAClosedWindow(unittest.TestCase):
     """Any EIO read as the window closing: a failing SD card answers EIO
@@ -2777,6 +2832,257 @@ class Crash(unittest.TestCase):
         self.assertNotIn("Traceback", err.getvalue() + out.getvalue())
 
 
+class TheSaveFolderLineFollowsTheChoice(unittest.TestCase):
+    """The save-folder line was the check's: with the system owning the
+    save file unticked, the result named a folder nothing was written to,
+    and kept, the Confirm screen did not name it."""
+
+    SAVES = {"event": "saves", "platform": "retroarch", "files": 1,
+             "folder": "/home/user/MySaves/<core>"}
+    LINE = "1 file read from RetroArch's save directory goes to ~/MySaves/<core>"
+
+    def said(self, lines) -> str:
+        return " ".join(" ".join(lines).split())
+
+    def run_to_result(self, confirm_saves: bool, install_saves: bool) -> install.TuiState:
+        state = checked_state()
+        state.saves = dict(self.SAVES)
+        state.must_confirm = True
+        state.stack.append("confirm")
+        install._start_run(state, "confirm")
+        dry = {**CHECK, "files": 3, "missing": 3, "wrong_hash": 0}
+        for event in [*([self.SAVES] if confirm_saves else []), dry]:
+            install.tui_receive(state, ("event", state.run_id, event))
+        install.tui_receive(state, ("ended", state.run_id, 1, []))
+        self.confirm = self.said(install.tui_view(state, 200, 40))
+        install._start_run(state, "install")
+        summary = {"event": "summary", "mode": "install", "installed": 3, "up_to_date": 0,
+                   "pending": 0, "failed": [], "skipped": [], "left_out": [], "kept": [],
+                   "interrupted": False}
+        for event in [*([self.SAVES] if install_saves else []), summary]:
+            install.tui_receive(state, ("event", state.run_id, event))
+        install.tui_receive(state, ("ended", state.run_id, 0, []))
+        return state
+
+    def test_unticked_it_is_named_nowhere(self):
+        state = self.run_to_result(confirm_saves=False, install_saves=False)
+        self.assertNotIn("save directory", self.confirm)
+        self.assertEqual(state.screen, "result")
+        self.assertNotIn("save directory", self.said(install.tui_outcome(state)[1]))
+
+    def test_kept_the_confirm_screen_and_the_result_name_it(self):
+        state = self.run_to_result(confirm_saves=True, install_saves=True)
+        self.assertIn(self.LINE, self.confirm)
+        self.assertIn(self.LINE, self.said(install.tui_outcome(state)[1]))
+
+
+@unittest.skipIf(os.name == "nt" or os.geteuid() == 0,
+                 "a folder without search permission is POSIX")
+class AFolderThisUserCannotOpen(unittest.TestCase):
+    """A BIOS folder under a folder this user cannot enter: given with
+    --dest the interface ended on a traceback before its first screen, and
+    typed on the folder screen it ended on a crash log."""
+
+    def setUp(self):
+        base = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp"))
+        self.addCleanup(shutil.rmtree, base, True)
+        self.lock = base / "lock"
+        self.lock.mkdir()
+        os.chmod(self.lock, 0o000)
+        self.addCleanup(os.chmod, self.lock, 0o755)
+        self.dest = self.lock / "system"
+
+    def said(self, state: install.TuiState) -> str:
+        return " ".join(" ".join(install.tui_view(state, 200, 40)).split())
+
+    def test_typed_it_stays_on_the_path_screen(self):
+        state = fresh_state()
+        state.platform = "retroarch"
+        state.stack = ["setup", "folder", "path"]
+        state.path_text = str(self.dest)
+        self.assertEqual(install.tui_update(state, Key("enter")), [])
+        self.assertEqual(state.screen, "path")
+        self.assertEqual(state.path_text, str(self.dest))
+        self.assertIn(f"This user cannot open {install._short_path(self.lock, state.home)} "
+                      "(Permission denied)", self.said(state))
+
+    def test_given_with_dest_opens_the_folder_screen(self):
+        args = _args(platform="retroarch", dest=str(self.dest))
+        with mock.patch.object(install, "detect_platforms", return_value=[]), \
+                mock.patch.object(install, "_mounted_cards", return_value=[]):
+            state, effects = install.tui_start(args, "linux")
+        self.assertEqual(effects, [])
+        self.assertEqual(state.screen, "folder")
+        self.assertIn(f"This user cannot open {install._short_path(self.lock, state.home)} "
+                      "(Permission denied)", self.said(state))
+        self.assertIn("choose another folder", self.said(state))
+
+    def test_chosen_from_the_setups(self):
+        state = fresh_state()
+        state.setups = [("retroarch", self.dest, "RetroArch")]
+        state.stack = ["setup"]
+        self.assertEqual(install.tui_update(state, Key("enter")), [])
+        self.assertEqual(state.screen, "folder")
+        self.assertIn(f"This user cannot open {install._short_path(self.lock, state.home)} "
+                      "(Permission denied)", self.said(state))
+
+    def test_the_nearest_folder_stops_above_it(self):
+        self.assertEqual(install._nearest_folder(self.dest / "deeper"), self.lock)
+
+
+# POSIX modules a Windows Python does not have: importing one raises there.
+POSIX_ONLY = ("termios", "tty", "pty", "fcntl", "select", "resource", "pwd", "grp")
+# The signal names a Windows Python defines.
+WINDOWS_SIGNALS = ("SIGABRT", "SIGBREAK", "SIGFPE", "SIGILL", "SIGINT", "SIGSEGV",
+                   "SIGTERM")
+
+
+class WindowsOs(types.ModuleType):
+    """The os module with os.name "nt", everything else delegated."""
+
+    def __init__(self) -> None:
+        super().__init__("os")
+        self.name = "nt"
+
+    def __getattr__(self, attribute: str) -> object:
+        return getattr(os, attribute)
+
+
+class SimulatedWindows:
+    """os.name "nt", the POSIX modules unimportable, the signal module cut
+    down to the names Windows has, and msvcrt reading the given keys."""
+
+    def __init__(self, keys: str = "") -> None:
+        self.keys = list(keys)
+        self.handlers: dict[int, object] = {}
+        self.locked: list[int] = []
+        self.msvcrt = types.SimpleNamespace(
+            kbhit=lambda: bool(self.keys), getwch=lambda: self.keys.pop(0),
+            locking=lambda fd, mode, size: self.locked.append(fd), LK_NBLCK=2)
+        self.signal = types.ModuleType("signal")
+        for number, name in enumerate(WINDOWS_SIGNALS, start=100):
+            setattr(self.signal, name, number)
+        self.signal.SIG_DFL, self.signal.SIG_IGN = 0, 1
+        self.signal.CTRL_C_EVENT, self.signal.CTRL_BREAK_EVENT = 0, 1
+        self.signal.signal = self._install
+        self.signal.getsignal = lambda number: self.handlers.get(number, 0)
+        self.stack = contextlib.ExitStack()
+
+    def _install(self, number: int, handler: object) -> object:
+        if number not in {getattr(self.signal, name) for name in WINDOWS_SIGNALS}:
+            raise ValueError(f"invalid signal value {number}")
+        previous = self.handlers.get(number, 0)
+        self.handlers[number] = handler
+        return previous
+
+    def __enter__(self) -> SimulatedWindows:
+        self.stack.enter_context(mock.patch.dict(
+            sys.modules, {**{name: None for name in POSIX_ONLY}, "msvcrt": self.msvcrt}))
+        # install.py alone sees "nt": pathlib here keeps making PosixPaths.
+        self.stack.enter_context(mock.patch.object(install, "os", WindowsOs()))
+        self.stack.enter_context(mock.patch.object(install, "signal", self.signal))
+        return self
+
+    def __exit__(self, *exc: object) -> None:
+        self.stack.close()
+
+
+class WindowsHasNoPosixModules(unittest.TestCase):
+    """_terminal_gone imported termios unguarded: on Windows a closed console
+    raised ModuleNotFoundError inside run_tui's except block, a traceback
+    instead of an exit code. Every path the interface takes on Windows runs
+    here with the POSIX modules gone."""
+
+    class Console(io.StringIO):
+        broken = False
+
+        def fileno(self) -> int:
+            return 0
+
+        def isatty(self) -> bool:
+            return True
+
+        def write(self, text: str) -> int:
+            if self.broken:
+                raise OSError(errno.EPIPE, "The pipe is being closed")
+            return super().write(text)
+
+    def test_the_simulation_refuses_posix_imports(self):
+        with SimulatedWindows():
+            for name in POSIX_ONLY:
+                with self.subTest(module=name), self.assertRaises(ImportError):
+                    __import__(name)
+            self.assertFalse(hasattr(install.signal, "SIGTSTP"))
+
+    def test_a_closed_console_is_a_closed_terminal(self):
+        console = self.Console()
+        with SimulatedWindows():
+            terminal = install._TuiTerminal(console, console)
+            self.assertTrue(install._closed_terminal(
+                OSError(errno.EPIPE, "The pipe is being closed"), terminal))
+            self.assertFalse(install._closed_terminal(
+                OSError(errno.ENOENT, "No such file"), terminal))
+
+    def test_keys_enter_leave_and_cleanup(self):
+        console = self.Console()
+        with SimulatedWindows("a\xe0H") as windows:
+            terminal = install._TuiTerminal(console, console)
+            terminal.enter()
+            self.assertEqual([(key.name, key.text) for key in terminal.keys(0.05)],
+                             [("char", "a"), ("up", "")])
+            handlers = {windows.signal.SIGINT: windows.signal.SIG_DFL}
+            self.assertEqual(install._tui_cleanup(terminal, [], handlers, fresh_state()), [])
+            self.assertIn(install._SCREEN_OFF, console.getvalue())
+
+    def test_a_child_is_interrupted_through_the_console(self):
+        run = install._EngineRun.__new__(install._EngineRun)
+        run.proc = mock.Mock(**{"poll.return_value": None})
+        with SimulatedWindows(), mock.patch.object(install.os, "kill") as kill:
+            run.interrupt(from_key=False)
+            kill.assert_not_called()
+            run.proc.send_signal.assert_not_called()
+            run.interrupt(from_key=True)
+            kill.assert_called_once_with(0, install.signal.CTRL_C_EVENT)
+
+    def test_work_folder_owner_and_partial_lock(self):
+        work = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp"))
+        self.addCleanup(shutil.rmtree, work, True)
+        (work / install.TUI_OWNER).write_text(str(os.getpid()), encoding="utf-8")
+        with SimulatedWindows() as windows:
+            self.assertIsNone(install._owner_alive(work))
+            handle = install._claim_partial(work / "x.bin.part")
+            self.assertIsNotNone(handle)
+            handle.close()
+            self.assertEqual(len(windows.locked), 1)
+
+    def test_run_tui_ends_on_the_closed_console_code(self):
+        console = self.Console()
+        real_draw = install._TuiTerminal.draw
+
+        def breaking_draw(terminal, lines, cols):
+            console.broken = True
+            real_draw(terminal, lines, cols)
+
+        work = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp"))
+        self.addCleanup(shutil.rmtree, work, True)
+        with SimulatedWindows() as windows, \
+                mock.patch.object(install._TuiTerminal, "draw", breaking_draw), \
+                mock.patch.object(install, "_drop_closed_terminal"), \
+                mock.patch.object(install.tempfile, "mkdtemp", return_value=str(work)), \
+                mock.patch.object(install, "tui_start", return_value=(fresh_state(), [])), \
+                mock.patch.object(install, "_open_local_sources"), \
+                mock.patch.object(install, "detect_os", return_value="windows"), \
+                mock.patch.object(sys, "stdin", console), \
+                mock.patch.object(sys, "stdout", console), \
+                mock.patch.object(sys, "stderr", console):
+            code = install.run_tui(_args())
+            # SIGHUP does not exist there: the closed window's code is 129.
+            self.assertEqual(code, 129)
+            self.assertFalse((work / "error.log").exists())
+            self.assertTrue(all(handler == 0 for handler in windows.handlers.values()),
+                            windows.handlers)
+
+
 # --- End to end, on a pseudo-terminal -------------------------------------
 
 

tests/test_save_root.py

@@ -25,6 +25,7 @@ import threading
 import unittest
 import zipfile
 from pathlib import Path
+from unittest import mock
 
 REPO_ROOT = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(REPO_ROOT / "scripts"))
@@ -552,6 +553,60 @@ class TheInstallerFindsTheSaveDirectory(unittest.TestCase):
                                            False, False))
         install._check_save_directory(manifest)
 
+    def other_spelling(self, spelled: Path) -> install.SavePlace | str:
+        """save_directory_for the BIOS folder written another way, the config
+        keyed as detection writes it, with savefile_directory set."""
+        saves = self.root / "MySaves"
+        saves.mkdir(exist_ok=True)
+        self.cfg.write_text(f'savefile_directory = "{saves}"\n')
+        install._RETROARCH_CONFIGS[self.bios] = (self.cfg, self.app)
+        self.addCleanup(install._SAVE_DIRS.pop, spelled, None)
+        with mock.patch.object(install, "_CONFIGS_READ", [True]):
+            return install.save_directory_for(spelled, self.MANIFEST)
+
+    def test_a_relative_bios_folder_finds_its_config(self):
+        """cd ~/.config/retroarch && install.py --dest system: the save files
+        went to a guessed saves/ beside it."""
+        self.addCleanup(os.chdir, os.getcwd())
+        os.chdir(self.app)
+        self.assertEqual(self.other_spelling(Path("system")),
+                         install.SavePlace(self.root / "MySaves", True))
+
+    def test_a_bios_folder_through_a_link_finds_its_config(self):
+        """EmuDeck links RetroArch's system folder to ~/Emulation/bios."""
+        link = self.root / "bios-link"
+        link.symlink_to(self.bios, target_is_directory=True)
+        self.assertEqual(self.other_spelling(link),
+                         install.SavePlace(self.root / "MySaves", True))
+        linked_app = self.root / "RA-link"
+        linked_app.symlink_to(self.app, target_is_directory=True)
+        self.assertEqual(self.other_spelling(linked_app / "system"),
+                         install.SavePlace(self.root / "MySaves", True))
+
+    def test_case_only_differs_where_the_system_folds_it(self):
+        spelled = Path(str(self.bios).upper())
+        with mock.patch.object(install.sys, "platform", "linux"):
+            self.assertNotEqual(self.other_spelling(spelled),
+                                install.SavePlace(self.root / "MySaves", True))
+        install._SAVE_DIRS.pop(spelled, None)
+        with mock.patch.object(install.sys, "platform", "darwin"):
+            self.assertEqual(self.other_spelling(spelled),
+                             install.SavePlace(self.root / "MySaves", True))
+
+    def test_the_cores_folder_through_another_spelling(self):
+        cores = self.root / "MyCores"
+        cores.mkdir()
+        self.cfg.write_text(f'libretro_directory = "{cores}"\n')
+        install._RETROARCH_CONFIGS[self.bios] = (self.cfg, self.app)
+        link = self.root / "bios-link"
+        link.symlink_to(self.bios, target_is_directory=True)
+        self.addCleanup(os.chdir, os.getcwd())
+        os.chdir(self.app)
+        with mock.patch.object(install, "_CONFIGS_READ", [True]):
+            for spelled in (Path("system"), link, Path("./system/")):
+                with self.subTest(spelled=str(spelled)):
+                    self.assertEqual(install.core_dirs("retroarch", spelled), [cores])
+
     def test_the_entry_path(self):
         saves = (self.app / "saves").resolve()
         install._SAVE_DIRS[self.bios] = install.SavePlace(self.app / "saves", True)