diff --git a/lib/builtin_content.py b/lib/builtin_content.py new file mode 100644 index 0000000..792644d --- /dev/null +++ b/lib/builtin_content.py @@ -0,0 +1,378 @@ +"""Builtin content seeding: the calibration/diagnostic sloppaks and the starter library. + +Carved VERBATIM out of server.py (R3b) — with ONE deliberate signature change, and it is +the whole reason this module is safe. + +━━━ WHY THE ROOT IS A PARAMETER ━━━ + +server.py had `_feedBack_server_root()` = `Path(__file__).resolve().parent`. That is +correct *in server.py*: the repo root in dev, resources/feedBack when bundled — the tree +that actually holds docs/ and data/. + +Move that body here unchanged and it keeps working, silently, and returns `lib/`. There is +no docs/diagnostics under lib/, so every seed would quietly find nothing and log "source +missing" — a verbatim move whose meaning changed because `__file__` did. Nothing would +fail; the starter library would just never appear. + +So this module CANNOT compute a root: it takes `server_root` as a parameter, and server.py +— the only place that legitimately knows where it lives — passes it in. The trap is now +structurally impossible rather than merely avoided. (_copy_builtin_packs already took the +root this way; the two seed helpers now do too.) + +Everything else is byte-identical. `log` is this module's own logger under the same +`feedBack.` hierarchy, and CONFIG_DIR is read late as `appstate.config_dir` — see appstate.py +for why those reads must be late-bound (tests monkeypatch it). +""" +import logging +import os +import secrets +import shutil +import stat +import tempfile +from pathlib import Path + +import appstate +from dlc_paths import _get_dlc_dir + +log = logging.getLogger("feedBack.builtin_content") + + +BUILTIN_DIAGNOSTIC_SUBDIR = "diagnostics-builtin" + + +BUILTIN_DIAGNOSTIC_SOURCES: list[tuple[str, str]] = [ + ( + "feedBack-diagnostic-basic-guitar.sloppak", + "docs/diagnostics/feedBack-diagnostic-basic-guitar.sloppak", + ), +] + + +def builtin_diagnostic_filename() -> str: + """Library filename (DLC-relative POSIX path) of the calibration sloppak — + the onboarding challenge target (spec 010).""" + return f"{BUILTIN_DIAGNOSTIC_SUBDIR}/{BUILTIN_DIAGNOSTIC_SOURCES[0][0]}" + + +def _copy_builtin_packs( + root: Path, + dest_dir: Path, + sources: list[tuple[str, str]], + label: str, + update_existing: bool = True, +) -> int: + """Symlink-safe, mtime-aware copy of bundled packs into ``dest_dir``. + + ``sources`` is a list of ``(dest_name, rel_source)`` pairs; each source is + resolved under ``root`` (the repo root in dev, ``resources/feedBack`` when + bundled). A pack is copied when its destination is missing. Never deletes + user files; refuses to follow a symlinked seed directory or destination and + refuses to clobber a non-regular destination (any would let a copy escape + ``dest_dir`` or destroy user data). Logs and continues on error. ``label`` + prefixes every log line. + + ``update_existing`` controls what happens when a *regular* destination file + already exists: when True (diagnostic seed) a bundle copy newer than the + destination refreshes it; when False (one-time starter content) an existing + file is always left as-is so the user's copy is never overwritten. + + Returns the number of ``sources`` that are present at their destination + afterwards (freshly seeded, refreshed, or already current) — so callers can + tell whether every pack made it. A skip (missing source, symlink/non-regular + refusal, copy error) does not count. + """ + # Refuse a symlinked seed directory: mkdir(exist_ok=True) would accept it + # and copies would land at the link target, outside the DLC tree. The + # per-file symlink guard below cannot catch this. + if dest_dir.is_symlink(): + log.warning("%s: %s is a symlink, skipping all seeding", label, dest_dir.name) + return 0 + dest_dir.mkdir(parents=True, exist_ok=True) + + # Pin the seed directory by an O_NOFOLLOW fd so a symlink swapped in for + # dest_dir *after* the check above cannot redirect the per-file stat / + # temp-create / replace outside the DLC tree (parent-directory TOCTOU). + # os.replace accepts dir_fd on POSIX even though it isn't listed in + # os.supports_dir_fd, so gate on os.rename (the reliable proxy); platforms + # without dir_fd/O_NOFOLLOW (e.g. Windows) fall back to path-based ops. + dir_fd = None + if ( + hasattr(os, "O_NOFOLLOW") + and hasattr(os, "O_DIRECTORY") + and os.open in os.supports_dir_fd + and os.rename in os.supports_dir_fd + ): + try: + dir_fd = os.open(dest_dir, os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY) + except OSError as exc: + log.warning("%s: cannot open seed dir %s: %s", label, dest_dir, exc) + return 0 + + try: + present = 0 + for dest_name, rel_source in sources: + source = root / rel_source + if not source.is_file(): + log.warning("%s: source missing, skipping %s (%s)", label, dest_name, source) + continue + + # lstat the destination without following symlinks. Pinned by dir_fd + # this resolves within the real seed dir, immune to a parent swap. + try: + if dir_fd is not None: + dstat = os.lstat(dest_name, dir_fd=dir_fd) + else: + dstat = os.lstat(dest_dir / dest_name) + dest_exists = True + dest_islink = stat.S_ISLNK(dstat.st_mode) + except FileNotFoundError: + dest_exists = False + dest_islink = False + except OSError as exc: + log.warning("%s: cannot stat %s: %s", label, dest_name, exc) + continue + + # Refuse to seed through a symlink at the destination name. + if dest_islink: + log.warning("%s: destination is a symlink, skipping %s", label, dest_name) + continue + + # A non-regular destination (directory, fifo, …) the user placed + # there: never clobber it, and never count it as present — otherwise + # a one-time seed would mark itself done without a real pack on disk. + if dest_exists and not stat.S_ISREG(dstat.st_mode): + log.warning("%s: destination is not a regular file, skipping %s", label, dest_name) + continue + + if dest_exists: + # A regular file is already there. One-time seeds (starter + # content) must never overwrite the user's copy; refreshing + # seeds (diagnostics) replace it only when the bundle is newer. + if not update_existing: + log.info("%s: already present %s", label, dest_name) + present += 1 + continue + try: + src_mtime = source.stat().st_mtime + except OSError as exc: + log.warning("%s: cannot stat source %s: %s", label, source, exc) + continue + if src_mtime <= dstat.st_mtime: + log.info("%s: already present %s", label, dest_name) + present += 1 + continue + action = "updated" + else: + action = "seeded" + + if _write_builtin_pack(source, dest_dir, dest_name, dir_fd): + present += 1 + log.info("%s: %s %s -> %s", label, action, source.name, dest_name) + else: + log.warning("%s: failed to copy %s -> %s/%s", label, source, dest_dir.name, dest_name) + + return present + finally: + if dir_fd is not None: + os.close(dir_fd) + + +def _write_builtin_pack( + source: Path, + dest_dir: Path, + dest_name: str, + dir_fd: int | None, +) -> bool: + """Atomically write ``source`` to ``dest_name`` inside ``dest_dir``. + + Writes to a temp file then ``os.replace()``s onto the final name so a + symlink raced in at the destination is overwritten (rename semantics), not + followed, and a crash never leaves a half-written pack. When ``dir_fd`` is + given, every step is anchored to that fd (O_NOFOLLOW temp create + dir_fd + replace), closing the parent-directory TOCTOU; otherwise falls back to + path-based temp+replace. Returns True on success. Never raises. + """ + # Unique per-attempt name (O_EXCL create) so a crash that orphans a temp + # can't permanently block later seeds via an EEXIST collision. + tmp_name = f".seed-{dest_name}.{os.getpid()}.{secrets.token_hex(4)}.tmp" + try: + src_stat = source.stat() + except OSError as exc: + log.debug("builtin pack: cannot stat source %s: %s", source, exc) + return False + if dir_fd is not None: + tmp_fd = None + try: + tmp_fd = os.open( + tmp_name, + os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, + 0o644, + dir_fd=dir_fd, + ) + with open(source, "rb") as sf, os.fdopen(tmp_fd, "wb") as tf: + tmp_fd = None # fdopen now owns the descriptor + shutil.copyfileobj(sf, tf) + os.replace(tmp_name, dest_name, src_dir_fd=dir_fd, dst_dir_fd=dir_fd) + # Preserve the bundle mtime (copyfileobj doesn't) so the mtime-based + # refresh check matches the shutil.copy2 fallback path. Best-effort. + try: + os.utime( + dest_name, + ns=(src_stat.st_atime_ns, src_stat.st_mtime_ns), + dir_fd=dir_fd, + follow_symlinks=False, + ) + except OSError as exc: + log.debug("builtin pack: could not set mtime on %s: %s", dest_name, exc) + return True + except OSError as exc: + log.debug("builtin pack write (dir_fd) failed for %s: %s", dest_name, exc) + if tmp_fd is not None: + try: + os.close(tmp_fd) + except OSError: + pass + try: + os.unlink(tmp_name, dir_fd=dir_fd) + except OSError: + pass + return False + + tmp = None + try: + fd, tmp = tempfile.mkstemp(dir=dest_dir, prefix=".seed-", suffix=".tmp") + os.close(fd) + shutil.copy2(source, tmp) + os.replace(tmp, dest_dir / dest_name) + tmp = None + return True + except OSError as exc: + log.debug("builtin pack write failed for %s: %s", dest_name, exc) + return False + finally: + if tmp is not None: + try: + os.unlink(tmp) + except OSError: + pass + + +def seed_builtin_diagnostic_sloppaks(server_root: Path, dlc: Path | None = None) -> None: + """Copy bundled diagnostic sloppaks into DLC before library scan. + + Creates ``DLC_DIR/diagnostics-builtin/`` and copies each bundled sloppak + when the destination is missing or older than the repo/bundle source. + Never deletes user files or touches manually copied paths (e.g. + ``diagnostics-test/``). Re-seeds whenever the destination is missing so the + diagnostic target is always available. Logs and continues on errors. + """ + try: + if dlc is None: + dlc = _get_dlc_dir() + if dlc is None: + log.debug("Builtin diagnostic seed: no DLC folder configured, skipping") + return + _copy_builtin_packs( + server_root, + dlc / BUILTIN_DIAGNOSTIC_SUBDIR, + BUILTIN_DIAGNOSTIC_SOURCES, + "Builtin diagnostic seed", + ) + except Exception: + log.warning("Builtin diagnostic seed: unexpected error", exc_info=True) + + +# Starter content: bundled songs copied into ``DLC_DIR/starter/`` exactly ONCE, +# on first run, as a welcome library so a fresh install isn't empty. Unlike the +# diagnostic seed this is one-time — guarded by a marker in CONFIG_DIR — so if +# the user deletes the starter song it stays gone. ``starter/`` is NOT in the +# library scan carve-out (unlike diagnostics-builtin/ / tutorials-builtin/), so +# seeded packs surface as ordinary library songs. +BUILTIN_STARTER_SUBDIR = "starter" + + +BUILTIN_STARTER_SOURCES: list[tuple[str, str]] = [ + ( + "beethoven-fur_elise.feedpak", + "content/starter/beethoven-fur_elise.feedpak", + ), + ( + "star_spangled_banner.feedpak", + "content/starter/star_spangled_banner.feedpak", + ), + ( + "the_adicts-ode-to-joy_vst_cover.feedpak", + "content/starter/the_adicts-ode-to-joy_vst_cover.feedpak", + ), +] + + +STARTER_SEED_MARKER = ".starter-content-seeded" + + +def seed_builtin_starter_content(server_root: Path, dlc: Path | None = None) -> None: + """Copy bundled starter songs into ``DLC_DIR/starter/`` exactly once. + + Guarded by ``CONFIG_DIR/.starter-content-seeded``: the first run with a DLC + folder configured seeds the packs and writes the marker; subsequent runs are + no-ops, so a user who deletes the starter song does not get it back on the + next launch. Symlink-safe; never deletes user files. Logs, never raises. + """ + try: + marker = appstate.config_dir / STARTER_SEED_MARKER + # Already seeded? The marker is a sentinel: any existing path there + # (regular file, or a symlink/dir a user deliberately planted to opt + # out) means "done" — lstat so we detect it without following a symlink. + # Worst case of a planted marker is simply no starter content, never a + # data write; the O_EXCL|O_NOFOLLOW create below refuses to write + # *through* a symlink regardless. + try: + os.lstat(marker) + return + except FileNotFoundError: + pass + except OSError as exc: + log.warning("Starter content seed: cannot stat marker %s: %s", marker, exc) + return + if dlc is None: + dlc = _get_dlc_dir() + if dlc is None: + # No DLC yet — leave the marker unwritten so we retry once a + # library folder is configured. + log.debug("Starter content seed: no DLC folder configured, skipping") + return + present = _copy_builtin_packs( + server_root, + dlc / BUILTIN_STARTER_SUBDIR, + BUILTIN_STARTER_SOURCES, + "Starter content seed", + update_existing=False, + ) + # Only mark seeding complete once every starter pack is actually in + # place. If a source was missing or a copy failed, leave the marker + # unwritten so the next launch retries rather than permanently skipping. + if present < len(BUILTIN_STARTER_SOURCES): + log.info( + "Starter content seed: %d/%d packs present, will retry next launch", + present, + len(BUILTIN_STARTER_SOURCES), + ) + return + # Record completion with an exclusive, no-follow create so a planted or + # raced symlink at the marker path can't redirect the write outside + # CONFIG_DIR. O_EXCL fails (EEXIST) on any existing path including a + # symlink, so we never write through one. + try: + appstate.config_dir.mkdir(parents=True, exist_ok=True) + flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_NOFOLLOW", 0) + fd = os.open(marker, flags, 0o644) + try: + os.write(fd, b"1\n") + finally: + os.close(fd) + except FileExistsError: + pass # already marked (or a non-regular path is squatting) — fine + except OSError as exc: + log.warning("Starter content seed: could not write marker %s: %s", marker, exc) + except Exception: + log.warning("Starter content seed: unexpected error", exc_info=True) diff --git a/server.py b/server.py index a52ba69..155007e 100644 --- a/server.py +++ b/server.py @@ -44,6 +44,7 @@ from dlc_paths import _get_dlc_dir, _resolve_dlc_path # `appstate.configure(...)` below publishes into the same namespace routers read. # Lives in lib/ because that is the one core dir every packaging path copies. import appstate +import builtin_content # Extracted route modules. They import `appstate`, never `server` — one-way graph. from routers import audio_effects, artist_aliases, loops, playlists, ws_highway, chart, wanted, library_extras, shop, progression, profile, stats, version, diagnostics from routers import tunings as tunings_router @@ -645,13 +646,6 @@ def _make_scan_executor(): ) -_BUILTIN_DIAGNOSTIC_SUBDIR = "diagnostics-builtin" -_BUILTIN_DIAGNOSTIC_SOURCES: list[tuple[str, str]] = [ - ( - "feedBack-diagnostic-basic-guitar.sloppak", - "docs/diagnostics/feedBack-diagnostic-basic-guitar.sloppak", - ), -] def _feedBack_server_root() -> Path: @@ -659,10 +653,6 @@ def _feedBack_server_root() -> Path: return Path(__file__).resolve().parent -def _builtin_diagnostic_filename() -> str: - """Library filename (DLC-relative POSIX path) of the calibration sloppak — - the onboarding challenge target (spec 010).""" - return f"{_BUILTIN_DIAGNOSTIC_SUBDIR}/{_BUILTIN_DIAGNOSTIC_SOURCES[0][0]}" # Progression content (spec 010): bundled JSON under data/progression/ (paths, @@ -694,329 +684,19 @@ def _get_progression_content() -> dict: # path is unchanged; routers call `appstate.get_progression_content()`. appstate.configure( get_progression_content=_get_progression_content, - builtin_diagnostic_filename=_builtin_diagnostic_filename, + builtin_diagnostic_filename=builtin_content.builtin_diagnostic_filename, tuning_providers=tuning_providers, ) -def _copy_builtin_packs( - root: Path, - dest_dir: Path, - sources: list[tuple[str, str]], - label: str, - update_existing: bool = True, -) -> int: - """Symlink-safe, mtime-aware copy of bundled packs into ``dest_dir``. - - ``sources`` is a list of ``(dest_name, rel_source)`` pairs; each source is - resolved under ``root`` (the repo root in dev, ``resources/feedBack`` when - bundled). A pack is copied when its destination is missing. Never deletes - user files; refuses to follow a symlinked seed directory or destination and - refuses to clobber a non-regular destination (any would let a copy escape - ``dest_dir`` or destroy user data). Logs and continues on error. ``label`` - prefixes every log line. - - ``update_existing`` controls what happens when a *regular* destination file - already exists: when True (diagnostic seed) a bundle copy newer than the - destination refreshes it; when False (one-time starter content) an existing - file is always left as-is so the user's copy is never overwritten. - - Returns the number of ``sources`` that are present at their destination - afterwards (freshly seeded, refreshed, or already current) — so callers can - tell whether every pack made it. A skip (missing source, symlink/non-regular - refusal, copy error) does not count. - """ - # Refuse a symlinked seed directory: mkdir(exist_ok=True) would accept it - # and copies would land at the link target, outside the DLC tree. The - # per-file symlink guard below cannot catch this. - if dest_dir.is_symlink(): - log.warning("%s: %s is a symlink, skipping all seeding", label, dest_dir.name) - return 0 - dest_dir.mkdir(parents=True, exist_ok=True) - - # Pin the seed directory by an O_NOFOLLOW fd so a symlink swapped in for - # dest_dir *after* the check above cannot redirect the per-file stat / - # temp-create / replace outside the DLC tree (parent-directory TOCTOU). - # os.replace accepts dir_fd on POSIX even though it isn't listed in - # os.supports_dir_fd, so gate on os.rename (the reliable proxy); platforms - # without dir_fd/O_NOFOLLOW (e.g. Windows) fall back to path-based ops. - dir_fd = None - if ( - hasattr(os, "O_NOFOLLOW") - and hasattr(os, "O_DIRECTORY") - and os.open in os.supports_dir_fd - and os.rename in os.supports_dir_fd - ): - try: - dir_fd = os.open(dest_dir, os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY) - except OSError as exc: - log.warning("%s: cannot open seed dir %s: %s", label, dest_dir, exc) - return 0 - - try: - present = 0 - for dest_name, rel_source in sources: - source = root / rel_source - if not source.is_file(): - log.warning("%s: source missing, skipping %s (%s)", label, dest_name, source) - continue - - # lstat the destination without following symlinks. Pinned by dir_fd - # this resolves within the real seed dir, immune to a parent swap. - try: - if dir_fd is not None: - dstat = os.lstat(dest_name, dir_fd=dir_fd) - else: - dstat = os.lstat(dest_dir / dest_name) - dest_exists = True - dest_islink = stat.S_ISLNK(dstat.st_mode) - except FileNotFoundError: - dest_exists = False - dest_islink = False - except OSError as exc: - log.warning("%s: cannot stat %s: %s", label, dest_name, exc) - continue - - # Refuse to seed through a symlink at the destination name. - if dest_islink: - log.warning("%s: destination is a symlink, skipping %s", label, dest_name) - continue - - # A non-regular destination (directory, fifo, …) the user placed - # there: never clobber it, and never count it as present — otherwise - # a one-time seed would mark itself done without a real pack on disk. - if dest_exists and not stat.S_ISREG(dstat.st_mode): - log.warning("%s: destination is not a regular file, skipping %s", label, dest_name) - continue - - if dest_exists: - # A regular file is already there. One-time seeds (starter - # content) must never overwrite the user's copy; refreshing - # seeds (diagnostics) replace it only when the bundle is newer. - if not update_existing: - log.info("%s: already present %s", label, dest_name) - present += 1 - continue - try: - src_mtime = source.stat().st_mtime - except OSError as exc: - log.warning("%s: cannot stat source %s: %s", label, source, exc) - continue - if src_mtime <= dstat.st_mtime: - log.info("%s: already present %s", label, dest_name) - present += 1 - continue - action = "updated" - else: - action = "seeded" - - if _write_builtin_pack(source, dest_dir, dest_name, dir_fd): - present += 1 - log.info("%s: %s %s -> %s", label, action, source.name, dest_name) - else: - log.warning("%s: failed to copy %s -> %s/%s", label, source, dest_dir.name, dest_name) - - return present - finally: - if dir_fd is not None: - os.close(dir_fd) -def _write_builtin_pack( - source: Path, - dest_dir: Path, - dest_name: str, - dir_fd: int | None, -) -> bool: - """Atomically write ``source`` to ``dest_name`` inside ``dest_dir``. - - Writes to a temp file then ``os.replace()``s onto the final name so a - symlink raced in at the destination is overwritten (rename semantics), not - followed, and a crash never leaves a half-written pack. When ``dir_fd`` is - given, every step is anchored to that fd (O_NOFOLLOW temp create + dir_fd - replace), closing the parent-directory TOCTOU; otherwise falls back to - path-based temp+replace. Returns True on success. Never raises. - """ - # Unique per-attempt name (O_EXCL create) so a crash that orphans a temp - # can't permanently block later seeds via an EEXIST collision. - tmp_name = f".seed-{dest_name}.{os.getpid()}.{secrets.token_hex(4)}.tmp" - try: - src_stat = source.stat() - except OSError as exc: - log.debug("builtin pack: cannot stat source %s: %s", source, exc) - return False - if dir_fd is not None: - tmp_fd = None - try: - tmp_fd = os.open( - tmp_name, - os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, - 0o644, - dir_fd=dir_fd, - ) - with open(source, "rb") as sf, os.fdopen(tmp_fd, "wb") as tf: - tmp_fd = None # fdopen now owns the descriptor - shutil.copyfileobj(sf, tf) - os.replace(tmp_name, dest_name, src_dir_fd=dir_fd, dst_dir_fd=dir_fd) - # Preserve the bundle mtime (copyfileobj doesn't) so the mtime-based - # refresh check matches the shutil.copy2 fallback path. Best-effort. - try: - os.utime( - dest_name, - ns=(src_stat.st_atime_ns, src_stat.st_mtime_ns), - dir_fd=dir_fd, - follow_symlinks=False, - ) - except OSError as exc: - log.debug("builtin pack: could not set mtime on %s: %s", dest_name, exc) - return True - except OSError as exc: - log.debug("builtin pack write (dir_fd) failed for %s: %s", dest_name, exc) - if tmp_fd is not None: - try: - os.close(tmp_fd) - except OSError: - pass - try: - os.unlink(tmp_name, dir_fd=dir_fd) - except OSError: - pass - return False - - tmp = None - try: - fd, tmp = tempfile.mkstemp(dir=dest_dir, prefix=".seed-", suffix=".tmp") - os.close(fd) - shutil.copy2(source, tmp) - os.replace(tmp, dest_dir / dest_name) - tmp = None - return True - except OSError as exc: - log.debug("builtin pack write failed for %s: %s", dest_name, exc) - return False - finally: - if tmp is not None: - try: - os.unlink(tmp) - except OSError: - pass -def _seed_builtin_diagnostic_sloppaks(dlc: Path | None = None) -> None: - """Copy bundled diagnostic sloppaks into DLC before library scan. - - Creates ``DLC_DIR/diagnostics-builtin/`` and copies each bundled sloppak - when the destination is missing or older than the repo/bundle source. - Never deletes user files or touches manually copied paths (e.g. - ``diagnostics-test/``). Re-seeds whenever the destination is missing so the - diagnostic target is always available. Logs and continues on errors. - """ - try: - if dlc is None: - dlc = _get_dlc_dir() - if dlc is None: - log.debug("Builtin diagnostic seed: no DLC folder configured, skipping") - return - _copy_builtin_packs( - _feedBack_server_root(), - dlc / _BUILTIN_DIAGNOSTIC_SUBDIR, - _BUILTIN_DIAGNOSTIC_SOURCES, - "Builtin diagnostic seed", - ) - except Exception: - log.warning("Builtin diagnostic seed: unexpected error", exc_info=True) -# Starter content: bundled songs copied into ``DLC_DIR/starter/`` exactly ONCE, -# on first run, as a welcome library so a fresh install isn't empty. Unlike the -# diagnostic seed this is one-time — guarded by a marker in CONFIG_DIR — so if -# the user deletes the starter song it stays gone. ``starter/`` is NOT in the -# library scan carve-out (unlike diagnostics-builtin/ / tutorials-builtin/), so -# seeded packs surface as ordinary library songs. -_BUILTIN_STARTER_SUBDIR = "starter" -_BUILTIN_STARTER_SOURCES: list[tuple[str, str]] = [ - ( - "beethoven-fur_elise.feedpak", - "content/starter/beethoven-fur_elise.feedpak", - ), - ( - "star_spangled_banner.feedpak", - "content/starter/star_spangled_banner.feedpak", - ), - ( - "the_adicts-ode-to-joy_vst_cover.feedpak", - "content/starter/the_adicts-ode-to-joy_vst_cover.feedpak", - ), -] -_STARTER_SEED_MARKER = ".starter-content-seeded" -def _seed_builtin_starter_content(dlc: Path | None = None) -> None: - """Copy bundled starter songs into ``DLC_DIR/starter/`` exactly once. - - Guarded by ``CONFIG_DIR/.starter-content-seeded``: the first run with a DLC - folder configured seeds the packs and writes the marker; subsequent runs are - no-ops, so a user who deletes the starter song does not get it back on the - next launch. Symlink-safe; never deletes user files. Logs, never raises. - """ - try: - marker = CONFIG_DIR / _STARTER_SEED_MARKER - # Already seeded? The marker is a sentinel: any existing path there - # (regular file, or a symlink/dir a user deliberately planted to opt - # out) means "done" — lstat so we detect it without following a symlink. - # Worst case of a planted marker is simply no starter content, never a - # data write; the O_EXCL|O_NOFOLLOW create below refuses to write - # *through* a symlink regardless. - try: - os.lstat(marker) - return - except FileNotFoundError: - pass - except OSError as exc: - log.warning("Starter content seed: cannot stat marker %s: %s", marker, exc) - return - if dlc is None: - dlc = _get_dlc_dir() - if dlc is None: - # No DLC yet — leave the marker unwritten so we retry once a - # library folder is configured. - log.debug("Starter content seed: no DLC folder configured, skipping") - return - present = _copy_builtin_packs( - _feedBack_server_root(), - dlc / _BUILTIN_STARTER_SUBDIR, - _BUILTIN_STARTER_SOURCES, - "Starter content seed", - update_existing=False, - ) - # Only mark seeding complete once every starter pack is actually in - # place. If a source was missing or a copy failed, leave the marker - # unwritten so the next launch retries rather than permanently skipping. - if present < len(_BUILTIN_STARTER_SOURCES): - log.info( - "Starter content seed: %d/%d packs present, will retry next launch", - present, - len(_BUILTIN_STARTER_SOURCES), - ) - return - # Record completion with an exclusive, no-follow create so a planted or - # raced symlink at the marker path can't redirect the write outside - # CONFIG_DIR. O_EXCL fails (EEXIST) on any existing path including a - # symlink, so we never write through one. - try: - CONFIG_DIR.mkdir(parents=True, exist_ok=True) - flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_NOFOLLOW", 0) - fd = os.open(marker, flags, 0o644) - try: - os.write(fd, b"1\n") - finally: - os.close(fd) - except FileExistsError: - pass # already marked (or a non-regular path is squatting) — fine - except OSError as exc: - log.warning("Starter content seed: could not write marker %s: %s", marker, exc) - except Exception: - log.warning("Starter content seed: unexpected error", exc_info=True) def _background_scan(): @@ -1038,8 +718,8 @@ def _background_scan(): log.warning("Scan: no DLC folder configured") return - _seed_builtin_diagnostic_sloppaks(dlc) - _seed_builtin_starter_content(dlc) + builtin_content.seed_builtin_diagnostic_sloppaks(_feedBack_server_root(), dlc) + builtin_content.seed_builtin_starter_content(_feedBack_server_root(), dlc) # Listing can fail on macOS without Full Disk Access, or on Docker if the # path isn't shared. Report the failure explicitly rather than silently diff --git a/tests/test_builtin_diagnostic_seed.py b/tests/test_builtin_diagnostic_seed.py index 408338f..6c2fa2a 100644 --- a/tests/test_builtin_diagnostic_seed.py +++ b/tests/test_builtin_diagnostic_seed.py @@ -7,6 +7,7 @@ import os import sys import time +import builtin_content import pytest @@ -23,13 +24,13 @@ def test_seed_creates_builtin_diagnostic_sloppak(tmp_path, server_mod): """First seed copies the bundled sloppak into diagnostics-builtin/.""" dlc = tmp_path / "dlc" dlc.mkdir() - source = server_mod._feedBack_server_root() / server_mod._BUILTIN_DIAGNOSTIC_SOURCES[0][1] + source = server_mod._feedBack_server_root() / builtin_content.BUILTIN_DIAGNOSTIC_SOURCES[0][1] if not source.is_file(): pytest.skip(f"source sloppak not present in checkout: {source}") - server_mod._seed_builtin_diagnostic_sloppaks(dlc) + builtin_content.seed_builtin_diagnostic_sloppaks(server_mod._feedBack_server_root(), dlc) - dest = dlc / server_mod._BUILTIN_DIAGNOSTIC_SUBDIR / server_mod._BUILTIN_DIAGNOSTIC_SOURCES[0][0] + dest = dlc / builtin_content.BUILTIN_DIAGNOSTIC_SUBDIR / builtin_content.BUILTIN_DIAGNOSTIC_SOURCES[0][0] assert dest.is_file() assert dest.stat().st_size == source.stat().st_size @@ -38,16 +39,16 @@ def test_seed_is_idempotent_when_destination_exists(tmp_path, server_mod): """Second seed leaves an up-to-date destination unchanged.""" dlc = tmp_path / "dlc" dlc.mkdir() - source = server_mod._feedBack_server_root() / server_mod._BUILTIN_DIAGNOSTIC_SOURCES[0][1] + source = server_mod._feedBack_server_root() / builtin_content.BUILTIN_DIAGNOSTIC_SOURCES[0][1] if not source.is_file(): pytest.skip(f"source sloppak not present in checkout: {source}") - server_mod._seed_builtin_diagnostic_sloppaks(dlc) - dest = dlc / server_mod._BUILTIN_DIAGNOSTIC_SUBDIR / server_mod._BUILTIN_DIAGNOSTIC_SOURCES[0][0] + builtin_content.seed_builtin_diagnostic_sloppaks(server_mod._feedBack_server_root(), dlc) + dest = dlc / builtin_content.BUILTIN_DIAGNOSTIC_SUBDIR / builtin_content.BUILTIN_DIAGNOSTIC_SOURCES[0][0] first_mtime = dest.stat().st_mtime_ns first_size = dest.stat().st_size - server_mod._seed_builtin_diagnostic_sloppaks(dlc) + builtin_content.seed_builtin_diagnostic_sloppaks(server_mod._feedBack_server_root(), dlc) assert dest.stat().st_mtime_ns == first_mtime assert dest.stat().st_size == first_size @@ -57,18 +58,18 @@ def test_seed_skips_when_destination_is_newer(tmp_path, server_mod): """An existing newer destination is not overwritten.""" dlc = tmp_path / "dlc" dlc.mkdir() - source = server_mod._feedBack_server_root() / server_mod._BUILTIN_DIAGNOSTIC_SOURCES[0][1] + source = server_mod._feedBack_server_root() / builtin_content.BUILTIN_DIAGNOSTIC_SOURCES[0][1] if not source.is_file(): pytest.skip(f"source sloppak not present in checkout: {source}") - dest_dir = dlc / server_mod._BUILTIN_DIAGNOSTIC_SUBDIR + dest_dir = dlc / builtin_content.BUILTIN_DIAGNOSTIC_SUBDIR dest_dir.mkdir(parents=True) - dest_name = server_mod._BUILTIN_DIAGNOSTIC_SOURCES[0][0] + dest_name = builtin_content.BUILTIN_DIAGNOSTIC_SOURCES[0][0] dest = dest_dir / dest_name dest.write_bytes(b"user-owned diagnostic copy") future = time.time() + 3600 os.utime(dest, (future, future)) - server_mod._seed_builtin_diagnostic_sloppaks(dlc) + builtin_content.seed_builtin_diagnostic_sloppaks(server_mod._feedBack_server_root(), dlc) assert dest.read_bytes() == b"user-owned diagnostic copy" @@ -77,18 +78,18 @@ def test_seed_refuses_to_follow_symlink_destination(tmp_path, server_mod): """A symlink at the destination is skipped, not written through.""" dlc = tmp_path / "dlc" dlc.mkdir() - source = server_mod._feedBack_server_root() / server_mod._BUILTIN_DIAGNOSTIC_SOURCES[0][1] + source = server_mod._feedBack_server_root() / builtin_content.BUILTIN_DIAGNOSTIC_SOURCES[0][1] if not source.is_file(): pytest.skip(f"source sloppak not present in checkout: {source}") outside = tmp_path / "outside.txt" outside.write_bytes(b"do not overwrite me") - dest_dir = dlc / server_mod._BUILTIN_DIAGNOSTIC_SUBDIR + dest_dir = dlc / builtin_content.BUILTIN_DIAGNOSTIC_SUBDIR dest_dir.mkdir(parents=True) - dest = dest_dir / server_mod._BUILTIN_DIAGNOSTIC_SOURCES[0][0] + dest = dest_dir / builtin_content.BUILTIN_DIAGNOSTIC_SOURCES[0][0] dest.symlink_to(outside) - server_mod._seed_builtin_diagnostic_sloppaks(dlc) + builtin_content.seed_builtin_diagnostic_sloppaks(server_mod._feedBack_server_root(), dlc) # The symlink target must be untouched and the link left as-is. assert outside.read_bytes() == b"do not overwrite me" @@ -101,11 +102,11 @@ def test_seed_refuses_symlinked_seed_directory(tmp_path, server_mod): dlc.mkdir() outside_dir = tmp_path / "outside_dir" outside_dir.mkdir() - (dlc / server_mod._BUILTIN_DIAGNOSTIC_SUBDIR).symlink_to( + (dlc / builtin_content.BUILTIN_DIAGNOSTIC_SUBDIR).symlink_to( outside_dir, target_is_directory=True ) - server_mod._seed_builtin_diagnostic_sloppaks(dlc) + builtin_content.seed_builtin_diagnostic_sloppaks(server_mod._feedBack_server_root(), dlc) # Nothing was written through the directory symlink into the link target. assert list(outside_dir.iterdir()) == [] @@ -116,11 +117,11 @@ def test_seed_missing_source_does_not_crash(tmp_path, server_mod, monkeypatch): dlc = tmp_path / "dlc" dlc.mkdir() monkeypatch.setattr( - server_mod, - "_BUILTIN_DIAGNOSTIC_SOURCES", + builtin_content, + "BUILTIN_DIAGNOSTIC_SOURCES", [("missing.sloppak", "docs/diagnostics/does-not-exist.sloppak")], ) - server_mod._seed_builtin_diagnostic_sloppaks(dlc) + builtin_content.seed_builtin_diagnostic_sloppaks(server_mod._feedBack_server_root(), dlc) - assert not (dlc / server_mod._BUILTIN_DIAGNOSTIC_SUBDIR / "missing.sloppak").exists() + assert not (dlc / builtin_content.BUILTIN_DIAGNOSTIC_SUBDIR / "missing.sloppak").exists() diff --git a/tests/test_builtin_starter_seed.py b/tests/test_builtin_starter_seed.py index 32baabf..f83c043 100644 --- a/tests/test_builtin_starter_seed.py +++ b/tests/test_builtin_starter_seed.py @@ -5,6 +5,7 @@ from __future__ import annotations import importlib import sys +import builtin_content import pytest @@ -21,15 +22,15 @@ def server_mod(tmp_path, monkeypatch, isolate_logging): def _source(server_mod): return ( server_mod._feedBack_server_root() - / server_mod._BUILTIN_STARTER_SOURCES[0][1] + / builtin_content.BUILTIN_STARTER_SOURCES[0][1] ) def _dest(server_mod, dlc): return ( dlc - / server_mod._BUILTIN_STARTER_SUBDIR - / server_mod._BUILTIN_STARTER_SOURCES[0][0] + / builtin_content.BUILTIN_STARTER_SUBDIR + / builtin_content.BUILTIN_STARTER_SOURCES[0][0] ) @@ -41,12 +42,12 @@ def test_seed_creates_starter_content_and_marker(tmp_path, server_mod): if not source.is_file(): pytest.skip(f"starter source not present in checkout: {source}") - server_mod._seed_builtin_starter_content(dlc) + builtin_content.seed_builtin_starter_content(server_mod._feedBack_server_root(), dlc) dest = _dest(server_mod, dlc) assert dest.is_file() assert dest.stat().st_size == source.stat().st_size - assert (server_mod.CONFIG_DIR / server_mod._STARTER_SEED_MARKER).is_file() + assert (server_mod.CONFIG_DIR / builtin_content.STARTER_SEED_MARKER).is_file() def test_seed_preserves_source_mtime(tmp_path, server_mod): @@ -58,7 +59,7 @@ def test_seed_preserves_source_mtime(tmp_path, server_mod): if not source.is_file(): pytest.skip(f"starter source not present in checkout: {source}") - server_mod._seed_builtin_starter_content(dlc) + builtin_content.seed_builtin_starter_content(server_mod._feedBack_server_root(), dlc) assert _dest(server_mod, dlc).stat().st_mtime_ns == source.stat().st_mtime_ns @@ -78,7 +79,7 @@ def test_seed_runs_only_once_and_respects_deletion(tmp_path, server_mod): if not source.is_file(): pytest.skip(f"starter source not present in checkout: {source}") - server_mod._seed_builtin_starter_content(dlc) + builtin_content.seed_builtin_starter_content(server_mod._feedBack_server_root(), dlc) dest = _dest(server_mod, dlc) assert dest.is_file() @@ -86,7 +87,7 @@ def test_seed_runs_only_once_and_respects_deletion(tmp_path, server_mod): dest.unlink() # A subsequent launch must not re-seed it. - server_mod._seed_builtin_starter_content(dlc) + builtin_content.seed_builtin_starter_content(server_mod._feedBack_server_root(), dlc) assert not dest.exists() @@ -98,15 +99,15 @@ def test_seed_deferred_until_dlc_configured(tmp_path, server_mod): pytest.skip(f"starter source not present in checkout: {source}") # dlc is None and DLC_DIR unset -> _get_dlc_dir() returns None. - server_mod._seed_builtin_starter_content(None) - assert not (server_mod.CONFIG_DIR / server_mod._STARTER_SEED_MARKER).exists() + builtin_content.seed_builtin_starter_content(server_mod._feedBack_server_root(), None) + assert not (server_mod.CONFIG_DIR / builtin_content.STARTER_SEED_MARKER).exists() # Now a DLC is configured: the deferred seed runs. dlc = tmp_path / "dlc" dlc.mkdir() - server_mod._seed_builtin_starter_content(dlc) + builtin_content.seed_builtin_starter_content(server_mod._feedBack_server_root(), dlc) assert _dest(server_mod, dlc).is_file() - assert (server_mod.CONFIG_DIR / server_mod._STARTER_SEED_MARKER).is_file() + assert (server_mod.CONFIG_DIR / builtin_content.STARTER_SEED_MARKER).is_file() def test_seed_refuses_symlinked_seed_directory(tmp_path, server_mod): @@ -119,13 +120,13 @@ def test_seed_refuses_symlinked_seed_directory(tmp_path, server_mod): outside_dir = tmp_path / "outside" outside_dir.mkdir() - (dlc / server_mod._BUILTIN_STARTER_SUBDIR).symlink_to(outside_dir) + (dlc / builtin_content.BUILTIN_STARTER_SUBDIR).symlink_to(outside_dir) - server_mod._seed_builtin_starter_content(dlc) + builtin_content.seed_builtin_starter_content(server_mod._feedBack_server_root(), dlc) assert list(outside_dir.iterdir()) == [] # An incomplete seed must NOT write the marker, so a later launch retries. - assert not (server_mod.CONFIG_DIR / server_mod._STARTER_SEED_MARKER).exists() + assert not (server_mod.CONFIG_DIR / builtin_content.STARTER_SEED_MARKER).exists() def test_seed_never_overwrites_an_existing_user_file(tmp_path, server_mod): @@ -140,11 +141,11 @@ def test_seed_never_overwrites_an_existing_user_file(tmp_path, server_mod): dest.write_bytes(b"user's own edited pack") _os.utime(dest, (1_000_000, 1_000_000)) # far older than the bundled source - server_mod._seed_builtin_starter_content(dlc) + builtin_content.seed_builtin_starter_content(server_mod._feedBack_server_root(), dlc) assert dest.read_bytes() == b"user's own edited pack" # untouched # counted as already-present, so the one-time seed considers itself done - assert (server_mod.CONFIG_DIR / server_mod._STARTER_SEED_MARKER).is_file() + assert (server_mod.CONFIG_DIR / builtin_content.STARTER_SEED_MARKER).is_file() def test_seed_does_not_mark_when_destination_is_a_directory(tmp_path, server_mod): @@ -160,10 +161,10 @@ def test_seed_does_not_mark_when_destination_is_a_directory(tmp_path, server_mod bogus.parent.mkdir(parents=True, exist_ok=True) bogus.mkdir() # user (or junk) placed a directory where the pack goes - server_mod._seed_builtin_starter_content(dlc) + builtin_content.seed_builtin_starter_content(server_mod._feedBack_server_root(), dlc) assert bogus.is_dir() # untouched - assert not (server_mod.CONFIG_DIR / server_mod._STARTER_SEED_MARKER).exists() + assert not (server_mod.CONFIG_DIR / builtin_content.STARTER_SEED_MARKER).exists() def test_seed_does_not_mark_when_source_missing(tmp_path, server_mod, monkeypatch): @@ -172,15 +173,15 @@ def test_seed_does_not_mark_when_source_missing(tmp_path, server_mod, monkeypatc dlc = tmp_path / "dlc" dlc.mkdir() monkeypatch.setattr( - server_mod, - "_BUILTIN_STARTER_SOURCES", + builtin_content, + "BUILTIN_STARTER_SOURCES", [("missing.feedpak", "content/starter/does-not-exist.feedpak")], ) - server_mod._seed_builtin_starter_content(dlc) + builtin_content.seed_builtin_starter_content(server_mod._feedBack_server_root(), dlc) - assert not (dlc / server_mod._BUILTIN_STARTER_SUBDIR / "missing.feedpak").exists() - assert not (server_mod.CONFIG_DIR / server_mod._STARTER_SEED_MARKER).exists() + assert not (dlc / builtin_content.BUILTIN_STARTER_SUBDIR / "missing.feedpak").exists() + assert not (server_mod.CONFIG_DIR / builtin_content.STARTER_SEED_MARKER).exists() def test_every_starter_source_file_is_present(server_mod): @@ -190,7 +191,7 @@ def test_every_starter_source_file_is_present(server_mod): the checkout is clean, so "on disk" == committed.""" root = server_mod._feedBack_server_root() missing = [ - rel for _, rel in server_mod._BUILTIN_STARTER_SOURCES + rel for _, rel in builtin_content.BUILTIN_STARTER_SOURCES if not (root / rel).is_file() ] assert not missing, f"listed starter sources missing on disk: {missing}" @@ -199,18 +200,18 @@ def test_every_starter_source_file_is_present(server_mod): def test_seed_lands_every_listed_starter_pack(tmp_path, server_mod): """A real seed run copies every listed pack into starter/ and marks done.""" root = server_mod._feedBack_server_root() - for _, rel in server_mod._BUILTIN_STARTER_SOURCES: + for _, rel in builtin_content.BUILTIN_STARTER_SOURCES: if not (root / rel).is_file(): pytest.skip(f"starter source not present in checkout: {rel}") dlc = tmp_path / "dlc" dlc.mkdir() - server_mod._seed_builtin_starter_content(dlc) + builtin_content.seed_builtin_starter_content(server_mod._feedBack_server_root(), dlc) - for dest_name, _ in server_mod._BUILTIN_STARTER_SOURCES: - dest = dlc / server_mod._BUILTIN_STARTER_SUBDIR / dest_name + for dest_name, _ in builtin_content.BUILTIN_STARTER_SOURCES: + dest = dlc / builtin_content.BUILTIN_STARTER_SUBDIR / dest_name assert dest.is_file(), f"pack not seeded: {dest_name}" - assert (server_mod.CONFIG_DIR / server_mod._STARTER_SEED_MARKER).is_file() + assert (server_mod.CONFIG_DIR / builtin_content.STARTER_SEED_MARKER).is_file() def test_no_unlisted_starter_pack_on_disk(server_mod): @@ -220,7 +221,7 @@ def test_no_unlisted_starter_pack_on_disk(server_mod): main before being wired up. In CI the checkout is clean, so this flags any stray/committed pack that isn't listed.""" root = server_mod._feedBack_server_root() - listed = {rel for _, rel in server_mod._BUILTIN_STARTER_SOURCES} + listed = {rel for _, rel in builtin_content.BUILTIN_STARTER_SOURCES} if not listed: pytest.skip("no starter sources declared") content_dir = (root / next(iter(listed))).parent # all sources share this dir diff --git a/tests/test_progression_api.py b/tests/test_progression_api.py index e995e5f..9de9ff8 100644 --- a/tests/test_progression_api.py +++ b/tests/test_progression_api.py @@ -5,6 +5,7 @@ import importlib import json import sys +import builtin_content import pytest from fastapi.testclient import TestClient @@ -192,7 +193,7 @@ def test_low_accuracy_play_does_not_complete_gated_challenge(client): def test_diagnostic_at_100_completes_calibration(client, server): - diag = server._builtin_diagnostic_filename() + diag = builtin_content.builtin_diagnostic_filename() # A near-miss leaves calibration pending. _scored_play(client, filename=diag, accuracy=0.97, score=500) assert client.get("/api/progression").json()["onboarding"]["calibration_status"] == "pending" @@ -207,7 +208,7 @@ def test_diagnostic_play_does_not_feed_challenges_or_quests(client, server): # The calibration run is a perfect guitar play — it must yield rank 1 # EXACTLY, advancing neither the guitar path nor the daily song quest. client.post("/api/progression/paths", json={"add": ["guitar"]}) - r = _scored_play(client, filename=server._builtin_diagnostic_filename(), + r = _scored_play(client, filename=builtin_content.builtin_diagnostic_filename(), accuracy=1.0, score=500) summary = r.json()["progression"] assert summary["calibration_completed"] is True @@ -228,7 +229,7 @@ def test_pathless_diagnostic_run_still_completes_calibration(client, server): run is an earned achievement and must count even before any path is selected (e.g. a pre-progression profile playing the diagnostic as a hardware test) — yielding a valid pathless rank-1 state.""" - _scored_play(client, filename=server._builtin_diagnostic_filename(), + _scored_play(client, filename=builtin_content.builtin_diagnostic_filename(), accuracy=1.0, score=500) data = client.get("/api/progression").json() assert data["onboarding"]["calibration_status"] == "completed" @@ -240,7 +241,7 @@ def test_diagnostic_upgrades_skipped_without_rank_change(client, server): client.post("/api/progression/paths", json={"add": ["guitar"]}) r = client.post("/api/progression/onboarding", json={"action": "skip"}) assert r.json()["onboarding"]["calibration_status"] == "skipped" - _scored_play(client, filename=server._builtin_diagnostic_filename(), accuracy=1.0, score=500) + _scored_play(client, filename=builtin_content.builtin_diagnostic_filename(), accuracy=1.0, score=500) data = client.get("/api/progression").json() assert data["onboarding"]["calibration_status"] == "completed" assert data["mastery_rank"] == 1