fix: refuse a hash scrape that reads no BIOS set

The MAME and FBNeo hash scrapers cached an empty parse for 24 hours and
reported OK. Both now raise before writing, and an empty cache is never
served as fresh.
This commit is contained in:
Abdessamad Derraz committed 2026-10-21 05:04:36 +02:00
1 parent 348c56e94c
commit 981468874c
3 files changed
+73 -1

No files matched your search

+6
View File
@@ -42,6 +42,8 @@ def _is_cache_fresh() -> bool:
return False
try:
data = json.loads(CACHE_PATH.read_text(encoding="utf-8"))
if not data.get("bios_sets"):
return False
fetched_at = datetime.fromisoformat(data["fetched_at"])
return datetime.now(timezone.utc) - fetched_at < timedelta(hours=STALE_HOURS)
except (json.JSONDecodeError, KeyError, ValueError):
@@ -178,6 +180,10 @@ def fetch_and_cache(force: bool = False) -> dict[str, Any]:
log.info("parsing source tree")
bios_sets = parse_fbneo_source_tree(str(CLONE_DIR))
if not bios_sets or not commit:
# An empty read was cached for a day and merged nothing, and
# the pipeline step reported OK.
raise RuntimeError(f"no BIOS set read at FBNeo {version} ({commit[:12]})")
cache: dict[str, Any] = {
"source": "finalburnneo/FBNeo",
+6 -1
View File
@@ -62,7 +62,7 @@ def _load_cache() -> dict[str, Any] | None:
def _is_stale(cache: dict[str, Any] | None) -> bool:
if cache is None:
if cache is None or not cache.get("bios_sets"):
return True
fetched_at = cache.get("fetched_at")
if not fetched_at:
@@ -292,6 +292,11 @@ def _fetch_hashes(force: bool) -> dict[str, Any]:
bios_sets = parse_mame_source_tree(str(_CLONE_DIR))
version = _parse_version_tag(tag)
commit = _get_commit()
if not bios_sets or not commit:
# An empty read was cached for a day and every profile entry
# came out "out of scope", so nothing was refreshed and the
# pipeline step reported OK.
raise RuntimeError(f"no BIOS root set read at {tag}")
data: dict[str, Any] = {
"source": "mamedev/mame",
+61
View File
@@ -880,5 +880,66 @@ class OneNativeIdOneSystem(unittest.TestCase):
ids = {_norm_system_id(s) for s in data["systems"]}
self.assertEqual(wanted & ids & machines, wanted)
class HashScraperRefusesAnEmptyRead(unittest.TestCase):
"""A source tree that yields no BIOS set is never cached as a result.
Cached, it was served for a day: every profile entry came out "out of
scope", nothing was refreshed and pipeline steps 2a and 2a2 said OK.
"""
def _fresh(self, sets: dict) -> dict:
from datetime import datetime, timezone # noqa: PLC0415
return {"fetched_at": datetime.now(timezone.utc).isoformat(),
"commit": "abc", "bios_sets": sets}
def test_mame_raises_and_writes_no_cache(self):
from scraper import mame_hash_scraper as m # noqa: PLC0415
with tempfile.TemporaryDirectory() as tmp:
cache = Path(tmp) / "mame-hashes.json"
with mock.patch.object(m, "_CACHE_PATH", cache), \
mock.patch.object(m, "_get_release_tag", return_value="mame0289"), \
mock.patch.object(m, "_sparse_clone"), \
mock.patch.object(m, "_cleanup"), \
mock.patch.object(m, "_get_commit", return_value="abc"), \
mock.patch.object(m, "parse_mame_source_tree", return_value={}):
with self.assertRaises(RuntimeError):
m._fetch_hashes(force=True)
self.assertFalse(cache.exists())
def test_mame_does_not_serve_an_empty_cache(self):
from scraper import mame_hash_scraper as m # noqa: PLC0415
self.assertTrue(m._is_stale(self._fresh({})))
self.assertFalse(m._is_stale(self._fresh({"neogeo": {}})))
def test_fbneo_raises_and_writes_no_cache(self):
from scraper import fbneo_hash_scraper as f # noqa: PLC0415
with tempfile.TemporaryDirectory() as tmp:
cache = Path(tmp) / "fbneo-hashes.json"
with mock.patch.object(f, "CACHE_PATH", cache), \
mock.patch.object(f, "_sparse_clone"), \
mock.patch.object(f, "_cleanup"), \
mock.patch.object(f, "_extract_version", return_value=("v1.0.0.03", "abc")), \
mock.patch.object(f, "parse_fbneo_source_tree", return_value={}):
with self.assertRaises(RuntimeError):
f.fetch_and_cache(force=True)
self.assertFalse(cache.exists())
def test_fbneo_does_not_serve_an_empty_cache(self):
from scraper import fbneo_hash_scraper as f # noqa: PLC0415
with tempfile.TemporaryDirectory() as tmp:
cache = Path(tmp) / "fbneo-hashes.json"
with mock.patch.object(f, "CACHE_PATH", cache):
cache.write_text(json.dumps(self._fresh({})))
self.assertFalse(f._is_cache_fresh())
cache.write_text(json.dumps(self._fresh({"neogeo": {}})))
self.assertTrue(f._is_cache_fresh())
if __name__ == "__main__":
unittest.main()