refactor(server): carve builtin-content seeding into lib/builtin_content.py (R3b) (#900)

lib/builtin_content.py (321 lines moved). server.py 2,418 -> 2,098.

The calibration/diagnostic sloppaks and the starter library: _copy_builtin_packs,
_write_builtin_pack, the two seed helpers, their source tables, and the seed marker.

━━━ THE ONE SIGNATURE CHANGE, AND WHY THE CARVE IS UNSAFE WITHOUT IT ━━━

server.py has:

    def _feedBack_server_root() -> Path:
        return 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 into lib/ unchanged and it keeps working, silently, and returns lib/. There
is no docs/diagnostics under lib/, so every seed would find nothing, log "source missing"
at debug, and return. Nothing raises. Nothing fails. The starter library simply never
appears, and the calibration sloppak is never seeded — on a fresh install, in the field.

A verbatim move whose MEANING changed because __file__ did.

So this module cannot compute a root: `server_root` is 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 that way; the two seed helpers now do too.)

Everything else is byte-identical. CONFIG_DIR is read late as appstate.config_dir and the
DLC root through dlc_paths._get_dlc_dir — the same seam every router in lib/routers/ uses,
late-bound because tests monkeypatch it.

━━━ PYFLAKES FOUND THREE MISSING IMPORTS THE TESTS WOULD HAVE FOUND ONE AT A TIME ━━━

The moved code uses `secrets`, `stat` and `tempfile`; none was in my import block. Each is
a NameError on a live path. `python3 -m pyflakes` names all three in one shot — this is the
Python twin of the no-undef gate that guarded every frontend carve, and it should run on
every server.py slice from here.

It also flagged a PRE-EXISTING one I deliberately did not touch: server.py's
TuningProviderRegistry.get_merged() calls `logger.exception(...)` in an except handler and
there is no `logger` in the module (it is `log`). So a raising tuning provider takes down
the merged-tunings call for everyone, with a NameError naming the wrong problem. Filed as
issue #899 rather than smuggled into a carve whose whole value is being behaviour-neutral.

The constants lost their underscore prefix: they cross a module boundary now (the seed
tests read them), so `_BUILTIN_STARTER_SOURCES` was a lie.

pytest 2397, pyflakes 0, Codex 0. Guarded by the plugin_context contract test (#898).

Refs #48

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Byron Gamatos
2026-07-12 00:50:50 +02:00
committed by GitHub
co-authored by Claude Opus 4.8
parent 1cef01d02c
commit 70dbe45e27
5 changed files with 441 additions and 380 deletions
+378
View File
@@ -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)
+4 -324
View File
@@ -44,6 +44,7 @@ from dlc_paths import _get_dlc_dir, _resolve_dlc_path
# `appstate.configure(...)` below publishes into the same namespace routers read. # `appstate.configure(...)` below publishes into the same namespace routers read.
# Lives in lib/ because that is the one core dir every packaging path copies. # Lives in lib/ because that is the one core dir every packaging path copies.
import appstate import appstate
import builtin_content
# Extracted route modules. They import `appstate`, never `server` — one-way graph. # 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 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 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: def _feedBack_server_root() -> Path:
@@ -659,10 +653,6 @@ def _feedBack_server_root() -> Path:
return Path(__file__).resolve().parent 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, # 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()`. # path is unchanged; routers call `appstate.get_progression_content()`.
appstate.configure( appstate.configure(
get_progression_content=_get_progression_content, get_progression_content=_get_progression_content,
builtin_diagnostic_filename=_builtin_diagnostic_filename, builtin_diagnostic_filename=builtin_content.builtin_diagnostic_filename,
tuning_providers=tuning_providers, 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(): def _background_scan():
@@ -1038,8 +718,8 @@ def _background_scan():
log.warning("Scan: no DLC folder configured") log.warning("Scan: no DLC folder configured")
return return
_seed_builtin_diagnostic_sloppaks(dlc) builtin_content.seed_builtin_diagnostic_sloppaks(_feedBack_server_root(), dlc)
_seed_builtin_starter_content(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 # 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 # path isn't shared. Report the failure explicitly rather than silently
+22 -21
View File
@@ -7,6 +7,7 @@ import os
import sys import sys
import time import time
import builtin_content
import pytest 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/.""" """First seed copies the bundled sloppak into diagnostics-builtin/."""
dlc = tmp_path / "dlc" dlc = tmp_path / "dlc"
dlc.mkdir() 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(): if not source.is_file():
pytest.skip(f"source sloppak not present in checkout: {source}") 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.is_file()
assert dest.stat().st_size == source.stat().st_size 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.""" """Second seed leaves an up-to-date destination unchanged."""
dlc = tmp_path / "dlc" dlc = tmp_path / "dlc"
dlc.mkdir() 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(): if not source.is_file():
pytest.skip(f"source sloppak not present in checkout: {source}") 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]
first_mtime = dest.stat().st_mtime_ns first_mtime = dest.stat().st_mtime_ns
first_size = dest.stat().st_size 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_mtime_ns == first_mtime
assert dest.stat().st_size == first_size 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.""" """An existing newer destination is not overwritten."""
dlc = tmp_path / "dlc" dlc = tmp_path / "dlc"
dlc.mkdir() 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(): if not source.is_file():
pytest.skip(f"source sloppak not present in checkout: {source}") 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_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 = dest_dir / dest_name
dest.write_bytes(b"user-owned diagnostic copy") dest.write_bytes(b"user-owned diagnostic copy")
future = time.time() + 3600 future = time.time() + 3600
os.utime(dest, (future, future)) 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" 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.""" """A symlink at the destination is skipped, not written through."""
dlc = tmp_path / "dlc" dlc = tmp_path / "dlc"
dlc.mkdir() 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(): if not source.is_file():
pytest.skip(f"source sloppak not present in checkout: {source}") pytest.skip(f"source sloppak not present in checkout: {source}")
outside = tmp_path / "outside.txt" outside = tmp_path / "outside.txt"
outside.write_bytes(b"do not overwrite me") 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_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) 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. # The symlink target must be untouched and the link left as-is.
assert outside.read_bytes() == b"do not overwrite me" 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() dlc.mkdir()
outside_dir = tmp_path / "outside_dir" outside_dir = tmp_path / "outside_dir"
outside_dir.mkdir() 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 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. # Nothing was written through the directory symlink into the link target.
assert list(outside_dir.iterdir()) == [] 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 = tmp_path / "dlc"
dlc.mkdir() dlc.mkdir()
monkeypatch.setattr( monkeypatch.setattr(
server_mod, builtin_content,
"_BUILTIN_DIAGNOSTIC_SOURCES", "BUILTIN_DIAGNOSTIC_SOURCES",
[("missing.sloppak", "docs/diagnostics/does-not-exist.sloppak")], [("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()
+32 -31
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import importlib import importlib
import sys import sys
import builtin_content
import pytest import pytest
@@ -21,15 +22,15 @@ def server_mod(tmp_path, monkeypatch, isolate_logging):
def _source(server_mod): def _source(server_mod):
return ( return (
server_mod._feedBack_server_root() server_mod._feedBack_server_root()
/ server_mod._BUILTIN_STARTER_SOURCES[0][1] / builtin_content.BUILTIN_STARTER_SOURCES[0][1]
) )
def _dest(server_mod, dlc): def _dest(server_mod, dlc):
return ( return (
dlc dlc
/ server_mod._BUILTIN_STARTER_SUBDIR / builtin_content.BUILTIN_STARTER_SUBDIR
/ server_mod._BUILTIN_STARTER_SOURCES[0][0] / 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(): if not source.is_file():
pytest.skip(f"starter source not present in checkout: {source}") 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) dest = _dest(server_mod, dlc)
assert dest.is_file() assert dest.is_file()
assert dest.stat().st_size == source.stat().st_size 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): 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(): if not source.is_file():
pytest.skip(f"starter source not present in checkout: {source}") 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 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(): if not source.is_file():
pytest.skip(f"starter source not present in checkout: {source}") 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) dest = _dest(server_mod, dlc)
assert dest.is_file() assert dest.is_file()
@@ -86,7 +87,7 @@ def test_seed_runs_only_once_and_respects_deletion(tmp_path, server_mod):
dest.unlink() dest.unlink()
# A subsequent launch must not re-seed it. # 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() 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}") pytest.skip(f"starter source not present in checkout: {source}")
# dlc is None and DLC_DIR unset -> _get_dlc_dir() returns None. # dlc is None and DLC_DIR unset -> _get_dlc_dir() returns None.
server_mod._seed_builtin_starter_content(None) builtin_content.seed_builtin_starter_content(server_mod._feedBack_server_root(), None)
assert not (server_mod.CONFIG_DIR / server_mod._STARTER_SEED_MARKER).exists() assert not (server_mod.CONFIG_DIR / builtin_content.STARTER_SEED_MARKER).exists()
# Now a DLC is configured: the deferred seed runs. # Now a DLC is configured: the deferred seed runs.
dlc = tmp_path / "dlc" dlc = tmp_path / "dlc"
dlc.mkdir() 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 _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): 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 = tmp_path / "outside"
outside_dir.mkdir() 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()) == [] assert list(outside_dir.iterdir()) == []
# An incomplete seed must NOT write the marker, so a later launch retries. # 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): 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") dest.write_bytes(b"user's own edited pack")
_os.utime(dest, (1_000_000, 1_000_000)) # far older than the bundled source _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 assert dest.read_bytes() == b"user's own edited pack" # untouched
# counted as already-present, so the one-time seed considers itself done # 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): 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.parent.mkdir(parents=True, exist_ok=True)
bogus.mkdir() # user (or junk) placed a directory where the pack goes 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 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): 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 = tmp_path / "dlc"
dlc.mkdir() dlc.mkdir()
monkeypatch.setattr( monkeypatch.setattr(
server_mod, builtin_content,
"_BUILTIN_STARTER_SOURCES", "BUILTIN_STARTER_SOURCES",
[("missing.feedpak", "content/starter/does-not-exist.feedpak")], [("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 (dlc / builtin_content.BUILTIN_STARTER_SUBDIR / "missing.feedpak").exists()
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_every_starter_source_file_is_present(server_mod): 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.""" the checkout is clean, so "on disk" == committed."""
root = server_mod._feedBack_server_root() root = server_mod._feedBack_server_root()
missing = [ 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() if not (root / rel).is_file()
] ]
assert not missing, f"listed starter sources missing on disk: {missing}" 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): 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.""" """A real seed run copies every listed pack into starter/ and marks done."""
root = server_mod._feedBack_server_root() 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(): if not (root / rel).is_file():
pytest.skip(f"starter source not present in checkout: {rel}") pytest.skip(f"starter source not present in checkout: {rel}")
dlc = tmp_path / "dlc" dlc = tmp_path / "dlc"
dlc.mkdir() 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: for dest_name, _ in builtin_content.BUILTIN_STARTER_SOURCES:
dest = dlc / server_mod._BUILTIN_STARTER_SUBDIR / dest_name dest = dlc / builtin_content.BUILTIN_STARTER_SUBDIR / dest_name
assert dest.is_file(), f"pack not seeded: {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): 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 main before being wired up. In CI the checkout is clean, so this flags any
stray/committed pack that isn't listed.""" stray/committed pack that isn't listed."""
root = server_mod._feedBack_server_root() 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: if not listed:
pytest.skip("no starter sources declared") pytest.skip("no starter sources declared")
content_dir = (root / next(iter(listed))).parent # all sources share this dir content_dir = (root / next(iter(listed))).parent # all sources share this dir
+5 -4
View File
@@ -5,6 +5,7 @@ import importlib
import json import json
import sys import sys
import builtin_content
import pytest import pytest
from fastapi.testclient import TestClient 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): 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. # A near-miss leaves calibration pending.
_scored_play(client, filename=diag, accuracy=0.97, score=500) _scored_play(client, filename=diag, accuracy=0.97, score=500)
assert client.get("/api/progression").json()["onboarding"]["calibration_status"] == "pending" 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 # The calibration run is a perfect guitar play — it must yield rank 1
# EXACTLY, advancing neither the guitar path nor the daily song quest. # EXACTLY, advancing neither the guitar path nor the daily song quest.
client.post("/api/progression/paths", json={"add": ["guitar"]}) 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) accuracy=1.0, score=500)
summary = r.json()["progression"] summary = r.json()["progression"]
assert summary["calibration_completed"] is True 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 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 selected (e.g. a pre-progression profile playing the diagnostic as a
hardware test) — yielding a valid pathless rank-1 state.""" 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) accuracy=1.0, score=500)
data = client.get("/api/progression").json() data = client.get("/api/progression").json()
assert data["onboarding"]["calibration_status"] == "completed" 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"]}) client.post("/api/progression/paths", json={"add": ["guitar"]})
r = client.post("/api/progression/onboarding", json={"action": "skip"}) r = client.post("/api/progression/onboarding", json={"action": "skip"})
assert r.json()["onboarding"]["calibration_status"] == "skipped" 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() data = client.get("/api/progression").json()
assert data["onboarding"]["calibration_status"] == "completed" assert data["onboarding"]["calibration_status"] == "completed"
assert data["mastery_rank"] == 1 assert data["mastery_rank"] == 1