chore: merge the round 4 installer fixes

tests/test_doc_pages.py

@@ -682,6 +682,16 @@ class FirstInstallWalkthrough(unittest.TestCase):
         self.text = _page("first-install.md")
         self.run = _page("installer.md")
 
+    def test_the_stop_keys_are_the_ones_the_screen_names(self):
+        """The page said "B or Ctrl+C stops it" for a computer, whose screen
+        says Esc or Ctrl-C, and where the b key does nothing."""
+        import install
+
+        section = _section(self.text, "4. Wait")
+        computer = install._STOP_KEYS["keys"][0].replace("Ctrl-C", "Ctrl+C")
+        self.assertIn(f"{computer} stops", section)
+        self.assertIn(f"{install._STOP_KEYS['gptokeyb'][0]} on a handheld", section)
+
     def test_no_hand_kept_size_before_the_run(self):
         import install
 

tests/test_install_cores.py

@@ -488,5 +488,39 @@ class TheOneLinerReadsInstalledCores(unittest.TestCase):
             self.assertEqual(repo.installed(), ["gba_bios.bin", "mgba/extra.bin"])
 
 
+class TheInterfaceFindsTheCoresLineModeFinds(unittest.TestCase):
+    """The interface names the BIOS folder with --dest, so its runs never
+    read retroarch.cfg: "only installed cores" found none where line mode,
+    detecting the same config, found them (macOS, Flatpak, any custom
+    system_directory)."""
+
+    def test_a_config_away_from_the_bios_folder(self):
+        with _CoreRepository() as repo:
+            (repo.root / "retroarch.cfg").unlink()
+            home = repo.root / "home"
+            config = home / ".config" / "retroarch"
+            config.mkdir(parents=True)
+            cores = repo.root / "elsewhere" / "my cores"
+            cores.mkdir(parents=True)
+            (cores / "mgba_libretro.so").write_bytes(b"")
+            (config / "retroarch.cfg").write_text(
+                f'system_directory = "{repo.dest}"\n'
+                f'libretro_directory = "{cores}"\n')
+            env = {**repo.env(), "HOME": str(home), "RETROBIOS_OS": "linux",
+                   "XDG_CONFIG_HOME": str(home / ".config")}
+            repo.dest.mkdir()
+            found = []
+            for given in ((), ("--dest", str(repo.dest))):
+                command = [sys.executable, str(REPO_ROOT / "install.py"), "--platform",
+                           "retroarch", *given, "--installed-cores", "--check"]
+                proc = subprocess.run(command, env=env, capture_output=True, text=True,
+                                      timeout=60, check=False, stdin=subprocess.DEVNULL)
+                self.assertNotIn("no installed core found", proc.stderr)
+                found.append([line for line in proc.stdout.splitlines()
+                              if "core installed" in line or "Narrowed" in line])
+            self.assertEqual(found[0], found[1])
+            self.assertIn("  1 core installed", found[1])
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_install_experience.py

@@ -507,6 +507,89 @@ class ARefusedFolderReportsEachFile(unittest.TestCase):
                          list(range(1, len(downloaded) + 1)))
 
 
+class AFolderThisUserCannotEnter(unittest.TestCase):
+    """A subfolder of the BIOS folder without search permission (made 0700
+    by an emulator that ran as root, or denied by a Windows ACL) raised
+    PermissionError from Path.exists() before any file was handled, in
+    --check and in an install alike."""
+
+    def _locked(self, repo: FakeRepository) -> Path:
+        (repo.dest / "pcsx").mkdir(parents=True)
+        (repo.dest / "pcsx" / "extra.bin").write_bytes(b"old")
+        os.chmod(repo.dest / "pcsx", 0o000)
+        return repo.dest / "pcsx"
+
+    @unittest.skipIf(os.name == "nt" or os.geteuid() == 0,
+                     "a folder without search permission is POSIX")
+    def test_check_names_the_file_and_reads_the_others(self):
+        with FakeRepository() as repo:
+            locked = self._locked(repo)
+            try:
+                proc = repo.run("--check")
+            finally:
+                os.chmod(locked, 0o755)
+        self.assertNotIn("Traceback", proc.stderr)
+        self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
+        self.assertIn("wrong hash: pcsx/extra.bin", proc.stdout)
+        self.assertIn("could not be read (Permission denied)", proc.stdout)
+        self.assertIn("missing: scph5501.bin", proc.stdout)
+
+    @unittest.skipIf(os.name == "nt" or os.geteuid() == 0,
+                     "a folder without search permission is POSIX")
+    def test_install_fails_that_file_alone(self):
+        with FakeRepository() as repo:
+            locked = self._locked(repo)
+            try:
+                proc = repo.run("--events")
+            finally:
+                os.chmod(locked, 0o755)
+            installed = repo.installed()
+        self.assertNotIn("Traceback", proc.stderr)
+        events = [json.loads(line) for line in proc.stdout.splitlines()]
+        failed = {e["dest"]: e["reason"] for e in events
+                  if e["event"] == "downloaded" and not e["ok"]}
+        self.assertEqual(list(failed), ["pcsx/extra.bin"], proc.stderr)
+        self.assertIn("this user may not have access", failed["pcsx/extra.bin"])
+        self.assertIn("scph5501.bin", installed)
+        self.assertIn("mgba/extra.bin", installed)
+
+    @unittest.skipIf(os.name == "nt" or os.geteuid() == 0,
+                     "a folder without search permission is POSIX")
+    def test_declared_state_reads_it_as_wrong(self):
+        with FakeRepository() as repo:
+            locked = self._locked(repo)
+            entry = {"dest": "pcsx/extra.bin", "sha1": "0" * 40, "size": 3}
+            try:
+                state = install._declared_state(entry, entry, repo.dest)
+            finally:
+                os.chmod(locked, 0o755)
+        self.assertEqual(state, "wrong_hash")
+
+
+class OutputToAFileAsksNothing(unittest.TestCase):
+    """`install.py > log.txt` at a terminal: the menu went into the file
+    and the run waited on an answer to a prompt nobody could see."""
+
+    @unittest.skipIf(os.name == "nt", "a pseudo-terminal for stdin")
+    def test_the_run_goes_on_as_with_no_input(self):
+        import pty
+
+        master, slave = pty.openpty()
+        try:
+            with FakeRepository() as repo:
+                proc = subprocess.run(
+                    repo.command(), env=repo.env(), stdin=slave, capture_output=True,
+                    text=True, timeout=60, check=False)
+                installed = repo.installed()
+        finally:
+            os.close(master)
+            os.close(slave)
+        self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
+        self.assertIn("nothing is asked", proc.stderr)
+        self.assertNotIn("Enter) install everything", proc.stdout)
+        self.assertEqual(len(installed), len(FakeRepository.PAYLOADS))
+
+
 class _Tty(io.StringIO):
     def isatty(self) -> bool:
         return True

tests/test_install_from.py

@@ -854,6 +854,67 @@ class AFolderAroundTheBiosFolder(FromFolder):
                          ["old card/copy.zip, inside as inner.bin"])
 
 
+    @unittest.skipUnless(Path("/proc/self/fd").is_dir(), "reads the open files from /proc")
+    def test_an_archive_the_list_replaces_is_not_held_open(self):
+        """Windows refuses to rename, replace or delete a file a process
+        holds open. The --from index kept the list's own wrong archive open
+        once its members had been listed, and the run failed to move it
+        aside, every run."""
+        good = io.BytesIO()
+        with zipfile.ZipFile(good, "w") as archive:
+            archive.writestr("a.bin", b"A" * 300)
+        good = good.getvalue()
+        card = self.root / "card"
+        bios = card / "bios"
+        _zip(bios / "neogeo.zip", {"x.bin": b"X" * 77})
+        (card / "backup").mkdir()
+        (card / "backup" / "neogeo.zip").write_bytes(good)
+        _zip(card / "backup" / "pack.zip", {"renamed.bin": GBA})
+        self.list.write_text(json.dumps({
+            "manifest_version": 2, "platform": "retroarch",
+            "files": [_entry("neogeo.zip", good, ["snk-neogeo"]),
+                      _entry("gba_bios.bin", GBA, ["nintendo-gba"])]}))
+
+        def held() -> set[str]:
+            out = set()
+            for fd in os.listdir("/proc/self/fd"):
+                with contextlib.suppress(OSError):
+                    out.add(os.path.realpath(os.readlink(f"/proc/self/fd/{fd}")))
+            return out
+
+        real_replace, real_rename, real_unlink = os.replace, os.rename, os.unlink
+
+        def refusing(real):
+            def call(*paths, **kwargs):
+                files = paths[:2] if real is not real_unlink else paths[:1]
+                if any(os.path.realpath(path) in held() for path in files):
+                    raise PermissionError(13, "The process cannot access the file",
+                                          str(files[0]))
+                return real(*paths, **kwargs)
+            return call
+
+        argv = ["install.py", "--platform", "retroarch", "--dest", str(bios),
+                "--manifest", str(self.list), "--from", str(card), "--no-input", "--yes"]
+        out = io.StringIO()
+        with mock.patch.object(sys, "argv", argv), \
+                mock.patch.dict(os.environ, {"RETROBIOS_INSTALL_OFFLINE": "1"}), \
+                mock.patch.object(os, "replace", refusing(real_replace)), \
+                mock.patch.object(os, "rename", refusing(real_rename)), \
+                mock.patch.object(os, "unlink", refusing(real_unlink)), \
+                mock.patch.object(os, "remove", refusing(real_unlink)), \
+                contextlib.redirect_stdout(out), contextlib.redirect_stderr(out):
+            try:
+                install.main()
+                code = 0
+            except SystemExit as exc:
+                code = exc.code
+            finally:
+                install.LOCAL.reset()
+        self.assertEqual(code, 0, out.getvalue())
+        self.assertEqual((bios / "neogeo.zip").read_bytes(), good)
+        self.assertEqual((bios / "gba_bios.bin").read_bytes(), GBA)
+
+
 class ManySameSizedFiles(FromFolder):
     """A card of 16 KB ROMs and a list of 16 KB entries: each entry sorted
     and walked every file of its size."""

tests/test_install_selection.py

@@ -374,6 +374,7 @@ class DecliningTheSizePointsAtTheChoice(unittest.TestCase):
                                   repo.env()["RETROBIOS_BASE_URL"] + "/{path}"), \
                 mock.patch.object(install, "CONFIRM_FROM_BYTES", 1), \
                 mock.patch("sys.stdin", mock.Mock(isatty=lambda: True)), \
+                mock.patch.object(install, "_is_terminal", return_value=True), \
                 mock.patch("builtins.input", lambda _prompt="": next(replies)), \
                 contextlib.redirect_stdout(out):
             try:
@@ -744,14 +745,26 @@ class TheOneLinerNarrows(unittest.TestCase):
             env = dict(repo.env(), HOME=str(home), USERPROFILE=str(home),
                        XDG_CONFIG_HOME=str(home), RETROBIOS_OS="linux")
             leader, follower = pty.openpty()
+            # The question needs the output on the terminal too: sent to a
+            # pipe, the run asks nothing.
             proc = subprocess.Popen(
                 [sys.executable, str(REPO_ROOT / "install.py")], stdin=follower,
-                stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, text=True,
+                stdout=follower, stderr=subprocess.PIPE, env=env, text=True,
             )
             os.close(follower)
             os.write(leader, b"\x04")
-            out, err = proc.communicate(timeout=60)
+            err = proc.communicate(timeout=60)[1]
+            chunks = []
+            while True:
+                try:
+                    chunk = os.read(leader, 65536)
+                except OSError:
+                    break
+                if not chunk:
+                    break
+                chunks.append(chunk)
             os.close(leader)
+            out = b"".join(chunks).decode("utf-8", "replace")
         self.assertEqual(proc.returncode, 130, out + err)
         self.assertIn("Stopped. Nothing was installed.", out)
         self.assertNotIn("carry on", out)

tests/test_install_tui.py

@@ -2205,6 +2205,33 @@ class AuditRoundThree(unittest.TestCase):
 
     # P2-1: a card too small for the download.
 
+    def test_a_stop_asked_for_says_only_the_outcome(self):
+        """Start+Select (SIGTERM) or SIGHUP stops the child before it writes
+        a summary: the farewell then listed the log's last lines."""
+        state = checked_state()
+        state.download = {"files": 50}
+        state.finished = [("scph5501.bin", True, "")]
+        state.tail = ["    [2/50] scph5501.bin ok",
+                      "    Stopping. Files already installed are kept."]
+        for code, shown in ((130, False), (1, True)):
+            with self.subTest(code=code):
+                state.code = code
+                _title, lines = install.tui_outcome(state)
+                self.assertEqual(any("[2/50]" in line for line in lines), shown, lines)
+
+    def test_a_pack_installs_from_the_pack_not_the_network(self):
+        """With --from-pack and no Wi-Fi, the plan said "to download", which
+        on a handheld with no network read as a failure on its way."""
+        state = checked_state()
+        with mock.patch.object(install.LOCAL, "pack", object()):
+            headline = str(install.tui_view(state, 80, 24)[0])
+            plan = "\n".join(str(line) for line in install.tui_view(state, 80, 24))
+            state.summary = {"installed": 3, "failed": [], "up_to_date": 0}
+            title, lines = install.tui_outcome(state)
+        self.assertIn("to install from the pack", plan, headline)
+        self.assertNotIn("to download", plan)
+        self.assertEqual(lines[0], "3 files installed from the pack.")
+
     def test_p2_1_another_folder_comes_first_when_space_is_short(self):
         state = checked_state({"free": 1024, "writable": True, "cores": []})
         self.assertEqual([row.label for row in install.tui_rows(state)[:2]],
@@ -2212,6 +2239,17 @@ class AuditRoundThree(unittest.TestCase):
         self.assertEqual(press(state, "enter"), [])
         self.assertEqual(state.screen, "folder")
 
+    def test_offline_with_nothing_to_copy_offers_no_install_anyway(self):
+        """Offline with nothing here to copy, Install is removed: a folder
+        short of space then put "Install anyway" back, a run that could
+        only fail every file."""
+        state = checked_state({"free": 1024, "writable": False, "cores": []})
+        with mock.patch.dict(os.environ, {install.OFFLINE_VARIABLE: "1"}):
+            labels = [row.label for row in install.tui_rows(state)]
+        self.assertEqual(labels[0], "Choose another folder")
+        self.assertNotIn("Install anyway", labels)
+        self.assertFalse(any(label.startswith("Install") for label in labels), labels)
+
     FULL_TAIL = [
         "  Checking existing files...",
         "    0/50 present (0 verified, 0 wrong hash)",
@@ -2557,6 +2595,34 @@ class IdleExit(unittest.TestCase):
                          ["RetroBIOS: No button pressed, closing."])
 
 
+class AReadErrorIsNotAClosedWindow(unittest.TestCase):
+    """Any EIO read as the window closing: a failing SD card answers EIO
+    too, and that run ended 129, no error.log, no launcher message."""
+
+    def test_eio_counts_only_once_the_terminal_hung_up(self):
+        import pty
+
+        master, slave = pty.openpty()
+        screen = open(os.dup(slave), "w", encoding="utf-8")  # noqa: SIM115
+        terminal = mock.Mock(fd=slave, stdout=screen)
+        try:
+            card = OSError(errno.EIO, "Input/output error", "/media/card/bios/x.bin")
+            self.assertFalse(install._closed_terminal(card, terminal))
+            self.assertFalse(install._closed_terminal(
+                OSError(errno.ENOENT, "No such file"), terminal))
+            os.close(master)
+            master = -1
+            self.assertTrue(install._closed_terminal(card, terminal))
+            self.assertTrue(install._closed_terminal(termios.error(errno.EIO, "gone"),
+                                                     terminal))
+        finally:
+            with contextlib.suppress(OSError):
+                screen.close()
+            os.close(slave)
+            if master >= 0:
+                os.close(master)
+
+
 class AuditRoundFour(unittest.TestCase):
     """tui_audit_3.md, each finding held where it was found."""
 
@@ -2935,8 +3001,13 @@ class EndToEnd(unittest.TestCase):
                     term.send("\r")
                     screen = term.wait("files to download")
                     self.assertIn("6 files to download", screen)
+                    # The second folder the run writes to, named before
+                    # anything is written, and again on the result.
+                    self.assertIn("1 file read from RetroArch's save directory goes to "
+                                  "~/MySaves", screen)
                     term.send("\r")
-                    term.wait("6 files installed.")
+                    screen = term.wait("6 files installed.")
+                    self.assertIn("save directory goes to ~/MySaves", screen)
                     term.send("\r")
                     self.assertEqual(term.close(), 0)
                 results.append(written(repo, saves))
@@ -3035,6 +3106,9 @@ class EndToEnd(unittest.TestCase):
         after = term.after()
         self.assertIn(f"Stopped: {on_disk} of 6 installed, run again to finish.", after)
         self.assertNotIn("Nothing was installed", after)
+        # The plain outcome only: the log's last lines are for a run that died.
+        self.assertIsNone(re.search(r"\[\d+/6\]", after), after)
+        self.assertNotIn("Stopping.", after)
         time.sleep(1)
         self.assertEqual(len(repo.installed()), on_disk)
 
@@ -3176,6 +3250,18 @@ class EndToEnd(unittest.TestCase):
         term.send("q")
         self.assertEqual(term.close(), 130)
 
+    def test_no_key_on_a_screen_reached_by_itself_closes_it(self):
+        """Given --platform, as the offline launcher route is, the check
+        runs at once and the plan follows without a key: the close counted
+        only on the first screen, and never came."""
+        repo = self.repo()
+        self.env[install.TUI_IDLE_ENV] = "1"
+        term = self.start("--platform", "retroarch", "--dest", str(repo.dest))
+        term.wait("> Install")
+        self.assertEqual(term.close(), 0)
+        self.assertIn("No button pressed, closing.", term.after())
+        self.assertEqual(repo.installed(), [])
+
     def test_a_desktop_names_its_keys_and_a_pad_its_buttons(self):
         self.repo()
         for env, bar in (({"DISPLAY": ":0"}, "Enter Continue  Esc Quit  ? Help"),