Commit 4197696d by PLN (Algolia)

feat(douanier): adopt the platform's path-derived scope convention (#43)

The audio API now speaks the platform scope convention (nechapi scopes.py):
access is hierarchical realm:domain:path and the required scope is DERIVED FROM
THE ROUTE, so it's maintenance-free — add an endpoint and its scope exists.

- scopes.py — vendored byte-for-byte from nechapi/_platform/scopes.py; a
  drift-guard test (test_scopes.py) fails if the two ever diverge, so the
  gateway and this service can never disagree on who's allowed in.
- app.py — replaced the per-route require_scope("emotion"|"features"|…) strings
  with ONE path-derived dependency: `require` computes api:audio:<path> from the
  request and checks it; `require_auth` covers /me (any identity). Also closed a
  footgun: a gateway-injected request with a MISSING X-Scopes header now defaults
  to NO scopes (was "*").
- auth.py — Principal.has_scope is now the hierarchical matcher (api:audio:*
  authorizes api:audio:analyze:emotion, etc.).
- tests — gateway-header tests grant api:audio:*; the scope-enforcement tests now
  prove real path-derivation (a sibling grant like api:audio:features → 403 on
  /grade and /onsets). +test_scopes.py for the matcher + drift guard. 67 passing.
- clients/README — scope table rewritten to the convention (api:audio:<path>,
  grant api:audio:* or api:* for breadth).

Validated end-to-end through https://api.nech.pl/audio/v1 with a freshly minted
api:* token: /me → scopes [api:*]; /features 200 (cache miss→hit); a sibling
scope 403s; no-identity 401s. Built + redeployed douanier:latest to erable
(seccomp=unconfined per DEPLOY.md).
parent fbe1742d
......@@ -18,6 +18,7 @@ from fastapi.middleware.cors import CORSMiddleware
import auth
import cache
import config
import scopes as scopelib
from engines import ears, feats, signal
from engines import loudness as loudness_eng
from engines import grade as grade_eng # VENDORED copy of the Foundry grader (see engines/grade.py)
......@@ -67,8 +68,8 @@ def _principal(request: Request) -> auth.Principal | None:
grants a wildcard, so `uvicorn` + curl works without the platform."""
tenant = request.headers.get(config.TENANT_HEADER)
if tenant:
scopes = request.headers.get(config.SCOPES_HEADER, "*")
return auth.Principal(tenant, frozenset(scopes.split()), None, None)
granted = request.headers.get(config.SCOPES_HEADER, "")
return auth.Principal(tenant, frozenset(granted.split()), None, None)
authz = request.headers.get("authorization", "")
kind, _, token = authz.partition(" ")
if (kind.lower() == "bearer" and token and config.DEV_TOKEN
......@@ -77,18 +78,37 @@ def _principal(request: Request) -> auth.Principal | None:
return None
def require_scope(scope: str | None):
"""Dependency factory: resolve the gateway identity and (if `scope` is given)
assert it carries that scope or the '*' wildcard. scope=None means "any
authenticated caller". 401 if no/unknown identity, 403 if scoped out."""
def dep(request: Request) -> auth.Principal:
principal = _principal(request)
if principal is None:
raise HTTPException(401, "no authenticated identity (gateway headers absent)")
if scope is not None and not principal.has_scope(scope):
raise HTTPException(403, f"caller lacks scope '{scope}'")
return principal
return dep
# Our domain in the platform scope tree, from root_path "/audio/v1" → "audio".
_DOMAIN = (config.ROOT_PATH.strip("/").split("/") or ["audio"])[0] or "audio"
def _required_scope(request: Request) -> str:
"""The platform scope this request needs, derived from its path — so access
control tracks the URL by construction (see scopes.py). The gateway already
stripped /audio/v1, so request.url.path is the route, e.g. /analyze/emotion
→ api:audio:analyze:emotion."""
return scopelib.required_for(_DOMAIN, request.url.path)
def require(request: Request) -> auth.Principal:
"""Path-derived scope gate: resolve the gateway identity and assert it carries
a scope authorizing THIS route. 401 if no identity, 403 if scoped out. The
platform gateway enforces the same rule centrally; this is defense-in-depth."""
principal = _principal(request)
if principal is None:
raise HTTPException(401, "no authenticated identity (gateway headers absent)")
required = _required_scope(request)
if not principal.has_scope(required):
raise HTTPException(403, f"caller lacks scope '{required}'")
return principal
def require_auth(request: Request) -> auth.Principal:
"""Any authenticated caller (no specific scope) — for identity echo endpoints."""
principal = _principal(request)
if principal is None:
raise HTTPException(401, "no authenticated identity (gateway headers absent)")
return principal
# ── health ──────────────────────────────────────────────────────────────────
......@@ -148,7 +168,7 @@ def metrics():
@v1.get("/me")
def whoami(principal: auth.Principal = Depends(require_scope(None))):
def whoami(principal: auth.Principal = Depends(require_auth)):
"""Echo the caller's gateway-resolved identity — handy for onboarding smoke tests."""
return {"account": principal.account, "scopes": sorted(principal.scopes),
"expires_at": principal.expires_at}
......@@ -187,7 +207,7 @@ def _cached(kind: str, params: dict, data: bytes, filename: str, compute, respon
@v1.post("/analyze/emotion")
async def analyze_emotion(response: Response, file: UploadFile = File(...),
_p: auth.Principal = Depends(require_scope("emotion"))):
_p: auth.Principal = Depends(require)):
"""Upload a short audio clip → valence/arousal + top emotions. Cached (#26):
the same audio returns instantly on a repeat instead of recomputing."""
data = await _read_upload(file)
......@@ -198,7 +218,7 @@ async def analyze_emotion(response: Response, file: UploadFile = File(...),
@v1.post("/features")
async def features(response: Response, file: UploadFile = File(...), rhythm: bool = True,
_p: auth.Principal = Depends(require_scope("features"))):
_p: auth.Principal = Depends(require)):
"""Upload a clip → the ~35-dim audio feature stack (spectral moments, MFCCs,
chroma/key, envelope, + rhythm when `rhythm=true`). Torch-free, cached."""
data = await _read_upload(file)
......@@ -210,7 +230,7 @@ async def features(response: Response, file: UploadFile = File(...), rhythm: boo
@v1.post("/analyze/samples")
async def analyze_samples(response: Response, file: UploadFile = File(...), name: str = "",
_p: auth.Principal = Depends(require_scope("samples"))):
_p: auth.Principal = Depends(require)):
"""Upload a one-shot/loop → per-sample EDA + MEASURED role (percs|bass|melodic|
tops|atmos) from the spectrum, never the name. Optional `name` only
disambiguates breaks/drums. Torch-free, cached."""
......@@ -223,7 +243,7 @@ async def analyze_samples(response: Response, file: UploadFile = File(...), name
@v1.post("/onsets")
async def onsets(response: Response, file: UploadFile = File(...),
_p: auth.Principal = Depends(require_scope("onsets"))):
_p: auth.Principal = Depends(require)):
"""Upload audio → onset hit times (s) + tempo + onset rate. The cheapest tier
(no model): rhythmic hits for visual sync / slicing. Torch-free, cached."""
data = await _read_upload(file)
......@@ -233,7 +253,7 @@ async def onsets(response: Response, file: UploadFile = File(...),
@v1.post("/waveform")
async def waveform(response: Response, file: UploadFile = File(...), bins: int = 800,
_p: auth.Principal = Depends(require_scope("waveform"))):
_p: auth.Principal = Depends(require)):
"""Upload audio → render-ready waveform: per-bin [min,max] + a 0..1 RMS energy
envelope (`?bins=`, ≤4000). For audio-reactive visuals. Torch-free, cached."""
data = await _read_upload(file)
......@@ -244,7 +264,7 @@ async def waveform(response: Response, file: UploadFile = File(...), bins: int =
@v1.post("/analyze")
async def analyze_all(response: Response, file: UploadFile = File(...),
_p: auth.Principal = Depends(require_scope("analyze"))):
_p: auth.Principal = Depends(require)):
"""One call → emotion + features + sample role for a clip (fewer round-trips
for hexa). Each is the same engine the dedicated routes use. Cached as a unit."""
data = await _read_upload(file)
......@@ -260,7 +280,7 @@ async def analyze_all(response: Response, file: UploadFile = File(...),
@v1.post("/grade")
async def grade(response: Response, file: UploadFile = File(...), role: str = "",
_p: auth.Principal = Depends(require_scope("grade"))):
_p: auth.Principal = Depends(require)):
"""Upload a loop/one-shot → the Foundry's MECHANICAL quality grade: composite
0..1 + S/A/B/C/D tier, per-rule sub-scores (seam click, zero-crossing
cleanliness, DC, bar self-consistency, level, bass mono-compat) + flags.
......@@ -274,7 +294,7 @@ async def grade(response: Response, file: UploadFile = File(...), role: str = ""
@v1.post("/loudness")
async def loudness(response: Response, file: UploadFile = File(...),
_p: auth.Principal = Depends(require_scope("loudness"))):
_p: auth.Principal = Depends(require)):
"""Upload audio → BS.1770 integrated LUFS + sample/true peak + crest + the
gain (dB) to hit each delivery target (-14 streaming, -9 club). Torch-free, cached."""
data = await _read_upload(file)
......@@ -286,7 +306,7 @@ async def loudness(response: Response, file: UploadFile = File(...),
@v1.post("/spectrum")
async def spectrum(response: Response, file: UploadFile = File(...),
bands: int = 64, frames: int = 200, mel: bool = True,
_p: auth.Principal = Depends(require_scope("spectrum"))):
_p: auth.Principal = Depends(require)):
"""FFT-as-a-service: a downsampled, render-ready spectrogram (`bands`×`frames`,
0..1). `mel=` perceptual vs log-linear; `frames=1` → a single averaged FFT
spectrum. Bounded payload → caches cheaply. Torch-free."""
......@@ -300,7 +320,7 @@ async def spectrum(response: Response, file: UploadFile = File(...),
@v1.post("/naming")
async def naming(response: Response, file: UploadFile = File(...), index: int = 1,
_p: auth.Principal = Depends(require_scope("naming"))):
_p: auth.Principal = Depends(require)):
"""Upload a sample → a convention-compliant name (NN_role_character) derived
from the MEASURED role + character, not the file name. Torch-free, cached."""
data = await _read_upload(file)
......@@ -318,7 +338,7 @@ async def naming(response: Response, file: UploadFile = File(...), index: int =
# ── #37 separation seam — no CPU path on erable; env-selected GPU backend later ──
@v1.post("/separate")
async def separate(_p: auth.Principal = Depends(require_scope("separate"))):
async def separate(_p: auth.Principal = Depends(require)):
"""Stem separation (demucs/roformer). erable has no GPU and no CPU-viable path,
so this is a deliberate 503 until a GPU-runner backend is wired (env-selected,
same response shape). The route exists now so hexa can code against it and get
......
......@@ -15,6 +15,7 @@ from dataclasses import dataclass
import config
import db
import scopes as _scopes
TTL_RE_DAYS = 86400.0
......@@ -32,7 +33,9 @@ class Principal:
expires_at: float | None
def has_scope(self, scope: str) -> bool:
return "*" in self.scopes or scope in self.scopes
"""Hierarchical match per the platform convention (see scopes.py):
a granted `api:audio:*` authorizes `api:audio:analyze:emotion`, etc."""
return _scopes.match_any(self.scopes, scope)
# ── minting / management (used by the douanier CLI) ─────────────────────────────
......
......@@ -43,22 +43,33 @@ export async function POST(req: Request) {
}
```
## Endpoints (scopes in parens)
## Scopes
| Method | Path | Returns | scope |
Access is by the platform's hierarchical scope convention. The scope a route
needs is **derived from its path**: `/audio/v1/analyze/emotion` requires
`api:audio:analyze:emotion`, `/audio/v1/grade` requires `api:audio:grade`, etc.
A grant of **`api:audio:*`** covers the whole audio API, and **`api:*`** covers
every API on the platform — so most tokens just carry one of those. (Full
convention: the platform's `RECIPE.md` / `scopes.py`.)
## Endpoints (required scope in parens)
| Method | Path | Returns | required scope |
|---|---|---|---|
| GET | `/healthz` | status + engine availability | — |
| POST | `/analyze/emotion` | valence/arousal + top emotions | `emotion` |
| POST | `/features` | ~35-dim feature stack | `features` |
| POST | `/analyze/samples` | per-sample EDA + measured role | `samples` |
| POST | `/grade` | mechanical loop quality (tier + flags) | `grade` |
| POST | `/onsets` | onset times + tempo | `onsets` |
| POST | `/waveform` | `[min,max]` peaks + RMS envelope | `waveform` |
| POST | `/spectrum` | downsampled spectrogram (FFT-as-a-service) | `spectrum` |
| POST | `/loudness` | BS.1770 LUFS + peak + gain-to-target | `loudness` |
| POST | `/naming` | convention-compliant sample name | `naming` |
| POST | `/analyze` | emotion + features + sample in one call | `analyze` |
| POST | `/separate` | **503** until a GPU runner | `separate` |
| GET | `/healthz` | status + engine availability | — (public) |
| POST | `/analyze/emotion` | valence/arousal + top emotions | `api:audio:analyze:emotion` |
| POST | `/features` | ~35-dim feature stack | `api:audio:features` |
| POST | `/analyze/samples` | per-sample EDA + measured role | `api:audio:analyze:samples` |
| POST | `/grade` | mechanical loop quality (tier + flags) | `api:audio:grade` |
| POST | `/onsets` | onset times + tempo | `api:audio:onsets` |
| POST | `/waveform` | `[min,max]` peaks + RMS envelope | `api:audio:waveform` |
| POST | `/spectrum` | downsampled spectrogram (FFT-as-a-service) | `api:audio:spectrum` |
| POST | `/loudness` | BS.1770 LUFS + peak + gain-to-target | `api:audio:loudness` |
| POST | `/naming` | convention-compliant sample name | `api:audio:naming` |
| POST | `/analyze` | emotion + features + sample in one call | `api:audio:analyze` |
| POST | `/separate` | **503** until a GPU runner | `api:audio:separate` |
(All covered by `api:audio:*` or `api:*`.)
## curl
......
"""Nech.PL APIs — the scope convention, in one place.
Scopes are hierarchical, ':'-delimited paths: realm:domain:seg:seg…
realm the capability class. `api` = the HTTP APIs; `admin` = the control
plane (tenant/token management); future siblings: `mail`, `data`, …
domain the sub-API (for the `api` realm): audio, geo, …
seg… the route path under that API, version stripped.
A route's REQUIRED scope is derived from its public path, so access control
tracks the URL by construction — add `/audio/foo/bar` and it simply requires
`api:audio:foo:bar`, no table to maintain:
/audio/v1/analyze/emotion -> api:audio:analyze:emotion
/audio/v1/grade -> api:audio:grade
/geo/v1/reverse -> api:geo:reverse
A GRANTED scope matches a required one if it is the same, or an ancestor ending
in `*` (a subtree grant):
* everything (break-glass superadmin)
api:* every HTTP API (present & future)
api:audio:* all of the audio API
api:audio:analyze:* the analyze subtree
api:audio:analyze:emotion that one route, exactly
admin:* the control plane (NOT granted by `api:*` or, by
convention, by a bare domain grant)
Note the deliberate asymmetry: a NON-wildcard grant is exact — `api:audio` does
NOT grant `api:audio:grade`. Only the `*`-suffixed form opens a subtree. This
keeps "give them exactly this route" safe and "give them everything under here"
explicit.
Pure stdlib, no deps — the same file is vendored verbatim into each service so
the gateway and every API agree on the rule. Keep it dependency-free and in
sync (services drift-guard against this copy).
"""
from __future__ import annotations
REALMS = ("api", "admin") # extensible: add "mail", "data", … as they appear
def required_for(domain: str, route: str, realm: str = "api") -> str:
"""The scope a request needs, from its (domain, route). `route` is the
path under the API with the version already stripped (e.g. "/analyze/emotion"
or "/"). Returns e.g. "api:audio:analyze:emotion"; "api:audio" for the root."""
segs = [s for s in route.strip("/").split("/") if s]
parts = [realm]
if domain:
parts.append(domain)
parts.extend(segs)
return ":".join(parts)
def matches(granted: str, required: str) -> bool:
"""True if a single GRANTED scope authorizes a REQUIRED scope.
Wildcard `*` (only ever the last segment) opens the whole subtree at that
level; otherwise the grant must equal the requirement exactly."""
g = granted.split(":")
r = required.split(":")
for i, seg in enumerate(g):
if seg == "*":
return True # subtree grant — covers this node & below
if i >= len(r) or seg != r[i]:
return False
return len(g) == len(r) # exhausted without '*' → must be exact
def match_any(granted, required: str) -> bool:
"""True if ANY of the caller's granted scopes authorizes `required`."""
return any(matches(g, required) for g in granted)
def validate(scope: str) -> bool:
"""A scope is well-formed if every segment is non-empty and `*` appears only
as the final segment (a `*` mid-path like `api:*:emotion` is meaningless)."""
if not scope:
return False
parts = scope.split(":")
for i, p in enumerate(parts):
if p == "":
return False
if p == "*" and i != len(parts) - 1:
return False
return True
......@@ -60,7 +60,7 @@ def test_endpoint_light_path_is_torch_free(monkeypatch):
data = fh.read()
r = client.post("/analyze/emotion",
files={"file": ("t.wav", data, "audio/wav")},
headers={"X-Tenant": "hexa", "X-Scopes": "emotion"})
headers={"X-Tenant": "hexa", "X-Scopes": "api:audio:*"})
assert r.status_code == 200
body = r.json()
assert body["emotion"]["engine"] == "light"
......
......@@ -37,7 +37,7 @@ def _tmp(freqs=(220, 440), secs=2.0, amp=0.4):
return str(p)
GW = lambda scope: {"X-Tenant": "hexa", "X-Scopes": scope}
GW = lambda scope="*": {"X-Tenant": "hexa", "X-Scopes": "api:audio:*"}
# ── loudness ──────────────────────────────────────────────────────────────────
......
......@@ -21,7 +21,7 @@ import config
from engines import grade as vendored
client = TestClient(APP.app)
GW = {"X-Tenant": "hexa", "X-Scopes": "grade"}
GW = {"X-Tenant": "hexa", "X-Scopes": "api:audio:*"}
def _wav_bytes(freqs=(110, 220), secs=1.5, sr=44100):
......@@ -47,8 +47,9 @@ def test_grade_endpoint_and_cache():
def test_grade_scope_enforced():
# api:audio:features doesn't cover /grade → 403 (path-derived scope).
r = client.post("/grade", files={"file": ("c.wav", _wav_bytes(), "audio/wav")},
headers={"X-Tenant": "hexa", "X-Scopes": "features"})
headers={"X-Tenant": "hexa", "X-Scopes": "api:audio:features"})
assert r.status_code == 403
......
"""The platform scope convention — matcher unit tests + drift guard (#43).
scopes.py is VENDORED byte-for-byte from the platform canonical
(nechapi/_platform/scopes.py) so the gateway and this service enforce the
identical rule. The drift guard fails if the two ever diverge.
"""
from pathlib import Path
import scopes
# ── required-scope derivation (access tracks the URL by construction) ──────────
def test_required_for_derives_from_path():
assert scopes.required_for("audio", "/analyze/emotion") == "api:audio:analyze:emotion"
assert scopes.required_for("audio", "/grade") == "api:audio:grade"
assert scopes.required_for("audio", "/") == "api:audio"
assert scopes.required_for("geo", "/reverse") == "api:geo:reverse"
# ── matching: wildcards open subtrees; non-wildcards are exact ─────────────────
def test_wildcards_open_subtrees():
req = "api:audio:analyze:emotion"
for g in ("*", "api:*", "api:audio:*", "api:audio:analyze:*", "api:audio:analyze:emotion"):
assert scopes.matches(g, req), g
def test_non_wildcard_is_exact_not_prefix():
# granting exactly api:audio does NOT grant its children — only api:audio:* does
assert not scopes.matches("api:audio", "api:audio:grade")
assert scopes.matches("api:audio", "api:audio")
def test_sibling_and_realm_isolation():
assert not scopes.matches("api:audio:features", "api:audio:grade")
assert not scopes.matches("api:*", "admin:tenants") # api:* never grants admin
assert not scopes.matches("admin:*", "api:audio:grade")
assert scopes.matches("*", "admin:anything") # bare * is superadmin
def test_match_any():
granted = {"api:audio:features", "api:audio:grade"}
assert scopes.match_any(granted, "api:audio:grade")
assert not scopes.match_any(granted, "api:audio:analyze:emotion")
# ── validation (the CLI rejects malformed grants on mint) ─────────────────────
def test_validate():
for ok in ("*", "api:*", "api:audio:*", "api:audio:analyze:emotion", "admin:*"):
assert scopes.validate(ok), ok
for bad in ("", "api::emotion", "api:*:emotion", "api:audio:"):
assert not scopes.validate(bad), bad
# ── drift guard: vendored copy must equal the platform canonical ──────────────
def test_vendored_scopes_matches_platform_canonical():
canonical = Path.home() / "Work" / "nechapi" / "_platform" / "scopes.py"
if not canonical.exists():
import pytest
pytest.skip("platform canonical not present (CI/container) — dev-only guard")
here = Path(scopes.__file__)
assert here.read_text() == canonical.read_text(), (
"armada/api/scopes.py drifted from nechapi/_platform/scopes.py — re-vendor it"
)
......@@ -64,7 +64,7 @@ def test_signal_pulls_no_torch():
# ── endpoints ─────────────────────────────────────────────────────────────────
GW = lambda scope: {"X-Tenant": "hexa", "X-Scopes": scope}
GW = lambda scope="*": {"X-Tenant": "hexa", "X-Scopes": "api:audio:*"}
def test_onsets_endpoint_and_cache():
......@@ -98,6 +98,7 @@ def test_analyze_composite(monkeypatch):
def test_signal_scope_enforced():
# api:audio:features doesn't cover /onsets → 403 (path-derived scope).
r = client.post("/onsets", files={"file": ("c.wav", _wav_bytes(_click_track()), "audio/wav")},
headers=GW("features"))
headers={"X-Tenant": "hexa", "X-Scopes": "api:audio:features"})
assert r.status_code == 403
......@@ -18,8 +18,9 @@ client = TestClient(APP.app)
# the local-dev fallback: a bearer matching DOUANIER_DEV_TOKEN (conftest) → wildcard.
DEV = {"Authorization": "Bearer test-secret"}
# how the platform gateway actually calls us: injected identity headers.
GW = {"X-Tenant": "hexa", "X-Scopes": "emotion"}
# how the platform gateway actually calls us: injected identity headers. A broad
# api:audio:* grant authorizes every audio route (scopes are path-derived now).
GW = {"X-Tenant": "hexa", "X-Scopes": "api:audio:*"}
def test_healthz_ok():
......@@ -31,11 +32,11 @@ def test_healthz_ok():
def test_me_via_gateway_headers():
r = client.get("/me", headers={"X-Tenant": "hexa", "X-Scopes": "emotion separate"})
r = client.get("/me", headers={"X-Tenant": "hexa", "X-Scopes": "api:audio:* admin:*"})
assert r.status_code == 200
body = r.json()
assert body["account"] == "hexa"
assert set(body["scopes"]) == {"emotion", "separate"}
assert set(body["scopes"]) == {"api:audio:*", "admin:*"}
def test_me_via_dev_bearer_fallback():
......@@ -60,10 +61,11 @@ def test_emotion_rejects_bad_bearer():
def test_emotion_scope_enforced():
# an authenticated tenant WITHOUT the emotion scope is 403, not 401.
# an authenticated tenant whose grant doesn't cover this route is 403, not 401.
# api:audio:features authorizes /features but NOT /analyze/emotion.
r = client.post("/analyze/emotion",
files={"file": ("c.wav", b"RIFF....", "audio/wav")},
headers={"X-Tenant": "hexa", "X-Scopes": "features"})
headers={"X-Tenant": "hexa", "X-Scopes": "api:audio:features"})
assert r.status_code == 403
......@@ -111,27 +113,27 @@ def _wav_bytes(freqs=(220, 277, 330), secs=1.5):
def test_features_endpoint_and_cache():
files = {"file": ("c.wav", _wav_bytes(), "audio/wav")}
r1 = client.post("/features", files=files, headers={"X-Tenant": "hexa", "X-Scopes": "features"})
r1 = client.post("/features", files=files, headers=GW)
assert r1.status_code == 200
assert r1.headers["X-Douanier-Cache"] == "miss"
body = r1.json()
assert "features" in body and body["engine"] == "light"
assert "spectral_centroid" in body["features"]
r2 = client.post("/features", files={"file": ("c.wav", _wav_bytes(), "audio/wav")},
headers={"X-Tenant": "hexa", "X-Scopes": "features"})
headers=GW)
assert r2.headers["X-Douanier-Cache"] == "hit"
def test_features_scope_enforced():
# api:audio:analyze:emotion authorizes /analyze/emotion but NOT /features.
r = client.post("/features", files={"file": ("c.wav", b"RIFF....", "audio/wav")},
headers={"X-Tenant": "hexa", "X-Scopes": "emotion"})
headers={"X-Tenant": "hexa", "X-Scopes": "api:audio:analyze:emotion"})
assert r.status_code == 403
def test_samples_endpoint_measures_role():
files = {"file": ("c.wav", _wav_bytes(freqs=(60,)), "audio/wav")}
r = client.post("/analyze/samples?name=lead", files=files,
headers={"X-Tenant": "hexa", "X-Scopes": "samples"})
r = client.post("/analyze/samples?name=lead", files=files, headers=GW)
assert r.status_code == 200
body = r.json()
# 60 Hz sine → bass by measurement, despite the "lead" name hint
......@@ -139,7 +141,7 @@ def test_samples_endpoint_measures_role():
def test_separate_is_503_until_gpu():
r = client.post("/separate", headers={"X-Tenant": "hexa", "X-Scopes": "separate"})
r = client.post("/separate", headers=GW)
assert r.status_code == 503
assert r.json()["detail"]["error"] == "compute_unavailable"
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment