Compare commits

..
Author SHA1 Message Date
ChrisBeWithYou 2281e6ce5e Merge remote-tracking branch 'origin/feat/drum-parts-loader' into feat/drum-part-picker 2026-07-20 23:01:21 -05:00
ChrisBeWithYou c7fa0ca3f1 Normalize drum part pointer identities 2026-07-20 23:01:06 -05:00
ChrisBeWithYou 7b85eef104 Update reconnect source contract test 2026-07-20 22:48:23 -05:00
ChrisBeWithYou e9be269bb7 Merge remote-tracking branch 'origin/feat/drum-parts-loader' into feat/drum-part-picker 2026-07-20 22:46:59 -05:00
ChrisBeWithYou 1b8186b076 Fix drum-part review findings 2026-07-20 21:59:22 -05:00
ChrisBeWithYouandClaude Opus 4.8 9a2e98ed74 feat(player): drum-part picker for multiple drum charts (feedpak 1.17.0)
The last mile of the multiple-drum-parts feature: let a player CHOOSE which
drum chart plays. #1020 taught the loader + highway WS to carry several drum
parts (song_info.drum_parts + ?drum_part=<id> + a part_id echo on drum_tab);
this adds the host-chrome selector that drives it.

A "Drum part" <select> sits beside the arrangement switcher in the advanced
settings popover, shown only when a song has 2+ drum charts (drum_parts is
always present — empty for non-drum songs — so single-drum / no-drum songs
hide the row and nothing changes for them). Selecting a part re-streams that
part's tab over the highway WS, exactly like an arrangement switch.

- static/highway.js:
  - reconnect() gains a third `drumPart` arg → sets `?drum_part=<id>` on the WS
    URL (mirrors the existing `arrangement` param one line up). Empty/undefined
    → the primary part, i.e. byte-identical to today for any pack untouched.
  - song_info handler populates #drum-part-select from msg.drum_parts and
    shows/hides #v3-drum-part-row on `length > 1` (parallel to the #arr-select
    block right above it).
  - drum_tab handler carries msg.part_id onto hwState.drumTab (plugins can read
    bundle.drumTab.part_id) and reflects it as the picker's selected value, so
    the dropdown stays honest even when the server resolves an unknown/absent
    selection to the primary.
- static/app.js:
  - changeArrangement() gains an optional `drumPart`; at reconnect it forwards
    the explicit part, else preserves the current picker selection — so an
    ARRANGEMENT switch keeps the chosen drum part (parts are song-level).
  - new changeDrumPart(id) delegates to changeArrangement with the current
    arrangement held + the new part applied (a part switch is the same
    re-stream, so it reuses all the transition ceremony). Exported on window.
- static/v3/index.html: the #drum-part-select row (hidden by default).

No plugin change: the drum renderers just draw whatever drum_tab streams.

RUNTIME-VERIFIED (Playwright, the core player, a 2-drum pack + a no-drum pack):
10/10 — the picker populates with both parts and shows for the multi-drum song;
song_info.drum_parts reaches getSongInfo(); the primary is pre-selected;
selecting the 2nd part drives highway.reconnect with the id and the WS URL
carries `?drum_part=drums-2`; the picker then reflects the server's part_id
echo; a no-drum song hides the row; no page errors. ESLint 0 errors (the two
max-lines warnings are pre-existing on these files). No pytest touched (JS-only).
Stacked on #1020.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017xGPjDBF8NTwTK7VQvizix
Signed-off-by: ChrisBeWithYou <christian.a.cowan@gmail.com>
2026-07-20 17:13:31 -05:00
ChrisBeWithYouandClaude Opus 4.8 e0f1e2b641 feat(sloppak): load multiple drum parts (feedpak 1.17.0 drums-as-arrangements)
A song can now ship SEVERAL drum charts (a second drummer, an aux-percussion
layer). The Arrangement Editor already writes them per the feedpak 1.17.0 FEP
(feedpak-spec#63): the primary stays the song-level `drum_tab:` key (what this
app has always played), and each part rides the manifest as a `type: drums`
arrangement entry carrying a per-arrangement `drum_tab` file pointer and NO
note `file` — an entry this loader's file/notation gate already skips, which
is exactly why old builds are unaffected by such packs.

lib/sloppak.py:
- The arrangements loop collects drum-part pointer entries instead of merely
  skipping them — but still NEVER turns one into a fretted Arrangement. That
  skip is the grading invariant (an empty drum chart must not reach the
  fretted pipeline / note-detection grading) and is now pinned by test.
- New `LoadedSloppak.drum_parts`: [{id, name, drum_tab}], primary FIRST. The
  entry aliasing the song-level file contributes its id/name but is never
  loaded twice (the primary's payload IS `loaded.drum_tab`, same object).
  Legacy single-drum packs read as a one-part list; a pointer-only pack (a
  writer omitted the alias) promotes its first part so has_drum_tab, the
  default stream, and the drum-only placeholder keep working.
- The song-level drum_tab loading block is extracted verbatim into
  `_load_drum_tab_file()` and shared by both paths, so every part gets the
  same permissive posture: missing file → that part silently absent;
  traversal / parse / validation failure → that part skipped with a warning,
  never an aborted load. (The 9 pinned drumtab-load tests pass unchanged.)

lib/routers/ws_highway.py:
- `song_info` gains `drum_parts` (names only; always a list, empty without
  drums) so a part picker can bind unconditionally.
- `?drum_part=<id>` on the WS URL selects which part's tab streams as the
  `drum_tab`/`drum_hits` messages; the default and any unknown id fall back
  to the primary — byte-identical legacy behavior. The `drum_tab` message
  carries `part_id` only when a parts list exists, keeping the legacy frame
  unchanged.

Tests: tests/test_sloppak_drum_parts.py (9) — the grading invariant +
parallel-ids pin, primary-first resolution with alias identity, legacy
one-part list, pointer-only promotion, per-part failure isolation (bad JSON,
path traversal, duplicate rels), and the drum-only placeholder with pointer
entries. Full suite: the only failures are 9 machine-environmental tests
(installed desktop plugins under LOCALAPPDATA, CRLF/path-shape assertions)
that fail identically on an untouched origin/main checkout on this box.
tools/check_spec_conformance.py passes against the spec's current HEAD
(`drum_tab` and `type` are declared keys); the semantics of the
per-arrangement placement land in feedpak-spec#63 — this PR should merge
after it.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017xGPjDBF8NTwTK7VQvizix
Signed-off-by: ChrisBeWithYou <christian.a.cowan@gmail.com>
2026-07-20 15:36:17 -05:00
25 changed files with 127 additions and 2169 deletions
-90
View File
@@ -1,90 +0,0 @@
name: Content packs
# Build & publish opt-in career venue packs as per-pack, versioned, immutable
# releases (convention: tag `venue-<id>-v<N>`, asset `<id>-pack-v<N>.zip`),
# then open a PR bumping venues.json url/sha256/bytes. This is automation so
# publishing packs is never a person's manual job (SLIM-NIGHTLY item 1b).
#
# Immutable tags → publishing is a deliberate, versioned act, so this runs on
# manual dispatch (not push): a media change means a new version, a human call.
on:
workflow_dispatch:
inputs:
venues:
description: "Space-separated venue ids to (re)publish, e.g. 'club arena'"
required: true
default: "club"
version:
description: "Pack version N (tag venue-<id>-vN). Bump for new media."
required: true
default: "1"
concurrency:
group: content-packs
cancel-in-progress: false
permissions:
contents: write
pull-requests: write
jobs:
publish:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
# Media lives in-tree today; add `lfs: true` once SLIM-NIGHTLY item 4
# moves venue-packs/** to Git LFS.
- uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Build & publish packs
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
# Never interpolate dispatch inputs straight into the shell — a crafted
# value would execute on the runner with this job's write token. Pass
# via env, validate the formats, and use a Bash argument array.
VENUES: ${{ github.event.inputs.venues }}
VERSION: ${{ github.event.inputs.version }}
run: |
set -euo pipefail
[[ "$VERSION" =~ ^[1-9][0-9]*$ ]] || { echo "::error::version must be a positive integer"; exit 1; }
[[ "$VENUES" =~ ^[a-z0-9][a-z0-9-]*(\ [a-z0-9][a-z0-9-]*)*$ ]] || { echo "::error::venues must be space-separated venue ids"; exit 1; }
read -r -a venues <<< "$VENUES"
dirs=()
for v in "${venues[@]}"; do
dirs+=("plugins/career/venue-packs/$v")
done
python tools/content_packs.py "${dirs[@]}" \
--version "$VERSION" \
--publish \
--manifest /tmp/packs-manifest.json
cat /tmp/packs-manifest.json
- name: Apply url/sha256/bytes to venues.json
run: |
python - <<'PY'
import json, pathlib
manifest = json.load(open("/tmp/packs-manifest.json"))
vpath = pathlib.Path("plugins/career/venues.json")
data = json.loads(vpath.read_text())
for v in data["venues"]:
m = manifest.get(v["id"])
if m and v.get("pack"):
v["pack"].update(url=m["url"], sha256=m["sha256"], bytes=m["bytes"])
vpath.write_text(json.dumps(data, indent=4) + "\n")
PY
- name: Open manifest-bump PR
uses: peter-evans/create-pull-request@v6
with:
commit-message: "career: refresh venue pack manifest (v${{ github.event.inputs.version }})"
title: "career: refresh venue pack manifest (v${{ github.event.inputs.version }})"
body: |
Automated by the content-packs workflow after publishing
`${{ github.event.inputs.venues }}` v${{ github.event.inputs.version }}
to their `venue-<id>-v${{ github.event.inputs.version }}` releases.
Bumps `venues.json` pack url/sha256/bytes to match the uploaded zips.
branch: content-packs/manifest-bump
delete-branch: true
+1 -52
View File
@@ -8,44 +8,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Added
- **Core reader for source rigs (feedpak 1.18.0).** A pack can declare what a
MIDI part should sound like by binding a rig; core now reads that binding and
hands it to the client instead of dropping it. Three parts: the
`tone_changes` WS message carries the pack's rig bindings (`base_rig`, and
`rig` per change) alongside the tone names it already sent; the manifest
`rigs:` key loads the pack's rig library (`rigs.json`, spec §7.9) verbatim;
and the binding precedence is resolved per spec §5.1/§5.2 — a manifest
arrangement entry's `tones` replaces the arrangement JSON's **wholesale**
(no field-level merge), while top-level `drum_tones` binds the primary drum
part as the fallback a `type: drums` entry's own `tones` outranks. Core
deliberately stops there: it does not select a realization or apply the
`intent.gm` floor, which belong to whatever actually voices the part. Packs
that bind no rig produce a byte-identical `tone_changes` payload, so existing
consumers are unaffected.
- **Opt-in career venue packs (#122)** — higher-tier venue crowd media
(`club`, `arena`) is no longer bundled; the app downloads each pack on demand
from its release when you reach the venue (sha256-verified), keeping the
starter `bar` venue bundled for offline play. Trims ~678 MB from the desktop
download; an unpublished pack shows "coming soon" and plays on the standard
stage until its release lands.
- **Session-sync relay WebSocket — `/ws/sync/{session_id}` (#1030).** A
deliberately dumb JSON fan-out room: a text frame received from one client is
forwarded verbatim to every other client on the same session id; the server
interprets nothing (message schemas are owned by consumers). Rooms are created
on first join and garbage-collected when the last socket leaves — no history,
no replay, no persistence, so a host that crashes and rejoins the same id
resumes publishing to reconnecting subscribers with no server-side
coordination. Session ids are client-generated (`[A-Za-z0-9_-]{4,64}`); DoS
hygiene for a LAN-exposed port via frame-size (16 KB), per-room (16 sockets),
total-room (32), and per-socket rate (120 msg/s sustained, 240 burst) caps —
over-limit sockets are closed with a policy code and the room carries on, and
a peer that dies — or stalls: fan-out sends are bounded by a 5 s timeout —
mid-fan-out is dropped without disturbing delivery to the rest. `main.py`
also caps inbound WS frames at the transport (`ws_max_size=64 KB`, down from
uvicorn's 16 MB default) so oversized frames never materialize server-side. First consumer: splitscreen's "pop out to LAN" follower mode
(feedBack-plugin-splitscreen#21), which relays playhead/playstate/song-change
frames from a host window to view-only followers on other LAN devices.
Implementation in `lib/routers/ws_sync.py`; tests in `tests/test_ws_sync.py`.
- **Drum-part picker (feedpak 1.17.0 "drums as arrangements").** When a song
carries several drum charts, a **Drum part** selector appears beside the
arrangement switcher (advanced settings) so a player can choose which drummer
@@ -68,9 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
loaded as fretted arrangements — the loader's file/notation gate keeps a drum
part out of the fretted pipeline (and out of note-detection grading), pinned
by test. Legacy single-drum packs read exactly as before, as a one-part list.
- **`chart-transform` capability domain (#952)** — plugins can now remap the
chart before rendering and scoring through a core-owned provider
coordinator. Synchronous transforms run after difficulty filtering; host
- **Chart-transform coordinator.** Synchronous transforms run after difficulty filtering; host
data is isolated from providers, accepted timelines are time-sorted, and
failures fall back to the original chart with a fixed public reason.
Effective chart arrays and metadata are available to 2D/custom renderers
@@ -313,17 +273,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
engine (`app.js`, `highway.js`, `playSong`, `showScreen`, the capability registry).
### Fixed
- **Count-in follows the song's meter and its pickup measure.** The count-in
(loop wrap, section practice, and the "Countdown before song" setting) always
clicked exactly four beats, so a 3/4 song was counted in 4/4, and a song
opening with a pickup (anacrusis) had the pickup enter where the downbeat
belonged — putting the player a beat ahead for the whole song. The bar length
now comes from the `song_timeline` beats already on the highway
(`measure >= 0` marks downbeats; no new plumbing, since the `time_signatures`
map is streamed to plugins rather than stored in the frontend), and a first
bar shorter than that meter shortens the count by its length: a 1-beat pickup
in 4/4 counts "1 2 3" and the music enters on 4. Songs without beats — pre-chart,
minigames, synthetic highways — still get four.
- **GP8 asset resolution honours the directory the registry named.**
`<EmbeddedFilePath>` is matched on filename stem so a format variant of the
same recording can win (an `.ogg` beside the declared `.mp3` is copied out
+1 -1
View File
@@ -690,7 +690,7 @@ The highway WebSocket at `/ws/highway/{filename}?arrangement={index}` streams th
| `anchors` | `{ type, data: [{ time, fret, width }] }` | Fret zoom anchors |
| `chord_templates` | `{ type, data: [{ name, frets: [6] }] }` | Named chord shapes |
| `lyrics` | `{ type, data: [{ w, t, d }], source }` | Syllables: `w`=word, `t`=time, `d`=duration. `-` joins to previous, `+` = line break. `source` is one of `"xml"`, `"whisperx"`, `"user"` — UI can use it to render an "auto-transcribed" badge for `whisperx`. Sloppaks always include `source` (legacy sloppaks without a `lyrics_source` manifest key default to `"xml"` at load time). Loose folders set it based on which extractor matched. Absent only when no lyrics fired the message at all |
| `tone_changes` | `{ type: 'tone_changes', base, base_rig?, data: [{ t, name, rig? }] }` | Optional — tone change events relative to the arrangement base tone; only sent if tones were found. Note the time key is **`t`**, not `time` (both the sloppak path and the legacy XML path emit `t`). `base_rig` and each entry's `rig` are the pack's **rig bindings** — ids into [`rigs.json`](https://github.com/got-feedback/feedpak-spec/blob/main/spec/feedpak-v1.md#79-rigsjson) (feedpak §6.9/§7.9), carried through verbatim and **not** resolved by core: selecting a realization and applying the `intent.gm` floor belong to whatever voices the part. Both are **omitted entirely** when the chart binds no rig, so consumers predating the rig model see the payload they always did. |
| `tone_changes` | `{ type: 'tone_changes', base, data: [{ time, name }] }` | Optional — tone change events relative to the arrangement base tone; only sent if tones were found |
| `notes` | `{ type, data: [{ t, s, f, sus, ho, po, sl, bn, ... }] }` | Single notes |
| `chords` | `{ type, data: [{ t, notes: [{ s, f, sus, ... }] }] }` | Chord events |
| `phrases` | `{ type, data: [{ start_time, end_time, max_difficulty, levels: [{ difficulty, notes, chords, anchors, handshapes }] }], total }` | Optional — per-phrase difficulty ladder for master-difficulty slider (feedBack#48). Only sent when the source chart carries multi-level phrase data (phrase-aware sloppak). Sent in chunks (`data` is a batch, `total` is the full count across messages) to avoid multi-MB single frames. Absent for GP imports and legacy sloppak; consumers must treat missing message as "single fixed difficulty — slider disabled". |
+9 -18
View File
@@ -26,7 +26,6 @@ from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from song import (
anchor_to_wire,
arrangement_is_bass,
arrangement_string_count,
base_open_string_midis,
chord_template_to_wire,
@@ -274,8 +273,9 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1,
bass_idxs = [
i
for i, a in enumerate(song.arrangements)
if arrangement_is_bass(a)
if getattr(a, "path_bass", False)
or (smart_names[i] or "").lower().startswith("bass")
or "bass" in (getattr(a, "name", "") or "").lower()
]
if bass_idxs:
# Among the bass parts: (1) honor the saved default-arrangement
@@ -774,29 +774,20 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1,
# (Arrangement.tones, populated by the converter), so read it straight
# off `arr` rather than walking for XML that doesn't exist.
if is_slop:
# `sloppak_tone_changes` builds the (base, base_rig, sorted
# changes) triple from `Arrangement.tones`, skipping non-string
# names, non-finite/non-numeric times, and unusable rig ids —
# unit-tested in test_tones.py.
# `sloppak_tone_changes` builds the (base, sorted changes) pair
# from `Arrangement.tones`, skipping non-string names and
# non-finite/non-numeric times — unit-tested in test_tones.py.
from tones import sloppak_tone_changes
base_name, base_rig, tone_changes = sloppak_tone_changes(
getattr(arr, "tones", None)
)
base_name, tone_changes = sloppak_tone_changes(getattr(arr, "tones", None))
# Send when there's a base tone OR timed changes — a single-tone
# arrangement has a base but no switches, and the highway should
# still be able to show the initial tone.
if tone_changes or base_name:
payload = {
await websocket.send_json({
"type": "tone_changes",
"base": base_name,
"data": tone_changes,
}
# `base_rig` is additive (feedpak-spec §6.9) — omitted entirely
# when the chart binds no rig, so consumers that predate the rig
# model see the exact payload they always did.
if base_rig:
payload["base_rig"] = base_rig
await websocket.send_json(payload)
})
else:
xml_paths = sorted(_xml_walk("*.xml"))
@@ -1023,7 +1014,7 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1,
# base[string] + offset + capo + fret (matches the tuner / open-string
# labels). arrangement_string_count is O(notes), so compute once here.
_base = base_open_string_midis(
arrangement_string_count(arr), arrangement_is_bass(arr))
arrangement_string_count(arr), "bass" in (arr.name or "").lower())
_capo = int(getattr(arr, "capo", 0) or 0)
def _fill_scale_degree(wire: dict, n, t: float) -> None:
-140
View File
@@ -1,140 +0,0 @@
"""Session-sync relay WebSocket — /ws/sync/{session_id} (feedBack#1030).
A deliberately dumb fan-out room: a JSON text frame received from one client
is forwarded verbatim to every OTHER client connected to the same session id.
The server interprets nothing beyond the limits below message schemas are
owned entirely by consumers. First consumer: splitscreen's LAN follower mode
(feedBack-plugin-splitscreen#21), which relays playhead/playstate/song-change
frames from a host window to view-only followers on other LAN devices.
Design points (full spec in the issue):
- Rooms are created on first join and garbage-collected when the last socket
leaves. No history, no replay, no persistence a late joiner simply waits
for the next frame. Consumers that need state on join re-send it themselves
(splitscreen answers every follower ``hello`` with a fresh ``config``).
- That statelessness is what makes consumer crash-recovery work: a host that
relaunches and rejoins the same session id resumes publishing to its
reconnecting subscribers with no server-side coordination, and an idle room
is indistinguishable from a nonexistent one.
- ``session_id`` is client-generated and opaque (``[A-Za-z0-9_-]{4,64}``);
consumers pick their own id policy (splitscreen uses a short typeable,
persistent room key).
- DoS hygiene for a port that may be LAN-exposed: frame-size cap, per-room and
total-room caps, and a per-socket inbound token-bucket rate cap. Over-limit
sockets are closed with a policy code; the room carries on. A peer that dies
mid-fan-out is dropped without wedging delivery to the rest.
"""
import asyncio
import logging
import re
import time
from fastapi import APIRouter, WebSocket
log = logging.getLogger("feedBack.server")
router = APIRouter()
_SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_-]{4,64}$")
# Limits. Sized generously above the first consumer's needs (splitscreen
# publishes time frames at ~15-20 Hz to a handful of viewers) while bounding
# what an open LAN port can be made to do. All module-level so tests (and a
# desperate operator) can override them.
MAX_FRAME_BYTES = 16 * 1024
MAX_CLIENTS_PER_ROOM = 16
MAX_ROOMS = 32
RATE_MSGS_PER_SEC = 120.0 # sustained inbound frames per socket
RATE_BURST = 240.0 # token-bucket burst headroom
# A peer that stops draining its socket would leave send_text() pending
# forever — and since publishers await the fan-out gather, one stalled peer
# would stall every publisher's receive loop behind it. Bounding the send
# turns the stall into an eviction through the normal failed-send drop path.
SEND_TIMEOUT_SECONDS = 5.0
# RFC 6455 close codes.
_WS_UNSUPPORTED_DATA = 1003 # binary frame on a text-only relay
_WS_POLICY_VIOLATION = 1008 # invalid session id / rate cap exceeded
_WS_MSG_TOO_BIG = 1009
_WS_TRY_AGAIN_LATER = 1013 # room or server at capacity
# session_id → {socket: per-socket send lock}. The lock serializes concurrent
# fan-out sends to the same peer (two publishers relaying at once must not
# interleave writes on a third socket's transport).
_rooms: dict[str, dict[WebSocket, asyncio.Lock]] = {}
async def _send_locked(peer: WebSocket, lock: asyncio.Lock, text: str) -> None:
async with lock:
await asyncio.wait_for(peer.send_text(text), timeout=SEND_TIMEOUT_SECONDS)
@router.websocket("/ws/sync/{session_id}")
async def sync_ws(websocket: WebSocket, session_id: str):
"""Join the fan-out room *session_id*; relay every inbound text frame."""
await websocket.accept()
if not _SESSION_ID_RE.fullmatch(session_id):
await websocket.close(code=_WS_POLICY_VIOLATION, reason="invalid session id")
return
# Capacity checks and insertion run with no await between them, so
# concurrent joiners on the event loop can't race past the caps.
room = _rooms.get(session_id)
if room is None:
if len(_rooms) >= MAX_ROOMS:
await websocket.close(code=_WS_TRY_AGAIN_LATER, reason="too many active sessions")
return
room = _rooms[session_id] = {}
log.debug("ws_sync: room %s created", session_id)
elif len(room) >= MAX_CLIENTS_PER_ROOM:
await websocket.close(code=_WS_TRY_AGAIN_LATER, reason="session full")
return
room[websocket] = asyncio.Lock()
tokens = RATE_BURST
last_refill = time.monotonic()
try:
while True:
message = await websocket.receive()
if message["type"] == "websocket.disconnect":
break
text = message.get("text")
if text is None:
await websocket.close(code=_WS_UNSUPPORTED_DATA, reason="text frames only")
break
if len(text.encode("utf-8", errors="ignore")) > MAX_FRAME_BYTES:
await websocket.close(code=_WS_MSG_TOO_BIG, reason="frame too large")
break
now = time.monotonic()
tokens = min(RATE_BURST, tokens + (now - last_refill) * RATE_MSGS_PER_SEC)
last_refill = now
tokens -= 1.0
if tokens < 0:
await websocket.close(code=_WS_POLICY_VIOLATION, reason="rate cap exceeded")
break
peers = [(ws, lock) for ws, lock in room.items() if ws is not websocket]
if not peers:
continue
results = await asyncio.gather(
*(_send_locked(ws, lock, text) for ws, lock in peers),
return_exceptions=True,
)
# A peer that failed mid-send is dropped from the room here; its
# own handler finishes cleanup (the finally below) when its
# receive loop observes the disconnect.
for (peer, _lock), result in zip(peers, results):
if isinstance(result, Exception):
room.pop(peer, None)
finally:
room.pop(websocket, None)
# Guard against deleting a NEW room another joiner created after this
# one emptied (only possible for a dict that is no longer ours).
if not room and _rooms.get(session_id) is room:
del _rooms[session_id]
log.debug("ws_sync: room %s closed", session_id)
+79 -206
View File
@@ -121,41 +121,6 @@ def partition_stems(stems: list[dict]) -> tuple[dict | None, list[dict]]:
return full, [s for s in stems if str(s.get("id", "")) != FULL_MIX_STEM_ID]
def _resolve_pack_path(source_dir: Path, rel: str, label: str) -> Path | None:
"""Resolve a manifest-relative path, contained inside the pack. None if not.
Every manifest key that names a file routes through here. A crafted manifest
must not read outside the sloppak directory via path traversal
(e.g. `../../etc`), and a symlink loop or permission error on `.resolve()`
must disable that one file rather than abort the whole load so both
failures are caught, and both are warnings rather than raises.
The two branches log differently on purpose: a `ValueError` means the path
resolved *outside* the pack (a crafted or broken manifest), an `OSError`
means it could not be resolved at all (symlink loop, permissions). Reading
"escapes source_dir" in the logs and reading "resolution failed" lead an
operator to very different places, so the distinction is worth two lines.
Returns the resolved path **existence is NOT checked here**. Callers
differ on that deliberately: a missing optional side-file is silent, while a
missing arrangement skips an entry, so each caller keeps its own `.exists()`
(or `.is_file()`) test and its own control flow.
`label` names the manifest key in the log message ("keys", "song_timeline",
a drum part's id, …).
"""
try:
p = (source_dir / rel).resolve()
p.relative_to(source_dir.resolve())
except ValueError:
log.warning("sloppak: %s path %r escapes source_dir — skipped", label, rel)
return None
except OSError as e:
log.warning("sloppak: %s path resolution failed (%s) — skipped", label, e)
return None
return p
def _legacy_full_mix(manifest: dict, source_dir: Path) -> str | None:
"""Full mix from the DEPRECATED `original_audio:` manifest key, or None.
@@ -187,8 +152,16 @@ def _legacy_full_mix(manifest: dict, source_dir: Path) -> str | None:
if not isinstance(rel_raw, str) or not rel_raw.strip():
return None
rel = rel_raw.strip()
target = _resolve_pack_path(source_dir, rel, "original_audio")
if target is None or not target.is_file():
try:
target = (source_dir / rel).resolve()
target.relative_to(source_dir.resolve())
except ValueError:
log.warning("sloppak: original_audio path %r escapes source_dir — skipped", rel)
return None
except OSError as e:
log.warning("sloppak: original_audio path resolution failed (%s) — skipped", e)
return None
if not target.is_file():
return None
log.info(
"sloppak: pack uses the deprecated `original_audio:` key (%r) — the full mix "
@@ -725,14 +698,6 @@ class LoadedSloppak:
# absent / unreadable / malformed. Streamed over the highway WS as a
# `keys` message; consumers (renderers, plugins) read it from there.
keys: dict | None = None
# Parsed `rigs.json` payload (manifest `rigs:` key, spec §7.9) — the pack's
# library of engine-agnostic signal chains: effect chains and, since
# feedpak 1.18.0, MIDI-voiced sound sources. Arrangements bind rigs to time
# by referencing a rig `id` from `tones.base_rig` / `tones.changes[].rig`
# (§6.9), which `lib/tones.py` carries onto the wire. None when absent /
# unreadable / malformed. Rig objects are kept verbatim — this loader does
# not select realizations or apply the `intent.gm` floor.
rigs: dict | None = None
# Sanitized song-level tempo + time-signature maps from `song_timeline.json`
# (feedpak 1.2.0). `tempos`: [{time, bpm}]; `time_signatures`: [{time, ts}].
# None when absent/empty. Streamed over the highway WS (`tempos` /
@@ -784,8 +749,20 @@ def _load_drum_tab_file(source_dir: Path, rel: str, label: str) -> dict | None:
permissive a missing file disables that part silently; a traversal,
parse, or validation failure disables it with a warning, never aborting
the load."""
dt_path = _resolve_pack_path(source_dir, rel, label)
if dt_path is None or not dt_path.exists():
# Constrain to source_dir to prevent a crafted manifest from reading
# files outside the sloppak directory via path traversal (e.g. ../../etc).
# Wrap both resolve() calls in a broad handler: symlink loops and
# permission errors on .resolve() should disable drums, not abort load.
try:
dt_path = (source_dir / rel).resolve()
dt_path.relative_to(source_dir.resolve())
except ValueError:
log.warning("sloppak: %s path %r escapes source_dir — skipped", label, rel)
return None
except OSError as e:
log.warning("sloppak: %s path resolution failed (%s) — skipped", label, e)
return None
if not dt_path.exists():
return None
try:
raw = load_json(dt_path)
@@ -799,109 +776,18 @@ def _load_drum_tab_file(source_dir: Path, rel: str, label: str) -> dict | None:
return raw
def _load_rigs_file(source_dir: Path, rel: str) -> dict | None:
"""Load the pack's rig library (manifest `rigs:` key, spec §7.9).
Returns `{"version": int, "rigs": [...]}` or None. Same permissive posture
as every other side-file: missing / unreadable / malformed -> None, never
fatal spec §7.9 is explicit that a rig library a Reader can't use MUST NOT
fail the pack.
Rig objects are kept **verbatim**. Only entries that could never be
addressed are dropped a rig is reachable solely by `id` (from
`tones.base_rig` / `changes[].rig`), so a non-dict entry or one without a
usable string id is unreferenceable by construction. Everything else,
including unknown `role` / `engine` / `kind` values and `ext` namespaces,
passes through untouched, because this loader does not interpret rigs:
realization selection and the `intent.gm` fallback belong to whatever
voices the part.
"""
r_path = _resolve_pack_path(source_dir, rel, "rigs")
if r_path is None or not r_path.exists():
return None
try:
raw = load_json(r_path)
except Exception as e:
log.warning("sloppak: failed to parse rigs %r: %s", rel, e)
return None
if not isinstance(raw, dict):
log.warning("sloppak: rigs %r ignored — expected dict, got %s",
rel, type(raw).__name__)
return None
if not isinstance(raw.get("rigs"), list):
log.warning("sloppak: rigs %r ignored — 'rigs' must be a list", rel)
return None
clean_rigs: list[dict] = []
seen: set[str] = set()
for rig in raw["rigs"]:
if not isinstance(rig, dict):
continue
rid = rig.get("id")
if not isinstance(rid, str) or not rid.strip():
continue
# Normalize the library side of the lookup the same way the reference
# side is normalized in lib/tones.py — otherwise a pack with padded ids
# fails to resolve against a stripped `base_rig` / `rig`.
rid = rid.strip()
# A duplicate id makes `tones.base_rig` ambiguous, which would surface
# as the wrong sound rather than an error. First wins, loudly.
if rid in seen:
log.warning("sloppak: rigs %r has duplicate rig id %r — later one ignored",
rel, rid)
continue
seen.add(rid)
clean_rigs.append({**rig, "id": rid})
# int only — a float version (incl. NaN/Inf, which json.loads accepts)
# would raise on int(); default rather than abort an optional side-file.
_ver = raw.get("version")
return {
"version": _ver if isinstance(_ver, int) and not isinstance(_ver, bool) else 1,
"rigs": clean_rigs,
}
def _entry_tones(entry: dict) -> dict | None:
"""A manifest entry's `tones` binding, or None when it doesn't carry one.
Spec §5.2: a manifest arrangement entry's `tones` overrides the arrangement
JSON's `tones` **wholesale** — no field-level merge. This normalizes the
"does it carry one" test for both the arrangement path and the drum path.
An empty dict reads as *absent*, not as "override to silence": it is what a
Writer emits by accident, `arrangement_from_wire` already normalizes the
in-JSON `{}` to None the same way, and treating it as an override would let
a stray empty object silently unbind a part's sound.
"""
tones = entry.get("tones")
return tones if isinstance(tones, dict) and tones else None
def _resolve_drum_parts(
source_dir: Path,
drum_tab_rel: object,
drum_tab_data: dict | None,
drum_pointer_entries: list[dict],
drum_tones: dict | None = None,
) -> tuple[dict | None, list[dict] | None]:
"""Resolve drum pointers into a primary-first list with unique ids.
Also binds each part's sound (feedpak 1.18.0). The precedence mirrors the
`drum_tab` alias rule this function already implements: a `type: drums`
entry's own `tones` wins for that part, and the song-level `drum_tones` is
the fallback for the **primary** part only. A Reader MUST NOT apply both to
the same part (spec §5.1/§5.2), which is why the primary picks one or the
other here rather than merging them.
"""
"""Resolve drum pointers into a primary-first list with unique ids."""
if drum_tab_data is None and not drum_pointer_entries:
return drum_tab_data, None
primary_id = "drums"
primary_name = None
# The primary's own binding, lifted from its alias pointer entry when it has
# one. Stays None if no entry claims the primary — `drum_tones` fills in.
primary_tones = None
extra_parts: list[dict] = []
seen_rels: set[str] = set()
# Use the same canonical, traversal-safe identity as zip member lookup so
@@ -925,11 +811,6 @@ def _resolve_drum_parts(
primary_id = entry_id
if entry_name:
primary_name = entry_name
# This entry IS the primary (an alias pointer at the same file), so
# its binding is the primary's — and it outranks `drum_tones`.
_alias_tones = _entry_tones(entry)
if _alias_tones is not None:
primary_tones = _alias_tones
continue
tab = _load_drum_tab_file(source_dir, rel, f"drum part {entry_id or rel}")
if tab is None:
@@ -940,9 +821,6 @@ def _resolve_drum_parts(
"name": entry_name
or (tab_name if isinstance(tab_name, str) and tab_name else "Drums"),
"drum_tab": tab,
# Non-primary parts bind through their own entry only; `drum_tones`
# is explicitly the primary's fallback, never theirs.
"tones": _entry_tones(entry),
})
parts: list[dict] = []
@@ -951,14 +829,7 @@ def _resolve_drum_parts(
if primary_name is None:
tab_name = drum_tab_data.get("name")
primary_name = tab_name if isinstance(tab_name, str) and tab_name else "Drums"
parts.append({
"id": primary_id,
"name": primary_name,
"drum_tab": drum_tab_data,
# Entry `tones` takes precedence; `drum_tones` is the fallback. One
# or the other, never both on the same part (spec §5.1).
"tones": primary_tones if primary_tones is not None else drum_tones,
})
parts.append({"id": primary_id, "name": primary_name, "drum_tab": drum_tab_data})
used_ids.add(primary_id)
next_generated_id = 2
@@ -1011,25 +882,16 @@ def load_song(
rel = rel_raw.strip() if isinstance(rel_raw, str) else ""
notation_raw = entry.get("notation")
has_notation_key = isinstance(notation_raw, str) and bool(notation_raw.strip())
_etype = str(entry.get("type") or "").strip().lower()
is_drums = _etype in ("drums", "drum")
# A drums-typed entry MUST NEVER become a fretted Arrangement (grading
# invariant, spec §5.2/§7.5): route on `type` FIRST, not on file
# absence — a malformed drums entry that also carries a note file/
# notation would otherwise fall through and grade as garbage.
if is_drums or (not rel and not has_notation_key):
if not rel and not has_notation_key:
# A DRUM-PART POINTER entry (feedpak 1.17.0 "drums as
# arrangements"): `type: drums` with a per-arrangement `drum_tab`
# file. Collect it for the drum-parts load after this loop.
if is_drums and isinstance(entry.get("drum_tab"), str):
# file and no note file. Collect it for the drum-parts load after
# this loop — but NEVER turn it into a fretted Arrangement: this
# skip is the grading invariant (a drum part must not reach the
# fretted pipeline, where its empty chart would grade as garbage).
_etype = str(entry.get("type") or "").strip().lower()
if _etype in ("drums", "drum") and isinstance(entry.get("drum_tab"), str):
drum_pointer_entries.append(entry)
elif is_drums:
# Drums-typed but no drum_tab pointer — drop it (any note
# file/notation it carries is ignored), never fret it.
log.warning(
"sloppak: drums-typed arrangement entry %r has no drum_tab pointer — dropped",
entry.get("id"),
)
elif isinstance(entry.get("drum_tab"), str):
log.warning(
"sloppak: arrangement entry has drum_tab %r but type=%r — ignored",
@@ -1038,8 +900,16 @@ def load_song(
continue
data = None
if rel:
arr_path = _resolve_pack_path(source_dir, rel, "arrangement")
if arr_path is None or not arr_path.exists():
try:
arr_path = (source_dir / rel).resolve()
arr_path.relative_to(source_dir.resolve())
except ValueError:
log.warning("sloppak: arrangement path %r escapes source_dir — skipped", rel)
continue
except OSError as e:
log.warning("sloppak: arrangement path resolution failed (%s) — skipped", e)
continue
if not arr_path.exists():
continue
try:
data = load_json(arr_path)
@@ -1056,11 +926,6 @@ def load_song(
# the arrangement JSON (name, tuning, capo, centOffset).
if entry.get("name"):
arr.name = str(entry["name"])
# Editor-authored instrument type (feedpak-spec §5.2 / editor PR #335).
# Drives arrangement_string_count's bass fallback so a bass authored on
# an arrangement whose NAME doesn't say "bass" still reports 4 strings.
if entry.get("type"):
arr.type = str(entry["type"]).strip().lower()
if "tuning" in entry:
arr.tuning = list(entry["tuning"])
if "capo" in entry:
@@ -1069,14 +934,6 @@ def load_song(
# _finite_float keeps a malformed manifest NaN/Infinity from
# poisoning the song_info JSON (same guard as the wire path).
arr.cent_offset = _finite_float(entry["centOffset"])
# `tones` overrides WHOLESALE, unlike the field-level overrides above:
# the entry's object replaces the arrangement JSON's entirely, with no
# per-field merge (spec §5.2). A Writer SHOULD NOT emit both, but when
# one does, a half-merged sound — this pack's base with that pack's
# changes — would be worse than either source alone.
_entry_tone_block = _entry_tones(entry)
if _entry_tone_block is not None:
arr.tones = _entry_tone_block
# Beats/sections can live on the arrangement itself in the wire format.
# If the manifest-level arrangement JSON carries them, pull them onto
@@ -1109,7 +966,15 @@ def load_song(
notation_rel = notation_rel.strip()
if not notation_rel:
continue
nt_path = _resolve_pack_path(source_dir, notation_rel, "notation")
try:
nt_path = (source_dir / notation_rel).resolve()
nt_path.relative_to(source_dir.resolve())
except ValueError:
log.warning("sloppak: notation path %r escapes source_dir — skipped", notation_rel)
nt_path = None
except OSError as e:
log.warning("sloppak: notation path resolution failed (%s) — skipped", e)
nt_path = None
raw_nt = None
if nt_path is not None and nt_path.exists():
try:
@@ -1141,15 +1006,8 @@ def load_song(
# Keep the dense compatibility logic independently testable and guarantee
# ids are unique before the highway exposes them as selectors.
# Top-level `drum_tones` (spec §5.1) binds the song-level drum part — the
# fallback for packs without `type: drums` arrangements. Same shape as an
# arrangement entry's `tones`; `_resolve_drum_parts` owns the precedence.
_raw_drum_tones = manifest.get("drum_tones")
drum_tones_data = _raw_drum_tones if isinstance(_raw_drum_tones, dict) and _raw_drum_tones else None
drum_tab_data, drum_parts = _resolve_drum_parts(
source_dir, drum_tab_rel, drum_tab_data, drum_pointer_entries,
drum_tones_data,
)
# Drum-only sloppak: every GP track was percussion, so it ships a
@@ -1189,7 +1047,15 @@ def load_song(
time_sigs_data: list | None = None
song_timeline_rel = manifest.get("song_timeline")
if isinstance(song_timeline_rel, str) and song_timeline_rel:
st_path = _resolve_pack_path(source_dir, song_timeline_rel, "song_timeline")
try:
st_path = (source_dir / song_timeline_rel).resolve()
st_path.relative_to(source_dir.resolve())
except ValueError:
log.warning("sloppak: song_timeline path %r escapes source_dir — skipped", song_timeline_rel)
st_path = None
except OSError as e:
log.warning("sloppak: song_timeline path resolution failed (%s) — skipped", e)
st_path = None
if st_path is not None and st_path.exists():
try:
raw = load_json(st_path)
@@ -1279,7 +1145,15 @@ def load_song(
# downstream through the WS path.
lyrics_rel = manifest.get("lyrics")
if isinstance(lyrics_rel, str) and lyrics_rel:
lyr_path = _resolve_pack_path(source_dir, lyrics_rel, "lyrics")
try:
lyr_path = (source_dir / lyrics_rel).resolve()
lyr_path.relative_to(source_dir.resolve())
except ValueError:
log.warning("sloppak: lyrics path %r escapes source_dir — skipped", lyrics_rel)
lyr_path = None
except OSError as e:
log.warning("sloppak: lyrics path resolution failed (%s) — skipped", e)
lyr_path = None
if lyr_path is not None and lyr_path.exists():
try:
raw = load_json(lyr_path)
@@ -1384,7 +1258,15 @@ def load_song(
keys_data: dict | None = None
keys_rel = manifest.get("keys")
if isinstance(keys_rel, str) and keys_rel:
k_path = _resolve_pack_path(source_dir, keys_rel, "keys")
try:
k_path = (source_dir / keys_rel).resolve()
k_path.relative_to(source_dir.resolve())
except ValueError:
log.warning("sloppak: keys path %r escapes source_dir — skipped", keys_rel)
k_path = None
except OSError as e:
log.warning("sloppak: keys path resolution failed (%s) — skipped", e)
k_path = None
if k_path is not None and k_path.exists():
try:
raw = load_json(k_path)
@@ -1429,14 +1311,6 @@ def load_song(
"events": clean_events,
}
# Optional rigs.json — the pack's rig library (manifest `rigs:` key,
# spec §7.9). Loaded here so the highway WS can hand it to whatever voices
# the part; the bindings that reference it ride the arrangement's `tones`.
rigs_data: dict | None = None
rigs_rel = manifest.get("rigs")
if isinstance(rigs_rel, str) and rigs_rel:
rigs_data = _load_rigs_file(source_dir, rigs_rel)
_fpv = manifest.get("feedpak_version")
# The pack's full mix. Normally the RESERVED `full` stem partitioned out
# above (spec §5.3) — no path work needed, it was validated with the other
@@ -1467,7 +1341,6 @@ def load_song(
tempos=tempos_data,
time_signatures=time_sigs_data,
keys=keys_data,
rigs=rigs_data,
notation_by_id=notation_by_id_data,
arrangement_ids=arrangement_ids_acc,
full_mix=full_mix_data,
+7 -43
View File
@@ -182,12 +182,6 @@ class Arrangement:
# `base`/`changes` drive the highway tone-change markers; `definitions`
# feed the Tones plugin gear panel.
tones: dict | None = None
# Editor-authored instrument type (feedpak-spec §5.2 / editor PR #335):
# "bass" | "guitar" | "piano" | "keys" | "drums" | "". First-class DATA that
# lets a user author an instrument on an arrangement whose NAME doesn't say
# so. Lifted from the sloppak manifest entry by sloppak.load_song(); "" for
# archive/loose sources, which instead carry the path_* flags below.
type: str = ""
# arrangement XML <arrangementProperties> flags for smart naming (feedBack feat/arrangement).
# Populated from the XML; default False/0 for sloppak / GP-imported sources.
path_lead: bool = False
@@ -509,8 +503,8 @@ def note_pitch_midi(arr: "Arrangement", note: "Note") -> int | None:
O(notes) via ``arrangement_string_count`` for a whole arrangement, hoist
the base with :func:`base_open_string_midis` and call :func:`pitch_from_base`
per note instead."""
base = base_open_string_midis(arrangement_string_count(arr),
arrangement_is_bass(arr))
is_bass = "bass" in (arr.name or "").lower()
base = base_open_string_midis(arrangement_string_count(arr), is_bass)
return pitch_from_base(base, int(getattr(arr, "capo", 0) or 0),
arr.tuning or [], note.string, note.fret)
@@ -639,23 +633,6 @@ def phrase_from_wire(d: dict) -> Phrase:
)
def arrangement_is_bass(arr: Arrangement) -> bool:
"""Whether ``arr`` is a bass, most-authoritative signal first: an
editor-authored ``type == "bass"`` (feedpak-spec §5.2 / editor PR #335,
lifted onto sloppaks by :func:`sloppak.load_song`), then the archive
``path_bass`` <arrangementProperties> flag, then the legacy "bass"
case-insensitive substring in the name. Single source of the bass decision
so string-count derivation and the open-string pitch base (via
:func:`base_open_string_midis`) agree a bass authored on an arrangement
whose NAME doesn't say "bass" must get both 4 lanes AND the bass MIDI base,
not 4 lanes on a guitar octave."""
return (
(arr.type or "").strip().lower() == "bass"
or bool(arr.path_bass)
or "bass" in (arr.name or "").lower()
)
def arrangement_string_count(arr: Arrangement) -> int:
"""Derive the active arrangement's string count.
@@ -673,17 +650,10 @@ def arrangement_string_count(arr: Arrangement) -> int:
But this is a LOWER BOUND only a 6-string lead chart that
never plays string 5 reports 5, undercounting by 1.
2. **Instrument-type fallback.** An arrangement whose authoritative
instrument signal says bass defaults to 4; everything else
defaults to 6. This catches the partial-string-usage case where
notes don't span all the instrument's strings. The bass signal is
any of: an editor-authored ``type == "bass"`` (feedpak-spec §5.2 /
editor PR #335 — lifted onto sloppaks by ``sloppak.load_song``),
the ``path_bass`` <arrangementProperties> flag (archive/DLC
sources), or the legacy "bass" case-insensitive substring in the
name. Trusting ``type``/``path_bass`` closes the gap where a user
authors a bass instrument on an arrangement whose NAME doesn't say
"bass" (the editor lays out 4 lanes; core must agree).
2. **Name-based fallback.** Arrangements named "Bass" (case-
insensitive substring match) default to 4; everything else
defaults to 6. This catches the partial-string-usage case
where notes don't span all the instrument's strings.
A third signal ``len(arr.tuning)`` when it isn't the arrangement XML
padded value of 6 folds in for sloppak / GP-imported sources
@@ -714,10 +684,6 @@ def arrangement_string_count(arr: Arrangement) -> int:
max(0, 4, 0) = 4
* Empty arrangement named "Lead" (tuning len 6)
max(0, 6, 0) = 6
* Editor-authored bass named "Low End" (type "bass", tuning len 6,
notes 0..3) name_based=4 max(4, 4, 0) = 4
* DLC bass named "Low End" (pathBass flag set, tuning len 6, notes
0..3) name_based=4 max(4, 4, 0) = 4
Topkoa's issue argues plugins shouldn't do arrangement-name
matching; server-side fallback IS the right place for it
@@ -733,9 +699,7 @@ def arrangement_string_count(arr: Arrangement) -> int:
if cn.string > max_s:
max_s = cn.string
notes_count = max_s + 1 if max_s >= 0 else 0
# Bass signal (type / pathBass / name substring) — see arrangement_is_bass.
# Any one being bass pulls the fallback to 4.
name_based = 4 if arrangement_is_bass(arr) else 6
name_based = 4 if "bass" in arr.name.lower() else 6
# Tuning-length signal — only trustworthy when NOT the arrangement XML
# padded value of 6. Length 4/5 indicates explicit bass / 5-string
# bass; length 7/8 indicates an extended-range guitar from GP.
+9 -25
View File
@@ -32,29 +32,20 @@ def tokens(s: str) -> set[str]:
return {t for t in re.split(r"[^a-z0-9]+", (s or "").lower()) if t}
def sloppak_tone_changes(arr_tones) -> tuple[str, str, list[dict]]:
def sloppak_tone_changes(arr_tones) -> tuple[str, list[dict]]:
"""Build the highway tone-change payload from an arrangement's tone block.
Given ``Arrangement.tones`` (the dict embedded in the sloppak, or ``None``),
returns ``(base, base_rig, changes)`` where ``base`` is the initial tone
name, ``base_rig`` is the ``rigs.json`` rig id bound to it (feedpak-spec
§6.9; ``""`` when absent), and ``changes`` is a time-sorted
``[{"t", "name", "rig"?}]`` list. Non-string names, non-dict entries, and
non-numeric / non-finite times are skipped a hand-edited or third-party
sloppak must not crash the highway WebSocket or emit NaN/inf (which the
client's ``JSON.parse`` rejects).
``rig`` / ``base_rig`` are carried through but NOT resolved against
``rigs.json`` here: this builder only preserves the binding the chart
declared. Realization selection and the ``intent.gm`` fallback (§7.9) belong
to the consumer that actually voices the part.
returns ``(base, changes)`` where ``base`` is the initial tone name and
``changes`` is a time-sorted ``[{"t", "name"}]`` list. Non-string names,
non-dict entries, and non-numeric / non-finite times are skipped a
hand-edited or third-party sloppak must not crash the highway WebSocket
or emit NaN/inf (which the client's ``JSON.parse`` rejects).
"""
if not isinstance(arr_tones, dict):
return "", "", []
return "", []
base_val = arr_tones.get("base", "")
base = base_val.strip() if isinstance(base_val, str) else ""
base_rig_val = arr_tones.get("base_rig", "")
base_rig = base_rig_val.strip() if isinstance(base_rig_val, str) else ""
changes: list[dict] = []
raw_changes = arr_tones.get("changes")
@@ -74,13 +65,6 @@ def sloppak_tone_changes(arr_tones) -> tuple[str, str, list[dict]]:
continue
if not math.isfinite(t):
continue
change = {"t": round(t, 3), "name": name}
# ponytail: `rig` only when it's a usable id — a non-string or blank
# value is dropped rather than forwarded, so a consumer can treat
# presence of the key as "this change binds a rig".
rig = c.get("rig")
if isinstance(rig, str) and rig.strip():
change["rig"] = rig.strip()
changes.append(change)
changes.append({"t": round(t, 3), "name": name})
changes.sort(key=lambda x: x["t"])
return base, base_rig, changes
return base, changes
-7
View File
@@ -44,13 +44,6 @@ def run() -> None:
# record — including early startup messages — passes through the same
# structured pipeline.
log_config=None,
# Cap inbound WebSocket frames at the transport, before uvicorn
# materializes them in memory (its default is 16 MB). No client sends
# large frames to this server: the highway WS receives only small
# control messages, and the /ws/sync relay enforces its own tighter
# 16 KB application cap (routers/ws_sync.py MAX_FRAME_BYTES) — this is
# the defense-in-depth bound above it.
ws_max_size=64 * 1024,
)
+2 -9
View File
@@ -104,13 +104,6 @@ def _bundled(venue_id):
return (_bundled_venue_dir(venue_id) / "manifest.json").is_file()
def _pack_published(pack):
"""A remote pack is downloadable only once a publish has stamped its real
size the committed manifest carries a 0-byte placeholder (and an all-zero
sha) until then, so don't offer a download that can't succeed yet."""
return bool(pack and (pack.get("bytes") or 0) > 0)
def _stars():
"""(total, per-song dict, detail rows). Accuracy is a 0..1 fraction."""
db = _state["meta_db"]
@@ -671,7 +664,7 @@ def setup(app, context):
"unlocked": stars_total >= v["star_threshold"],
"installed": _installed(v["id"]),
"bundled": _bundled(v["id"]),
"has_pack": _bundled(v["id"]) or _pack_published(v.get("pack")),
"has_pack": _bundled(v["id"]) or bool(v.get("pack")),
"download": dl,
})
return {
@@ -941,7 +934,7 @@ def setup(app, context):
if venue is None:
raise HTTPException(404, "Unknown venue.")
pack = venue.get("pack")
if not _pack_published(pack):
if not pack:
raise HTTPException(404, "No pack published for this venue yet.")
stars_total, _, _ = _stars()
if stars_total < venue["star_threshold"]:
+2 -10
View File
@@ -17,22 +17,14 @@
"name": "Velvet Room",
"description": "A proper club stage. People actually came to hear you.",
"star_threshold": 50,
"pack": {
"url": "https://github.com/got-feedBack/feedBack/releases/download/venue-club-v1/club-pack-v1.zip",
"sha256": "0000000000000000000000000000000000000000000000000000000000000000",
"bytes": 0
}
"pack": null
},
{
"id": "arena",
"name": "Feedback Arena",
"description": "Ten thousand seats. Try not to think about it.",
"star_threshold": 150,
"pack": {
"url": "https://github.com/got-feedBack/feedBack/releases/download/venue-arena-v1/arena-pack-v1.zip",
"sha256": "8898c7912c84123559ecfd8b0afd3be19da9a0a4b04aa9dbb5aa1e7f4e3e1669",
"bytes": 351284599
}
"pack": null
}
]
}
+1 -7
View File
@@ -49,7 +49,7 @@ import demo_mode
import scan
import tailwind_rebuild
# Extracted route modules. They import `appstate`, never `server` — one-way graph.
from routers import audio_effects, artist_aliases, loops, playlists, ws_highway, ws_sync, 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
import enrichment
from routers import art as art_router
@@ -1618,12 +1618,6 @@ app.include_router(media_router.router)
app.include_router(ws_highway.router)
# ── Session-sync relay WebSocket (feedBack#1030) ─────────────────────────────
# Dumb JSON fan-out rooms for cross-device followers (splitscreen LAN mode).
# Implementation in lib/routers/ws_sync.py.
app.include_router(ws_sync.router)
# ── Audio serving ─────────────────────────────────────────────────────────────
+5 -78
View File
@@ -1,4 +1,4 @@
// Count-in — the one-bar click before playback, plus the song-credits overlay that
// Count-in — the 1-2-3-4 click before playback, plus the song-credits overlay that
// shares its lifecycle and timers.
//
// The third slice out of app.js's strongly-connected core, and the first that had to
@@ -40,75 +40,6 @@ export function playClick(high = false) {
osc.stop(_audioCtx.currentTime + 0.08);
}
// ── How many clicks lead into `startT` ──────────────────────────────────
// One bar, derived from the song_timeline beats: `window.highway.getBeats()`
// is the only meter data the frontend holds (the `time_signatures` map is
// streamed to plugins, not stored here). Beats carry `measure >= 0` on
// downbeats, so the gap between consecutive downbeats IS the bar length —
// which is why a 3/4 song no longer gets four clicks.
//
// A first bar shorter than that is a pickup (anacrusis), and the count is
// shortened by its length so the music enters on its real beat: a 1-beat
// pickup in 4/4 counts "1 2 3" and the pickup lands on 4. Counting a full
// four there puts the pickup where the downbeat belongs, and the player comes
// in a beat late for the whole song.
export function countInBeats(startT) {
const DEFAULT = 4; // pre-chart, synthetic highway (minigames), or no beats
let beats = null;
try {
if (window.highway && typeof window.highway.getBeats === 'function') {
beats = window.highway.getBeats();
}
} catch (_) { /* fall through to the default */ }
if (!Array.isArray(beats) || beats.length < 2) return DEFAULT;
const downbeats = [];
for (let i = 0; i < beats.length; i++) {
if (beats[i] && beats[i].measure >= 0) downbeats.push(i);
}
if (downbeats.length < 2) return DEFAULT;
// Bar length = the most common gap between downbeats. The mode rather than
// the first gap: it ignores a short pickup bar and a short final bar, and
// survives an isolated meter change mid-song. The beats trailing the last
// downbeat count as a candidate too — otherwise a song of pickup + one bar
// offers only the pickup's own gap and the count collapses to it.
const gapCounts = new Map();
const addGap = (gap) => gapCounts.set(gap, (gapCounts.get(gap) || 0) + 1);
for (let k = 1; k < downbeats.length; k++) {
addGap(downbeats[k] - downbeats[k - 1]);
}
addGap(beats.length - downbeats[downbeats.length - 1]);
let barLen = DEFAULT;
let bestCount = 0;
for (const [gap, n] of gapCounts) {
// Tie → the longer bar: a pickup's short gap must not outvote the
// real meter when the song is too short to repeat it.
if (n > bestCount || (n === bestCount && gap > barLen)) {
barLen = gap;
bestCount = n;
}
}
// The beat playback resumes on. The 50 ms tolerance matches the seek
// precision the loop-wrap path already assumes.
const startIdx = beats.findIndex(b => b && b.time >= startT - 0.05);
if (startIdx === -1) return barLen; // past the last beat
if (!(beats[startIdx].measure >= 0)) return barLen; // resuming mid-bar
const nextDownbeat = downbeats.find(d => d > startIdx);
if (nextDownbeat === undefined) return barLen; // the last downbeat
const thisBar = nextDownbeat - startIdx;
if (thisBar <= 0) return barLen;
// Only the song's FIRST bar can be a pickup. A short bar anywhere else is
// a meter change (or a truncated final bar), and counting it as a pickup
// would leave almost no count-in at all — so elsewhere we simply count
// that bar's own length, which is also what a mid-song meter change wants.
if (startIdx === downbeats[0] && thisBar < barLen) return barLen - thisBar;
return thisBar;
}
let _countingIn = false;
let _countOverlay = null;
// Generation token so teardown can cancel an in-progress count-in. Each
@@ -342,15 +273,12 @@ export async function startCountIn(opts = {}) {
function beginCount() {
const bpm = window.highway.getBPM(loopA);
const beatInterval = 60 / bpm;
// One bar of the meter at loop A (a short bar there is counted short,
// same as the song-start pickup).
const clicks = countInBeats(loopA);
let count = 0;
function tick() {
if (gen !== _countInGen) return; // teardown mid-count
count++;
if (count > clicks) {
if (count > 4) {
hideCountOverlay();
_countingIn = false;
if (window._juceMode) {
@@ -392,7 +320,7 @@ export async function startCountIn(opts = {}) {
}
}
// Start-of-song count-in: a one-bar click before playback begins, gated by the
// Start-of-song count-in: a 4-beat click before playback begins, gated by the
// "Countdown before song" setting (Gameplay tab). Mirrors the loop count-in's
// overlay + click + gen-token cancellation, but counts from the song's current
// position (0 at song start) with no loop A/B rewind. startCountIn() is loop-
@@ -412,15 +340,14 @@ export async function startSongCountIn() {
if (gen !== _countInGen) return; // teardown during pause
const startT = S.lastAudioTime || 0;
let bpm = window.highway.getBPM(startT);
// Pre-chart / malformed-tempo fallback: 120 BPM (500 ms per beat).
// Pre-chart / malformed-tempo fallback: 4 beats at 120 BPM (500 ms each).
if (!Number.isFinite(bpm) || bpm <= 0) bpm = 120;
const beatInterval = 60 / bpm;
const clicks = countInBeats(startT);
let count = 0;
function tick() {
if (gen !== _countInGen) return; // teardown mid-count
count++;
if (count > clicks) {
if (count > 4) {
hideCountOverlay();
_countingIn = false;
// Hand off to the normal play path — togglePlay() flips isPlaying,
-189
View File
@@ -1,189 +0,0 @@
// Verify `countInBeats(startT)` in static/js/count-in.js sizes the count-in
// to the song's own bar rather than a hardcoded four clicks.
//
// Two behaviours are under test:
// 1. Meter — a 3/4 song gets three clicks, not four.
// 2. Pickup (anacrusis) — a first bar shorter than the meter shortens the
// count so the pickup enters on its real beat (1-beat pickup in 4/4 →
// "1 2 3", music on 4). A full four there puts the pickup where the
// downbeat belongs and the player comes in a beat late all song.
//
// The meter is read from the song_timeline beats (`window.highway.getBeats()`,
// `measure >= 0` on downbeats) because that is the only meter data the
// frontend holds — the `time_signatures` map is streamed to plugins, not
// stored here.
//
// Same extraction approach as loop_restart.test.js: pull the function source
// out of the module and evaluate it in a vm sandbox with a stubbed highway,
// rather than loading the ESM module and its DOM-coupled imports.
const { test } = require('node:test');
const assert = require('node:assert/strict');
const fs = require('node:fs');
const path = require('node:path');
const vm = require('node:vm');
const COUNT_IN_JS = path.join(__dirname, '..', '..', 'static', 'js', 'count-in.js');
// Brace-match the function body out of the source. Brittle by design:
// a rename fails loudly here rather than silently skipping coverage.
function extractFunction(src, signature) {
const start = src.indexOf(signature);
if (start === -1) throw new Error(`extractFunction: '${signature}' not found`);
const openBrace = src.indexOf('{', start + signature.length);
let depth = 1;
let i = openBrace + 1;
while (i < src.length && depth > 0) {
const ch = src[i];
if (ch === '{') depth++;
else if (ch === '}') depth--;
i++;
}
if (depth !== 0) throw new Error(`extractFunction: unbalanced braces after '${signature}'`);
return src.slice(start, i);
}
const src = fs.readFileSync(COUNT_IN_JS, 'utf8');
// Drop the `export` keyword so the body evaluates as a plain declaration.
const fnSrc = extractFunction(src, 'export function countInBeats')
.replace(/^export\s+/, '');
// `beats` is the song_timeline shape: {time, measure}, measure >= 0 only on
// downbeats. `getBeats` may also be absent entirely (pre-chart / minigame).
function load(beats) {
const sandbox = {
window: beats === undefined
? { highway: {} }
: { highway: { getBeats: () => beats } },
};
vm.createContext(sandbox);
vm.runInContext(`${fnSrc}; globalThis.__fn = countInBeats;`, sandbox);
return sandbox.__fn;
}
// Build a beats array: `bars` full bars of `beatsPerBar`, optionally preceded
// by a pickup of `pickup` beats. One beat per 0.5 s throughout.
function makeBeats({ beatsPerBar = 4, bars = 4, pickup = 0 } = {}) {
const out = [];
let t = 0;
let measure = 0;
if (pickup > 0) {
for (let i = 0; i < pickup; i++) {
out.push({ time: t, measure: i === 0 ? measure : -1 });
t += 0.5;
}
measure++;
}
for (let b = 0; b < bars; b++) {
for (let i = 0; i < beatsPerBar; i++) {
out.push({ time: t, measure: i === 0 ? measure : -1 });
t += 0.5;
}
measure++;
}
return out;
}
// ── Meter ────────────────────────────────────────────────────────────────
test('countInBeats counts a full bar in 4/4', () => {
const countInBeats = load(makeBeats({ beatsPerBar: 4 }));
assert.equal(countInBeats(0), 4);
});
test('countInBeats counts three in 3/4 (was hardcoded four)', () => {
const countInBeats = load(makeBeats({ beatsPerBar: 3 }));
assert.equal(countInBeats(0), 3);
});
test('countInBeats counts six in 6/8', () => {
const countInBeats = load(makeBeats({ beatsPerBar: 6 }));
assert.equal(countInBeats(0), 6);
});
// ── Pickup (anacrusis) ───────────────────────────────────────────────────
test('countInBeats shortens the count by a 1-beat pickup in 4/4', () => {
const countInBeats = load(makeBeats({ beatsPerBar: 4, pickup: 1 }));
assert.equal(countInBeats(0), 3, 'counts 1-2-3 so the pickup lands on 4');
});
test('countInBeats shortens the count by a 2-beat pickup in 4/4', () => {
const countInBeats = load(makeBeats({ beatsPerBar: 4, pickup: 2 }));
assert.equal(countInBeats(0), 2);
});
test('countInBeats handles a pickup in 3/4', () => {
const countInBeats = load(makeBeats({ beatsPerBar: 3, pickup: 1 }));
assert.equal(countInBeats(0), 2);
});
test('countInBeats finds the meter when the song is only a pickup plus one bar', () => {
// Gap counts tie (one 1-beat gap, one 4-beat gap) — the longer bar is the
// meter, so this must be 3 rather than 0.
const countInBeats = load(makeBeats({ beatsPerBar: 4, bars: 1, pickup: 1 }));
assert.equal(countInBeats(0), 3);
});
// ── Resuming somewhere other than the song top ───────────────────────────
test('countInBeats counts a full bar at a mid-song downbeat, pickup notwithstanding', () => {
const beats = makeBeats({ beatsPerBar: 4, pickup: 1 });
const countInBeats = load(beats);
// Index 5 is the downbeat of the second full bar (1 pickup + 4 beats).
assert.equal(beats[5].measure >= 0, true, 'fixture sanity: index 5 is a downbeat');
assert.equal(countInBeats(beats[5].time), 4);
});
test('countInBeats counts a mid-song meter change by that bar, not as a pickup', () => {
// 4/4 throughout, except one 3-beat bar at index 8. Treating a short bar
// anywhere but the song's first as a pickup would count a single click.
const beats = [];
let t = 0;
const push = (n, measure) => {
for (let i = 0; i < n; i++) { beats.push({ time: t, measure: i === 0 ? measure : -1 }); t += 0.5; }
};
push(4, 0); push(4, 1); push(3, 2); push(4, 3); push(4, 4);
const countInBeats = load(beats);
assert.equal(beats[8].measure, 2, 'fixture sanity: index 8 opens the 3-beat bar');
assert.equal(countInBeats(beats[8].time), 3, 'counts the short bar itself');
assert.equal(countInBeats(0), 4, 'the 4/4 opening is unaffected');
});
test('countInBeats counts a full bar when resuming mid-bar', () => {
const beats = makeBeats({ beatsPerBar: 4 });
const countInBeats = load(beats);
assert.equal(countInBeats(beats[2].time), 4); // third beat of bar 1
});
test('countInBeats tolerates a start time slightly past the beat (seek slop)', () => {
const countInBeats = load(makeBeats({ beatsPerBar: 4, pickup: 1 }));
assert.equal(countInBeats(0.02), 3);
});
// ── Fallbacks ────────────────────────────────────────────────────────────
test('countInBeats falls back to four without a beats array', () => {
assert.equal(load(undefined)(0), 4, 'no getBeats (pre-chart / minigame)');
assert.equal(load([])(0), 4, 'empty beats');
assert.equal(load(null)(0), 4, 'null beats');
});
test('countInBeats falls back to four when beats carry no downbeat labels', () => {
const beats = [0, 0.5, 1.0, 1.5, 2.0].map(time => ({ time, measure: -1 }));
assert.equal(load(beats)(0), 4);
});
test('countInBeats falls back to four with only one downbeat', () => {
const beats = [
{ time: 0, measure: 0 },
{ time: 0.5, measure: -1 },
{ time: 1.0, measure: -1 },
];
assert.equal(load(beats)(0), 4);
});
test('countInBeats counts a full bar past the last beat', () => {
const beats = makeBeats({ beatsPerBar: 3 });
assert.equal(load(beats)(9999), 3);
});
-4
View File
@@ -88,10 +88,6 @@ function buildSandbox() {
playClick: () => {},
showCountOverlay: () => {},
hideCountOverlay: () => {},
// beginCount sizes the count to the bar at loop A; the wrap-path
// assertions below don't depend on how many clicks it decides on.
// Covered directly in count_in_beats.test.js.
countInBeats: () => 4,
// Stubbed DOM access. Anything querying for a button just gets a
// permissive object that ignores writes.
+2 -61
View File
@@ -83,25 +83,10 @@ def test_download_without_published_pack_404s(client):
def test_download_locked_venue_403s(client, monkeypatch):
club = career_routes._venue("club")
monkeypatch.setitem(club, "pack", {"url": "http://x/pack.zip", "sha256": "0" * 64, "bytes": 123})
monkeypatch.setitem(club, "pack", {"url": "http://x/pack.zip", "sha256": "0" * 64})
assert client.post("/api/plugins/career/packs/club/download").status_code == 403
def test_placeholder_pack_is_not_offered_until_published(client, monkeypatch):
# A committed manifest carries a 0-byte placeholder until its release is
# published. Such a pack must not be offered (has_pack False) and its
# download must 404 — else the UI shows a button that can only fail.
monkeypatch.setattr(career_routes, "_bundled", lambda vid: False)
club = career_routes._venue("club")
monkeypatch.setitem(club, "pack",
{"url": "http://x/c.zip", "sha256": "0" * 64, "bytes": 0})
by_id = {v["id"]: v for v in client.get("/api/plugins/career/state").json()["venues"]}
assert by_id["club"]["has_pack"] is False # placeholder → not offered
assert by_id["arena"]["has_pack"] is True # arena ships real bytes
# Even forced, an unpublished pack won't start a download.
assert client.post("/api/plugins/career/packs/club/download").status_code == 404
def test_bundled_bar_pack_is_installed_and_served(client):
state = client.get("/api/plugins/career/state").json()
bar = {v["id"]: v for v in state["venues"]}["bar"]
@@ -182,53 +167,9 @@ def test_download_worker_end_to_end(client, tmp_path):
assert "sha256" in bad["error"]
def test_content_packs_build_roundtrips_through_download(client, tmp_path):
# tools/content_packs.py must produce a zip the real career worker accepts:
# build_pack → manifest_entry → _download_pack → installed.
from tools import content_packs
src = tmp_path / "bar"
src.mkdir()
for s in career_routes.REQUIRED_LOOPS:
(src / f"{s}.mp4").write_bytes(b"fake-" + s.encode())
(src / "cheer.mp4").write_bytes(b"fake-cheer")
(src / "manifest.json").write_text(json.dumps({
"venue": "bar", "version": 1,
"loops": {s: f"{s}.mp4" for s in career_routes.REQUIRED_LOOPS},
"stingers": {"cheer": "cheer.mp4"},
}))
out_dir = tmp_path / "packs"
zip_path = out_dir / content_packs.pack_asset("bar", 1)
info = content_packs.build_pack(src, zip_path)
entry = content_packs.manifest_entry(zip_path, zip_path.resolve().as_uri())
assert entry["sha256"] == info["sha256"] and entry["bytes"] == info["bytes"]
progress = {"status": "running", "bytes_done": 0, "bytes_total": 0, "error": None}
career_routes._download_pack("bar", entry, progress)
assert progress["status"] == "done", progress["error"]
assert career_routes._installed("bar")
def test_content_packs_rejects_files_the_downloader_would_refuse(tmp_path):
# A stray file (e.g. macOS .DS_Store) must fail the build, not get published
# and then break every client's download at _validate_pack_dir.
from tools import content_packs
src = tmp_path / "bar"
src.mkdir()
(src / "bored.mp4").write_bytes(b"fake")
(src / ".DS_Store").write_bytes(b"junk")
try:
content_packs.build_pack(src, tmp_path / "bar-pack-v1.zip")
except ValueError as e:
assert "downloader will reject" in str(e)
else:
raise AssertionError("build_pack accepted a .DS_Store the downloader rejects")
def test_double_download_409s(client, monkeypatch):
bar = career_routes._venue("bar")
monkeypatch.setitem(bar, "pack", {"url": "http://x/pack.zip", "sha256": "0" * 64, "bytes": 123})
monkeypatch.setitem(bar, "pack", {"url": "http://x/pack.zip", "sha256": "0" * 64})
# Pretend one is already running.
career_routes._state["downloads"]["bar"] = {"status": "running"}
assert client.post("/api/plugins/career/packs/bar/download").status_code == 409
-91
View File
@@ -1,91 +0,0 @@
"""VST-pack slicing (tools/content_packs.build_vst_pack): each platform pack
keeps only its own binaries + the shared bundle files, drops the rest, and is
reproducible."""
import zipfile
from pathlib import Path
from tools import content_packs
def _fake_vst_tree(root: Path):
# One fat .vst3 with all three platform binaries + shared files, plus a
# src/ build tree that must never ship.
c = root / "amps" / "Foo.vst3" / "Contents"
(c / "MacOS").mkdir(parents=True)
(c / "x86_64-win").mkdir(parents=True)
(c / "x86_64-linux").mkdir(parents=True)
(c / "Resources").mkdir(parents=True)
(c / "MacOS" / "Foo").write_bytes(b"mac-binary")
(c / "x86_64-win" / "Foo.vst3").write_bytes(b"win-binary")
(c / "x86_64-linux" / "Foo.so").write_bytes(b"linux-binary")
(c / "Info.plist").write_bytes(b"<plist/>")
(c / "Resources" / "moduleinfo.json").write_bytes(b"{}")
(root / "src" / "build").mkdir(parents=True)
(root / "src" / "build" / "junk.o").write_bytes(b"objfile")
def _names(zip_path):
with zipfile.ZipFile(zip_path) as zf:
return set(zf.namelist())
def test_slice_keeps_target_platform_and_shared_drops_foreign(tmp_path):
root = tmp_path / "vst"
_fake_vst_tree(root)
content_packs.build_vst_pack(root, tmp_path / "mac.zip", "mac")
names = _names(tmp_path / "mac.zip")
base = "amps/Foo.vst3/Contents"
assert f"{base}/MacOS/Foo" in names # target binary kept
assert f"{base}/Info.plist" in names # shared kept
assert f"{base}/Resources/moduleinfo.json" in names # shared kept
assert f"{base}/x86_64-win/Foo.vst3" not in names # foreign dropped
assert f"{base}/x86_64-linux/Foo.so" not in names # foreign dropped
assert not any(n.startswith("src/") for n in names) # build trees never ship
def test_each_platform_gets_its_own_binary(tmp_path):
root = tmp_path / "vst"
_fake_vst_tree(root)
wanted = {"mac": "MacOS/Foo", "win": "x86_64-win/Foo.vst3", "linux": "x86_64-linux/Foo.so"}
for plat, rel in wanted.items():
content_packs.build_vst_pack(root, tmp_path / f"{plat}.zip", plat)
names = _names(tmp_path / f"{plat}.zip")
assert f"amps/Foo.vst3/Contents/{rel}" in names
others = [v for k, v in wanted.items() if k != plat]
for o in others:
assert f"amps/Foo.vst3/Contents/{o}" not in names
def test_slice_is_reproducible(tmp_path):
root = tmp_path / "vst"
_fake_vst_tree(root)
a = content_packs.build_vst_pack(root, tmp_path / "a.zip", "linux")
b = content_packs.build_vst_pack(root, tmp_path / "b.zip", "linux")
assert a == b and a["sha256"]
def test_slice_pins_create_system_for_cross_runner_reproducibility(tmp_path, monkeypatch):
# ZipInfo defaults create_system from the host OS (0 on Windows, 3 on Unix),
# and it lands in the central directory — so without an explicit pin the same
# tree hashes differently on a Windows runner, breaking the precomputable-hash
# guarantee exactly where it matters (native .vst3 are built on Windows). A
# same-machine reproducibility test can't catch that; simulate win32 and
# assert the pin forces 3 regardless.
monkeypatch.setattr(zipfile.sys, "platform", "win32")
root = tmp_path / "vst"
_fake_vst_tree(root)
content_packs.build_vst_pack(root, tmp_path / "w.zip", "linux")
with zipfile.ZipFile(tmp_path / "w.zip") as zf:
assert all(i.create_system == 3 for i in zf.infolist())
def test_unknown_platform_rejected(tmp_path):
root = tmp_path / "vst"
_fake_vst_tree(root)
try:
content_packs.build_vst_pack(root, tmp_path / "x.zip", "bsd")
except ValueError as e:
assert "unknown platform" in str(e)
else:
raise AssertionError("build_vst_pack accepted an unknown platform")
+1 -35
View File
@@ -19,7 +19,6 @@ a per-arrangement `drum_tab` file pointer and NO note `file`. The loader:
from __future__ import annotations
import json
import logging
from pathlib import Path
import yaml
@@ -102,28 +101,6 @@ def test_pointer_entries_never_become_fretted_arrangements(tmp_path: Path):
assert loaded.arrangement_ids == ["lead"]
def test_drums_typed_entry_with_note_file_never_frets(tmp_path: Path):
# A malformed entry: type:drums but ALSO carrying a note `file`. Keying the
# skip on file absence would let it through as a fretted, selectable,
# gradeable Arrangement (spec §5.2/§7.5 MUST-NOT). Routing on `type` first
# drops it instead — it never reaches song.arrangements.
bogus = {
"name": "Bogus", "tuning": [0, 0, 0, 0, 0, 0], "capo": 0,
"notes": [], "chords": [], "anchors": [], "handshapes": [],
"templates": [], "beats": [], "sections": [],
}
pak = _write_pak(tmp_path, {
"arrangements": [
{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"},
{"id": "bad", "name": "Bogus", "type": "drums",
"file": "arrangements/bogus.json"},
],
}, {"arrangements/bogus.json": bogus})
loaded = _load(pak, tmp_path)
assert [a.name for a in loaded.song.arrangements] == ["Lead"]
assert loaded.arrangement_ids == ["lead"]
# ── Parts resolution ─────────────────────────────────────────────────────────
def test_two_parts_resolve_primary_first_with_alias_identity(tmp_path: Path):
@@ -254,18 +231,7 @@ def test_drum_pointer_with_wrong_type_logs_warning(tmp_path: Path, caplog):
{"id": "typo", "type": "druns", "drum_tab": "drum_tab_typo.json"},
],
}, {"drum_tab_typo.json": _tab("Typo")})
# feedBack sets propagate=False, so pytest's root capture sees nothing from
# it — attach caplog's handler to the feedBack logger and pin WARNING
# regardless of ambient level (a sibling test can leak ERROR onto this tree).
lg = logging.getLogger("feedBack")
orig_level = lg.level
lg.addHandler(caplog.handler)
lg.setLevel(logging.WARNING)
try:
loaded = _load(pak, tmp_path)
finally:
lg.removeHandler(caplog.handler)
lg.setLevel(orig_level)
loaded = _load(pak, tmp_path)
assert loaded.drum_parts is None
assert "has drum_tab" in caplog.text and "type='druns'" in caplog.text
-207
View File
@@ -1,207 +0,0 @@
"""End-to-end test for the sloppak loader recognising a `rigs:` manifest key
(rigs.json the pack-level library of engine-agnostic rigs, spec §7.9) and
surfacing the payload on the LoadedSloppak.
The governing posture: rig objects pass through VERBATIM. This loader does not
select realizations or apply the `intent.gm` floor it only makes the library
addressable by `id`, which is what `tones.base_rig` / `tones.changes[].rig`
reference."""
from __future__ import annotations
import json
from pathlib import Path
import yaml
import sloppak as sloppak_mod
def _write_dir_sloppak(root: Path, manifest_extras: dict, rigs_payload) -> Path:
"""Minimal directory-form sloppak; writes rigs.json when a payload is given.
Unique filename per test (tmp_path leaf) so the module-level
resolve_source_dir cache isn't poisoned across tests."""
pak = root / f"{root.name}.sloppak"
pak.mkdir()
arr_dir = pak / "arrangements"
arr_dir.mkdir()
arr = {
"name": "Lead", "tuning": [0, 0, 0, 0, 0, 0], "capo": 0,
"notes": [], "chords": [], "anchors": [], "handshapes": [],
"templates": [], "beats": [], "sections": [],
}
(arr_dir / "lead.json").write_text(json.dumps(arr))
manifest = {
"title": "Test", "artist": "Tester", "album": "", "year": 2026,
"duration": 10.0,
"arrangements": [{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"}],
"stems": [{"id": "full", "file": "stems/full.ogg", "default": True}],
}
manifest.update(manifest_extras)
(pak / "manifest.yaml").write_text(yaml.safe_dump(manifest, sort_keys=False))
if rigs_payload is not None:
(pak / "rigs.json").write_text(json.dumps(rigs_payload))
return pak
def _load(pak_path: Path, tmp_path: Path):
dlc_root = pak_path.parent
cache = tmp_path / "cache"
cache.mkdir()
return sloppak_mod.load_song(pak_path.name, dlc_root, cache)
# ── Happy path ───────────────────────────────────────────────────────────────
def test_load_song_attaches_rigs_when_manifest_opts_in(tmp_path: Path):
"""A source rig (spec §7.9 1.18.0) survives the load intact — including the
`soundfont` realization and the `intent.gm` floor a consumer needs to voice
the part."""
payload = {
"version": 1,
"rigs": [
{
"id": "grand-piano",
"name": "Grand Piano",
"instrument": "keys",
"blocks": [
{
"role": "source",
"name": "Concert Grand",
"intent": {"kind": "instrument", "gm": {"program": 0}},
"realizations": [
{"engine": "soundfont", "format": "sf2",
"ref": "sounds/grand.sf2", "bank": 0, "program": 0},
],
},
],
},
],
}
pak = _write_dir_sloppak(tmp_path, {"rigs": "rigs.json"}, payload)
loaded = _load(pak, tmp_path)
assert loaded.rigs is not None
assert loaded.rigs["version"] == 1
assert loaded.rigs["rigs"] == payload["rigs"]
def test_load_song_rigs_absent_without_manifest_key(tmp_path: Path):
"""The file alone must not opt a pack in — the manifest is the opt-in
(spec §9.1, "manifest opt-in, file off to the side")."""
pak = _write_dir_sloppak(tmp_path, {}, {"version": 1, "rigs": []})
assert _load(pak, tmp_path).rigs is None
# ── Verbatim passthrough ─────────────────────────────────────────────────────
def test_load_song_preserves_unknown_rig_content(tmp_path: Path):
"""Unknown `role` / `engine` / `kind` values and `ext` namespaces MUST
survive (spec §7.9) core does not interpret rigs, so it must not prune
what a newer writer or a plugin put there."""
payload = {
"version": 2,
"rigs": [
{
"id": "future-rig",
"blocks": [
{"role": "quantum-flux", "intent": {"kind": "not-yet-invented"},
"realizations": [{"engine": "some-future-engine", "ref": "x.bin"}],
"ext": {"vendor.custom": {"anything": [1, 2, 3]}}},
],
"graph": {"nodes": ["input", "output"], "edges": [["input", "output"]]},
"ext": {"vendor.rig": "kept"},
},
],
}
pak = _write_dir_sloppak(tmp_path, {"rigs": "rigs.json"}, payload)
loaded = _load(pak, tmp_path)
assert loaded.rigs["version"] == 2
assert loaded.rigs["rigs"] == payload["rigs"]
# ── Addressability ───────────────────────────────────────────────────────────
def test_load_song_drops_unaddressable_rigs_and_normalizes_ids(tmp_path: Path):
"""A rig is reachable only by `id`, so entries without a usable one are
unreferenceable by construction. Ids are stripped to match the reference
side, which lib/tones.py strips before it reaches the wire."""
payload = {
"rigs": [
"not-a-dict",
{"name": "no id at all"},
{"id": "", "name": "blank id"},
{"id": " ", "name": "whitespace id"},
{"id": 7, "name": "non-string id"},
{"id": " padded-rig ", "name": "Padded"},
{"id": "plain-rig", "name": "Plain"},
],
}
pak = _write_dir_sloppak(tmp_path, {"rigs": "rigs.json"}, payload)
loaded = _load(pak, tmp_path)
assert [r["id"] for r in loaded.rigs["rigs"]] == ["padded-rig", "plain-rig"]
# Everything except the normalized id is untouched.
assert loaded.rigs["rigs"][0]["name"] == "Padded"
# `version` defaults when the file omits it.
assert loaded.rigs["version"] == 1
def test_load_song_first_rig_wins_on_duplicate_id(tmp_path: Path):
"""A duplicate id makes `tones.base_rig` ambiguous, which would surface as
the wrong sound rather than an error."""
payload = {
"rigs": [
{"id": "dupe", "name": "First"},
{"id": "dupe", "name": "Second"},
{"id": " dupe ", "name": "Third, padded into a collision"},
],
}
pak = _write_dir_sloppak(tmp_path, {"rigs": "rigs.json"}, payload)
loaded = _load(pak, tmp_path)
assert len(loaded.rigs["rigs"]) == 1
assert loaded.rigs["rigs"][0]["name"] == "First"
# ── Permissive posture (spec §7.9: never fail the pack) ──────────────────────
def test_load_song_survives_malformed_rigs(tmp_path: Path):
"""Malformed / missing / traversing rig libraries disable rigs, never the
pack the song itself must still load."""
cases = [
{"version": 1, "rigs": "not-a-list"}, # wrong `rigs` type
["top-level-not-a-dict"], # wrong document type
{"version": 1}, # no `rigs` key at all
]
for i, payload in enumerate(cases):
sub = tmp_path / f"case{i}"
sub.mkdir()
pak = _write_dir_sloppak(sub, {"rigs": "rigs.json"}, payload)
loaded = _load(pak, sub)
assert loaded.rigs is None, f"case {i} should disable rigs"
assert loaded.song is not None, f"case {i} must not fail the pack"
def test_load_song_survives_unparseable_rigs(tmp_path: Path):
pak = _write_dir_sloppak(tmp_path, {"rigs": "rigs.json"}, None)
(pak / "rigs.json").write_text("{ not json at all ")
loaded = _load(pak, tmp_path)
assert loaded.rigs is None
assert loaded.song is not None
def test_load_song_survives_missing_rigs_file(tmp_path: Path):
pak = _write_dir_sloppak(tmp_path, {"rigs": "rigs.json"}, None)
loaded = _load(pak, tmp_path)
assert loaded.rigs is None
assert loaded.song is not None
def test_load_song_rejects_traversing_rigs_path(tmp_path: Path):
"""A crafted manifest must not read outside the pack."""
(tmp_path / "outside.json").write_text(json.dumps({"rigs": [{"id": "leaked"}]}))
pak = _write_dir_sloppak(tmp_path, {"rigs": "../outside.json"}, None)
loaded = _load(pak, tmp_path)
assert loaded.rigs is None
assert loaded.song is not None
-230
View File
@@ -1,230 +0,0 @@
"""Loader coverage for the manifest-vs-in-JSON `tones` precedence cascade
(feedpak 1.18.0, spec §5.1 / §5.2).
Two rules, both about *which* sound binding wins, neither about interpreting it:
- A manifest arrangement entry's `tones` replaces the arrangement JSON's
`tones` **WHOLESALE** no field-level merge. A half-merged block (this
source's `base` with that source's `changes`) would be a sound nobody
authored, so the two never blend.
- Top-level `drum_tones` binds the song-level (primary) drum part and is the
fallback; a `type: drums` entry's own `tones` takes precedence, and a
Reader MUST NOT apply both to the same part.
"""
from __future__ import annotations
import json
from pathlib import Path
import yaml
import sloppak as sloppak_mod
IN_JSON_TONES = {
"base": "In-JSON Clean",
"base_rig": "injson-clean",
"changes": [{"t": 5.0, "name": "In-JSON Lead", "rig": "injson-lead"}],
}
ENTRY_TONES = {
"base": "Entry Grand",
"base_rig": "entry-grand",
"changes": [{"t": 9.0, "name": "Entry Rhodes", "rig": "entry-rhodes"}],
}
def _tab(name: str) -> dict:
return {
"version": 1,
"name": name,
"kit": [{"id": "kick", "name": "Kick"}],
"hits": [{"t": 1.0, "p": "kick", "v": 100}],
}
def _write_pak(root: Path, manifest_extras: dict, arr_tones: dict | None = None,
files: dict[str, dict] | None = None) -> Path:
"""Directory-form sloppak with one Lead arrangement, optionally carrying an
in-JSON `tones` block, plus any extra files."""
pak = root / f"{root.name}.sloppak"
pak.mkdir()
arr_dir = pak / "arrangements"
arr_dir.mkdir()
arr = {
"name": "Lead", "tuning": [0, 0, 0, 0, 0, 0], "capo": 0,
"notes": [], "chords": [], "anchors": [], "handshapes": [],
"templates": [], "beats": [], "sections": [],
}
if arr_tones is not None:
arr["tones"] = arr_tones
(arr_dir / "lead.json").write_text(json.dumps(arr))
manifest = {
"title": "Test", "artist": "Tester", "album": "", "year": 2026,
"duration": 10.0,
"arrangements": [{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"}],
"stems": [{"id": "full", "file": "stems/full.ogg", "default": True}],
}
manifest.update(manifest_extras)
(pak / "manifest.yaml").write_text(yaml.safe_dump(manifest, sort_keys=False))
for rel, payload in (files or {}).items():
(pak / rel).write_text(json.dumps(payload))
return pak
def _load(pak_path: Path, tmp_path: Path):
cache = tmp_path / "cache"
cache.mkdir()
return sloppak_mod.load_song(pak_path.name, pak_path.parent, cache)
# ── Arrangement entry vs in-JSON (§5.2) ──────────────────────────────────────
def test_entry_tones_replaces_in_json_wholesale(tmp_path: Path):
"""The entry object replaces the in-JSON one entirely — no key survives
from the loser, not even ones the winner doesn't define."""
entry_tones = {"base": "Entry Only"} # no base_rig, no changes
pak = _write_pak(
tmp_path,
{"arrangements": [{"id": "lead", "name": "Lead",
"file": "arrangements/lead.json",
"tones": entry_tones}]},
arr_tones=IN_JSON_TONES,
)
arr = _load(pak, tmp_path).song.arrangements[0]
assert arr.tones == entry_tones
# The in-JSON `base_rig` and `changes` must NOT have been merged in.
assert "base_rig" not in arr.tones
assert "changes" not in arr.tones
def test_in_json_tones_survive_when_entry_has_none(tmp_path: Path):
pak = _write_pak(tmp_path, {}, arr_tones=IN_JSON_TONES)
assert _load(pak, tmp_path).song.arrangements[0].tones == IN_JSON_TONES
def test_empty_entry_tones_is_absent_not_an_override(tmp_path: Path):
"""`{}` reads as "didn't specify", not "override to silence" — otherwise a
stray empty object silently unbinds the part's sound."""
pak = _write_pak(
tmp_path,
{"arrangements": [{"id": "lead", "name": "Lead",
"file": "arrangements/lead.json", "tones": {}}]},
arr_tones=IN_JSON_TONES,
)
assert _load(pak, tmp_path).song.arrangements[0].tones == IN_JSON_TONES
def test_malformed_entry_tones_is_ignored(tmp_path: Path):
"""A non-dict `tones` must not override, and must not crash the load."""
pak = _write_pak(
tmp_path,
{"arrangements": [{"id": "lead", "name": "Lead",
"file": "arrangements/lead.json",
"tones": ["not", "a", "dict"]}]},
arr_tones=IN_JSON_TONES,
)
assert _load(pak, tmp_path).song.arrangements[0].tones == IN_JSON_TONES
def test_entry_tones_binds_a_notation_only_arrangement(tmp_path: Path):
"""§5.2: entry `tones` is available whether or not the arrangement has a
`file` a keys part is a notation-only entry, and binding its sound is the
whole point of the 1.18.0 work."""
notation = {"version": 1, "measures": []}
pak = _write_pak(
tmp_path,
{"arrangements": [{"id": "keys", "name": "Keys",
"notation": "notation_keys.json",
"tones": ENTRY_TONES}]},
files={"notation_keys.json": notation},
)
arr = _load(pak, tmp_path).song.arrangements[0]
assert arr.name == "Keys"
assert arr.tones == ENTRY_TONES
# ── drum_tones vs entry tones (§5.1) ─────────────────────────────────────────
def test_drum_tones_binds_the_primary_part(tmp_path: Path):
pak = _write_pak(
tmp_path,
{"drum_tab": "drum_tab.json", "drum_tones": ENTRY_TONES},
files={"drum_tab.json": _tab("Drums")},
)
parts = _load(pak, tmp_path).drum_parts
assert len(parts) == 1
assert parts[0]["tones"] == ENTRY_TONES
def test_entry_tones_outrank_drum_tones_on_the_primary(tmp_path: Path):
"""An alias pointer entry naming the same file IS the primary, so its own
binding wins and `drum_tones` must not also be applied."""
alias_tones = {"base": "Alias Kit", "base_rig": "alias-kit"}
pak = _write_pak(
tmp_path,
{
"drum_tab": "drum_tab.json",
"drum_tones": ENTRY_TONES,
"arrangements": [
{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"},
{"id": "drums", "name": "Drums", "type": "drums",
"drum_tab": "drum_tab.json", "tones": alias_tones},
],
},
files={"drum_tab.json": _tab("Drums")},
)
parts = _load(pak, tmp_path).drum_parts
assert len(parts) == 1
assert parts[0]["tones"] == alias_tones
def test_drum_tones_does_not_leak_to_secondary_parts(tmp_path: Path):
"""`drum_tones` is the PRIMARY's fallback only. A second drummer with no
binding of its own gets None not the primary's kit."""
live_tones = {"base": "Live Kit", "base_rig": "live-kit"}
pak = _write_pak(
tmp_path,
{
"drum_tab": "drum_tab.json",
"drum_tones": ENTRY_TONES,
"arrangements": [
{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"},
{"id": "drums-live", "name": "Drums (Live)", "type": "drums",
"drum_tab": "drum_tab_live.json", "tones": live_tones},
{"id": "drums-prog", "name": "Drums (Prog)", "type": "drums",
"drum_tab": "drum_tab_prog.json"},
],
},
files={
"drum_tab.json": _tab("Drums"),
"drum_tab_live.json": _tab("Drums Live"),
"drum_tab_prog.json": _tab("Drums Prog"),
},
)
parts = {p["id"]: p for p in _load(pak, tmp_path).drum_parts}
assert parts["drums"]["tones"] == ENTRY_TONES # primary, from drum_tones
assert parts["drums-live"]["tones"] == live_tones # own entry
assert parts["drums-prog"]["tones"] is None # no binding, no leak
def test_drum_parts_carry_none_when_pack_binds_nothing(tmp_path: Path):
"""A pack with drums and no sound binding at all still loads, with the key
present and None consumers can read `part["tones"]` unconditionally."""
pak = _write_pak(
tmp_path,
{"drum_tab": "drum_tab.json"},
files={"drum_tab.json": _tab("Drums")},
)
parts = _load(pak, tmp_path).drum_parts
assert parts[0]["tones"] is None
def test_malformed_drum_tones_is_ignored(tmp_path: Path):
pak = _write_pak(
tmp_path,
{"drum_tab": "drum_tab.json", "drum_tones": "not-a-dict"},
files={"drum_tab.json": _tab("Drums")},
)
assert _load(pak, tmp_path).drum_parts[0]["tones"] is None
-79
View File
@@ -329,31 +329,6 @@ def test_note_pitch_midi_bass_uses_bass_base():
assert note_pitch_midi(bass, Note(time=0, string=0, fret=0)) == 28
def test_note_pitch_midi_authored_bass_type_uses_bass_base():
"""Editor PR #335: a bass authored via `type` on an arrangement whose NAME
doesn't say "bass". arrangement_string_count now returns 4 for it, so the
open-string base MUST also be the bass base (low E1 = 28), not the guitar
octave (40). Pre-fix note_pitch_midi keyed is_bass off the name only, so
this returned 40 (4 lanes on a guitar octave the exact inconsistency)."""
bass = Arrangement(
name="Low End", type="bass",
tuning=[0, 0, 0, 0],
notes=[Note(time=float(i), string=i, fret=0) for i in range(4)],
)
assert note_pitch_midi(bass, Note(time=0, string=0, fret=0)) == 28
def test_note_pitch_midi_path_bass_flag_uses_bass_base():
"""Same guarantee via the archive <arrangementProperties> pathBass flag on a
non-"bass"-named arrangement: bass base (28), not guitar (40)."""
bass = Arrangement(
name="Low End", path_bass=True,
tuning=[0, 0, 0, 0],
notes=[Note(time=float(i), string=i, fret=0) for i in range(4)],
)
assert note_pitch_midi(bass, Note(time=0, string=0, fret=0)) == 28
def test_note_pitch_midi_out_of_range_string_is_none():
arr = Arrangement(name="Lead", tuning=[0, 0, 0, 0, 0, 0])
assert note_pitch_midi(arr, Note(time=0, string=9, fret=0)) is None
@@ -1178,60 +1153,6 @@ def test_string_count_ignores_rs_padded_tuning_for_bass():
assert arrangement_string_count(arr) == 4
def test_string_count_4_for_path_bass_flag_without_bass_in_name():
# Archive/DLC bass whose manifest ArrangementName isn't "Bass" but whose
# <arrangementProperties> pathBass flag is set. Notes on 0..3, tuning
# padded to the arrangement-XML length of 6. Pre-fix, name_based forced
# 6 (name has no "bass"), so this returned 6 despite the authoritative
# instrument flag saying bass.
arr = Arrangement(
name="Low End",
path_bass=True,
tuning=[0, 0, 0, 0, 0, 0],
notes=[Note(time=float(i), string=i, fret=0) for i in range(4)],
)
assert arrangement_string_count(arr) == 4
def test_string_count_4_for_authored_bass_type_without_bass_in_name():
# Editor PR #335: an instrument `type` authored as bass on an arrangement
# whose NAME does not contain "bass" (the sloppak loader lifts the manifest
# `type` onto Arrangement.type). Notes on 0..3, tuning padded to 6.
# The editor lays out 4 lanes off the type; core must agree.
arr = Arrangement(
name="Low End",
type="bass",
tuning=[0, 0, 0, 0, 0, 0],
notes=[Note(time=float(i), string=i, fret=0) for i in range(4)],
)
assert arrangement_string_count(arr) == 4
def test_string_count_6_for_authored_guitar_type_no_regression():
# A non-bass authored type on a generic name still resolves to the
# canonical 6 — the type signal only pulls DOWN to 4 for bass.
arr = Arrangement(
name="Track 1",
type="guitar",
notes=[Note(time=float(i), string=i, fret=0) for i in range(5)],
)
assert arrangement_string_count(arr) == 6
def test_arrangement_is_bass_signal_safety():
# The manifest `type` is lifted onto arr.type verbatim; the helper must be
# safe against the messy shapes a hand-edited/loose source can produce.
from song import arrangement_is_bass
assert arrangement_is_bass(Arrangement(name="Low End", type="bass"))
assert arrangement_is_bass(Arrangement(name="Low End", type=" BASS ")) # ws/case
assert arrangement_is_bass(Arrangement(name="Low End", path_bass=True))
assert arrangement_is_bass(Arrangement(name="Slap Bass")) # legacy name
# Non-bass / absent signals stay False (back-compat: no bass signal → 6).
assert not arrangement_is_bass(Arrangement(name="Lead", type=""))
assert not arrangement_is_bass(Arrangement(name="Rhythm", type="guitar"))
assert not arrangement_is_bass(Arrangement(name="", type=""))
# ── compute_smart_names ───────────────────────────────────────────────────────
def _sarr(path_lead=False, path_rhythm=False, path_bass=False,
+8 -57
View File
@@ -6,17 +6,16 @@ from tones import sloppak_tone_changes
# ── sloppak_tone_changes (highway payload builder) ───────────────────────────
def test_sloppak_tone_changes_sorts_and_returns_base():
base, base_rig, changes = sloppak_tone_changes({
base, changes = sloppak_tone_changes({
"base": "Clean",
"changes": [{"t": 12.5, "name": "Drive"}, {"t": 3.0, "name": "Clean"}],
})
assert base == "Clean"
assert base_rig == ""
assert changes == [{"t": 3.0, "name": "Clean"}, {"t": 12.5, "name": "Drive"}]
def test_sloppak_tone_changes_skips_malformed_markers():
_, _, changes = sloppak_tone_changes({
_, changes = sloppak_tone_changes({
"changes": [
{"t": "nan", "name": "BadStr"},
{"t": float("inf"), "name": "Inf"},
@@ -30,66 +29,18 @@ def test_sloppak_tone_changes_skips_malformed_markers():
def test_sloppak_tone_changes_handles_none_and_bad_base():
assert sloppak_tone_changes(None) == ("", "", [])
base, base_rig, changes = sloppak_tone_changes({"base": 123, "changes": []})
assert base == "" and base_rig == "" and changes == []
assert sloppak_tone_changes(None) == ("", [])
base, changes = sloppak_tone_changes({"base": 123, "changes": []})
assert base == "" and changes == []
def test_sloppak_tone_changes_non_dict_input():
"""A truthy non-dict payload must not crash."""
assert sloppak_tone_changes(["not", "a", "dict"]) == ("", "", [])
assert sloppak_tone_changes("nope") == ("", "", [])
assert sloppak_tone_changes(["not", "a", "dict"]) == ("", [])
assert sloppak_tone_changes("nope") == ("", [])
def test_sloppak_tone_changes_non_list_changes():
"""A truthy non-list `changes` value must not raise on iteration."""
base, _, changes = sloppak_tone_changes({"base": "Clean", "changes": 1})
base, changes = sloppak_tone_changes({"base": "Clean", "changes": 1})
assert base == "Clean" and changes == []
# ── rig bindings (feedpak-spec 1.18.0 §6.9) ──────────────────────────────────
def test_sloppak_tone_changes_carries_rig_bindings():
"""`base_rig` and per-change `rig` reach the wire — the binding a chart
declares is what core must hand the consumer that voices the part."""
base, base_rig, changes = sloppak_tone_changes({
"base": "Clean Rhythm",
"base_rig": "clean-rhythm",
"changes": [
{"t": 12.5, "name": "Lead Drive", "rig": "lead-drive"},
{"t": 48.0, "name": "Clean Rhythm", "rig": "clean-rhythm"},
],
})
assert base == "Clean Rhythm"
assert base_rig == "clean-rhythm"
assert changes == [
{"t": 12.5, "name": "Lead Drive", "rig": "lead-drive"},
{"t": 48.0, "name": "Clean Rhythm", "rig": "clean-rhythm"},
]
def test_sloppak_tone_changes_omits_unusable_rig_ids():
"""A non-string or blank `rig` is dropped rather than forwarded, so a
consumer can treat presence of the key as "this change binds a rig"."""
_, base_rig, changes = sloppak_tone_changes({
"base_rig": " ",
"changes": [
{"t": 1.0, "name": "A", "rig": 7},
{"t": 2.0, "name": "B", "rig": ""},
{"t": 3.0, "name": "C", "rig": None},
{"t": 4.0, "name": "D", "rig": " padded-id "},
],
})
assert base_rig == ""
assert changes == [
{"t": 1.0, "name": "A"},
{"t": 2.0, "name": "B"},
{"t": 3.0, "name": "C"},
{"t": 4.0, "name": "D", "rig": "padded-id"},
]
def test_sloppak_tone_changes_non_string_base_rig():
"""A non-string `base_rig` must not crash or leak a non-id onto the wire."""
_, base_rig, _ = sloppak_tone_changes({"base": "Clean", "base_rig": 42})
assert base_rig == ""
-2
View File
@@ -60,14 +60,12 @@ def test_the_failure_is_actually_logged(registry, caplog):
# capture_logger() context manager for this, but it is not importable from here:
# pyproject pins pythonpath to [".", "lib"], so `tests` is not a package.)
lg = logging.getLogger("feedBack")
orig_level = lg.level
lg.addHandler(caplog.handler)
lg.setLevel(logging.ERROR)
try:
registry.get_merged()
finally:
lg.removeHandler(caplog.handler)
lg.setLevel(orig_level) # restore, or ERROR leaks onto the feedBack tree
assert any("bad-plugin" in r.getMessage() for r in caplog.records), (
"the raising provider was never named in the logs"
-234
View File
@@ -1,234 +0,0 @@
"""Tests for the session-sync relay WebSocket (/ws/sync/{session_id}).
Behavior tests run against a minimal FastAPI app carrying just the router
(fast no full-server import); one integration test imports the real server
to pin that the route is actually mounted there.
Covers the feedBack#1030 acceptance list: bidirectional fan-out, late join,
sender never echoed, room garbage collection, and the limit closes (invalid
session id, binary frames, frame size, room size, room count, rate cap)
including that one client tripping a limit doesn't disturb the others.
"""
from __future__ import annotations
import asyncio
import importlib
import sys
import time
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from starlette.websockets import WebSocketDisconnect
from routers import ws_sync
@pytest.fixture(autouse=True)
def _clean_rooms():
ws_sync._rooms.clear()
yield
ws_sync._rooms.clear()
@pytest.fixture()
def client():
app = FastAPI()
app.include_router(ws_sync.router)
with TestClient(app) as c:
yield c
def _expect_close(ws, code):
with pytest.raises(WebSocketDisconnect) as exc:
ws.receive_text()
assert exc.value.code == code
# ── Fan-out semantics ────────────────────────────────────────────────────────
def test_two_clients_relay_both_directions_and_no_echo(client):
with client.websocket_connect("/ws/sync/ROOM01") as a, \
client.websocket_connect("/ws/sync/ROOM01") as b:
a.send_text('{"type":"time","t":1.5}')
assert b.receive_text() == '{"type":"time","t":1.5}'
b.send_text('{"type":"hello"}')
# A's first inbound frame is B's hello — NOT an echo of its own send.
assert a.receive_text() == '{"type":"hello"}'
def test_late_joiner_receives_subsequent_frames(client):
with client.websocket_connect("/ws/sync/ROOM02") as a, \
client.websocket_connect("/ws/sync/ROOM02") as b:
a.send_text("f1")
assert b.receive_text() == "f1"
with client.websocket_connect("/ws/sync/ROOM02") as c:
a.send_text("f2")
assert b.receive_text() == "f2"
assert c.receive_text() == "f2"
def test_rooms_are_isolated(client):
with client.websocket_connect("/ws/sync/ROOMA1") as a, \
client.websocket_connect("/ws/sync/ROOMB1") as b, \
client.websocket_connect("/ws/sync/ROOMA1") as a2:
a.send_text("for-room-a")
assert a2.receive_text() == "for-room-a"
# B (other room) got nothing: prove it by relaying within B's room.
with client.websocket_connect("/ws/sync/ROOMB1") as b2:
b2.send_text("for-room-b")
assert b.receive_text() == "for-room-b"
def test_client_disconnect_does_not_disrupt_remaining(client):
with client.websocket_connect("/ws/sync/ROOM03") as a, \
client.websocket_connect("/ws/sync/ROOM03") as b:
with client.websocket_connect("/ws/sync/ROOM03") as c:
a.send_text("before")
assert b.receive_text() == "before"
assert c.receive_text() == "before"
# C is gone; relay between A and B continues.
a.send_text("after")
assert b.receive_text() == "after"
def test_room_garbage_collected_when_last_client_leaves(client):
with client.websocket_connect("/ws/sync/ROOM04") as a:
with client.websocket_connect("/ws/sync/ROOM04") as b:
a.send_text("x")
assert b.receive_text() == "x"
assert "ROOM04" in ws_sync._rooms
assert "ROOM04" not in ws_sync._rooms
assert ws_sync._rooms == {}
# ── Limit enforcement ────────────────────────────────────────────────────────
@pytest.mark.parametrize("bad_id", ["abc", "x" * 65, "has space", "bad$id", "nope!"])
def test_invalid_session_id_closed_with_policy_code(client, bad_id):
with client.websocket_connect(f"/ws/sync/{bad_id}") as ws:
_expect_close(ws, 1008)
assert ws_sync._rooms == {}
def test_binary_frame_closes_with_unsupported_data(client):
with client.websocket_connect("/ws/sync/ROOM05") as ws:
ws.send_bytes(b"\x00\x01")
_expect_close(ws, 1003)
def test_oversized_frame_closes_sender_only(client):
with client.websocket_connect("/ws/sync/ROOM06") as a, \
client.websocket_connect("/ws/sync/ROOM06") as b, \
client.websocket_connect("/ws/sync/ROOM06") as c:
a.send_text("x" * (ws_sync.MAX_FRAME_BYTES + 1))
_expect_close(a, 1009)
# The room carries on without A.
b.send_text("still-alive")
assert c.receive_text() == "still-alive"
def test_room_client_cap(client, monkeypatch):
monkeypatch.setattr(ws_sync, "MAX_CLIENTS_PER_ROOM", 2)
with client.websocket_connect("/ws/sync/ROOM07") as a, \
client.websocket_connect("/ws/sync/ROOM07") as b, \
client.websocket_connect("/ws/sync/ROOM07") as c:
_expect_close(c, 1013)
a.send_text("two-is-fine")
assert b.receive_text() == "two-is-fine"
def test_total_room_cap(client, monkeypatch):
monkeypatch.setattr(ws_sync, "MAX_ROOMS", 1)
with client.websocket_connect("/ws/sync/ROOM08"):
with client.websocket_connect("/ws/sync/ROOM09") as overflow:
_expect_close(overflow, 1013)
# Joining the EXISTING room is still fine at the room cap.
with client.websocket_connect("/ws/sync/ROOM08"):
pass
def test_rate_cap_closes_flooding_sender(client, monkeypatch):
monkeypatch.setattr(ws_sync, "RATE_BURST", 3.0)
monkeypatch.setattr(ws_sync, "RATE_MSGS_PER_SEC", 0.0)
with client.websocket_connect("/ws/sync/ROOM10") as a, \
client.websocket_connect("/ws/sync/ROOM10") as b:
for i in range(3):
a.send_text(f"burst-{i}")
for i in range(3):
assert b.receive_text() == f"burst-{i}"
a.send_text("one-too-many")
_expect_close(a, 1008)
# The over-limit frame was dropped, not relayed, and B lives on.
with client.websocket_connect("/ws/sync/ROOM10") as c:
c.send_text("fresh-socket")
assert b.receive_text() == "fresh-socket"
class _StalledPeer:
"""A fake room member whose send never completes (peer stopped draining)."""
async def send_text(self, text):
await asyncio.Event().wait()
def test_stalled_peer_is_evicted_and_healthy_peers_still_receive(client, monkeypatch):
monkeypatch.setattr(ws_sync, "SEND_TIMEOUT_SECONDS", 0.2)
with client.websocket_connect("/ws/sync/ROOM11") as a, \
client.websocket_connect("/ws/sync/ROOM11") as b:
# Wait for both handlers to have registered in the room, then inject
# the stalled peer directly (a real stalled TCP peer isn't
# constructible under TestClient).
deadline = time.monotonic() + 2.0
while len(ws_sync._rooms.get("ROOM11", {})) < 2:
assert time.monotonic() < deadline, "room never filled"
time.sleep(0.01)
stalled = _StalledPeer()
ws_sync._rooms["ROOM11"][stalled] = asyncio.Lock()
# Healthy delivery is not blocked behind the stalled peer, and by the
# time a second frame has round-tripped, the first fan-out's timeout
# has fired and evicted it.
a.send_text("f1")
assert b.receive_text() == "f1"
a.send_text("f2")
assert b.receive_text() == "f2"
assert stalled not in ws_sync._rooms["ROOM11"]
def test_main_run_caps_uvicorn_ws_max_size():
"""main.py must bound inbound WS frames at the transport (uvicorn defaults
to 16 MB, which would let a client materialize frames far past the relay's
16 KB application cap before the handler ever sees them)."""
import unittest.mock
import main
with (
unittest.mock.patch("logging_setup.configure_logging"),
unittest.mock.patch("uvicorn.run") as mock_run,
):
main.run()
kwargs = mock_run.call_args.kwargs
assert kwargs.get("ws_max_size") == 64 * 1024
assert kwargs["ws_max_size"] >= ws_sync.MAX_FRAME_BYTES
# ── Real-app integration ─────────────────────────────────────────────────────
def test_route_mounted_on_real_server(tmp_path, monkeypatch):
monkeypatch.setenv("CONFIG_DIR", str(tmp_path / "config"))
monkeypatch.setenv("DLC_DIR", str(tmp_path / "dlc"))
monkeypatch.setenv("FEEDBACK_SYNC_STARTUP", "1")
sys.modules.pop("server", None)
server = importlib.import_module("server")
monkeypatch.setattr(server, "load_plugins", lambda *a, **kw: None)
monkeypatch.setattr(server, "startup_scan", lambda: None)
with TestClient(server.app) as client:
with client.websocket_connect("/ws/sync/REALAPP") as a, \
client.websocket_connect("/ws/sync/REALAPP") as b:
a.send_text('{"type":"time","t":0}')
assert b.receive_text() == '{"type":"time","t":0}'
-294
View File
@@ -1,294 +0,0 @@
#!/usr/bin/env python3
"""Build & publish opt-in content packs (career venue media, rig VST slices).
Flat-zips a pack directory, sha256s it, and emits the ``{url, sha256, bytes}``
block that the career and rig_builder download paths consume
(``plugins/career/routes.py`` ``_download_pack``). Two modes:
--local <dir> write zips + a file:// manifest (dev/CI/tests; no network)
--publish create/upload each pack's per-pack release; emit release URLs
The zip is flat (files at the archive root) to satisfy career's zip-slip guard
(``PACK_FILENAME_RE``) and ``_validate_pack_dir``. This module is the reusable
core the content-packs CI workflow calls, so building packs is automation
never a person's manual job.
Run ``python tools/content_packs.py --selfcheck`` for the built-in round-trip.
"""
import argparse
import hashlib
import json
import re
import subprocess
import sys
import zipfile
from pathlib import Path
REPO = "got-feedBack/feedBack" # where the content-packs release lives (public)
# Must mirror career's download-time whitelist (plugins/career/routes.py
# PACK_FILENAME_RE). If the builder packs a name the downloader rejects (e.g. a
# stray .DS_Store), the published pack fails _validate_pack_dir for every client.
PACK_FILENAME_RE = re.compile(r"^[a-z0-9_-]{1,64}\.(mp4|webm|mp3|json)$")
def build_pack(src_dir: Path, out_zip: Path) -> dict:
"""Flat-zip every file directly under src_dir; return {sha256, bytes}.
Only regular files at the top level are included (venue packs are flat).
Subdirectories are skipped a nested tree would trip career's zip-slip
guard on download anyway.
The build is REPRODUCIBLE: identical file contents always yield a
byte-identical zip (fixed name order, fixed mtime, fixed permissions,
ZIP_STORED). So a sha256 computed on any machine matches the zip the CI
workflow or another contributor produces anyone can precompute the
manifest values without having to be the one who uploads the asset.
"""
files = sorted((p for p in src_dir.iterdir() if p.is_file()),
key=lambda p: p.name)
if not files:
raise ValueError(f"no files to pack in {src_dir}")
bad = [p.name for p in files if not PACK_FILENAME_RE.fullmatch(p.name)]
if bad:
raise ValueError(
f"{src_dir}: files the downloader will reject: {bad} "
f"(allowed: {PACK_FILENAME_RE.pattern})")
out_zip.parent.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(out_zip, "w", zipfile.ZIP_STORED) as zf:
# ZIP_STORED: the media (mp4/mp3) and .vst3 binaries are already
# compressed; deflating just burns CPU for ~0 gain.
for p in files:
# Fixed mtime (the zip epoch, 1980-01-01) + fixed perms so the
# bytes don't depend on the checkout's file timestamps.
info = zipfile.ZipInfo(p.name, date_time=(1980, 1, 1, 0, 0, 0))
info.compress_type = zipfile.ZIP_STORED
# Pin create_system: ZipInfo defaults it from the host OS (0 on
# Windows, 3 on Unix), which would otherwise make the same pack
# hash differently across runners. 3 = Unix.
info.create_system = 3
info.external_attr = 0o644 << 16
zf.writestr(info, p.read_bytes())
data = out_zip.read_bytes()
return {"sha256": hashlib.sha256(data).hexdigest(), "bytes": len(data)}
# rig_builder ships "fat" .vst3 bundles carrying all three platforms inside
# Contents/. A pack for one platform keeps that platform's binary dir + the
# shared bundle files, and drops the other two.
VST_PLATFORM_DIRS = {"mac": "MacOS", "win": "x86_64-win", "linux": "x86_64-linux"}
def build_vst_pack(vst_root: Path, out_zip: Path, platform: str) -> dict:
"""Reproducibly zip the VST tree keeping only `platform`'s binaries.
Slices each fat .vst3: everything is kept except the two foreign platform
dirs (MacOS / x86_64-win / x86_64-linux) and the vst/src build trees. Arc
names are relative to vst_root so the download endpoint extracts straight
into <plugin>/vst/. Same reproducible-build guarantees as build_pack.
"""
if platform not in VST_PLATFORM_DIRS:
raise ValueError(f"unknown platform {platform!r} (want mac/win/linux)")
foreign = set(VST_PLATFORM_DIRS.values()) - {VST_PLATFORM_DIRS[platform]}
files = []
for p in sorted(vst_root.rglob("*"), key=lambda q: q.as_posix()):
if not p.is_file():
continue
rel = p.relative_to(vst_root)
if rel.parts and rel.parts[0] == "src": # skip C++/JUCE build trees
continue
if set(rel.parts) & foreign: # drop foreign-platform binaries
continue
files.append((p, rel))
if not files:
raise ValueError(f"no VST files to pack for {platform} in {vst_root}")
out_zip.parent.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(out_zip, "w", zipfile.ZIP_STORED) as zf:
for p, rel in files:
info = zipfile.ZipInfo(rel.as_posix(), date_time=(1980, 1, 1, 0, 0, 0))
info.compress_type = zipfile.ZIP_STORED
# Pin create_system like build_pack: ZipInfo defaults it from the
# host OS (0 on Windows, 3 on Unix), which would otherwise make the
# same pack hash differently across runners. VST packs are the most
# likely to be built on Windows (native .vst3), so without this pin
# the precomputable-hash guarantee breaks exactly where it's needed.
info.create_system = 3
info.external_attr = 0o644 << 16
zf.writestr(info, p.read_bytes())
data = out_zip.read_bytes()
return {"sha256": hashlib.sha256(data).hexdigest(), "bytes": len(data)}
def manifest_entry(out_zip: Path, url: str) -> dict:
"""Pack info as the download-path expects it: {url, sha256, bytes}."""
return {"url": url,
"sha256": hashlib.sha256(out_zip.read_bytes()).hexdigest(),
"bytes": out_zip.stat().st_size}
# Per-pack, versioned, immutable release convention (matches what the team
# already published, e.g. tag `venue-arena-v1` / asset `arena-pack-v1.zip`).
def pack_tag(pack_id: str, version: int) -> str:
return f"venue-{pack_id}-v{version}"
def pack_asset(pack_id: str, version: int) -> str:
return f"{pack_id}-pack-v{version}.zip"
def pack_url(pack_id: str, version: int, repo: str = REPO) -> str:
return (f"https://github.com/{repo}/releases/download/"
f"{pack_tag(pack_id, version)}/{pack_asset(pack_id, version)}")
# VST packs use the same immutable per-pack convention, keyed by platform:
# tag `vst-<plat>-v<N>`, asset `vst-<plat>-pack-v<N>.zip`. The manifest they
# emit is keyed by platform (mac/win/linux) — the shape the rig_builder plugin's
# data/vst_packs.json consumes.
def vst_tag(platform: str, version: int) -> str:
return f"vst-{platform}-v{version}"
def vst_asset(platform: str, version: int) -> str:
return f"vst-{platform}-pack-v{version}.zip"
def vst_url(platform: str, version: int, repo: str = REPO) -> str:
return (f"https://github.com/{repo}/releases/download/"
f"{vst_tag(platform, version)}/{vst_asset(platform, version)}")
def _publish_release(tag: str, zip_path: Path, title: str, notes: str,
repo: str = REPO) -> None:
"""Create the per-pack release if missing, then upload the versioned zip.
Tags are immutable: a media change means a new version (v1 v2), never a
re-upload so no --clobber. gh errors if the asset already exists, which is
the right guard against overwriting a published, referenced pack.
"""
if subprocess.run(["gh", "release", "view", tag, "--repo", repo],
capture_output=True).returncode != 0:
subprocess.run(
["gh", "release", "create", tag, "--repo", repo, "--latest=false",
"--title", title, "--notes", notes],
check=True)
subprocess.run(
["gh", "release", "upload", tag, str(zip_path), "--repo", repo], check=True)
def publish(pack_id: str, version: int, zip_path: Path, repo: str = REPO) -> None:
_publish_release(pack_tag(pack_id, version), zip_path,
f"{pack_id.capitalize()} venue pack v{version}",
"Opt-in career venue pack. Not a code release.", repo)
def _pack_id(src_dir: Path) -> str:
return src_dir.name
def main(argv=None) -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("src", nargs="*", type=Path,
help="pack source dirs (e.g. plugins/career/venue-packs/club)")
ap.add_argument("--version", type=int, default=1,
help="pack version (tag venue-<id>-v<N>); default 1")
ap.add_argument("--local", type=Path, metavar="DIR",
help="write zips here + a file:// manifest.json; no upload")
ap.add_argument("--publish", action="store_true",
help="create/upload the per-pack release; emit release URLs")
ap.add_argument("--vst", action="store_true",
help="slice one rig VST root (src[0]) into per-platform "
"vst-<plat>-v<N> packs; manifest keyed by platform "
"(the shape rig_builder's data/vst_packs.json wants)")
ap.add_argument("--manifest", type=Path,
help="write the {id: {url,sha256,bytes}} map here (default: stdout)")
ap.add_argument("--selfcheck", action="store_true", help="run the round-trip demo and exit")
args = ap.parse_args(argv)
if args.selfcheck:
return _selfcheck()
if not args.src or (not args.local and not args.publish):
ap.error("need one or more src dirs and either --local or --publish")
out_dir = args.local if args.local else Path(args.src[0]).parent / "_packs"
manifest = {}
if args.vst:
vst_root = args.src[0]
for plat in VST_PLATFORM_DIRS:
zip_path = out_dir / vst_asset(plat, args.version)
build_vst_pack(vst_root, zip_path, plat)
if args.publish:
_publish_release(vst_tag(plat, args.version), zip_path,
f"Rig VST pack ({plat}) v{args.version}",
"Opt-in per-platform rig VST pack. Not a code release.")
url = vst_url(plat, args.version)
else:
url = (out_dir.resolve() / zip_path.name).as_uri()
manifest[plat] = manifest_entry(zip_path, url)
else:
for src in args.src:
pid = _pack_id(src)
zip_path = out_dir / pack_asset(pid, args.version)
build_pack(src, zip_path)
if args.publish:
publish(pid, args.version, zip_path)
url = pack_url(pid, args.version)
else:
url = (out_dir.resolve() / zip_path.name).as_uri()
manifest[pid] = manifest_entry(zip_path, url)
out = json.dumps(manifest, indent=2)
if args.manifest:
args.manifest.write_text(out + "\n", encoding="utf-8")
else:
print(out)
return 0
def _selfcheck() -> int:
"""Build a pack and confirm build_pack/manifest_entry agree on the digest."""
import tempfile
with tempfile.TemporaryDirectory() as td:
td = Path(td)
src = td / "bar"
src.mkdir()
(src / "manifest.json").write_text('{"venue":"bar"}')
(src / "bored.mp4").write_bytes(b"\x00fake-video")
zip_path = td / pack_asset("bar", 1)
info = build_pack(src, zip_path)
# Reproducible: a second build (into a different path) is byte-identical.
info2 = build_pack(src, td / "again.zip")
assert info2["sha256"] == info["sha256"], "build is not reproducible"
entry = manifest_entry(zip_path, pack_url("bar", 1))
assert entry["sha256"] == info["sha256"], "digest mismatch"
assert entry["bytes"] == info["bytes"]
assert entry["url"] == (
f"https://github.com/{REPO}/releases/download/venue-bar-v1/bar-pack-v1.zip")
# Round-trip: the zip must be flat (names == basenames).
with zipfile.ZipFile(zip_path) as zf:
names = zf.namelist()
assert set(names) == {"manifest.json", "bored.mp4"}, names
# VST slice: keep target platform + shared, drop foreign, reproducible.
c = td / "vst" / "Foo.vst3" / "Contents"
for d in ("MacOS", "x86_64-win", "x86_64-linux", "Resources"):
(c / d).mkdir(parents=True)
(c / "MacOS" / "Foo").write_bytes(b"mac")
(c / "x86_64-linux" / "Foo.so").write_bytes(b"linux")
(c / "Info.plist").write_bytes(b"<plist/>")
vzip = td / vst_asset("linux", 1)
vinfo = build_vst_pack(td / "vst", vzip, "linux")
assert vinfo == build_vst_pack(td / "vst", td / "v2.zip", "linux"), \
"vst slice is not reproducible"
with zipfile.ZipFile(vzip) as zf:
vnames = set(zf.namelist())
assert "Foo.vst3/Contents/x86_64-linux/Foo.so" in vnames
assert "Foo.vst3/Contents/Info.plist" in vnames
assert not any("MacOS" in n for n in vnames), vnames
print("content_packs selfcheck: ok")
return 0
if __name__ == "__main__":
sys.exit(main())