uj-mastering-master/scripts/webapi_smoke.py

#!/usr/bin/env python3
"""End-to-end check of the web API against a running server.

Not a unit test — it drives the real HTTP surface with real audio and asserts on
the contract the browser depends on:

  * every registered metric returns a drawable payload (curves or a heatmap)
  * the numbers agree with an independent computation (librosa RMS, sample peak)
  * view toggles (lin/log, absolute/relative) never change the underlying data
  * **no audio is written to the host**: uploads are spooled to temp, and neither
    an application directory nor a temp file survives the request
  * overlay, metadata, delete and error paths behave

Usage:  python3 scripts/webapi_smoke.py [base_url] [audio_dir]
"""

from __future__ import annotations

import glob
import hashlib
import json
import math
import os
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

BASE = sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:8137"
AUDIO = Path(sys.argv[2] if len(sys.argv) > 2 else "/tmp/ujm-test")

PASS, FAIL = [], []


def check(name: str, ok: bool, detail: str = "") -> None:
  (PASS if ok else FAIL).append(name + (f" — {detail}" if detail else ""))
  print(f"{'PASS' if ok else 'FAIL'}  {name}{' — ' + detail if detail else ''}")


def get(path: str, **params) -> dict:
  url = BASE + path + (("?" + urllib.parse.urlencode(params)) if params else "")
  with urllib.request.urlopen(url, timeout=600) as res:
    return json.loads(res.read())


def post(path: str, payload=None, **params) -> dict:
  url = BASE + path + (("?" + urllib.parse.urlencode(params)) if params else "")
  data = json.dumps(payload).encode() if payload is not None else b""
  req = urllib.request.Request(url, data=data, method="POST",
                               headers={"Content-Type": "application/json"})
  with urllib.request.urlopen(req, timeout=600) as res:
    return json.loads(res.read())


def put_audio(path: Path, metric: str = "rms_power", file_id: str | None = None) -> dict:
  """Send a raw audio body, the way the browser does."""
  data = path.read_bytes()
  qs = {"name": path.name, "metric": metric}
  if file_id:
    qs["file_id"] = file_id
  req = urllib.request.Request(
    f"{BASE}/api/analyse?" + urllib.parse.urlencode(qs), data=data, method="POST",
    headers={"Content-Type": "application/octet-stream"})
  with urllib.request.urlopen(req, timeout=900) as res:
    return json.loads(res.read())


def expect_http_error(fn, code: int, name: str) -> None:
  try:
    fn()
  except urllib.error.HTTPError as exc:
    check(name, exc.code == code, f"HTTP {exc.code}")
  except Exception as exc:  # noqa: BLE001
    check(name, False, f"raised {type(exc).__name__}: {exc}")
  else:
    check(name, False, "no error raised")


def temp_leftovers() -> list[str]:
  """Any ujm- spools sitting in the temp dir (would mean a leak)."""
  return glob.glob(os.path.join(tempfile.gettempdir(), "ujm-*"))


def main() -> int:
  files = sorted(AUDIO.glob("*.wav")) + sorted(AUDIO.glob("*.mp3"))
  if not files:
    print(f"no audio in {AUDIO}")
    return 2

  # ---- catalogue, and the promise that there is no server-side audio store ----
  cat = get("/api/metrics")
  ids = [m["id"] for m in cat["metrics"]]
  check("metric catalogue lists all 7 metrics", len(ids) == 7, ", ".join(ids))
  check("catalogue marks a default", cat["default"] in ids, cat["default"])
  sess = get("/api/session")
  check("session advertises client-side audio storage",
        "client" in sess.get("audio_storage", ""), sess.get("audio_storage", ""))
  check("session exposes no upload path", "upload_dir" not in sess)

  before = set(temp_leftovers())

  # ---- analysis, one track per request (the browser's model) ----
  uploads = {}
  for f in files:
    res = put_audio(f)
    uploads[f.name] = res
    check(f"analyse {f.name} returns a drawable payload",
          bool(res["payload"]["series"] or res["payload"]["segments"]
               or res["payload"]["heatmap"]))
    check(f"analyse {f.name} returns a content-hash id",
          res["file"]["file_id"] == hashlib.sha256(f.read_bytes()).hexdigest()[:32],
          res["file"]["file_id"][:12])

  sine = next(r for n, r in uploads.items() if "sine440" in n)
  noise = next(r for n, r in uploads.items() if "pinknoise" in n)
  # The tagged MP3 needs an encoder to create, so it may be absent on a machine
  # without ffmpeg (e.g. the container image). Skip its assertions rather than
  # failing the suite on a missing fixture — that would be a false alarm.
  tagged = next((r for r in uploads.values()
                 if r["file"]["song_name"].startswith("Test Artist")), None)

  check("duration read from container", abs(sine["file"]["duration"] - 12.0) < 0.05,
        f"{sine['file']['duration']}s")
  check("native sample rate preserved (sr=None)", sine["file"]["sample_rate"] == 44100)
  if tagged is None:
    print("SKIP  ID3 tag assertions — no tagged MP3 fixture (needs ffmpeg)")
  else:
    check("ID3 tags resolve to a song name",
          tagged["file"]["song_name"] == "Test Artist - Test Title",
          tagged["file"]["song_name"])
    check("CJK-containing filename handled",
          tagged["file"]["name"].endswith(".mp3"), tagged["file"]["name"])
  check("analysis starts mono, as documented", sine["file"]["channels"] == 1)
  check("size is reported back to the client", sine["file"]["size_bytes"] > 0,
        f"{sine['file']['size_bytes']} bytes")

  # ---- the no-persistence guarantee ----
  after = set(temp_leftovers())
  check("no spool file survives a completed analysis",
        not (after - before), ", ".join(sorted(after - before)) or "clean")

  # re-sending the same bytes must hit the cache, not re-decode
  t0 = time.perf_counter()
  again = put_audio(Path(AUDIO / "sine440.wav"), metric="waveform")
  warm = time.perf_counter() - t0
  check("re-sent bytes keep the same file_id (content addressed)",
        again["file"]["file_id"] == sine["file"]["file_id"])
  check("cached metric is served without re-decoding", warm < 3.0, f"{warm:.2f}s")

  # ---- every metric draws, and the client can request it per track ----
  for mid in ids:
    # the plot endpoint only renders what has been analysed; analyse this metric
    # for the sine first, exactly as the UI does on a dropdown change.
    put_audio(Path(AUDIO / "sine440.wav"), metric=mid)
    # restrict to the sine so per-metric shape is unambiguous
    post("/api/overlay", {"file_ids": [sine["file"]["file_id"]]})
    plot = get("/api/plot", metric=mid, x_mode="absolute")
    ds = plot["datasets"]
    if not ds:
      check(f"metric {mid} returns a dataset", False, "no datasets")
      continue
    payload = ds[0]["payload"]
    has_drawable = bool(payload["series"] or payload["segments"] or payload["heatmap"])
    x_len = len(payload["series"][0]["x"]) if payload["series"] else (
      len(payload["segments"][0]["x"]) if payload["segments"] else
      (len(payload["heatmap"]["x"]) if payload["heatmap"] else 0))
    check(f"metric {mid} draws", has_drawable and x_len > 1,
          f"{len(payload['series'])} curves, {len(payload['segments'])} bands, "
          f"{'heatmap' if payload['heatmap'] else 'no heatmap'}, x={x_len}")
    if payload["heatmap"]:
      hm = payload["heatmap"]
      # TARGET_COLUMNS is a ceiling, not a floor: the hop is whichever is
      # larger, N_FFT//4 (1024 samples) or len(y)/TARGET_COLUMNS.
      check("spectrogram respects the 4000-column ceiling",
            len(hm["x"]) <= 4000 and len(hm["x"]) > 1, f"{len(hm['x'])} columns")
      check("spectrogram serialises finite dB",
            hm["z_max"] == 0.0 and hm["z_min"] == -80.0,
            f"z range {hm['z_min']}..{hm['z_max']}")
      check("spectrogram ships frequency extent for the client log axis",
            hm["dy"] > 0 and hm["y0"] > 0, f"y0={hm['y0']:.2f} dy={hm['dy']:.3f}")

  # ---- physics sanity: independent recomputation ----
  import librosa
  import numpy as np

  post("/api/overlay", {"file_ids": [sine["file"]["file_id"]]})
  put_audio(Path(AUDIO / "sine440.wav"), metric="rms_power")
  y, sr = librosa.load(str(AUDIO / "sine440.wav"), sr=None)
  y = librosa.to_mono(y)

  rms_plot = get("/api/plot", metric="rms_power")
  check("isolation left exactly one dataset on the plot",
        len(rms_plot["datasets"]) == 1, f"{len(rms_plot['datasets'])} datasets")
  band = rms_plot["datasets"][0]["payload"]["segments"][0]
  # Reproduce the metric's own framing: librosa.feature.rms at 10 s / 2 s.
  expected = librosa.feature.rms(y=y, frame_length=10 * sr, hop_length=2 * sr).ravel()
  reported = np.array([v for v in band["hi"] if v is not None], dtype=float)
  check("RMS power matches librosa's own RMS frame-for-frame",
        len(reported) == len(expected)
        and float(np.max(np.abs(reported - expected))) < 1e-4,
        f"{len(reported)} frames, max Δ {float(np.max(np.abs(reported - expected))):.2e}")

  put_audio(Path(AUDIO / "sine440.wav"), metric="waveform")
  wseg = get("/api/plot", metric="waveform")["datasets"][0]["payload"]["segments"][0]
  # 529200 samples into ~4000 columns: each column spans ~132 samples of a
  # constant-amplitude sine, so the envelope must *reach* the true peak exactly.
  check("waveform envelope reaches the true extremes",
        abs(max(wseg["hi"]) - float(np.max(y))) < 1e-6
        and abs(min(wseg["lo"]) - float(np.min(y))) < 1e-6,
        f"hi {max(wseg['hi']):.6f} vs peak {float(np.max(y)):.6f}")

  put_audio(Path(AUDIO / "sine440.wav"), metric="true_peak")
  tp = get("/api/plot", metric="true_peak")["datasets"][0]["payload"]
  note = next(n for n in tp["annotations"] if "dBTP" in n)
  dbtp = float(note.replace("Max:", "").replace("dBTP", "").strip())
  expected_dbtp = 20 * math.log10(float(np.max(np.abs(y))) + 1e-12)
  check("true peak matches the sample peak for a 440 Hz tone (no ISP)",
        abs(dbtp - expected_dbtp) < 0.5,
        f"api {dbtp:.2f} dBTP vs {expected_dbtp:.2f} dBFS")

  put_audio(Path(AUDIO / "sine440.wav"), metric="lufs")
  lp = get("/api/plot", metric="lufs")["datasets"][0]["payload"]
  integrated_line = next((h for h in lp["hlines"] if h["label"].startswith("Integrated:")), None)
  check("integrated LUFS is drawn as its own labelled reference line",
        integrated_line is not None,
        integrated_line["label"] if integrated_line else "missing")
  lu = float(integrated_line["label"].split(":")[1].split()[0]) if integrated_line else None
  check("integrated LUFS is sane for a -12 dBFS sine",
        lu is not None and -50 < lu < 0, f"{lu} LUFS")
  check("the -14 LUFS streaming target line is present",
        any(abs(h["y"] + 14.0) < 1e-9 for h in lp["hlines"]))
  check("LRA is reported", any("LU" in n for n in lp["annotations"]),
        "; ".join(lp["annotations"]))

  # ---- view toggles must not touch the data ----
  put_audio(Path(AUDIO / "sine440.wav"), metric="spectrogram")
  lin = get("/api/plot", metric="spectrogram", y_log="false")
  log = get("/api/plot", metric="spectrogram", y_log="true")
  check("y_log toggle only changes the axis flag",
        lin["datasets"][0]["payload"]["heatmap"]["z"]
        == log["datasets"][0]["payload"]["heatmap"]["z"])
  check("y_log=true is reflected in axes",
        log["axes"]["y_log"] is True and lin["axes"]["y_log"] is False)

  put_audio(Path(AUDIO / "sine440.wav"), metric="rms_power")
  abs_x = get("/api/plot", metric="rms_power", x_mode="absolute")["datasets"][0]["payload"]["segments"][0]["x"]
  rel = get("/api/plot", metric="rms_power", x_mode="relative")
  rel_x = rel["datasets"][0]["payload"]["segments"][0]["x"]
  check("relative mode normalises x to 0..100",
        abs(rel_x[0]) < 1e-6 and abs(rel_x[-1] - 100.0) < 1e-6,
        f"{rel_x[0]:.4f} .. {rel_x[-1]:.4f}")
  check("relative mode relabels the x axis",
        rel["axes"]["x_label"] == "Position (%)", rel["axes"]["x_label"])
  check("relative mode keeps the same sample count", len(abs_x) == len(rel_x),
        f"{len(abs_x)} vs {len(rel_x)}")

  # ---- caching: a repeat view render must not recompute ----
  t0 = time.perf_counter()
  first = get("/api/plot", metric="spectrogram")
  t_cold = time.perf_counter() - t0
  t0 = time.perf_counter()
  second = get("/api/plot", metric="spectrogram")
  t_warm = time.perf_counter() - t0
  check("repeat plot render is cache-served and identical",
        first["datasets"][0]["payload"]["heatmap"]["z"]
        == second["datasets"][0]["payload"]["heatmap"]["z"],
        f"cold {t_cold:.2f}s vs warm {t_warm:.2f}s")

  # ---- overlay ----
  all_ids = [r["file"]["file_id"] for r in uploads.values()]
  post("/api/overlay", {"file_ids": all_ids})
  for f in files:
    put_audio(f, metric="crest_factor")
  overlay_all = get("/api/plot", metric="crest_factor")
  check("overlay draws every selected track", len(overlay_all["datasets"]) == len(files),
        f"{len(overlay_all['datasets'])} datasets")
  labels = [d["label"] for d in overlay_all["datasets"]]
  check("each overlay dataset is labelled with its song", all(labels), ", ".join(labels))
  check("datasets carry a distinct palette slot",
        len({d["palette_index"] for d in overlay_all["datasets"]}) == len(files))

  post("/api/overlay", {"file_ids": [sine["file"]["file_id"]]})
  check("narrowing the overlay removes the others",
        len(get("/api/plot", metric="crest_factor")["datasets"]) == 1)

  post("/api/overlay", {"file_ids": all_ids})
  spec = get("/api/plot", metric="crest_factor")
  check("crest factor overlays all tracks (unlike the heatmap)",
        len(spec["datasets"]) == len(files), f"{len(spec['datasets'])} datasets")
  spec = get("/api/plot", metric="spectrogram")
  check("spectrogram sends one track at a time",
        len(spec["datasets"]) == 1 and bool(spec["warnings"]),
        spec["warnings"][0] if spec["warnings"] else "no warning")

  # a metric not yet computed for a track is reported, not silently skipped
  post("/api/overlay", {"file_ids": all_ids})
  plot = get("/api/plot", metric="psr")
  unresolved = [w for w in plot["warnings"] if "not computed yet" in w]
  check("un-analysed metrics are flagged rather than faked",
        len(unresolved) >= 1 or len(plot["datasets"]) >= 1,
        f"{len(unresolved)} pending, {len(plot['datasets'])} ready")

  # ---- reference lines ----
  refs = json.dumps([{"value": -14, "label": "target", "color": "#ff0000", "style": "dash"}])
  withrefs = get("/api/plot", metric="rms_power", refs=refs)
  check("reference lines round-trip",
        withrefs["refs"][0]["value"] == -14.0 and withrefs["refs"][0]["label"] == "target")
  badrefs = get("/api/plot", metric="rms_power", refs=json.dumps([{"nope": 1}]))
  check("malformed reference entries are dropped, not fatal", badrefs["refs"] == [])

  # ---- metadata endpoint ----
  meta = get("/api/metadata", file_id=sine["file"]["file_id"])
  check("metadata endpoint returns the snapshot",
        meta["file_id"] == sine["file"]["file_id"]
        and meta["name"] == sine["file"]["name"], meta["name"])

  # ---- error paths ----
  expect_http_error(lambda: get("/api/plot", metric="not_a_metric"), 400,
                    "unknown metric rejected")
  expect_http_error(lambda: get("/api/metadata", file_id="0" * 32), 404,
                    "unknown track rejected")
  expect_http_error(lambda: post("/api/overlay", {"file_ids": "nope"}), 400,
                    "overlay requires a list")
  expect_http_error(
    lambda: urllib.request.urlopen(urllib.request.Request(
      f"{BASE}/api/analyse?name=evil.exe&metric=rms_power", data=b"xx", method="POST"),
      timeout=30), 400, "unsupported extension rejected")

  # ---- delete ----
  post("/api/overlay", {"file_ids": [noise["file"]["file_id"]]})
  deleted = json.loads(urllib.request.urlopen(urllib.request.Request(
    f"{BASE}/api/file?" + urllib.parse.urlencode({"file_id": noise["file"]["file_id"]}),
    method="DELETE"), timeout=60).read())
  check("delete forgets the track", noise["file"]["file_id"] not in deleted["overlay"])
  check("delete explains the audio is still the client's",
        "browser" in deleted["note"], deleted["note"])
  check("deleted track is gone from the session",
        not any(f["file_id"] == noise["file"]["file_id"]
                for f in get("/api/session")["files"]))

  final = set(temp_leftovers())
  check("temp dir holds no spool after the whole suite", not (final - before),
        ", ".join(sorted(final - before)) or "clean")

  print(f"\n{len(PASS)} passed, {len(FAIL)} failed")
  for f in FAIL:
    print("  FAILED:", f)
  return 1 if FAIL else 0


if __name__ == "__main__":
  sys.exit(main())