Compare commits

..
Author SHA1 Message Date
=Scr4tch= 196008a98e fix(sloppak/ws_highway): drums stay in picker and show correct hit count
Problem:
The editor strips drum arrangements from the sloppak manifest, converting
them to a drum_tab sidecar.  This left two visible bugs:

1. "Drums (0)" — drum-only packs showed zero notes in the arrangement
   picker even though the drum highway displayed thousands of hits.  The
   sloppak loader synthesized a placeholder Arrangement(name="Drums")
   with no notes whenever the arrangements list was empty, and the
   WebSocket highway always reported that placeholder's note count (0).

2. Drums disappear alongside other arrangements — when a drum+bass or
   drum+guitar pack was loaded, the placeholder was never created because
   the arrangements list was not empty (the other instrument was present).
   The drum arrangement had been removed by the editor during drum_tab
   conversion, so drums vanished entirely from the arrangement picker.

Fix (sloppak.py):
Replaced the "arrangements list is empty" trigger with "no drum
arrangement exists".  The loader now checks whether any arrangement name
contains 'drum' or 'percussion' (case-insensitive substring match using
_DRUM_KEYWORDS) and synthesizes the placeholder when a drum_tab is
present but no matching arrangement is found.  This ensures drums appear
in the picker even alongside bass, guitar, or other pitched instruments.

Fix (ws_highway.py):
When building the arrangement list for the song_info WebSocket message,
the highway now reads the drum_tab hit count from the loaded sloppak and
substitutes it for any empty arrangement whose name matches the drum
keywords.  The placeholder has no notes of its own (it exists only to
carry the drum_tab through to the drum highway), so the real hit count
from drum_tab is surfaced instead — e.g. "Drums (1922)" instead of
"Drums (0)".

Guard against false positives:
The hit-count override is scoped to drum/percussion arrangements only.
Without this guard, an empty "Vocals" or "Keys" track in the same
pack would incorrectly inherit the drum hit count.  The same keyword set
_DRUM_KEYWORDS = ('drum', 'percussion') is used in both sloppak.py and
ws_highway.py to keep the detection consistent.

Signed-off-by: =Scr4tch= <305609711+0-Scr4tch-0@users.noreply.github.com>
2026-07-20 12:05:37 -04:00
21 changed files with 84 additions and 1529 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
-46
View File
@@ -8,52 +8,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Added
- **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
to play. Selecting one re-streams that part's tab over the highway WS
(`?drum_part=<id>`, mirroring the arrangement switch); the choice persists
across an arrangement change, and the picker reflects the server's
authoritative part (unknown/absent selection falls back to the primary). The
row hides for single-drum and non-drum songs, so nothing changes there. Builds
on the loader below; no plugin change needed — the drum renderer just draws
whatever tab streams.
- **Multiple drum parts (feedpak 1.17.0 "drums as arrangements").** The sloppak
loader now reads `type: drums` arrangement entries carrying per-arrangement
`drum_tab` file pointers — a song can ship several drum charts (a second
drummer, an aux-percussion layer). Parts surface as `LoadedSloppak.drum_parts`
(primary first; the entry aliasing the song-level `drum_tab:` key is the
primary and is never loaded twice), the highway WS `song_info` gains a
`drum_parts` name list, and `?drum_part=<id>` on the WS URL selects which
part's tab streams (`drum_tab` messages carry `part_id` when multiple parts
exist; unknown ids fall back to the primary). Pointer entries are **never**
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
+25 -52
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,
@@ -144,21 +143,9 @@ def _sanitize_authors(manifest: dict | None) -> list[dict]:
return out
def _drum_part_id_for_wire(drum_parts: list[dict] | None, selected_id: str | None) -> str | None:
"""Expose a part id only when the pack genuinely has multiple parts."""
return selected_id if selected_id is not None and len(drum_parts or []) > 1 else None
@router.websocket("/ws/highway/{filename:path}")
async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1,
naming_mode: str = "legacy", drum_part: str = ""):
"""Stream song data for the highway renderer over WebSocket.
`drum_part` selects WHICH drum part's tab streams when the pack carries
several (feedpak 1.17.0 "drums as arrangements") — a part id from
song_info's `drum_parts`. Empty / unknown ids fall back to the primary,
so a stale or mistyped selection degrades to today's behavior instead of
silencing drums."""
async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1, naming_mode: str = "legacy"):
"""Stream song data for the highway renderer over WebSocket."""
await websocket.accept()
structlog.contextvars.bind_contextvars(ws_conn_id=uuid.uuid4().hex[:8])
@@ -274,8 +261,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
@@ -490,12 +478,24 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1,
_evict_audio_cache()
# Send song metadata
# For drum-only sloppaks the placeholder arrangement has 0 notes
# (drum_tab hits live separately). Surface the real hit count so
# the UI shows e.g. "Drums (1922)" instead of "Drums (0)".
_DRUM_KEYWORDS = ("drum", "percussion")
_dt_hit_count = 0
if is_slop and loaded_slop is not None and loaded_slop.drum_tab is not None:
_dt_hit_count = len(loaded_slop.drum_tab.get("hits") or [])
arr_list = [
{
"index": i,
"name": a.name,
"smart_name": smart_names[i],
"notes": len(a.notes) + sum(len(c.notes) for c in a.chords),
"notes": (
_dt_hit_count
if (len(a.notes) == 0 and not a.chords and _dt_hit_count > 0
and any(kw in (a.name or "").lower() for kw in _DRUM_KEYWORDS))
else len(a.notes) + sum(len(c.notes) for c in a.chords)
),
}
for i, a in enumerate(song.arrangements)
]
@@ -576,15 +576,6 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1,
"has_drum_tab": bool(
is_slop and loaded_slop is not None and loaded_slop.drum_tab is not None
),
# The song's DRUM PARTS (feedpak 1.17.0 "drums as arrangements"),
# primary first — names only; the selected part's payload streams
# as the `drum_tab`/`drum_hits` messages below. Always a list
# (empty when the pack has no drums, and a single entry for a
# legacy one-drum pack), so a part picker can bind unconditionally.
"drum_parts": [
{"id": p["id"], "name": p["name"]}
for p in (loaded_slop.drum_parts or [])
] if is_slop and loaded_slop is not None else [],
"has_notation": bool(
is_slop
and loaded_slop is not None
@@ -608,36 +599,18 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1,
# client-side drums plugin keeps a fallback decoder for them.
if is_slop and loaded_slop is not None and loaded_slop.drum_tab is not None:
dt = loaded_slop.drum_tab
# Multiple drum parts: `?drum_part=<id>` picks which part's tab
# streams; the default (and any unknown id) is the PRIMARY —
# exactly the pre-parts behavior, so legacy clients notice nothing.
_dt_part_id = None
if loaded_slop.drum_parts:
_dt_part_id = loaded_slop.drum_parts[0]["id"]
if drum_part:
for _p in loaded_slop.drum_parts:
if _p["id"] == drum_part:
dt = _p["drum_tab"]
_dt_part_id = _p["id"]
break
kit = drums_mod.normalise_kit(dt.get("kit"))
hits_wire = drums_mod.hits_to_wire(dt.get("hits") or [])
_dt_name = dt.get("name")
_dt_name = _dt_name if isinstance(_dt_name, str) and _dt_name else "Drums"
_dt_msg = {
"type": "drum_tab",
"version": int(dt.get("version", drums_mod.SCHEMA_VERSION)),
"name": _dt_name,
"kit": kit,
"total": len(hits_wire),
}
# Only multi-part packs identify a part on the wire. Legacy packs
# synthesize a one-item list internally but keep their old frame.
_wire_part_id = _drum_part_id_for_wire(loaded_slop.drum_parts, _dt_part_id)
if _wire_part_id is not None:
_dt_msg["part_id"] = _wire_part_id
try:
await websocket.send_json(_dt_msg)
await websocket.send_json({
"type": "drum_tab",
"version": int(dt.get("version", drums_mod.SCHEMA_VERSION)),
"name": _dt_name,
"kit": kit,
"total": len(hits_wire),
})
for i in range(0, len(hits_wire), 500):
await websocket.send_json({
"type": "drum_hits",
@@ -1014,7 +987,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)
+40 -165
View File
@@ -730,125 +730,6 @@ class LoadedSloppak:
# separated stems the moment one drops below 100% — demucs recombination is
# lossy, so the mixdown is strictly the better audio when nothing is muted.
full_mix: str | None = None
# The song's DRUM PARTS (feedpak 1.17.0 "drums as arrangements"): one dict
# {"id", "name", "drum_tab"} per part, primary FIRST. A part comes from a
# `type: drums` arrangement entry carrying a per-arrangement `drum_tab`
# file pointer and NO note `file` — entries this loader deliberately never
# turns into fretted Arrangements (see the file/notation gate in
# load_song; that skip IS the grading invariant). The primary part's
# payload is the SAME object as `drum_tab` above (the song-level key is
# its back-compat alias). None when the pack has no drums at all; a
# single-part list for a legacy pack with only the song-level key.
drum_parts: list[dict] | None = None
def _load_drum_tab_file(source_dir: Path, rel: str, label: str) -> dict | None:
"""Load + schema-validate one drum-tab JSON named by a manifest-relative
path. Shared by the song-level `drum_tab:` key and the per-arrangement
drum-part pointers (feedpak 1.17.0), so every tab gets the same posture:
permissive — a missing file disables that part silently; a traversal,
parse, or validation failure disables it with a warning, never aborting
the load."""
# 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)
except Exception as e:
log.warning("sloppak: failed to parse %s %r: %s", label, rel, e)
return None
ok, reason = drums_mod.validate_drum_tab(raw)
if not ok:
log.warning("sloppak: %s %r failed validation: %s", label, rel, reason)
return None
return raw
def _resolve_drum_parts(
source_dir: Path,
drum_tab_rel: object,
drum_tab_data: dict | None,
drum_pointer_entries: list[dict],
) -> tuple[dict | None, list[dict] | None]:
"""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
extra_parts: list[dict] = []
seen_rels: set[str] = set()
# Use the same canonical, traversal-safe identity as zip member lookup so
# equivalent spellings ("x.json", "./x.json", or backslashes) identify
# one file. Otherwise an alias pointer can reload and duplicate the primary.
primary_rel_key = (
_zip_member_key(drum_tab_rel.strip())
if isinstance(drum_tab_rel, str) and drum_tab_rel.strip() else None
)
for entry in drum_pointer_entries:
rel = str(entry.get("drum_tab") or "").strip()
rel_key = _zip_member_key(rel) if rel else None
rel_identity = rel_key or rel
if not rel or rel_identity in seen_rels:
continue
seen_rels.add(rel_identity)
entry_id = str(entry.get("id") or "").strip()
entry_name = str(entry.get("name") or "").strip()
if primary_rel_key is not None and rel_key == primary_rel_key:
if entry_id:
primary_id = entry_id
if entry_name:
primary_name = entry_name
continue
tab = _load_drum_tab_file(source_dir, rel, f"drum part {entry_id or rel}")
if tab is None:
continue
tab_name = tab.get("name")
extra_parts.append({
"id": entry_id,
"name": entry_name
or (tab_name if isinstance(tab_name, str) and tab_name else "Drums"),
"drum_tab": tab,
})
parts: list[dict] = []
used_ids: set[str] = set()
if drum_tab_data is not None:
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})
used_ids.add(primary_id)
next_generated_id = 2
for part in extra_parts:
part_id = part["id"]
if not part_id or part_id in used_ids:
while f"drums-{next_generated_id}" in used_ids:
next_generated_id += 1
part_id = f"drums-{next_generated_id}"
next_generated_id += 1
part["id"] = part_id
used_ids.add(part_id)
parts.append(part)
if not parts:
return drum_tab_data, None
if drum_tab_data is None:
drum_tab_data = parts[0]["drum_tab"]
return drum_tab_data, parts
def load_song(
@@ -873,7 +754,6 @@ def load_song(
notation_acc: dict[str, dict] = {}
any_notation = False
arrangement_ids_acc: list[str | None] = [] # parallel to song.arrangements
drum_pointer_entries: list[dict] = [] # feedpak 1.17.0 drum-part pointers
for entry in manifest.get("arrangements", []) or []:
if not isinstance(entry, dict):
log.warning("sloppak: non-dict arrangement entry skipped (%r)", type(entry).__name__)
@@ -882,30 +762,7 @@ 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):
# 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):
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",
entry.get("drum_tab"), entry.get("type"),
)
if not rel and not has_notation_key:
continue
data = None
if rel:
@@ -935,11 +792,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:
@@ -1016,24 +868,48 @@ def load_song(
drum_tab_data: dict | None = None
drum_tab_rel = manifest.get("drum_tab")
if isinstance(drum_tab_rel, str) and drum_tab_rel:
drum_tab_data = _load_drum_tab_file(source_dir, drum_tab_rel, "drum_tab")
# 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 / drum_tab_rel).resolve()
dt_path.relative_to(source_dir.resolve())
except ValueError:
log.warning("sloppak: drum_tab path %r escapes source_dir — skipped", drum_tab_rel)
dt_path = None
except OSError as e:
log.warning("sloppak: drum_tab path resolution failed (%s) — skipped", e)
dt_path = None
if dt_path is not None and dt_path.exists():
try:
raw = load_json(dt_path)
except Exception as e:
log.warning("sloppak: failed to parse drum_tab %r: %s", drum_tab_rel, e)
raw = None
if raw is not None:
ok, reason = drums_mod.validate_drum_tab(raw)
if ok:
drum_tab_data = raw
else:
log.warning("sloppak: drum_tab %r failed validation: %s",
drum_tab_rel, reason)
# Keep the dense compatibility logic independently testable and guarantee
# ids are unique before the highway exposes them as selectors.
drum_tab_data, drum_parts = _resolve_drum_parts(
source_dir, drum_tab_rel, drum_tab_data, drum_pointer_entries,
)
# Drum-only sloppak: every GP track was percussion, so it ships a
# drum_tab but no pitched arrangements. The highway WS rejects an empty
# arrangements list with "No arrangements found" *before* it serves the
# drum_tab, leaving the drums unplayable even in the drum highway.
# Synthesize a minimal placeholder arrangement so the stream proceeds and
# the drum_tab reaches the drum highway. It carries no notes (the guitar
# highway just shows an empty board) and, when the manifest omits a
# Drum sloppak: when a drum_tab is present but no existing arrangement is
# a drum part, synthesize a minimal placeholder so the drums appear in the
# arrangement picker and the drum_tab reaches the drum highway. The editor
# strips drum arrangements out of the manifest (converting them to
# drum_tab), so without this a drum+bass sloppak would show only Bass in
# the picker with no way to reach the drums. It carries no notes (the
# guitar highway just shows an empty board) and, when the manifest omits a
# duration, derives a song length from the last drum hit so the timeline
# isn't zero-length.
if not song.arrangements and drum_tab_data is not None:
_DRUM_KEYWORDS = ("drum", "percussion")
_has_drum_arr = any(
any(kw in (getattr(a, "name", "") or "").lower() for kw in _DRUM_KEYWORDS)
for a in song.arrangements
)
if not _has_drum_arr and drum_tab_data is not None:
if song.song_length <= 0:
# validate_drum_tab() intentionally does NOT type-check individual
# hits (they're sanitized at WS-stream time), so a hit may carry a
@@ -1350,7 +1226,6 @@ def load_song(
manifest=manifest,
feedpak_version=_fpv if isinstance(_fpv, str) and _fpv else None,
drum_tab=drum_tab_data,
drum_parts=drum_parts,
song_timeline=song_timeline_data,
tempos=tempos_data,
time_signatures=time_sigs_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.
-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 ─────────────────────────────────────────────────────────────
+3 -30
View File
@@ -1150,7 +1150,7 @@ window.feedBack.on('song:ready', () => {
let _arrBusyGen = 0;
let _arrBusyTimeout = null;
async function changeArrangement(index, drumPart) {
async function changeArrangement(index) {
if (currentFilename) {
// Tear down any pending fresh-load credits before switching: the
// no-count-in hold timer would otherwise fire togglePlay() against the
@@ -1276,38 +1276,11 @@ async function changeArrangement(index, drumPart) {
_resetSectionPracticeLog();
invalidateParentCount();
// Carry the selected drum part across the re-stream. An explicit
// `drumPart` (a drum-part switch, from changeDrumPart) wins; otherwise
// preserve the current picker selection so an ARRANGEMENT switch keeps
// the chosen part (drum parts are song-level, not per-arrangement).
const part = drumPart !== undefined
? drumPart
: (document.getElementById('drum-part-select')?.value || '');
window.highway.reconnect(currentFilename, index, part);
window.highway.reconnect(currentFilename, index);
window.feedBack.emit('arrangement:changed', { index, filename: currentFilename });
}
}
// Switch which drum part plays (feedpak 1.17.0 "drums as arrangements"). A part
// switch re-streams the same song with a different drum tab — the same
// transition as an arrangement switch — so it delegates to changeArrangement
// with the CURRENT arrangement held and the new part applied. Wired to
// #drum-part-select's onchange; the select is populated + shown by
// highway.js's song_info handler only when the song has 2+ drum parts.
async function changeDrumPart(partId) {
if (!currentFilename) return;
let index = 0;
const si = window.highway && typeof window.highway.getSongInfo === 'function'
? window.highway.getSongInfo() : null;
if (si && typeof si.arrangement_index === 'number' && si.arrangement_index >= 0) {
index = si.arrangement_index;
} else {
const arrSel = document.getElementById('arr-select');
if (arrSel && arrSel.value !== '') index = Number(arrSel.value) || 0;
}
return changeArrangement(index, partId);
}
// Restart the current song from the beginning (or from loop A when an AB
// loop is armed). Uses the canonical _audioSeek funnel only — never touches
// audio.currentTime directly and never reloads via playSong().
@@ -2352,7 +2325,7 @@ configureHost({
Object.assign(window, {
_confirmDialog, _getArrangementNamingMode, _libraryLocalFilename, _librarySongArtUrl,
_librarySongId, _onHeaderClick, _onNamingModeChange, _trapFocusInModal,
changeArrangement, changeDrumPart, checkPluginUpdates, clearLibFilters, clearLoop,
changeArrangement, checkPluginUpdates, clearLibFilters, clearLoop,
deleteSelectedLoop, esc, exportDiagnostics, exportSettings, filterFavorites,
filterLibrary, fullRescanLibrary, goFavPage, handleSliderInput,
hideScanBanner, importSettings, loadPlugins, loadSavedLoop,
+1 -46
View File
@@ -2298,31 +2298,6 @@ function createHighway() {
sel.appendChild(opt);
}
}
// Drum-part picker (feedpak 1.17.0 "drums as
// arrangements"): a song can carry several drum
// charts. Populate the picker beside the
// arrangement switcher; show it only when there
// are 2+ parts to choose between. `drum_parts`
// is always present (empty for non-drum songs),
// so a single-drum / no-drum song hides it. The
// currently-streaming part is marked selected by
// the `drum_tab` handler below (authoritative
// `part_id`), so we don't guess here.
{
const dpSel = document.getElementById('drum-part-select');
if (dpSel) {
const parts = Array.isArray(msg.drum_parts) ? msg.drum_parts : [];
dpSel.textContent = '';
for (const p of parts) {
const opt = document.createElement('option');
opt.value = p.id;
opt.textContent = p.name || p.id;
dpSel.appendChild(opt);
}
const dpRow = document.getElementById('v3-drum-part-row');
if (dpRow) dpRow.classList.toggle('hidden', parts.length <= 1);
}
}
}
// Plugin context API — broadcast current song state
if (window.feedBack) {
@@ -2405,22 +2380,7 @@ function createHighway() {
name: (typeof msg.name === 'string' && msg.name) ? msg.name : 'Drums',
kit: Array.isArray(msg.kit) ? msg.kit : [],
hits: [],
// Which drum part this stream carries (feedpak
// 1.17.0). Present only for multi-part packs;
// null otherwise. Plugins can read it via
// bundle.drumTab.part_id.
part_id: (typeof msg.part_id === 'string' && msg.part_id) ? msg.part_id : null,
};
// Reflect the authoritative streaming part in the
// picker (the server resolves an unknown/absent
// selection to the primary, so this keeps the
// dropdown honest even after a fallback).
if (hwState.drumTab.part_id) {
const dpSel = document.getElementById('drum-part-select');
if (dpSel && dpSel.value !== hwState.drumTab.part_id) {
dpSel.value = hwState.drumTab.part_id;
}
}
break;
case 'drum_hits':
if (hwState.drumTab && Array.isArray(msg.data)) {
@@ -2813,7 +2773,7 @@ function createHighway() {
localStorage.setItem('showFingerHints', String(hwState._showFingerHints));
},
reconnect(filename, arrangement, drumPart) {
reconnect(filename, arrangement) {
// Close old WS but keep audio + animation running
if (hwState.ws) { hwState.ws.close(); hwState.ws = null; }
hwState.ready = false;
@@ -2839,11 +2799,6 @@ function createHighway() {
_resetChordRenderState();
const wsParams = new URLSearchParams();
if (arrangement !== undefined) wsParams.set('arrangement', arrangement);
// Multiple drum parts (feedpak 1.17.0 "drums as arrangements"):
// carry the selected part id so the WS streams ITS drum tab. Empty
// / undefined → the primary part (server default), i.e. today's
// one-drum behavior for any pack the picker never touched.
if (drumPart) wsParams.set('drum_part', drumPart);
let namingMode = 'smart';
if (typeof window._getArrangementNamingMode === 'function') {
const v = window._getArrangementNamingMode();
-4
View File
@@ -1194,10 +1194,6 @@
<button id="arr-default-pin" type="button" onclick="pinCurrentArrangementDefault()" aria-pressed="false" aria-label="Select an arrangement to make it the default" class="v3-pop-pin" title="Select an arrangement to make it the default"></button>
</span>
</div>
<div class="v3-pop-row hidden" id="v3-drum-part-row">
<span class="v3-pop-label">Drum part</span>
<select id="drum-part-select" onchange="changeDrumPart(this.value)" class="v3-pop-select max-w-[130px]" title="Which drum chart to play — a song can carry several"></select>
</div>
<div class="v3-pop-row">
<span class="v3-pop-label" id="mastery-slider-label">Difficulty</span>
<span class="flex items-center gap-2">
+1 -1
View File
@@ -316,7 +316,7 @@ test('anchor zoom helpers read the staged anchors first', () => {
test('init and reconnect clear the stage but keep the provider', () => {
const src = fs.readFileSync(highwayJs, 'utf8');
const initBody = extractBlock(src, 'init(canvasEl, container)');
const reconnectBody = extractBlock(src, 'reconnect(filename, arrangement, drumPart)');
const reconnectBody = extractBlock(src, 'reconnect(filename, arrangement)');
assert.match(initBody, /_clearChartTransformStage\(\);/, 'init clears the stage');
assert.match(reconnectBody, /_clearChartTransformStage\(\);/, 'reconnect clears the stage');
assert.ok(!/init\([\s\S]{0,2000}_xfProvider = null/.test(src.slice(src.indexOf('const api = {'))),
+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
-17
View File
@@ -1,17 +0,0 @@
"""Wire-compatibility coverage for selectable drum parts."""
from routers.ws_highway import _drum_part_id_for_wire
def test_single_synthesized_part_keeps_legacy_frame_without_part_id():
parts = [{"id": "drums", "name": "Drums", "drum_tab": {}}]
assert _drum_part_id_for_wire(parts, "drums") is None
def test_multiple_parts_expose_selected_part_id():
parts = [
{"id": "drums", "name": "Drums", "drum_tab": {}},
{"id": "drums-2", "name": "Aux", "drum_tab": {}},
]
assert _drum_part_id_for_wire(parts, "drums-2") == "drums-2"
assert _drum_part_id_for_wire(parts, None) is None
-295
View File
@@ -1,295 +0,0 @@
"""Loader coverage for MULTIPLE drum parts (feedpak 1.17.0 "drums as
arrangements").
A drum part rides the manifest as a `type: drums` arrangement entry carrying
a per-arrangement `drum_tab` file pointer and NO note `file`. The loader:
- NEVER turns a pointer entry into a fretted Arrangement that skip is
the grading invariant (an empty drum chart must not reach the fretted
pipeline, where note detection would grade it as garbage);
- resolves the parts into `LoadedSloppak.drum_parts`, primary FIRST: the
entry aliasing the song-level `drum_tab:` file contributes its id/name
but is never loaded twice (its payload IS `loaded.drum_tab`);
- loads each extra part's file with the same permissive posture as the
song-level tab (a bad part disables that part only, never the load);
- copes with a pointer-only pack (no song-level key): the first part
becomes the primary so every legacy consumer keeps working.
"""
from __future__ import annotations
import json
import logging
from pathlib import Path
import yaml
import sloppak as sloppak_mod
def _tab(name: str, hits: list[dict] | None = None) -> dict:
return {
"version": 1,
"name": name,
"kit": [{"id": "kick", "name": "Kick"}],
"hits": hits if hits is not None else [{"t": 1.0, "p": "kick", "v": 100}],
}
def _write_pak(root: Path, manifest_extras: dict, files: dict[str, dict | str]) -> Path:
"""A minimal directory-form sloppak with one Lead arrangement plus the
given extra files ({relpath: json-dict-or-raw-text})."""
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))
for rel, payload in files.items():
text = payload if isinstance(payload, str) else json.dumps(payload)
(pak / rel).write_text(text)
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)
def _two_part_manifest() -> dict:
"""The exact shape the editor writes: primary alias entry + one extra."""
return {
"drum_tab": "drum_tab.json",
"arrangements": [
{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"},
{"id": "drums", "name": "Drums", "type": "drums",
"drum_tab": "drum_tab.json"},
{"id": "drums-2", "name": "Drums (Live)", "type": "drums",
"drum_tab": "drum_tab_drums-2.json"},
],
}
# ── The grading invariant ────────────────────────────────────────────────────
def test_pointer_entries_never_become_fretted_arrangements(tmp_path: Path):
pak = _write_pak(tmp_path, _two_part_manifest(), {
"drum_tab.json": _tab("Drums"),
"drum_tab_drums-2.json": _tab("Drums (Live)"),
})
loaded = _load(pak, tmp_path)
# Only the Lead chart is an Arrangement — neither drum part enters the
# fretted pipeline (song.arrangements is what note detection grades).
assert [a.name for a in loaded.song.arrangements] == ["Lead"]
# And the ids list stays parallel to song.arrangements (skipped entries
# contribute nothing) — a misalignment here would remap every chart edit.
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):
pak = _write_pak(tmp_path, _two_part_manifest(), {
"drum_tab.json": _tab("Drums"),
"drum_tab_drums-2.json": _tab("Drums (Live)", [{"t": 2.0, "p": "kick", "v": 90}]),
})
loaded = _load(pak, tmp_path)
assert loaded.drum_parts is not None
assert [(p["id"], p["name"]) for p in loaded.drum_parts] == [
("drums", "Drums"), ("drums-2", "Drums (Live)"),
]
# The primary's payload IS the song-level tab — same object, loaded once.
assert loaded.drum_parts[0]["drum_tab"] is loaded.drum_tab
assert loaded.drum_parts[1]["drum_tab"]["hits"][0]["t"] == 2.0
def test_primary_pointer_equivalent_path_is_not_duplicated(tmp_path: Path):
manifest = {
"drum_tab": "drum_tab.json",
"arrangements": [
{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"},
{"id": "kit", "name": "Live Kit", "type": "drums",
"drum_tab": "./drum_tab.json"},
],
}
pak = _write_pak(tmp_path, manifest, {"drum_tab.json": _tab("Drums")})
loaded = _load(pak, tmp_path)
assert loaded.drum_parts is not None
assert [(p["id"], p["name"]) for p in loaded.drum_parts] == [
("kit", "Live Kit"),
]
assert loaded.drum_parts[0]["drum_tab"] is loaded.drum_tab
def test_legacy_single_drum_pack_gets_a_one_part_list(tmp_path: Path):
pak = _write_pak(tmp_path, {"drum_tab": "drum_tab.json"}, {
"drum_tab.json": _tab("Drums"),
})
loaded = _load(pak, tmp_path)
assert loaded.drum_parts is not None and len(loaded.drum_parts) == 1
assert loaded.drum_parts[0]["id"] == "drums"
assert loaded.drum_parts[0]["drum_tab"] is loaded.drum_tab
def test_no_drums_means_no_parts(tmp_path: Path):
pak = _write_pak(tmp_path, {}, {})
loaded = _load(pak, tmp_path)
assert loaded.drum_parts is None
assert loaded.drum_tab is None
def test_pointer_only_pack_promotes_the_first_part_to_primary(tmp_path: Path):
# A writer that omitted the song-level alias: readers must cope (the
# spec keeps the alias, but a reader never crashes on its absence).
pak = _write_pak(tmp_path, {
"arrangements": [
{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"},
{"id": "kit", "name": "Kit", "type": "drums",
"drum_tab": "drum_tab_kit.json"},
],
}, {"drum_tab_kit.json": _tab("Kit")})
loaded = _load(pak, tmp_path)
assert loaded.drum_parts is not None and len(loaded.drum_parts) == 1
# The part's tab becomes THE drum tab, so has_drum_tab / the default
# stream / the drum-only placeholder all keep working.
assert loaded.drum_tab is loaded.drum_parts[0]["drum_tab"]
assert loaded.drum_parts[0]["id"] == "kit"
# ── Permissive per-part failure ──────────────────────────────────────────────
def test_a_bad_extra_part_disables_that_part_only(tmp_path: Path):
manifest = _two_part_manifest()
manifest["arrangements"].append(
{"id": "drums-3", "name": "Broken", "type": "drums",
"drum_tab": "drum_tab_broken.json"})
pak = _write_pak(tmp_path, manifest, {
"drum_tab.json": _tab("Drums"),
"drum_tab_drums-2.json": _tab("Drums (Live)"),
"drum_tab_broken.json": "not json {{{",
})
loaded = _load(pak, tmp_path)
assert [p["id"] for p in loaded.drum_parts] == ["drums", "drums-2"]
def test_a_traversal_part_path_is_skipped(tmp_path: Path):
manifest = _two_part_manifest()
manifest["arrangements"][2]["drum_tab"] = "../outside.json"
(tmp_path / "outside.json").write_text(json.dumps(_tab("Evil")))
pak = _write_pak(tmp_path, manifest, {"drum_tab.json": _tab("Drums")})
loaded = _load(pak, tmp_path)
assert [p["id"] for p in loaded.drum_parts] == ["drums"]
def test_duplicate_pointer_rels_load_once(tmp_path: Path):
manifest = _two_part_manifest()
manifest["arrangements"].append(
{"id": "drums-dup", "name": "Dup", "type": "drums",
"drum_tab": "drum_tab_drums-2.json"})
pak = _write_pak(tmp_path, manifest, {
"drum_tab.json": _tab("Drums"),
"drum_tab_drums-2.json": _tab("Drums (Live)"),
})
loaded = _load(pak, tmp_path)
assert [p["id"] for p in loaded.drum_parts] == ["drums", "drums-2"]
def test_duplicate_part_ids_are_made_unique(tmp_path: Path):
manifest = _two_part_manifest()
manifest["arrangements"][2]["id"] = "drums"
manifest["arrangements"].append(
{"id": "drums-2", "name": "Aux", "type": "drums",
"drum_tab": "drum_tab_aux.json"})
pak = _write_pak(tmp_path, manifest, {
"drum_tab.json": _tab("Drums"),
"drum_tab_drums-2.json": _tab("Drums (Live)"),
"drum_tab_aux.json": _tab("Aux"),
})
loaded = _load(pak, tmp_path)
assert [p["id"] for p in loaded.drum_parts] == ["drums", "drums-2", "drums-3"]
def test_drum_pointer_with_wrong_type_logs_warning(tmp_path: Path, caplog):
pak = _write_pak(tmp_path, {
"arrangements": [
{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"},
{"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)
assert loaded.drum_parts is None
assert "has drum_tab" in caplog.text and "type='druns'" in caplog.text
# ── Drum-only pack with parts ────────────────────────────────────────────────
def test_drum_only_pack_with_pointer_entries_still_synthesizes_placeholder(tmp_path: Path):
# No pitched arrangements at all, drums via pointer entries only: the
# placeholder "Drums" arrangement must still appear so the highway WS
# proceeds and the tab reaches the drum highway.
pak = _write_pak(tmp_path, {
"arrangements": [
{"id": "kit", "name": "Kit", "type": "drums",
"drum_tab": "drum_tab_kit.json"},
],
}, {"drum_tab_kit.json": _tab("Kit", [{"t": 5.0, "p": "kick", "v": 100}])})
# Remove the Lead arrangement _write_pak added to the manifest.
manifest_path = pak / "manifest.yaml"
manifest = yaml.safe_load(manifest_path.read_text())
manifest["arrangements"] = [e for e in manifest["arrangements"] if e.get("id") != "lead"]
manifest.pop("duration", None)
manifest_path.write_text(yaml.safe_dump(manifest, sort_keys=False))
loaded = _load(pak, tmp_path)
assert [a.name for a in loaded.song.arrangements] == ["Drums"]
assert loaded.drum_parts is not None and loaded.drum_parts[0]["id"] == "kit"
# Song length derived from the last hit (the drum-only path's rule).
assert loaded.song.song_length > 5.0
-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,
-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}'
-191
View File
@@ -1,191 +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)}
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)}")
def publish(pack_id: str, version: int, zip_path: Path, 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.
"""
tag = pack_tag(pack_id, version)
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", f"{pack_id.capitalize()} venue pack v{version}",
"--notes", "Opt-in career venue pack. Not a code release."],
check=True)
subprocess.run(
["gh", "release", "upload", tag, str(zip_path), "--repo", repo], check=True)
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("--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 = {}
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
print("content_packs selfcheck: ok")
return 0
if __name__ == "__main__":
sys.exit(main())