mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-23 20:29:06 -05:00
Abdessamad Derraz
fac47dc4c0
Without jsonschema the step printed only the missing module, and the run then read COMPLETE while PR validation runs the same check. The skip now names the install command and the PR check.
404 lines
17 KiB
Python
404 lines
17 KiB
Python
"""What the maintainer tools print: names a user can act on, counts that add up."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import io
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
|
|
|
import common # noqa: E402
|
|
import diff_truth # noqa: E402
|
|
import pipeline # noqa: E402
|
|
import truth # noqa: E402
|
|
import verify # noqa: E402
|
|
|
|
|
|
class ABuildbotNameSelectsItsProfile(unittest.TestCase):
|
|
"""`--emulator pcfx` was refused and pointed at two PlayStation cores,
|
|
though pcfx is in the cores: of beetle_pcfx."""
|
|
|
|
PROFILES = {
|
|
"beetle_pcfx": {"type": "libretro", "cores": ["pcfx"]},
|
|
"pcsx1": {"type": "libretro"},
|
|
"pcsx2": {"type": "standalone"},
|
|
"a": {"type": "libretro", "cores": ["shared"]},
|
|
"b": {"type": "libretro", "cores": ["shared"]},
|
|
}
|
|
|
|
def test_a_core_name_serves_the_profile(self):
|
|
selected = common.select_emulator_profiles(["pcfx"], self.PROFILES)
|
|
self.assertEqual([name for name, _profile in selected], ["beetle_pcfx"])
|
|
|
|
def test_a_name_two_profiles_claim_names_both(self):
|
|
with self.assertRaises(common.ProfileSelectionError) as caught:
|
|
common.select_emulator_profiles(["shared"], self.PROFILES)
|
|
self.assertIn("a, b", str(caught.exception))
|
|
|
|
def test_an_unknown_name_still_gets_a_hint(self):
|
|
with self.assertRaises(common.ProfileSelectionError) as caught:
|
|
common.select_emulator_profiles(["pcsx3"], self.PROFILES)
|
|
self.assertIn("Did you mean", str(caught.exception))
|
|
|
|
def test_cross_reference_reports_the_resolved_profile(self):
|
|
"""cross_reference filtered on the typed name, found no profile under
|
|
it and answered `pcfx is not run by None`."""
|
|
def report(name: str) -> subprocess.CompletedProcess:
|
|
return subprocess.run(
|
|
[sys.executable, "scripts/cross_reference.py", "--emulator", name],
|
|
cwd=REPO_ROOT, capture_output=True, text=True, timeout=600,
|
|
)
|
|
by_core, by_profile = report("pcfx"), report("beetle_pcfx")
|
|
self.assertEqual(by_core.returncode, 0, by_core.stderr)
|
|
self.assertEqual(by_core.stdout, by_profile.stdout)
|
|
|
|
|
|
class AMistypedTargetGetsTheClosestName(unittest.TestCase):
|
|
"""`--target swich` printed every target and alias on two lines."""
|
|
|
|
def test_did_you_mean(self):
|
|
with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as tmp:
|
|
targets = Path(tmp) / "targets"
|
|
targets.mkdir()
|
|
names = [f"board-{n}" for n in range(30)] + ["nintendo-switch"]
|
|
(targets / "p.yml").write_text(
|
|
"targets:\n" + "".join(f" {n}:\n cores: [a]\n" for n in names)
|
|
)
|
|
(targets / "_overrides.yml").write_text(
|
|
"p:\n targets:\n nintendo-switch:\n aliases: [switch]\n"
|
|
)
|
|
with self.assertRaises(ValueError) as caught:
|
|
common.load_target_config("p", "swich", tmp)
|
|
message = str(caught.exception)
|
|
self.assertIn("Did you mean: switch?", message)
|
|
self.assertIn("lists all 31", message)
|
|
self.assertNotIn("board-0", message)
|
|
|
|
|
|
class TheTruthDiffShowsWhereAndAsksNothing(unittest.TestCase):
|
|
"""The markdown wrote `**Remove** hiscore.dat (phantom)` beside six
|
|
`**Add** hiscore.dat`, none with a path, for files profiled cores read."""
|
|
|
|
def _report(self) -> dict:
|
|
truth_sys = {"files": [
|
|
{"name": "hiscore.dat", "path": "mame2003/hiscore.dat",
|
|
"md5": "a" * 32, "_cores": ["mame2003"]},
|
|
], "_coverage": {"cores_unprofiled": []}}
|
|
scraped_sys = {"files": [
|
|
{"name": "hiscore.dat", "destination": "fbneo/hiscore.dat", "md5": "b" * 32},
|
|
]}
|
|
report = truth.diff_platform_truth(
|
|
{"systems": {"arcade": truth_sys}}, {"systems": {"arcade": scraped_sys}}
|
|
)
|
|
report["platform"] = "p"
|
|
return report
|
|
|
|
def test_entries_carry_their_destination(self):
|
|
div = self._report()["divergences"]["arcade"]
|
|
self.assertEqual(div["extra_phantom"][0]["destination"], "fbneo/hiscore.dat")
|
|
self.assertEqual(div["missing"][0]["destination"], "mame2003/hiscore.dat")
|
|
|
|
def test_neither_format_asks_for_a_removal(self):
|
|
report = self._report()
|
|
for text in (diff_truth._format_markdown(report),
|
|
diff_truth._format_terminal(report)):
|
|
self.assertNotIn("Remove", text)
|
|
self.assertNotIn("phantom", text.lower())
|
|
self.assertIn("fbneo/hiscore.dat", text)
|
|
self.assertIn("mame2003/hiscore.dat", text)
|
|
|
|
|
|
class TheVerifySummaryCountsDiscrepancies(unittest.TestCase):
|
|
"""19 DISCREPANCY lines above a last line that read clean."""
|
|
|
|
def test_the_closing_line_names_them(self):
|
|
details = [
|
|
{"system": "arcade", "name": "aes.zip", "status": verify.Status.OK,
|
|
"discrepancy": "file present (OK) but hash differs"},
|
|
# Printed as MISSING, so not a second time as a discrepancy.
|
|
{"system": "arcade", "name": "x.zip", "status": verify.Status.MISSING,
|
|
"discrepancy": "unmet"},
|
|
{"system": "arcade", "name": "x.zip", "status": verify.Status.MISSING},
|
|
]
|
|
summary = verify._overall_summary([
|
|
({"severity_counts": {}, "details": details}, ["RetroArch"]),
|
|
({"severity_counts": {}, "details": []}, ["Batocera"]),
|
|
])
|
|
self.assertIn("1 discrepancy (RetroArch)", summary)
|
|
|
|
def test_a_shared_destination_names_the_path_and_both_sides(self):
|
|
result = {"name": "disk.rom", "status": verify.Status.UNTESTED}
|
|
entry = {"name": "disk.rom", "md5": "793f8678" + "0" * 24}
|
|
self.assertTrue(verify._twin_unmet(
|
|
result, "coco", entry, "nec-pc-88", "bios/disk.rom"
|
|
))
|
|
message = result["discrepancy"]
|
|
for part in ("bios/disk.rom", "nec-pc-88", "coco", "md5 793f8678"):
|
|
self.assertIn(part, message)
|
|
|
|
|
|
class ThePipelineCountsItsRealSteps(unittest.TestCase):
|
|
"""Labels read N/8 for a run of 24 steps, and the summary used keys
|
|
the log never printed."""
|
|
|
|
def test_every_recorded_step_has_a_label(self):
|
|
import re # noqa: PLC0415
|
|
|
|
source = (REPO_ROOT / "scripts" / "pipeline.py").read_text(encoding="utf-8")
|
|
keys = set(re.findall(r'results\["(\w+)"\] =', source))
|
|
self.assertEqual(keys - set(pipeline.STEP_LABELS), set())
|
|
|
|
def test_a_label_written_out_is_one_of_the_list(self):
|
|
import re # noqa: PLC0415
|
|
|
|
source = (REPO_ROOT / "scripts" / "pipeline.py").read_text(encoding="utf-8")
|
|
written = re.findall(r'run\([^()]*?, "([^"]+)"\)', source)
|
|
numbers = {label.split(" ", 1)[0] for label in pipeline.STEP_LABELS.values()}
|
|
for label in written:
|
|
with self.subTest(label=label):
|
|
# A step's own label, or a second command of a listed step.
|
|
self.assertIn(label.split(" ", 1)[0], numbers)
|
|
|
|
def test_no_title_counts_eight_steps(self):
|
|
"""Three print() headers of step 5 still read "5/8" beside the
|
|
computed "5 consistency check [20/24]"."""
|
|
import re # noqa: PLC0415
|
|
|
|
source = (REPO_ROOT / "scripts" / "pipeline.py").read_text(encoding="utf-8")
|
|
self.assertEqual(re.findall(r'"[^"\n]*---[^"\n]*\b\d+[a-z]?/\d+\b[^"\n]*"', source), [])
|
|
|
|
def test_progress_comes_from_the_list(self):
|
|
total = len(pipeline.STEP_LABELS)
|
|
title = pipeline.step_title(pipeline.STEP_LABELS["render_site"])
|
|
self.assertEqual(title, f"8b render site [{total}/{total}]")
|
|
self.assertNotIn("/8", pipeline.step_title(pipeline.STEP_LABELS["verify"]))
|
|
|
|
def test_a_skipped_schema_check_says_how_to_run_it(self):
|
|
"""Step 0 said only "jsonschema not installed", then the run read
|
|
COMPLETE and the schema failed in PR validation."""
|
|
argv = ["pipeline.py", "--offline", "--skip-packs", "--skip-docs"]
|
|
out = io.StringIO()
|
|
real_find_spec = pipeline.importlib.util.find_spec
|
|
with mock.patch.object(sys, "argv", argv), mock.patch.object(
|
|
pipeline, "run", return_value=(True, "")
|
|
), mock.patch.object(
|
|
pipeline.importlib.util, "find_spec",
|
|
side_effect=lambda name, *a: None if name == "jsonschema"
|
|
else real_find_spec(name, *a),
|
|
), contextlib.redirect_stdout(out), contextlib.suppress(SystemExit):
|
|
pipeline.main()
|
|
line = next(
|
|
row for row in out.getvalue().splitlines()
|
|
if "0 validate schemas" in row and "SKIPPED" in row
|
|
)
|
|
self.assertIn('pip install "jsonschema[format-nongpl]"', line)
|
|
self.assertIn("PR validation runs this check", line)
|
|
|
|
def test_an_offline_buildbot_check_has_a_summary_row(self):
|
|
argv = ["pipeline.py", "--offline", "--check-buildbot", "--skip-packs",
|
|
"--skip-docs"]
|
|
out = io.StringIO()
|
|
with mock.patch.object(sys, "argv", argv), mock.patch.object(
|
|
pipeline, "run", return_value=(True, "RetroArch: 1/1 OK\n")
|
|
), contextlib.redirect_stdout(out), contextlib.suppress(SystemExit):
|
|
pipeline.main()
|
|
summary = out.getvalue().split("=" * 60)[1]
|
|
self.assertIn("2b check_buildbot", summary)
|
|
# The number the run printed for the step leads its summary row.
|
|
self.assertIn("4d bundled_only", summary)
|
|
self.assertTrue(
|
|
pipeline.step_title(pipeline.STEP_LABELS["bundled_only"]).startswith("4d ")
|
|
)
|
|
|
|
|
|
class AHashConflictNamesBothCores(unittest.TestCase):
|
|
"""The warning named the second core only, and not which value stayed."""
|
|
|
|
def test_both_cores_and_the_kept_value(self):
|
|
truth._CONFLICTS_SHOWN.clear()
|
|
system: dict = {}
|
|
err = io.StringIO()
|
|
with contextlib.redirect_stderr(err):
|
|
truth._merge_file_into_system(
|
|
system, {"name": "G.yaml", "path": "r/G.yaml", "sha1": "5e" * 20},
|
|
"pcsx2", None,
|
|
)
|
|
truth._merge_file_into_system(
|
|
system, {"name": "G.yaml", "path": "r/G.yaml", "sha1": "7e" * 20},
|
|
"yaps2", None,
|
|
)
|
|
self.assertIn(f"{'5e' * 20} (pcsx2) kept", err.getvalue())
|
|
self.assertIn(f"{'7e' * 20} (yaps2)", err.getvalue())
|
|
|
|
def test_the_origin_is_not_written_to_the_model(self):
|
|
profile = {"type": "libretro", "systems": ["sony-playstation"],
|
|
"files": [{"name": "a.bin", "system": "sony-playstation",
|
|
"md5": "a" * 32}]}
|
|
model = truth.generate_platform_truth(
|
|
"p", {"cores": ["c"], "systems": {}}, {}, {"c": profile}
|
|
)
|
|
entries = [f for s in model["systems"].values() for f in s.get("files", [])]
|
|
self.assertTrue(entries)
|
|
self.assertFalse(any("_hash_by" in f for f in entries))
|
|
|
|
|
|
class HelpTextsWrapBetweenWords(unittest.TestCase):
|
|
def test_check_freshness_areas(self):
|
|
proc = subprocess.run(
|
|
[sys.executable, "scripts/check_freshness.py", "--help"],
|
|
capture_output=True, text=True, cwd=REPO_ROOT, check=False,
|
|
env={**os.environ, "COLUMNS": "80"}, timeout=60,
|
|
)
|
|
self.assertEqual(proc.returncode, 0, proc.stderr)
|
|
self.assertIn("coreinfo", " ".join(proc.stdout.split()))
|
|
|
|
def test_docstring_descriptions_keep_their_layout(self):
|
|
"""romset_recipes ran its two questions into one paragraph and both
|
|
help texts printed reST double backquotes."""
|
|
for script, line in (("romset_recipes.py", "--identify"),
|
|
("validate_site.py", None)):
|
|
with self.subTest(script=script):
|
|
proc = subprocess.run(
|
|
[sys.executable, f"scripts/{script}", "--help"],
|
|
capture_output=True, text=True, cwd=REPO_ROOT, check=False,
|
|
env={**os.environ, "COLUMNS": "80"}, timeout=60,
|
|
)
|
|
self.assertEqual(proc.returncode, 0, proc.stderr)
|
|
self.assertNotIn("``", proc.stdout)
|
|
if line:
|
|
self.assertIn(line, proc.stdout.splitlines())
|
|
|
|
|
|
class ListPlatformsTakesTheSharedFlag(unittest.TestCase):
|
|
"""verify, generate_pack, slots and pipeline say --include-archived."""
|
|
|
|
def test_both_spellings(self):
|
|
for flag in ("--include-archived", "--all"):
|
|
with self.subTest(flag=flag):
|
|
env = {k: v for k, v in os.environ.items() if k != "GITHUB_OUTPUT"}
|
|
proc = subprocess.run(
|
|
[sys.executable, "scripts/list_platforms.py", flag],
|
|
capture_output=True, text=True, cwd=REPO_ROOT, check=False,
|
|
env=env, timeout=60,
|
|
)
|
|
self.assertEqual(proc.returncode, 0, proc.stderr)
|
|
self.assertIn("retropie", proc.stdout)
|
|
|
|
def test_the_pack_list_marks_archived_platforms(self):
|
|
"""generate_pack --list printed retropie like a platform with a
|
|
release pack."""
|
|
proc = subprocess.run(
|
|
[sys.executable, "scripts/generate_pack.py", "--list"],
|
|
capture_output=True, text=True, cwd=REPO_ROOT, check=False, timeout=120,
|
|
)
|
|
self.assertEqual(proc.returncode, 0, proc.stderr)
|
|
lines = proc.stdout.splitlines()
|
|
self.assertIn("retropie (archived)", lines)
|
|
self.assertIn("retroarch", lines)
|
|
|
|
|
|
class NoScriptReadsTheTokenOnItsOwn(unittest.TestCase):
|
|
"""Nine scripts read GITHUB_TOKEN from the environment after
|
|
github_token() existed, and went anonymous when run on their own.
|
|
download.py is the user's downloader: it runs without the repository and
|
|
a user's gh session is not its business."""
|
|
|
|
READS = ('environ.get("GITHUB_TOKEN"', "environ['GITHUB_TOKEN']",
|
|
'environ["GITHUB_TOKEN"]', 'environ.get("GH_TOKEN"')
|
|
OWNERS = {"scripts/common.py", "scripts/download.py"}
|
|
|
|
def test_every_reader_asks_github_token(self):
|
|
for path in sorted((REPO_ROOT / "scripts").rglob("*.py")):
|
|
relative = path.relative_to(REPO_ROOT).as_posix()
|
|
if relative in self.OWNERS:
|
|
continue
|
|
for number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
|
if " = " in line and line.split(" = ")[0].strip().startswith("env["):
|
|
continue
|
|
with self.subTest(file=relative, line=number):
|
|
self.assertFalse(any(read in line for read in self.READS), line.strip())
|
|
|
|
|
|
class TheGitHubTokenComesFromGh(unittest.TestCase):
|
|
"""Only refresh_stale asked gh: profile_sync refused the run and the
|
|
scrapers went anonymous for a maintainer logged in to gh."""
|
|
|
|
def setUp(self):
|
|
common._gh_cli_token.cache_clear()
|
|
self.addCleanup(common._gh_cli_token.cache_clear)
|
|
env = {k: v for k, v in os.environ.items()
|
|
if k not in ("GITHUB_TOKEN", "GH_TOKEN", "RETROBIOS_OFFLINE")}
|
|
patcher = mock.patch.dict(os.environ, env, clear=True)
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
@staticmethod
|
|
def _gh(token: str = "gho_fromgh", code: int = 0):
|
|
return mock.patch(
|
|
"subprocess.run",
|
|
return_value=subprocess.CompletedProcess(["gh"], code, token, ""),
|
|
)
|
|
|
|
def test_gh_answers_when_the_variable_is_unset(self):
|
|
with self._gh():
|
|
self.assertEqual(common.github_token(), "gho_fromgh")
|
|
# Subprocesses of the run inherit it.
|
|
self.assertEqual(os.environ.get("GITHUB_TOKEN"), "gho_fromgh")
|
|
|
|
def test_the_variable_wins(self):
|
|
os.environ["GITHUB_TOKEN"] = "set"
|
|
with self._gh() as run:
|
|
self.assertEqual(common.github_token(), "set")
|
|
run.assert_not_called()
|
|
|
|
def test_no_gh_means_anonymous(self):
|
|
with mock.patch("subprocess.run", side_effect=FileNotFoundError("gh")):
|
|
self.assertEqual(common.github_token(), "")
|
|
with self._gh("", 1):
|
|
common._gh_cli_token.cache_clear()
|
|
self.assertEqual(common.github_token(), "")
|
|
|
|
def test_offline_runs_never_ask(self):
|
|
os.environ["RETROBIOS_OFFLINE"] = "1"
|
|
with self._gh() as run:
|
|
self.assertEqual(common.github_token(), "")
|
|
run.assert_not_called()
|
|
|
|
def test_every_api_caller_reads_it(self):
|
|
import profile_sync # noqa: PLC0415
|
|
import upstream # noqa: PLC0415
|
|
from scraper import base_scraper # noqa: PLC0415
|
|
|
|
with self._gh():
|
|
self.assertEqual(
|
|
upstream._headers("https://api.github.com/repos/o/n")["Authorization"],
|
|
"token gho_fromgh",
|
|
)
|
|
os.environ.pop("GITHUB_TOKEN")
|
|
with self._gh():
|
|
self.assertEqual(base_scraper.github_headers()["Authorization"],
|
|
"Bearer gho_fromgh")
|
|
os.environ.pop("GITHUB_TOKEN")
|
|
with self._gh(), contextlib.redirect_stderr(io.StringIO()):
|
|
profile_sync._check_quota(606, offline=False)
|
|
|
|
def test_another_forge_never_triggers_gh(self):
|
|
import upstream # noqa: PLC0415
|
|
|
|
with self._gh() as run:
|
|
headers = upstream._headers("https://gitlab.com/api/v4/projects/1")
|
|
self.assertNotIn("Authorization", headers)
|
|
run.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|