libretro/tests/test_install_experience.py

"""What the one-liner shows and how it recovers.

A live progress line with speed and time left, a plan before a large
download, a dry run, a quiet mode, a version, a pause between retries that
honours Retry-After, and large files that resume where they stopped.
"""

from __future__ import annotations

import contextlib
import hashlib
import http.server
import io
import json
import os
import subprocess
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from unittest import mock

REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT))
sys.path.insert(0, str(REPO_ROOT / "tests"))

import install  # noqa: E402
from test_install_selection import FakeRepository  # noqa: E402

PAYLOAD = bytes(range(256)) * 400  # 102400 bytes


class _RangeHandler(http.server.BaseHTTPRequestHandler):
    """Serves PAYLOAD with Range support and fails first when asked to."""

    ranges: list[str] = []
    fail_first: list[int] = []

    def log_message(self, *args) -> None:  # noqa: D102
        pass

    def do_GET(self) -> None:  # noqa: N802, D102
        if self.fail_first:
            code = self.fail_first.pop(0)
            self.send_response(code)
            self.send_header("Retry-After", "0")
            self.end_headers()
            return
        header = self.headers.get("Range")
        self.ranges.append(header or "")
        start = 0
        if header and header.startswith("bytes="):
            start = int(header[6:].split("-")[0])
            self.send_response(206)
            self.send_header("Content-Range", f"bytes {start}-{len(PAYLOAD) - 1}/{len(PAYLOAD)}")
        else:
            self.send_response(200)
        body = PAYLOAD[start:]
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


class _Server:
    def __enter__(self):
        _RangeHandler.ranges = []
        _RangeHandler.fail_first = []
        self.httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _RangeHandler)
        threading.Thread(target=self.httpd.serve_forever, daemon=True).start()
        port = self.httpd.server_address[1]
        self.patch = mock.patch.object(install, "RAW_FILE_URL", f"http://127.0.0.1:{port}/{{path}}")
        self.patch.start()
        return self

    def __exit__(self, *exc):
        self.patch.stop()
        self.httpd.shutdown()
        self.httpd.server_close()


ENTRY = {
    "dest": "big.bin", "repo_path": "bios/big.bin", "size": len(PAYLOAD),
    "sha256": hashlib.sha256(PAYLOAD).hexdigest(),
}


class LargeFilesResume(unittest.TestCase):
    def test_a_partial_copy_is_completed_with_a_range_request(self):
        with _Server(), tempfile.TemporaryDirectory() as tmp, \
                mock.patch.object(install, "RESUME_FROM_SIZE", 1024):
            part = Path(tmp) / ".big.bin.part"
            part.write_bytes(PAYLOAD[:40000])
            dest, ok, reason = install._download_one(dict(ENTRY), Path(tmp))
            self.assertEqual((dest, ok, reason), ("big.bin", True, ""))
            self.assertEqual((Path(tmp) / "big.bin").read_bytes(), PAYLOAD)
            self.assertEqual(_RangeHandler.ranges, ["bytes=40000-"])
            self.assertFalse(part.exists())

    def test_a_small_file_starts_over(self):
        with _Server(), tempfile.TemporaryDirectory() as tmp:
            dest, ok, _ = install._download_one(dict(ENTRY), Path(tmp))
            self.assertTrue(ok)
            self.assertEqual(_RangeHandler.ranges, [""])

    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
    def test_a_link_at_the_partial_name_is_never_written_through(self):
        with _Server(), tempfile.TemporaryDirectory() as tmp, \
                mock.patch.object(install, "RESUME_FROM_SIZE", 1024):
            victim = Path(tmp) / "victim.txt"
            victim.write_bytes(b"keep me")
            part = Path(tmp) / "bios" / ".big.bin.part"
            part.parent.mkdir()
            part.symlink_to(victim)
            dest, ok, _ = install._download_one(dict(ENTRY), Path(tmp) / "bios")
            self.assertTrue(ok)
            self.assertEqual(victim.read_bytes(), b"keep me")
            final = Path(tmp) / "bios" / "big.bin"
            self.assertFalse(final.is_symlink())
            self.assertEqual(final.read_bytes(), PAYLOAD)
            self.assertEqual(_RangeHandler.ranges, [""])


class TwoRunsOneFolder(unittest.TestCase):
    """Two installs into one BIOS folder appended to the same partial copy:
    each hashed the other's bytes and one renamed a file still growing."""

    @unittest.skipIf(os.name == "nt", "flock semantics")
    def test_a_partial_held_by_another_run_is_left_alone(self):
        with _Server(), tempfile.TemporaryDirectory() as tmp, \
                mock.patch.object(install, "RESUME_FROM_SIZE", 1024):
            part = Path(tmp) / ".big.bin.part"
            part.write_bytes(PAYLOAD[:40000])
            other = install._claim_partial(part)
            self.assertIsNotNone(other)
            try:
                dest, ok, reason = install._download_one(dict(ENTRY), Path(tmp))
                self.assertEqual((ok, reason), (True, ""))
                self.assertEqual((Path(tmp) / "big.bin").read_bytes(), PAYLOAD)
                self.assertEqual(part.read_bytes(), PAYLOAD[:40000])
                self.assertEqual(_RangeHandler.ranges, [""])
                self.assertIsNone(install._claim_partial(part))
            finally:
                other.close()
            self.assertIsNotNone(install._claim_partial(part))


class APartialRenamedBeforeTheLock(unittest.TestCase):
    """Open and lock are two steps. The run holding the partial could finish
    in between, rename it into place and release it: the second run then
    locked the installed BIOS, took it for a finished partial and truncated it."""

    @unittest.skipIf(os.name == "nt", "flock semantics")
    def test_the_installed_file_is_not_claimed(self):
        import fcntl

        with tempfile.TemporaryDirectory() as tmp:
            part = Path(tmp) / ".big.bin.part"
            final = Path(tmp) / "big.bin"
            part.write_bytes(PAYLOAD)
            real_flock = fcntl.flock

            def finish_first(fd, op):
                os.replace(part, final)
                return real_flock(fd, op)

            with mock.patch.object(fcntl, "flock", finish_first):
                claimed = install._claim_partial(part)
            self.assertIsNone(claimed)
            self.assertEqual(final.read_bytes(), PAYLOAD)


class InstalledFileModes(unittest.TestCase):
    """A file came out 0600, the mode of the temporary it was written to: run
    with sudo for a root-owned BIOS folder, the emulator's own user could not
    read it, and RetroDECK's pico8 and vox launchers were never executable."""

    def _mode(self, entry: dict) -> int:
        with _Server(), tempfile.TemporaryDirectory() as tmp:
            dest, ok, _ = install._download_one(dict(entry), Path(tmp))
            self.assertTrue(ok)
            return (Path(tmp) / dest).stat().st_mode & 0o777

    def test_the_umask_decides_as_for_any_new_file(self):
        umask = os.umask(0o022)
        try:
            self.assertEqual(self._mode(ENTRY), 0o644)
            self.assertEqual(self._mode({**ENTRY, "executable": True}), 0o755)
        finally:
            os.umask(umask)


class ProgramsAreMarked(unittest.TestCase):
    """The manifest says which files are programs, read from their header:
    the git mode cannot, hundreds of MAME archives carrying an execute bit."""

    def test_header_decides(self):
        sys.path.insert(0, str(REPO_ROOT / "scripts"))
        import generate_pack

        with tempfile.TemporaryDirectory() as tmp:
            for data, expected in ((b"\x7fELF\x02", True), (b"#!/bin/sh\n", True),
                                   (b"PK\x03\x04", False), (b"", False)):
                path = Path(tmp) / "f"
                path.write_bytes(data)
                with self.subTest(head=data):
                    self.assertEqual(generate_pack._is_program(str(path)), expected)

    def test_retrodeck_launchers_are_executable(self):
        manifest = json.loads((REPO_ROOT / "install" / "retrodeck.json").read_text())
        launchers = [f for f in manifest["files"]
                     if f["dest"].endswith(("pico-8/pico8", "voxatron/vox"))]
        if not launchers:
            self.skipTest("no launcher in the RetroDECK manifest")
        for entry in launchers:
            with self.subTest(dest=entry["dest"]):
                self.assertIs(entry.get("executable"), True)


class ManifestHashesMatchTheScan(unittest.TestCase):
    """A manifest entry's sha1 is checked against the database scan of the
    same path: under load this host hashed a 252 MB DSi NAND to a sha1 no
    file holds, the same wrong value in two runs."""

    def test_a_contradicted_hash_is_refused(self):
        sys.path.insert(0, str(REPO_ROOT / "scripts"))
        import generate_pack

        with tempfile.TemporaryDirectory() as tmp:
            path = Path(tmp) / "nand.bin"
            path.write_bytes(b"real")
            real = hashlib.sha1(b"real").hexdigest()
            good = {"files": {real: {"path": str(path)}}}
            self.assertEqual(generate_pack._stable_hashes(str(path), good)["sha1"], real)
            bad = {"files": {"0" * 40: {"path": str(path)}}}
            with self.assertRaises(RuntimeError):
                generate_pack._stable_hashes(str(path), bad)
            self.assertEqual(generate_pack._stable_hashes(str(path), {"files": {}})["sha1"], real)

    def test_a_wrong_first_reading_is_read_again_from_disk(self):
        """The retries went through the hash cache and got the first, wrong
        value back: one read of the file, then a refusal a second read would
        have avoided."""
        sys.path.insert(0, str(REPO_ROOT / "scripts"))
        import generate_pack
        import hashing

        real_compute = hashing._compute_hashes
        reads: list[str] = []

        def flaky(filepath, algos):
            reads.append(str(filepath))
            result = real_compute(filepath, algos)
            if len(reads) == 1:
                result = {**result, "sha1": "0" * 40}
            return result

        with tempfile.TemporaryDirectory() as tmp:
            path = Path(tmp) / "nand.bin"
            path.write_bytes(b"flaky")
            real = hashlib.sha1(b"flaky").hexdigest()
            db = {"files": {real: {"path": str(path)}}}
            with mock.patch.object(hashing, "_compute_hashes", flaky):
                self.assertEqual(generate_pack._stable_hashes(str(path), db)["sha1"], real)
        self.assertEqual(len(reads), 2)

    def test_the_cache_is_keyed_by_what_was_checked(self):
        """Keyed by id(db), a database freed and replaced at the same address
        got hashes verified against the first one back, unchecked."""
        source = (REPO_ROOT / "scripts" / "generate_pack.py").read_text(encoding="utf-8")
        body = source[source.index("def _stable_hashes("):]
        body = body[: body.index("\ndef ")]
        key_line = next(line for line in body.splitlines() if line.strip().startswith("key ="))
        self.assertNotIn("id(", key_line)
        self.assertIn("expected", key_line)


class RetriesWait(unittest.TestCase):
    def test_a_server_error_is_retried_after_the_delay_it_asks(self):
        with _Server(), tempfile.TemporaryDirectory() as tmp:
            _RangeHandler.fail_first = [503]
            dest, ok, _ = install._download_one(dict(ENTRY), Path(tmp))
            self.assertTrue(ok)

    def test_the_delay_doubles_and_follows_retry_after(self):
        self.assertEqual([install._retry_delay(n, None) for n in (1, 2, 3, 4, 5)],
                         [1.0, 2.0, 4.0, 8.0, 8.0])
        self.assertEqual(install._retry_delay(1, "5"), 5.0)
        self.assertEqual(install._retry_delay(1, "999"), 60.0)

    def test_a_refusal_names_the_rate_limit(self):
        with _Server(), tempfile.TemporaryDirectory() as tmp, \
                mock.patch.object(install, "MAX_RETRIES", 1):
            _RangeHandler.fail_first = [403]
            _dest, ok, reason = install._download_one(dict(ENTRY), Path(tmp))
            self.assertFalse(ok)
            self.assertIn("rate limit", reason)


class ProgressOutput(unittest.TestCase):
    def _term(self, quiet=False):
        stream = io.StringIO()
        term = install.Terminal(stream, quiet=quiet)
        return stream, term

    def test_a_log_gets_one_line_per_file_and_failures_say_why(self):
        stream, term = self._term()
        progress = install.Progress(2, 10, term)
        progress.done("a.bin", True)
        progress.done("b.bin", False, "HTTP 404")
        out = stream.getvalue()
        self.assertIn("[1/2] a.bin ok", out)
        self.assertIn("FAILED b.bin (HTTP 404)", out)
        self.assertNotIn("\033", out)

    def test_quiet_leaves_failures_to_the_summary(self):
        stream, term = self._term(quiet=True)
        progress = install.Progress(2, 10, term)
        progress.done("a.bin", True)
        progress.done("b.bin", False, "HTTP 404")
        self.assertEqual(stream.getvalue(), "")

    def test_durations_read_like_a_clock(self):
        self.assertEqual([install.format_duration(s) for s in (0, 42, 185, 3729)],
                         ["0:00", "0:42", "3:05", "1:02:09"])

    def test_the_summary_names_each_failure(self):
        buf = io.StringIO()
        with contextlib.redirect_stdout(buf), self.assertRaises(SystemExit):
            install._report_outcome(
                [Path("/b")], 1, 0, 1, 0, False,
                failures=[("x.bin", "HTTP 404")], transferred=2048, seconds=2.0,
            )
        out = buf.getvalue()
        self.assertIn("failed: x.bin (HTTP 404)", out)
        self.assertIn("2.0 KB downloaded in 0:02", out)


class CommandLine(unittest.TestCase):
    def test_a_dry_run_writes_nothing(self):
        with FakeRepository() as repo:
            proc = repo.run("--dry-run")
            self.assertEqual(proc.returncode, 0, proc.stderr)
            self.assertIn("Would download", proc.stdout)
            self.assertIn("Nothing was written", proc.stdout)
            self.assertEqual(repo.installed(), [])

    def test_quiet_prints_the_summary_only(self):
        with FakeRepository() as repo:
            proc = repo.run("--quiet")
            self.assertEqual(proc.returncode, 0, proc.stderr)
            self.assertNotIn("Checking existing files", proc.stdout)
            self.assertIn("Done.", proc.stdout)
            self.assertTrue(repo.installed())

    def test_version_names_the_file_the_bootstraps_pin(self):
        proc = subprocess.run(
            [sys.executable, str(REPO_ROOT / "install.py"), "--version"],
            capture_output=True, text=True, timeout=60, check=True,
        )
        digest = hashlib.sha256((REPO_ROOT / "install.py").read_bytes()).hexdigest()
        self.assertIn(digest[:12], proc.stdout)



class ReviewCases(unittest.TestCase):
    """Saying no to the large-download question ended on Done, --quiet hid
    fatal messages and swallowed --check, and a resumed copy inflated the speed."""

    def test_declining_is_not_done(self):
        buf = io.StringIO()
        with contextlib.redirect_stdout(buf):
            install._report_outcome([Path("/b")], 0, 3, 0, 0, False, declined=12)
        out = buf.getvalue()
        self.assertNotIn("Done.", out)
        self.assertIn("12 files not downloaded", out)
        self.assertIn("3 files already up to date", out)

    def test_quiet_refuses_modes_it_would_silence(self):
        for mode in ("--check", "--dry-run", "--list-platforms", "--list-systems"):
            proc = subprocess.run(
                [sys.executable, str(REPO_ROOT / "install.py"), "--quiet", mode],
                capture_output=True, text=True, timeout=60, check=False,
            )
            with self.subTest(mode=mode):
                self.assertEqual(proc.returncode, 2)
                self.assertIn("--quiet applies to an install", proc.stderr)

    def test_quiet_shows_what_a_failed_run_printed(self):
        with tempfile.TemporaryDirectory() as home:
            env = dict(os.environ, HOME=home, USERPROFILE=home, RETROBIOS_OS="linux",
                       XDG_CONFIG_HOME=home)
            proc = subprocess.run(
                [sys.executable, str(REPO_ROOT / "install.py"), "--quiet"],
                capture_output=True, text=True, timeout=60, check=False,
                env=env, stdin=subprocess.DEVNULL,
            )
        self.assertEqual(proc.returncode, 1)
        self.assertIn("No supported platform detected", proc.stderr)

    def test_a_resumed_copy_is_not_counted_as_fetched(self):
        progress = install.Progress(1, 100, install.Terminal(io.StringIO()))
        progress.add(60, resumed=True)
        progress.add(40)
        self.assertEqual(progress.transferred(), 40)

class _HeldHandler(http.server.BaseHTTPRequestHandler):
    """Serves the manifest and first.bin at once, every other file once released."""

    files: dict[str, bytes] = {}
    manifest = b""
    release = threading.Event()

    def log_message(self, *args) -> None:  # noqa: D102
        pass

    def do_GET(self) -> None:  # noqa: N802, D102
        if self.path.endswith(".json"):
            body = self.manifest
        else:
            name = self.path.rsplit("/", 1)[-1]
            if name != "first.bin":
                self.release.wait(30)
            body = self.files[name]
        self.send_response(200)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


class AStoppedRunCountsWhatLanded(unittest.TestCase):
    """Stopped at 8 of 118 files, the summary said "118 files installed":
    every queued file was counted, written or not, and a user who read it
    did not run the command again."""

    @unittest.skipIf(os.name == "nt", "SIGINT is sent the POSIX way")
    def test_only_the_written_file_is_counted(self):
        import signal  # noqa: PLC0415
        import time  # noqa: PLC0415

        files = {name: name.encode() * 64 for name in ("first.bin", "second.bin", "third.bin")}
        manifest = {"manifest_version": 2, "platform": "retroarch", "files": [
            {"dest": name, "size": len(data), "sha1": hashlib.sha1(data).hexdigest(),
             "repo_path": f"bios/{name}", "cores": None}
            for name, data in files.items()
        ]}
        _HeldHandler.files, _HeldHandler.manifest = files, json.dumps(manifest).encode()
        _HeldHandler.release = threading.Event()
        httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _HeldHandler)
        threading.Thread(target=httpd.serve_forever, daemon=True).start()
        self.addCleanup(httpd.server_close)
        self.addCleanup(httpd.shutdown)
        self.addCleanup(_HeldHandler.release.set)
        with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as tmp:
            dest = Path(tmp) / "bios"
            env = dict(os.environ,
                       RETROBIOS_BASE_URL=f"http://127.0.0.1:{httpd.server_address[1]}")
            log = Path(tmp) / "out.txt"
            # The held files are released once the run says it is stopping,
            # not after a fixed pause: under load the stop landed after the
            # release, all three files arrived, and nothing was stopped.
            with open(log, "w", encoding="utf-8") as out_file:
                proc = subprocess.Popen(
                    [sys.executable, str(REPO_ROOT / "install.py"), "--platform", "retroarch",
                     "--dest", str(dest), "--no-input"],
                    env=env, stdin=subprocess.DEVNULL, stdout=out_file,
                    stderr=subprocess.STDOUT, text=True,
                )
                deadline = time.monotonic() + 30
                while not (dest / "first.bin").exists() and time.monotonic() < deadline:
                    time.sleep(0.05)
                time.sleep(0.2)
                proc.send_signal(signal.SIGINT)
                while ("Stopping." not in log.read_text(encoding="utf-8")
                       and time.monotonic() < deadline):
                    time.sleep(0.05)
                _HeldHandler.release.set()
                proc.wait(timeout=60)
            out = err = log.read_text(encoding="utf-8")
            written = sorted(p.name for p in dest.iterdir() if not p.name.startswith("."))
        self.assertEqual(proc.returncode, 130, out + err)
        self.assertEqual(written, ["first.bin"])
        self.assertIn("Stopped. 1 file installed.", out)


class ARefusedFolderReportsEachFile(unittest.TestCase):
    """A read-only folder ended the run on its first refusal, and --events
    wrote one `downloaded` line for seven failed files, while the events
    table promises one per finished file."""

    @unittest.skipIf(os.name == "nt" or os.geteuid() == 0,
                     "a read-only folder is refused by POSIX permissions")
    def test_one_event_per_file_with_the_reason(self):
        with FakeRepository() as repo:
            repo.dest.mkdir()
            os.chmod(repo.dest, 0o555)
            try:
                proc = repo.run("--events")
            finally:
                os.chmod(repo.dest, 0o755)
        events = [json.loads(line) for line in proc.stdout.splitlines()]
        downloaded = [e for e in events if e["event"] == "downloaded"]
        self.assertEqual(proc.returncode, 2, proc.stderr)
        self.assertEqual(len(downloaded), len(FakeRepository.PAYLOADS))
        self.assertEqual({e["reason"] for e in downloaded}, {install.NOT_WRITABLE})
        self.assertEqual(sorted(e["files_done"] for e in downloaded),
                         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")

    @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 ADetectedFolderUnderAClosedOne(unittest.TestCase):
    """retroarch.cfg naming a system_directory under a folder this user
    cannot open, with no --dest: the case probe raised PermissionError and
    the run, and the interface before its first screen, ended on a
    traceback. The refusal --dest gets is the one detection gets."""

    @unittest.skipIf(os.name == "nt" or os.geteuid() == 0,
                     "a folder without search permission is POSIX")
    def test_it_is_refused_in_words(self):
        with FakeRepository() as repo:
            home = repo.root / "home"
            lock = repo.root / "lock"
            (home / ".config" / "retroarch").mkdir(parents=True)
            lock.mkdir()
            (home / ".config" / "retroarch" / "retroarch.cfg").write_text(
                f'system_directory = "{lock / "system"}"\n')
            env = {**repo.env(), "HOME": str(home), "XDG_CONFIG_HOME": str(home / ".config"),
                   "RETROBIOS_OS": "linux"}
            command = [sys.executable, str(REPO_ROOT / "install.py"), "--platform",
                       "retroarch", "--no-input"]
            os.chmod(lock, 0o000)
            try:
                runs = {name: subprocess.run(command + extra, env=env, capture_output=True,
                                             text=True, timeout=60, stdin=subprocess.DEVNULL)
                        for name, extra in (("check", ["--check"]), ("install", []))}
                kinds = install.tell_apart([("retroarch", lock / "system", "RetroArch"),
                                            ("retroarch", repo.dest, "RetroArch")])
            finally:
                os.chmod(lock, 0o755)
        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.assertEqual(len(kinds), 2)


class AHomeThisUserCannotOpen(unittest.TestCase):
    """HOME kept across `su` without `-`: detection stat'ed files under it
    and ended the run on a PermissionError traceback before any line."""

    @unittest.skipIf(os.name == "nt" or os.geteuid() == 0,
                     "a folder without search permission is POSIX")
    def test_detection_says_so_and_the_run_goes_on(self):
        with FakeRepository() as repo:
            home = repo.root / "home"
            home.mkdir()
            env = {**repo.env(), "HOME": str(home), "RETROBIOS_OS": "linux"}
            runs = {}
            os.chmod(home, 0o000)
            try:
                for name, extra in (("named", ["--platform", "retroarch", "--dest",
                                               str(repo.dest)]),
                                    ("detected", [])):
                    runs[name] = subprocess.run(
                        [sys.executable, str(REPO_ROOT / "install.py"), "--check",
                         "--no-input", *extra],
                        env=env, capture_output=True, text=True, timeout=60,
                        stdin=subprocess.DEVNULL)
            finally:
                os.chmod(home, 0o755)
        for name, proc in runs.items():
            with self.subTest(run=name):
                self.assertNotIn("Traceback", proc.stderr)
        # A named folder needs no detection and is checked.
        self.assertIn("0/5 present", runs["named"].stdout)
        self.assertIn(f"Could not look for setups: this user cannot open {home}",
                      runs["detected"].stderr)

    @unittest.skipIf(os.name == "nt" or os.geteuid() == 0,
                     "a folder without search permission is POSIX")
    def test_an_assumed_folder_is_not_called_detected(self):
        """The refusal ended "than the one detected here" for the built-in
        default taken because RetroArch was not detected."""
        with FakeRepository() as repo:
            home = repo.root / "home"
            home.mkdir()
            env = {**repo.env(), "HOME": str(home), "RETROBIOS_OS": "linux"}
            os.chmod(home, 0o000)
            try:
                proc = subprocess.run(
                    [sys.executable, str(REPO_ROOT / "install.py"), "--check",
                     "--no-input", "--platform", "retroarch"],
                    env=env, capture_output=True, text=True, timeout=60,
                    stdin=subprocess.DEVNULL)
            finally:
                os.chmod(home, 0o755)
        self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
        said = " ".join(proc.stderr.split())
        self.assertNotIn("the one detected here", said)
        self.assertIn("than the one taken as RetroArch's default, RetroArch not being "
                      "found here (name another with --dest).", said)


class APartlyClosedHome(unittest.TestCase):
    """~/.config/EmuDeck at 000 (a sudo run leaves it owned by root) made
    detection return nothing: the readable retroarch.cfg beside it was
    never read, and the run fell back to ~/.config/retroarch/system, a
    folder that RetroArch does not read."""

    @unittest.skipIf(os.name == "nt" or os.geteuid() == 0,
                     "a folder without search permission is POSIX")
    def test_retroarch_is_found_past_the_closed_folder(self):
        with FakeRepository() as repo:
            home = repo.root / "home"
            closed = home / ".config" / "EmuDeck"
            closed.mkdir(parents=True)
            cfg = home / ".config" / "retroarch" / "retroarch.cfg"
            cfg.parent.mkdir()
            system = repo.root / "elsewhere" / "system"
            system.mkdir(parents=True)
            cfg.write_text(f'system_directory = "{system}"\n')
            env = {**repo.env(), "HOME": str(home), "RETROBIOS_OS": "linux"}
            runs = {}
            os.chmod(closed, 0o000)
            try:
                for name, extra in (("detected", []),
                                    ("named", ["--platform", "retroarch"]),
                                    ("events", ["--events"])):
                    runs[name] = subprocess.run(
                        [sys.executable, str(REPO_ROOT / "install.py"), "--check",
                         "--no-input", *extra],
                        env=env, capture_output=True, text=True, timeout=60,
                        stdin=subprocess.DEVNULL)
            finally:
                os.chmod(closed, 0o755)
        refusal = (f"Could not look for setups: this user cannot open {closed} "
                   "(Permission denied)")
        for name, proc in runs.items():
            with self.subTest(run=name):
                self.assertNotIn("Traceback", proc.stderr)
                self.assertEqual(" ".join(proc.stderr.split()).count(refusal), 1,
                                 proc.stderr)
                self.assertNotIn(str(home / ".config" / "retroarch" / "system"),
                                 proc.stdout + proc.stderr)
        self.assertIn(f"Found RetroArch at {system}", runs["detected"].stdout)
        self.assertIn("0/5 present", runs["named"].stdout)
        self.assertNotIn("not detected", runs["named"].stdout)
        events = [json.loads(line) for line in runs["events"].stdout.splitlines()]
        kinds = [event["event"] for event in events]
        self.assertLess(kinds.index("unsearched"), kinds.index("detect"))
        unsearched = events[kinds.index("unsearched")]
        self.assertEqual((unsearched["folder"], unsearched["reason"]),
                         (str(closed), "Permission denied"))
        detect = events[kinds.index("detect")]
        self.assertEqual([entry["bios_dir"] for entry in detect["platforms"]], [str(system)])


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))

    @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:
        return True


class ColourFollowsNoColorOrg(unittest.TestCase):
    """no-color.org: only a NO_COLOR that is set and not empty turns colour
    off. An empty one, as `NO_COLOR= cmd` leaves, turned it off too."""

    def _colour(self, **env: str) -> bool:
        clean = {k: v for k, v in os.environ.items()
                 if k not in ("NO_COLOR", "FORCE_COLOR", "CLICOLOR_FORCE", "TERM")}
        with mock.patch.dict(os.environ, {**clean, **env}, clear=True), \
                mock.patch.object(install, "_enable_ansi", return_value=True):
            return install.Terminal(_Tty()).colour

    def test_an_empty_no_color_leaves_colour_on(self):
        self.assertTrue(self._colour())
        self.assertTrue(self._colour(NO_COLOR=""))
        self.assertFalse(self._colour(NO_COLOR="1"))

    def test_force_color_paints_a_pipe(self):
        with mock.patch.object(install, "_enable_ansi", return_value=True):
            for env, expected in (({}, False), ({"FORCE_COLOR": "1"}, True),
                                  ({"CLICOLOR_FORCE": "1"}, True),
                                  ({"FORCE_COLOR": "0"}, False),
                                  ({"FORCE_COLOR": "1", "NO_COLOR": "1"}, False)):
                clean = {k: v for k, v in os.environ.items()
                         if k not in ("NO_COLOR", "FORCE_COLOR", "CLICOLOR_FORCE")}
                with self.subTest(env=env), \
                        mock.patch.dict(os.environ, {**clean, **env}, clear=True):
                    term = install.Terminal(io.StringIO())
                    self.assertEqual(term.colour, expected)
                    self.assertFalse(term.live)


class SmallWordings(unittest.TestCase):
    def test_check_asks_which_to_check(self):
        """--check with several platforms asked "Install for:", and
        installs nothing."""
        out = io.StringIO()
        with mock.patch("builtins.input", lambda _prompt="": "1"), \
                contextlib.redirect_stdout(out):
            install._prompt_platform_choice(
                [("emudeck", Path("/e")), ("retroarch", Path("/r"))], "Check:")
        self.assertIn("Check:", out.getvalue())
        self.assertNotIn("Install for:", out.getvalue())

    def test_nothing_detected_is_said(self):
        with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as home:
            env = dict(os.environ, HOME=home, USERPROFILE=home, XDG_CONFIG_HOME=home,
                       RETROBIOS_OS="linux")
            proc = subprocess.run(
                [sys.executable, str(REPO_ROOT / "install.py"), "--list-platforms"],
                env=env, capture_output=True, text=True, timeout=60, check=False,
            )
        self.assertEqual(proc.returncode, 0, proc.stderr)
        self.assertIn("Detected on this system: none", proc.stdout)


class WindowsMandatoryLock(unittest.TestCase):
    """Issue 82: on Windows the partial copy's lock is mandatory, so hashing
    it by reopening its path failed with a sharing violation, and every file
    of 16 MB or more was reported as "folder not writable"."""

    def test_the_locked_partial_is_hashed_through_its_handle(self):
        real_open = open

        def locked_open(path, *args, **kwargs):
            if str(path).endswith(".part"):
                raise PermissionError(33, "The process cannot access the file")
            return real_open(path, *args, **kwargs)

        with _Server(), tempfile.TemporaryDirectory() as tmp, \
                mock.patch.object(install, "RESUME_FROM_SIZE", 1024), \
                mock.patch("builtins.open", locked_open):
            dest, ok, reason = install._download_one(dict(ENTRY), Path(tmp))
        self.assertEqual((dest, ok, reason), ("big.bin", True, ""))

    def test_a_read_only_file_in_place_is_named_as_such(self):
        with _Server(), tempfile.TemporaryDirectory() as tmp, \
                mock.patch.object(install.os, "replace",
                                  side_effect=PermissionError(5, "Access is denied")), \
                mock.patch.object(install, "_read_only_file", return_value=True):
            _dest, ok, reason = install._download_one(dict(ENTRY), Path(tmp))
        self.assertFalse(ok)
        self.assertEqual(reason, install.READ_ONLY)

    def test_a_read_only_file_in_place_is_made_writable_on_windows(self):
        with tempfile.TemporaryDirectory() as tmp:
            dest = Path(tmp) / "big.bin"
            dest.write_bytes(b"old")
            os.chmod(dest, 0o444)
            with mock.patch.object(install.os, "name", "nt"):
                install._clear_read_only(dest)
            self.assertTrue(os.access(dest, os.W_OK))

    def test_elsewhere_the_attribute_is_left_alone(self):
        with tempfile.TemporaryDirectory() as tmp:
            dest = Path(tmp) / "big.bin"
            dest.write_bytes(b"old")
            os.chmod(dest, 0o444)
            with mock.patch.object(install.os, "name", "posix"):
                install._clear_read_only(dest)
            self.assertFalse(os.access(dest, os.W_OK))
            os.chmod(dest, 0o644)

    def test_a_refused_folder_keeps_its_own_reason(self):
        with _Server(), tempfile.TemporaryDirectory() as tmp, \
                mock.patch.object(install.os, "replace",
                                  side_effect=PermissionError(13, "Permission denied")), \
                mock.patch.object(install, "_read_only_file", return_value=False):
            _dest, ok, reason = install._download_one(dict(ENTRY), Path(tmp))
        self.assertEqual((ok, reason), (False, install.NOT_WRITABLE))


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