From ba93a830f167ec48beb0fd698efc20a31e1985a0 Mon Sep 17 00:00:00 2001 From: nichinichisou Date: Wed, 30 Sep 2026 07:23:19 +0800 Subject: [PATCH] fix(jp): address review regressions Defer historical APK validation until a local resource cache miss, and reuse bounded APKS/XAPK expansions without invalidating active readers. Bind web preflight checks to the requested region, normalize schemeless TLS API roots, and preserve raw settings for config-check diagnostics. Add synthetic regressions for replay, archive lifetime and invalidation, region selection, API origin restrictions, and diagnostic JSON output. Co-Authored-By: Claude Code --- src/nnnotes/apkset.py | 56 ++++++++++++++----- src/nnnotes/catalog.py | 19 ++++++- src/nnnotes/cli.py | 4 +- src/nnnotes/cli_assets.py | 27 ++++----- src/nnnotes/jp.py | 4 +- tests/test_apkset.py | 115 ++++++++++++++++++++++++++++++++++++++ tests/test_configfile.py | 24 ++++++++ tests/test_jp.py | 108 +++++++++++++++++++++++++++++++++++ 8 files changed, 321 insertions(+), 36 deletions(-) create mode 100644 tests/test_apkset.py diff --git a/src/nnnotes/apkset.py b/src/nnnotes/apkset.py index 1852e98..52534bd 100644 --- a/src/nnnotes/apkset.py +++ b/src/nnnotes/apkset.py @@ -1,29 +1,60 @@ """Read a base APK and its splits, an APK directory, or an APKS/XAPK archive. The base manifest wins over split manifests. Resource members are resolved across the set; -no merged APK is produced. Nested APKs are spooled so large asset packs need not stay in RAM. +no merged APK is produced. Nested APKs are expanded once per archive revision to disk, with +at most two idle APK sets retained per process; active readers keep their set alive until closed. """ from __future__ import annotations import shutil import tempfile +import threading import zipfile from contextlib import ExitStack +from functools import lru_cache from pathlib import Path +from .cache import file_id + + +_EXPAND_LOCK = threading.Lock() + + +@lru_cache(maxsize=2) +def _expanded_apks(identity): + """An owned temporary directory and its APK paths; readers retain it across cache eviction.""" + with zipfile.ZipFile(identity[0]) as outer: + names = [n for n in outer.namelist() if n.lower().endswith(".apk")] + bases = [n for n in names if Path(n).name == "base.apk"] + if len(bases) != 1: + raise ValueError("APK archive must contain exactly one base.apk") + directory = tempfile.TemporaryDirectory(prefix="nnnotes-apks-") + try: + paths = [] + for i, name in enumerate(bases + sorted(n for n in names if n not in bases)): + path = Path(directory.name) / f"{i}.apk" + with outer.open(name) as stream, path.open("wb") as target: + shutil.copyfileobj(stream, target) + paths.append(path) + return directory, paths + except BaseException: + directory.cleanup() + raise + class ApkSet: def __init__(self, source): self.source = source self._stack = ExitStack() self._members = {} + self._expanded = None def __enter__(self): try: self._open() return self except BaseException: - self._stack.close() + self.__exit__(None, None, None) raise def _add(self, source): @@ -42,18 +73,9 @@ def _open(self): raise FileNotFoundError("APK directory has no base.apk") paths = [base] + sorted(p for p in path.glob("*.apk") if p != base) elif path.suffix.lower() in (".apks", ".xapk"): - outer = self._stack.enter_context(zipfile.ZipFile(path)) - names = [n for n in outer.namelist() if n.lower().endswith(".apk")] - bases = [n for n in names if Path(n).name == "base.apk"] - if len(bases) != 1: - raise ValueError("APK archive must contain exactly one base.apk") - for name in bases + sorted(n for n in names if n not in bases): - tmp = self._stack.enter_context(tempfile.SpooledTemporaryFile(max_size=16 * 1024 * 1024)) - with outer.open(name) as stream: - shutil.copyfileobj(stream, tmp) - tmp.seek(0) - self._add(tmp) - return + with _EXPAND_LOCK: # lru_cache alone permits duplicate concurrent expansions + self._expanded = _expanded_apks(file_id(path.resolve())) + paths = self._expanded[1] else: paths = [path] if path.name == "base.apk": @@ -74,4 +96,8 @@ def read(self, name): return archive.read(info) def __exit__(self, *args): - return self._stack.__exit__(*args) + try: + return self._stack.__exit__(*args) + finally: + self._members.clear() + self._expanded = None diff --git a/src/nnnotes/catalog.py b/src/nnnotes/catalog.py index 73a9c45..55e4c21 100644 --- a/src/nnnotes/catalog.py +++ b/src/nnnotes/catalog.py @@ -76,10 +76,11 @@ class Catalog: `cdn`: the region's CDN base (needed only to download what the cache lacks); `bundle_key`: the bundle decryption key (needed only for bundles not yet in the cache). Either may be a function that returns it, called the first time it is needed (a ConfigError it raises is raised naming the file that needed the setting). + `apk_catalog`: the stored catalog to replay without opening an APK; JP local reads verify it before extraction. """ def __init__(self, catalog_bytes: bytes, cache_dir: Path, *, cdn=None, bundle_key=None, apk: Path | None = None, - source=None, session=None): + source=None, session=None, apk_catalog: bytes | None = None): self._settings = {"cdn": cdn, "bundle_key": bundle_key} self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(parents=True, exist_ok=True) @@ -88,10 +89,22 @@ def __init__(self, catalog_bytes: bytes, cache_dir: Path, *, cdn=None, bundle_ke self._sources = {"remote": catalog_bytes} self._locations: list[dict] | None = None self._parsed: tuple | None = None - if self.apk is not None: + if apk_catalog is not None: + self._sources["apk"] = apk_catalog + elif self.apk is not None: with ApkSet(self.apk) as z: self._sources["apk"] = z.read(APK_CATALOG) + def check_apk(self, archive) -> None: + """Before extracting a JP local file, verify the APK against this catalog's snapshot.""" + if self.source is not None and "apk" in self._sources: + try: + matches = archive.read(APK_CATALOG) == self._sources["apk"] + except KeyError: + matches = False + if not matches: + raise ConfigError("JP historical catalog needs the APK set it was imported with") + def local_cache_dir(self) -> Path: """JP embedded files also depend on the APK catalog, independently of the CDN snapshot.""" if self.source is not None and "apk" in self._sources: @@ -278,6 +291,7 @@ def fetch(self, b: Bundle) -> Path: raise apk_missing(f"bundle {b.name}") rel = b.internal_id[len(LOCAL_PREFIX):].lstrip("/") with ApkSet(self.apk) as z: + self.check_apk(z) data = z.read(APK_AA_DIR + rel) key = self._setting("bundle_key", b.name) if data[:7] != b"UnityFS" else None _write_atomic(dst, _unityfs(data, b.name, key)) @@ -329,6 +343,7 @@ def apk_bundle(self, name_contains: str) -> Path: dst = self.local_cache_dir() / "bundles" / name if not (dst.exists() and dst.stat().st_size > 0): dst.parent.mkdir(parents=True, exist_ok=True) + self.check_apk(z) data = z.read(names[0]) key = self._setting("bundle_key", name) if data[:7] != b"UnityFS" else None _write_atomic(dst, _unityfs(data, name, key)) diff --git a/src/nnnotes/cli.py b/src/nnnotes/cli.py index 4207a87..c7f549b 100644 --- a/src/nnnotes/cli.py +++ b/src/nnnotes/cli.py @@ -89,6 +89,8 @@ def load_config(args) -> Config: overrides = {k: getattr(args, dest, None) for k, (dest, _) in FLAG_SETTINGS.items()} cfg = Config.load(getattr(args, "config", None), overrides=overrides, flags={k: flag for k, (_, flag) in FLAG_SETTINGS.items()}) + if getattr(args, "func", None) is cmd_config_check: + return use(cfg) # diagnose raw settings before applying runtime defaults if cfg.provider() == "jp" and not cfg.has("catalog", "language"): cfg = cfg.for_region(cfg.region()) return use(cfg) @@ -617,7 +619,7 @@ def cmd_web(args, cfg): if args.web_regions or args.all_regions else None) # None: the one [catalog] region if (stories or models) and regions and len({cfg.provider(r) for r in regions}) > 1: raise ConfigError("build JP stories/models in a separate site directory from international releases") - if (stories or models) and regions and cfg.provider(regions[0]) == "jp": + if (stories or models) and regions: cfg = use(cfg.for_region(regions[0])) base = {"region": regions[0]} if regions else {} unknown = web.unknown_pairs(cfg, args.pair, regions) if args.pair else [] diff --git a/src/nnnotes/cli_assets.py b/src/nnnotes/cli_assets.py index 11b89f4..4b66606 100644 --- a/src/nnnotes/cli_assets.py +++ b/src/nnnotes/cli_assets.py @@ -301,12 +301,10 @@ def catalog(self, vid: str): session = Session(cfg, cfg.region()) cache = source.cache_dir(cache) cat = Catalog(remote, cache, cdn=lambda: cfg.cdn(cfg.region()), - bundle_key=lambda: _bundle_key(cfg), apk=cfg.path("paths", "apk"), + bundle_key=lambda: _bundle_key(cfg), apk_catalog=apk, source=source, session=session) - if source is not None and apk is not None and cat.sources().get("apk") != apk: - raise ConfigError("JP historical catalog needs the APK set it was imported with") - # Replay the APK catalog imported with this version, not today's catalog offsets. - cat._sources = {"remote": remote, **({"apk": apk} if apk is not None else {})} + # Replay imported offsets; the current APK is needed only for uncached local files. + cat.apk = cfg.path("paths", "apk") hit = self._catalogs[vid] = (cat, catalogdb.by_id(catalogdb.index(remote, apk))) return hit @@ -321,23 +319,20 @@ def fetch_location(self, vid: str, lid: str) -> Path: if loc["kind"] == "bundle": return cat.fetch(Bundle(0, iid, file_name(iid), remote_path(iid) is not None)) if remote_path(iid) is None: - return self._apk_file(iid, apk=cat.apk, cache=cat.local_cache_dir()) + return self._apk_file(iid, cat) return cat.fetch_raw({"internal_id": iid}) - def _apk_file(self, internal_id: str, *, apk=None, cache=None) -> Path: - """A raw file of the APK as stored, through the cache (raw/, where a CDN file of that path would be); KeyError when the APK does not hold it.""" + def _apk_file(self, internal_id: str, cat) -> Path: + """A raw APK file through its snapshot's cache; only a cache miss needs the matching APK.""" from .cache import write_atomic from .catalog import APK_AA_DIR - if self.cache is None: - raise self.cfg.missing("paths", "cache") - apk = apk or self.cfg.path("paths", "apk") - if apk is None: - raise self.cfg.missing("paths", "apk") rel = apk_rel(internal_id) - dst = (cache or self.cache) / "raw" / rel + dst = cat.local_cache_dir() / "raw" / rel if not (dst.is_file() and dst.stat().st_size > 0): - with ApkSet(apk) as z: + if cat.apk is None: + raise self.cfg.missing("paths", "apk") + with ApkSet(cat.apk) as z: + cat.check_apk(z) data = z.read(APK_AA_DIR + rel) dst.parent.mkdir(parents=True, exist_ok=True) write_atomic(dst, data) diff --git a/src/nnnotes/jp.py b/src/nnnotes/jp.py index 842d93f..d7b6a47 100644 --- a/src/nnnotes/jp.py +++ b/src/nnnotes/jp.py @@ -160,8 +160,8 @@ def __init__(self, cfg, region): def observe(self, *, timeout=gameapi.TIMEOUT): with self._lock: section = f"servers.{self.region}" - api = gameapi.api_root(self.cfg, section) - origin(api) + api = gameapi.api_root(self.cfg, section).strip() + api = origin(api if "://" in api else "https://" + api) allowed = origin(self.cfg.cdn(self.region)) client = gameapi.client_version(self.cfg, self.region) headers = {} diff --git a/tests/test_apkset.py b/tests/test_apkset.py new file mode 100644 index 0000000..a02cc49 --- /dev/null +++ b/tests/test_apkset.py @@ -0,0 +1,115 @@ +"""Small synthetic APK sets: expansion reuse, invalidation and reader lifetimes.""" +from concurrent.futures import ThreadPoolExecutor +import os +from pathlib import Path +from unittest.mock import Mock + +import pytest + +from nnnotes import apkset +from nnnotes.catalog import APK_AA_DIR, APK_CATALOG, Catalog +from test_jp import cat_bytes, zip_bytes + + +@pytest.fixture(autouse=True) +def clear_expansions(): + apkset._expanded_apks.cache_clear() + yield + apkset._expanded_apks.cache_clear() + + +def package(path, payload=b"asset"): + path.write_bytes(zip_bytes({ + "base.apk": zip_bytes({"AndroidManifest.xml": b"base"}), + "split_data.apk": zip_bytes({"asset": payload, "AndroidManifest.xml": b"split"}), + })) + return path + + +@pytest.mark.parametrize("suffix", [".apks", ".xapk"]) +def test_repeated_and_concurrent_opens_expand_once(tmp_path, monkeypatch, suffix): + source = package(tmp_path / ("game" + suffix)) + copy = Mock(wraps=apkset.shutil.copyfileobj) + monkeypatch.setattr(apkset.shutil, "copyfileobj", copy) + def read(_): + with apkset.ApkSet(source) as archive: + assert archive.read("AndroidManifest.xml") == b"base" + return archive.read("asset") + with ThreadPoolExecutor(4) as pool: + assert list(pool.map(read, range(8))) == [b"asset"] * 8 + assert read(None) == b"asset" + assert copy.call_count == 2 + + +def test_replaced_archive_uses_new_expansion(tmp_path): + source = package(tmp_path / "game.apks") + with apkset.ApkSet(source) as old: + package(source, b"replacement") + with apkset.ApkSet(source) as new: + assert new.read("asset") == b"replacement" + assert old.read("asset") == b"asset" + + +def test_same_size_archive_update_invalidates_by_mtime(tmp_path): + source = package(tmp_path / "game.apks", b"first") + before = source.stat() + with apkset.ApkSet(source) as archive: + assert archive.read("asset") == b"first" + package(source, b"other") + assert source.stat().st_size == before.st_size + os.utime(source, ns=(before.st_atime_ns, before.st_mtime_ns + 2_000_000_000)) + with apkset.ApkSet(source) as archive: + assert archive.read("asset") == b"other" + + +def test_eviction_does_not_remove_an_active_readers_files(tmp_path): + source = package(tmp_path / "first.apks") + with apkset.ApkSet(source) as first: + expanded = Path(first._expanded[0].name) + for name in ("second", "third"): + with apkset.ApkSet(package(tmp_path / f"{name}.apks")) as other: + assert other.read("asset") == b"asset" + assert apkset._expanded_apks.cache_info().currsize == 2 + apkset._expanded_apks.cache_clear() + assert expanded.is_dir() + assert first.read("asset") == b"asset" + assert not expanded.exists() + + +def test_failed_expansion_is_cleaned_and_can_be_retried(tmp_path, monkeypatch): + source = package(tmp_path / "game.apks") + directories = [] + make_dir = apkset.tempfile.TemporaryDirectory + def temporary(**kwargs): + directory = make_dir(**kwargs) + directories.append(Path(directory.name)) + return directory + monkeypatch.setattr(apkset.tempfile, "TemporaryDirectory", temporary) + copy = apkset.shutil.copyfileobj + def fail(stream, target): + copy(stream, target) + raise OSError("synthetic copy failure") + monkeypatch.setattr(apkset.shutil, "copyfileobj", fail) + with pytest.raises(OSError, match="synthetic copy failure"): + with apkset.ApkSet(source): + pytest.fail("partial expansion should not open") + assert directories and all(not p.exists() for p in directories) + assert apkset._expanded_apks.cache_info().currsize == 0 + monkeypatch.setattr(apkset.shutil, "copyfileobj", copy) + with apkset.ApkSet(source) as archive: + assert archive.read("asset") == b"asset" + + +def test_cached_apk_bundle_does_not_reexpand_the_archive(tmp_path, monkeypatch): + source = tmp_path / "game.apks" + source.write_bytes(zip_bytes({ + "base.apk": zip_bytes({"AndroidManifest.xml": b"base"}), + "split_data.apk": zip_bytes({APK_CATALOG: cat_bytes(), + APK_AA_DIR + "Android/local.bundle": b"UnityFS\0local"}), + })) + cat = Catalog(cat_bytes(), tmp_path / "cache", apk=source) + copy = Mock(wraps=apkset.shutil.copyfileobj) + monkeypatch.setattr(apkset.shutil, "copyfileobj", copy) + assert cat.apk_bundle("local").read_bytes() == b"UnityFS\0local" + assert cat.apk_bundle("local").read_bytes() == b"UnityFS\0local" + assert copy.call_count == 0 diff --git a/tests/test_configfile.py b/tests/test_configfile.py index 131574f..be276fc 100644 --- a/tests/test_configfile.py +++ b/tests/test_configfile.py @@ -180,6 +180,30 @@ def test_edit_refuses_what_it_cannot_edit(): configfile.edit('[paths]\nfonts.ja = "a"\n', [("paths.fonts", "ja", "b")], "f") +@pytest.mark.parametrize("provider", ['"jpp"', '42']) +def test_check_reports_invalid_provider_as_json(tmp_path, capsys, provider): + conf = tmp_path / "nnnotes.toml" + conf.write_text(f'[catalog]\nregion="jp"\n[servers.jp]\nprovider={provider}\n', encoding="utf-8") + code, out, err = run(["--config", str(conf), "config", "check", "--json"], capsys) + assert code == 1 and not err + report = json.loads(out) + assert report["problems"] == 1 + item = next(s for s in report["settings"] if s["name"] == "servers.jp.provider") + assert item["status"] == "invalid" + + +def test_check_does_not_turn_jp_language_defaults_into_flags(tmp_path, capsys): + conf = tmp_path / "nnnotes.toml" + conf.write_text('[catalog]\nregion="jp"\n[servers.jp]\nprovider="jp"\n', encoding="utf-8") + code, out, _ = run(["--config", str(conf), "config", "check", "--json"], capsys) + assert code == 0 + settings = {s["name"]: s for s in json.loads(out)["settings"]} + assert settings["catalog.region"]["origin"] == "file" + assert settings["catalog.language"]["status"] == "unset" + cfg = cli.load_config(cli.build_parser().parse_args(["--config", str(conf), "master", "version"])) + assert cfg.require("catalog", "language") == "ja" # normal commands still apply the JP default + + def test_check_reports_states_not_values(tmp_path, capsys, monkeypatch): (tmp_path / "nnnotes.toml").write_text( f'[bundle]\nkey = "{KEY}"\n[master]\niv = "00"\n[catalog]\nregion = "xx"\n[servers.tw]\ncdn = "ftp://h"\n' diff --git a/tests/test_jp.py b/tests/test_jp.py index e67e3e6..400d7b9 100644 --- a/tests/test_jp.py +++ b/tests/test_jp.py @@ -90,6 +90,29 @@ def test_version_headers_and_master_without_asset(upstream): assert result.version == MASTER and result.resource_version == "" and result.resource_hash is None +@pytest.mark.parametrize("api", ["api.invalid:443", " api.invalid:8443/ ", "[::1]:443"]) +def test_api_root_defaults_to_https(api, monkeypatch): + cfg = Config({"catalog": {"region": "jp"}, "servers": {"jp": { + "api": api, "cdn": "https://cdn.invalid", "client_version": "1.0.4"}}}, environ={}) + calls = [] + def call(root, *args, response_metadata, **kwargs): + calls.append(root) + response_metadata.update({"x-sirius-env": "https://cdn.invalid", "x-sirius-cred": "synthetic"}) + return version_response(MASTER) + monkeypatch.setattr(gameapi, "call", call) + configfile.validate(configfile.resolve("servers.jp.api")[0], api) + assert jp.Session(cfg, "jp").observe().version.version == MASTER + assert calls == [jp.origin("https://" + api.strip())] + + +@pytest.mark.parametrize("api", ["http://api.invalid", "https://user:pass@api.invalid", "https://api.invalid/path"]) +def test_api_origin_restrictions_still_apply(api, monkeypatch): + cfg = Config({"catalog": {"region": "jp"}, "servers": {"jp": {"api": api}}}, environ={}) + monkeypatch.setattr(gameapi, "call", lambda *a, **kw: pytest.fail("invalid origin reached the API")) + with pytest.raises(ConfigError): + jp.Session(cfg, "jp").observe() + + def test_live_selection_numeric_and_no_fallback(): def entry(client, version): return {"minClientVersion": client, "version": version, "Android": HASH} @@ -248,6 +271,66 @@ def test_apk_update_with_same_asset_snapshot_does_not_reuse_embedded_cache(tmp_p assert len(db.versions()) == 2 +def local_catalog(revision=1): + return synth.CatalogWriter().build([ + ("Local", "Assets/local", [1]), + ("local.bundle", synth.local("local.bundle"), []), + ("local.acb", synth.local("local.acb"), []), + ], build_hash=str(revision) * 32) + + +@pytest.mark.parametrize("apk_state", ["unset", "missing", "different", "broken"]) +@pytest.mark.parametrize("kind", ["bundles", "rawFiles"]) +def test_remote_replay_does_not_open_the_apk(upstream, tmp_path, apk_state, kind): + from nnnotes.catalog import APK_CATALOG + cfg = upstream.cfg + cfg._data["bundle"] = {"key": "00" * 16, "nonce_seed": "11"} + cfg._data["paths"] = {"cache": str(tmp_path / "cache")} + apk = tmp_path / "base.apk" + if apk_state != "unset": + cfg._data["paths"]["apk"] = str(apk) + if apk_state == "different": + apk.write_bytes(zip_bytes({APK_CATALOG: local_catalog(2)})) + elif apk_state == "broken": + apk.write_bytes(b"not a zip") + source = jp.Session(cfg, "jp").observe().source + db = catalogdb.CatalogDB(tmp_path / "store") + record = db.add(cat_bytes(), local_catalog(), source=source.to_dict(), region="jp") + entry = next(e for e in db.index(record)[kind] if e["remote"]) + upstream.body = b"UnityFS\0synthetic resource" + fetcher = CatalogFetcher(tmp_path / "store", tmp_path / "cache", cfg) + assert fetcher.fetch_location(record["id"], entry["location"]).read_bytes() == upstream.body + assert len(upstream.seen) == 1 + + +@pytest.mark.parametrize("kind", ["bundles", "rawFiles"]) +@pytest.mark.parametrize("state", ["matching", "different", "no-catalog", "cached"]) +def test_local_replay_validates_only_on_cache_miss(tmp_path, kind, state): + from nnnotes.catalog import APK_CATALOG, APK_AA_DIR + source = jp.Source("jp", "1.0", HASH, "https://cdn.invalid") + apk = tmp_path / "base.apk" + members = {APK_AA_DIR + "Android/local.bundle": b"UnityFS\0local", APK_AA_DIR + "Android/local.acb": b"@UTFraw"} + if state != "no-catalog": + members[APK_CATALOG] = local_catalog(2 if state == "different" else 1) + apk.write_bytes(zip_bytes(members)) + cfg = Config({"catalog": {"region": "jp"}, "paths": {"apk": str(apk)}}, environ={}) + db = catalogdb.CatalogDB(tmp_path / "store") + record = db.add(cat_bytes(), local_catalog(), source=source.to_dict(), region="jp") + entry = next(e for e in db.index(record)[kind] if not e["remote"]) + fetcher = CatalogFetcher(tmp_path / "store", tmp_path / "cache", cfg) + if state in ("different", "no-catalog"): + with pytest.raises(ConfigError, match="APK set it was imported with"): + fetcher.fetch_location(record["id"], entry["location"]) + return + expected = members[APK_AA_DIR + ("Android/local.bundle" if kind == "bundles" else "Android/local.acb")] + assert fetcher.fetch_location(record["id"], entry["location"]).read_bytes() == expected + if state == "cached": + apk.unlink() + cfg._data["paths"].pop("apk") + replay = CatalogFetcher(tmp_path / "store", tmp_path / "cache", cfg) + assert replay.fetch_location(record["id"], entry["location"]).read_bytes() == expected + + def zip_bytes(files): output = io.BytesIO() with zipfile.ZipFile(output, "w") as z: @@ -287,6 +370,31 @@ def test_per_region_apk_and_client_settings(tmp_path): assert configfile.resolve("servers.jp.client_version")[0].kind == "string" +@pytest.mark.parametrize("kind", ["--all-live2d", "--all-stories"]) +@pytest.mark.parametrize("region", ["jp", "tw"]) +def test_web_preflight_uses_the_selected_regions_apk(tmp_path, monkeypatch, kind, region): + from nnnotes import storysite, tmpfont, web, webmodel + conf = tmp_path / "nnnotes.toml" + default = "jp" if region == "tw" else "tw" + conf.write_text(f'[catalog]\nregion="{default}"\nlanguage="ja"\n' + f'[servers.{default}]\ncdn="https://default.invalid"\n' + f'[servers.{region}]\ncdn="https://target.invalid"\napk="target.apk"\n' + '[paths]\nplayer="player"\n', encoding="utf-8") + (tmp_path / "target.apk").write_bytes(b"synthetic") + monkeypatch.setattr(tmpfont, "require_extra", lambda *a, **kw: None) + monkeypatch.setattr(web, "check_player", lambda *a, **kw: tmp_path / "player") + monkeypatch.setattr(cli, "open_catalog", lambda *a, **kw: object()) + monkeypatch.setattr(webmodel, "catalog_models", lambda *a, **kw: {}) + seen = [] + def build(out, selection, cfg, *args, **kwargs): + seen.append((cfg.region(), cfg.path("paths", "apk"))) + return {} + monkeypatch.setattr(webmodel, "build", build) + monkeypatch.setattr(storysite, "build", build) + cli.main(["--config", str(conf), "web", str(tmp_path / "site"), kind, "--region", region]) + assert seen == [(region, tmp_path / "target.apk")] + + def test_datapack_is_loaded_in_the_boot_environment(tmp_path, monkeypatch): from nnnotes import player boot = SimpleNamespace(objects=[SimpleNamespace(type=SimpleNamespace(name="GraphicsSettings"),