Merge feat/playlists-profiles: YouTube playlists + Lidarr profile hardening

- Playlist URLs download each track to per-artist folders (CLI + REST API).
  One playlist = one job; done if >=1 track succeeds ("Downloaded N/M tracks").
- REST API /fetch now routes URL/playlist queries to download jobs.
- Lidarr metadata/quality profiles selected by name with env overrides
  (LIDARR_METADATA_PROFILE/LIDARR_QUALITY_PROFILE), no more position-luck.
This commit is contained in:
2026-06-09 00:13:54 -07:00
11 changed files with 1014 additions and 17 deletions

View File

@@ -0,0 +1,645 @@
# Playlists + Profile Hardening Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** (1) Harden Lidarr profile selection (pick metadata + quality profile by name with env overrides, not by array position). (2) Add YouTube playlist support that downloads each track to its own per-artist folder, via both the CLI and the REST API (one playlist = one job, `done` if ≥1 track succeeds).
**Architecture:** New helpers in the single-file `musicfetch` binary (profile lookup by name; `is_playlist_url`/`expand_playlist`/`download_playlist`/`download_single`; `yt_download`/`act_youtube` return success bools). `server/mf.py` re-exports the new URL helpers; `server/jobs.py` gains callable `done_message` (so a batch can report `N/M`); `server/actions.py` + `server/app.py` route URL/playlist `q` to a download job. Tests import the binary via the existing `server.mf` loader (`musicfetch_core`).
**Tech Stack:** Python 3.10+, stdlib `urllib.parse`, `requests`/`ytmusicapi`/`yt-dlp` (already deps), FastAPI, pytest+monkeypatch. No new deps.
---
## Context for the implementer
Work from `/home/zhering/Documents/musicfetch` on branch `feat/playlists-profiles` (already checked out). The `musicfetch` binary (no `.py` ext) already has, verified at these locations:
- `get_default_metadata_profile_id()` (line ~447): returns `profiles[0]["id"]` — to be replaced.
- `add_artist()` (line ~457): payload hardcodes `"qualityProfileId": 1` (line ~466) and calls `get_default_metadata_profile_id()` (line ~467).
- `yt_download(url_or_query, target_folder, quality, dry_run, hit=None)` (line ~579): builds the yt-dlp cmd, `subprocess.run(cmd)` at the end, returns None. `--no-playlist` is in the cmd.
- `act_youtube(hit, root, quality, dry_run)` (line ~611): builds `music.youtube` URL + per-first-artist folder, calls `yt_download`, returns None.
- `run_yt_dlp_get_metadata(url)` (line ~623), `get_artist_from_metadata(meta)` (line ~635), `handle_url(url, root, quality, dry_run)` (line ~644).
- `is_url(s)` (early), `Hit` dataclass, `_ytm_artists(item)` (in YouTube-search section), module-level `YTMusic` (None if not installed), `subprocess`, `json`, `os`, `requests`, `RequestException`, `dbg`, `err`, `lidarr_get`, `lidarr_post`.
`server/mf.py` re-exports a fixed symbol list + `__all__`; `server/jobs.py` has `run_job(job_id, fn, done_message, fail_message=...)` where `done_message` is currently a str; `server/app.py` `fetch()` treats `q` only as a search term; `server/actions.py` has `perform_fetch`, `started_message`, `done_message`, `failed_message`.
Tests: `import server.mf # noqa: F401` then `import musicfetch_core as mf`; monkeypatch `mf.lidarr_get`, `mf.act_youtube`, `mf.subprocess`, `mf.YTMusic`, and `monkeypatch.setenv`.
Add to the top imports block of `musicfetch` (Task 2): `from urllib.parse import urlparse, parse_qs`.
---
### Task 1: Lidarr profile hardening
**Files:**
- Modify: `musicfetch` (replace `get_default_metadata_profile_id`; add `_profile_id_by_name` and `get_quality_profile_id`; change `add_artist` payload)
- Test: `tests/test_profiles.py`
- [ ] **Step 1: Write the failing test**
Create `tests/test_profiles.py`:
```python
import server.mf # noqa: F401
import musicfetch_core as mf
META = [{"id": 1, "name": "Standard"}, {"id": 2, "name": "None"}, {"id": 3, "name": "OST"}]
QUAL = [{"id": 1, "name": "Any"}, {"id": 2, "name": "Lossless"}]
def test_metadata_profile_default_standard_by_name(monkeypatch):
monkeypatch.delenv("LIDARR_METADATA_PROFILE", raising=False)
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: META)
assert mf.get_default_metadata_profile_id() == 1 # "Standard", not position-luck
def test_metadata_profile_env_override(monkeypatch):
monkeypatch.setenv("LIDARR_METADATA_PROFILE", "OST")
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: META)
assert mf.get_default_metadata_profile_id() == 3
def test_metadata_profile_unknown_name_falls_back_to_first(monkeypatch):
monkeypatch.setenv("LIDARR_METADATA_PROFILE", "Nonexistent")
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: META)
assert mf.get_default_metadata_profile_id() == 1
def test_quality_profile_default_any_by_name(monkeypatch):
monkeypatch.delenv("LIDARR_QUALITY_PROFILE", raising=False)
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: QUAL)
assert mf.get_quality_profile_id() == 1
def test_quality_profile_env_override(monkeypatch):
monkeypatch.setenv("LIDARR_QUALITY_PROFILE", "Lossless")
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: QUAL)
assert mf.get_quality_profile_id() == 2
def test_profile_fetch_error_returns_one(monkeypatch):
def boom(path, timeout=10):
raise mf.RequestException("down")
monkeypatch.setattr(mf, "lidarr_get", boom)
assert mf.get_default_metadata_profile_id() == 1
assert mf.get_quality_profile_id() == 1
```
- [ ] **Step 2: Run test to verify it fails**
Run: `pytest tests/test_profiles.py -v`
Expected: FAIL — `AttributeError: ... 'get_quality_profile_id'` and metadata env tests fail.
- [ ] **Step 3: Implement**
In `musicfetch`, replace `get_default_metadata_profile_id` with:
```python
def _profile_id_by_name(path: str, env_var: str, default_name: str) -> int:
"""Return the id of the profile whose name matches env_var (default
default_name, case-insensitive). Fall back to the first profile, then 1."""
name = os.environ.get(env_var, default_name)
try:
profiles = lidarr_get(path, timeout=10)
except RequestException as e:
dbg(f"{path} fetch failed: {e}")
return 1
if not profiles:
return 1
for p in profiles:
if p.get("name", "").casefold() == name.casefold():
return p["id"]
dbg(f"profile '{name}' not found at {path}; using first ('{profiles[0].get('name')}')")
return profiles[0]["id"]
def get_default_metadata_profile_id() -> int:
return _profile_id_by_name("/api/v1/metadataprofile", "LIDARR_METADATA_PROFILE", "Standard")
def get_quality_profile_id() -> int:
return _profile_id_by_name("/api/v1/qualityprofile", "LIDARR_QUALITY_PROFILE", "Any")
```
In `add_artist`, change the payload line `"qualityProfileId": 1,` to:
```python
"qualityProfileId": get_quality_profile_id(),
```
- [ ] **Step 4: Run test to verify it passes**
Run: `pytest tests/test_profiles.py -v`
Expected: PASS (6 passed)
- [ ] **Step 5: Commit**
```bash
git add musicfetch tests/test_profiles.py
git commit -m "fix(lidarr): select metadata/quality profiles by name with env overrides"
```
---
### Task 2: Playlist core in `musicfetch`
**Files:**
- Modify: `musicfetch` (add `from urllib.parse import urlparse, parse_qs`; add `is_playlist_url`, `_playlist_id`, `expand_playlist`, `download_playlist`, `download_single`; make `yt_download` + `act_youtube` return a success bool; rewrite `handle_url`)
- Test: `tests/test_playlist.py`
- [ ] **Step 1: Write the failing test**
Create `tests/test_playlist.py`:
```python
import server.mf # noqa: F401
import musicfetch_core as mf
# ---- is_playlist_url ----
def test_pure_playlist_url_is_playlist():
assert mf.is_playlist_url("https://music.youtube.com/playlist?list=PLabc") is True
assert mf.is_playlist_url("https://www.youtube.com/playlist?list=PLabc") is True
def test_watch_with_list_is_not_playlist():
assert mf.is_playlist_url("https://www.youtube.com/watch?v=abc&list=PLx") is False
def test_plain_watch_is_not_playlist():
assert mf.is_playlist_url("https://www.youtube.com/watch?v=abc") is False
def test_non_url_is_not_playlist():
assert mf.is_playlist_url("Daft Punk - Discovery") is False
# ---- expand_playlist (yt-dlp fallback path) ----
class _CP:
def __init__(self, stdout):
self.stdout = stdout
self.returncode = 0
def test_expand_playlist_ytdlp_fallback(monkeypatch):
import json as _json
monkeypatch.setattr(mf, "YTMusic", None) # force yt-dlp path
payload = {"title": "My Mix", "entries": [
{"id": "v1", "title": "Song One", "uploader": "Artist A"},
{"id": "v2", "title": "Song Two", "channel": "Artist B"},
{"id": None, "title": "skip"},
]}
monkeypatch.setattr(mf.subprocess, "run",
lambda *a, **k: _CP(_json.dumps(payload)))
title, hits = mf.expand_playlist("https://www.youtube.com/playlist?list=PLx")
assert title == "My Mix"
assert [h.payload["videoId"] for h in hits] == ["v1", "v2"]
assert hits[0].artist == "Artist A"
# ---- download_playlist ----
def test_download_playlist_counts_ok_and_total(monkeypatch):
h1 = mf.Hit(source="youtube", kind="track", title="A", artist="X", payload={"videoId": "1"})
h2 = mf.Hit(source="youtube", kind="track", title="B", artist="Y", payload={"videoId": "2"})
h3 = mf.Hit(source="youtube", kind="track", title="C", artist="Z", payload={"videoId": "3"})
monkeypatch.setattr(mf, "expand_playlist", lambda url: ("PL Title", [h1, h2, h3]))
def fake_act(hit, root, quality, dry_run):
return hit.title != "B" # B "fails"
monkeypatch.setattr(mf, "act_youtube", fake_act)
ok, total, title = mf.download_playlist("u", "/tmp", "best", False)
assert (ok, total, title) == (2, 3, "PL Title")
def test_download_playlist_track_exception_counts_as_failure(monkeypatch):
h1 = mf.Hit(source="youtube", kind="track", title="A", artist="X", payload={"videoId": "1"})
h2 = mf.Hit(source="youtube", kind="track", title="B", artist="Y", payload={"videoId": "2"})
monkeypatch.setattr(mf, "expand_playlist", lambda url: ("T", [h1, h2]))
def fake_act(hit, root, quality, dry_run):
if hit.title == "B":
raise RuntimeError("boom")
return True
monkeypatch.setattr(mf, "act_youtube", fake_act)
ok, total, _ = mf.download_playlist("u", "/tmp", "best", False)
assert (ok, total) == (1, 2)
# ---- yt_download returns success bool ----
def test_yt_download_returns_true_on_zero_exit(monkeypatch):
monkeypatch.setattr(mf.os, "makedirs", lambda *a, **k: None)
monkeypatch.setattr(mf.subprocess, "run", lambda *a, **k: _CP("")) # returncode 0
assert mf.yt_download("u", "/tmp/x", "best", False) is True
def test_yt_download_dry_run_returns_true(monkeypatch):
assert mf.yt_download("u", "/tmp/x", "best", True) is True
```
- [ ] **Step 2: Run test to verify it fails**
Run: `pytest tests/test_playlist.py -v`
Expected: FAIL — `AttributeError: ... 'is_playlist_url'`.
- [ ] **Step 3: Implement**
Add to the top imports block of `musicfetch`:
```python
from urllib.parse import urlparse, parse_qs
```
Make `yt_download` return a success bool. Change its tail (the `if dry_run:` block and the final `subprocess.run`) to:
```python
if dry_run:
print(f"[dry-run] mkdir -p {target_folder}")
print(f"[dry-run] {' '.join(cmd)}")
return True
os.makedirs(target_folder, exist_ok=True)
print(f"Downloading via yt-dlp -> {target_folder}")
return subprocess.run(cmd).returncode == 0
```
Make `act_youtube` return the bool — change its last line `yt_download(url, target, quality, dry_run, hit=hit)` to:
```python
return yt_download(url, target, quality, dry_run, hit=hit)
```
Add the playlist functions (place them in the URL-path section, after `handle_url`'s helpers / near `handle_url`):
```python
def _playlist_id(url: str) -> str:
return parse_qs(urlparse(url).query).get("list", [""])[0]
def is_playlist_url(url: str) -> bool:
"""True for a pure playlist URL (/playlist?list=… or list= without v=).
A watch?v=…&list=… URL is treated as a single track, not a batch."""
if not is_url(url):
return False
parsed = urlparse(url)
qs = parse_qs(parsed.query)
if "/playlist" in parsed.path:
return True
return "list" in qs and "v" not in qs
def expand_playlist(url: str) -> tuple[str, list[Hit]]:
"""Return (playlist_title, [track Hits]). Prefer ytmusicapi; fall back to
yt-dlp --flat-playlist. Returns ("", []) on failure."""
pid = _playlist_id(url)
if YTMusic is not None and pid:
try:
pl = YTMusic().get_playlist(pid, limit=None)
hits = []
for t in pl.get("tracks", []):
vid = t.get("videoId")
if not vid:
continue
alb = t.get("album")
album = alb.get("name", "") if isinstance(alb, dict) else (alb or "")
hits.append(Hit(source="youtube", kind="track", title=t.get("title", ""),
artist=_ytm_artists(t), album=album,
year=str(t.get("year") or ""), payload={"videoId": vid}))
if hits:
return pl.get("title", ""), hits
except Exception as e: # noqa: BLE001
dbg(f"ytmusicapi playlist expand failed: {e}")
try:
result = subprocess.run(["yt-dlp", "--flat-playlist", "-J", url],
capture_output=True, text=True, check=True)
data = json.loads(result.stdout)
except (subprocess.CalledProcessError, json.JSONDecodeError) as e:
err(f"yt-dlp playlist expand failed: {e}")
return "", []
hits = []
for entry in data.get("entries", []):
vid = entry.get("id")
if not vid:
continue
hits.append(Hit(source="youtube", kind="track", title=entry.get("title", ""),
artist=entry.get("uploader") or entry.get("channel") or "",
payload={"videoId": vid}))
return data.get("title", ""), hits
def download_playlist(url: str, root: str, quality: str, dry_run: bool) -> tuple[int, int, str]:
"""Download each playlist track via act_youtube. Returns (ok, total, title)."""
title, hits = expand_playlist(url)
ok = 0
for h in hits:
try:
if act_youtube(h, root, quality, dry_run):
ok += 1
except Exception as e: # noqa: BLE001 — one bad track shouldn't abort the batch
err(f"track failed ({h.title}): {e}")
return ok, len(hits), title
def download_single(url: str, root: str, quality: str, dry_run: bool) -> dict:
"""Download a single URL. Returns {title, artist, ok}."""
meta = run_yt_dlp_get_metadata(url)
artist = get_artist_from_metadata(meta) if meta else "Unknown Artist"
title = (meta or {}).get("title", "")
target = os.path.join(root, artist, "youtube")
ok = yt_download(url, target, quality, dry_run)
return {"title": title, "artist": artist, "ok": ok}
```
Rewrite `handle_url` to route playlists:
```python
def handle_url(url: str, root: str, quality: str, dry_run: bool):
if is_playlist_url(url):
ok, total, title = download_playlist(url, root, quality, dry_run)
label = f" from '{title}'" if title else ""
print(f"Downloaded {ok}/{total} tracks{label}")
return
download_single(url, root, quality, dry_run)
```
- [ ] **Step 4: Run test to verify it passes**
Run: `pytest tests/test_playlist.py -v`
Expected: PASS (9 passed)
- [ ] **Step 5: Full suite + compile**
Run: `pytest -q` (prior 43 + 6 profiles + 9 playlist = 58) and `python3 -m py_compile musicfetch`.
Expected: all green, clean compile.
- [ ] **Step 6: Commit**
```bash
git add musicfetch tests/test_playlist.py
git commit -m "feat(youtube): playlist expansion + per-track download, success bools"
```
---
### Task 3: Re-exports + callable job message
**Files:**
- Modify: `server/mf.py` (re-export new URL helpers)
- Modify: `server/jobs.py` (`run_job` accepts a callable `done_message`)
- Test: `tests/test_jobs.py` (add a callable-message test)
- [ ] **Step 1: Write the failing test**
Append to `tests/test_jobs.py`:
```python
def test_run_job_callable_done_message():
job = jobs.create_job(hit={}, message="m")
jobs.run_job(job.id, lambda: {"ok": 2, "total": 3},
done_message=lambda res: f"{res['ok']}/{res['total']} done")
j = _wait(job.id, "done")
assert j.message == "2/3 done"
```
Also add a re-export check — create `tests/test_mf_url_exports.py`:
```python
import server.mf as smf
def test_url_helpers_reexported():
assert callable(smf.is_url)
assert callable(smf.is_playlist_url)
assert callable(smf.download_playlist)
assert callable(smf.download_single)
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `pytest tests/test_jobs.py::test_run_job_callable_done_message tests/test_mf_url_exports.py -v`
Expected: FAIL (callable message not supported; `smf.is_playlist_url` missing).
- [ ] **Step 3: Implement**
In `server/jobs.py`, inside `run_job`'s `_task`, change the success branch to support a callable:
```python
result = fn()
msg = done_message(result) if callable(done_message) else done_message
_touch(job, status="done", result=result, message=msg)
```
(Update the `run_job` signature/type hint to `done_message` being `str | Callable[[dict], str]`; import `Callable` is already present.)
In `server/mf.py`, add to the re-export assignments and `__all__`:
```python
is_url = _mod.is_url
is_playlist_url = _mod.is_playlist_url
download_playlist = _mod.download_playlist
download_single = _mod.download_single
```
Add those four names to the `__all__` list.
- [ ] **Step 4: Run tests to verify they pass**
Run: `pytest tests/test_jobs.py tests/test_mf_url_exports.py -v`
Expected: PASS.
- [ ] **Step 5: Commit**
```bash
git add server/jobs.py server/mf.py tests/test_jobs.py tests/test_mf_url_exports.py
git commit -m "feat(server): re-export URL helpers; callable job done_message"
```
---
### Task 4: REST API URL/playlist routing
**Files:**
- Modify: `server/actions.py` (add `url_started_message`, `url_done_message`, `playlist_done_message`, `perform_url_fetch`)
- Modify: `server/app.py` (route URL `q` to a download job)
- Test: `tests/test_api_url.py`
- [ ] **Step 1: Write the failing test**
Create `tests/test_api_url.py`:
```python
import time
import pytest
from server import jobs as jobs_mod
@pytest.fixture(autouse=True)
def _clear_jobs():
jobs_mod.JOBS.clear()
yield
jobs_mod.JOBS.clear()
def _wait_done(client, auth, job_id, timeout=2.0):
end = time.time() + timeout
while time.time() < end:
b = client.get(f"/jobs/{job_id}", headers=auth).json()
if b["status"] in ("done", "failed"):
return b
time.sleep(0.01)
raise AssertionError("job never finished")
def test_playlist_url_batch_job(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.download_playlist",
lambda url, root, quality, dry_run: (2, 3, "My Mix"))
r = client.post("/fetch", params={"q": "https://music.youtube.com/playlist?list=PLx"}, headers=auth)
assert r.status_code == 200
body = r.json()
assert body["status"] == "queued"
assert body["hit"]["kind"] == "playlist"
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "done"
assert "2/3" in done["message"]
assert done["result"]["ok"] == 2
def test_playlist_zero_success_fails(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.download_playlist",
lambda url, root, quality, dry_run: (0, 3, "Dead Mix"))
body = client.post("/fetch", params={"q": "https://www.youtube.com/playlist?list=PLy"}, headers=auth).json()
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "failed"
def test_single_video_url_download(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.download_single",
lambda url, root, quality, dry_run: {"title": "Song", "artist": "A", "ok": True})
body = client.post("/fetch", params={"q": "https://music.youtube.com/watch?v=abc"}, headers=auth).json()
assert body["hit"]["kind"] == "track"
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "done"
assert "Song" in done["message"]
def test_search_query_still_works(client, auth, monkeypatch):
from server import mf
hit = mf.Hit(source="youtube", kind="track", title="T", artist="A", payload={"videoId": "x"})
monkeypatch.setattr("server.app.mf.build_combined_hits",
lambda q, limit, yt_first, lidarr_only, yt_only: [hit])
monkeypatch.setattr("server.app.mf.pick", lambda hits, q, ni, yf: hits[0])
monkeypatch.setattr("server.app.actions.perform_fetch",
lambda chosen, hits, quality, root: {"path": "/x", "lidarr_album_id": None})
r = client.post("/fetch", params={"q": "Daft Punk - Discovery"}, headers=auth)
assert r.status_code == 200
assert r.json()["status"] == "queued"
```
- [ ] **Step 2: Run test to verify it fails**
Run: `pytest tests/test_api_url.py -v`
Expected: FAIL (URL `q` currently goes to search → no `download_playlist`/`download_single` calls; kind not "playlist").
- [ ] **Step 3: Implement**
In `server/actions.py`, add:
```python
def url_started_message(kind: str, title: str = "") -> str:
if kind == "playlist":
return (f"Fetching playlist '{title}'. Downloading tracks now."
if title else "Fetching playlist. Downloading tracks now.")
return f"Fetching '{title}'. Downloading now." if title else "Fetching track. Downloading now."
def playlist_done_message(result: dict) -> str:
ok, total = result.get("ok", 0), result.get("total", 0)
failed = total - ok
return f"Downloaded {ok}/{total} tracks" + (f" ({failed} failed)." if failed else ".")
def url_done_message(result: dict) -> str:
title = result.get("title", "")
return f"Downloaded '{title}'." if title else "Download complete."
def perform_url_fetch(url: str, quality: str, root: str) -> dict:
"""Download a URL (playlist → batch, else single). Raises if nothing
downloaded so the job is marked failed."""
if mf.is_playlist_url(url):
ok, total, title = mf.download_playlist(url, root, quality, False)
if ok == 0:
raise RuntimeError(f"No tracks downloaded from playlist '{title}'." if title
else "No tracks downloaded from playlist.")
return {"kind": "playlist", "title": title, "ok": ok, "total": total,
"path": None, "lidarr_album_id": None}
info = mf.download_single(url, root, quality, False)
if not info.get("ok"):
raise RuntimeError("Download failed.")
return {"kind": "track", "title": info["title"], "artist": info["artist"],
"ok": 1, "total": 1, "path": None, "lidarr_album_id": None}
```
In `server/app.py` `fetch()`, add a URL branch BEFORE the search logic (after the `quality` validation; keep the existing `quality not in mf.QUALITY_CHOICES` 422 check above it). Insert:
```python
if mf.is_url(q):
kind = "playlist" if mf.is_playlist_url(q) else "track"
syn = mf.Hit(source="youtube", kind=kind, title="", artist="")
job = jobs.create_job(hit=syn, message=actions.url_started_message(kind))
response = _job_public(job)
done_msg = actions.playlist_done_message if kind == "playlist" else actions.url_done_message
jobs.run_job(
job.id,
lambda: actions.perform_url_fetch(q, quality, ROOT),
done_message=done_msg,
fail_message="Download failed.",
)
return response
```
(The existing `source` validation can stay; it's ignored for URLs. Leave the search path untouched below this branch.)
- [ ] **Step 4: Run test to verify it passes**
Run: `pytest tests/test_api_url.py -v`
Expected: PASS (4 passed)
- [ ] **Step 5: Full suite**
Run: `pytest -q`
Expected: all green (58 + callable/export + 4 api-url ≈ 64).
- [ ] **Step 6: Commit**
```bash
git add server/actions.py server/app.py tests/test_api_url.py
git commit -m "feat(server): route URL/playlist /fetch to download jobs"
```
---
### Task 5: Live verification
**Files:** none (controller-run).
- [ ] **Step 1: Profiles** — read-only confirm name selection against the real Lidarr:
```bash
cd /home/zhering/Documents/musicfetch
env LIDARR_URL=http://10.2.1.16:8686 LIDARR_API_KEY=49cf02acb4c7436b842df2150056d468 \
python3 -c "import server.mf, musicfetch_core as mf; print('meta', mf.get_default_metadata_profile_id(), 'qual', mf.get_quality_profile_id())"
```
Expected: `meta 1 qual 1` (Standard / Any). Then with `LIDARR_METADATA_PROFILE=OST``meta 3`.
- [ ] **Step 2: Playlist (CLI dry-run)** — confirm expansion + per-track routing without downloading. Pick a small real YT Music playlist URL:
```bash
./musicfetch -d "<small-playlist-url>"
```
Expected: prints a `[dry-run] yt-dlp …` line per track, each targeting `/media/music/<artist>/youtube`.
- [ ] **Step 3: Playlist (real, small)** — with user approval, run the API against a 3-5 track playlist:
```bash
fuser -k 6769/tcp 2>/dev/null; sleep 1
env MUSICFETCH_API_KEY=testkey MUSICFETCH_ROOT=/tmp \
python3 -m uvicorn server.app:app --host 127.0.0.1 --port 6769 --log-level warning &
sleep 4
curl -s -X POST 'http://127.0.0.1:6769/fetch?q=<small-playlist-url>' -H 'X-API-Key: testkey'
# poll /jobs/{id} → expect "Downloaded N/M tracks", files under /tmp/<artist>/youtube/
fuser -k 6769/tcp 2>/dev/null
```
---
## Self-Review
**Spec coverage:**
- Profile-by-name + env overrides + add_artist uses quality profile → Task 1. ✅
- `is_playlist_url` (watch?v&list → single) → Task 2. ✅
- `expand_playlist` (ytmusicapi → yt-dlp fallback) → Task 2. ✅
- `download_playlist` per-track via `act_youtube`, ok/total counting, per-track failures caught → Task 2. ✅
- `yt_download`/`act_youtube` success bools → Task 2. ✅
- CLI `handle_url` playlist routing → Task 2. ✅
- Re-exports + callable batch message → Task 3. ✅
- API URL routing, playlist batch job, `done` if ok≥1 else `failed`, single-URL job, Siri messages, search path unchanged → Task 4. ✅
- Live checks (profiles + playlist) → Task 5. ✅
- Out-of-scope (per-track fan-out, resume/dedup) excluded. ✅
**Placeholder scan:** none — all code/tests complete (the only `<…>` are real user-supplied URLs in the manual Task 5 steps).
**Type consistency:** `download_playlist -> (int,int,str)` consumed as `(ok,total,title)` in CLI + `perform_url_fetch`. `download_single -> {title,artist,ok}` consumed in `perform_url_fetch`. `yt_download`/`act_youtube` now return bool; `act_youtube`'s only other caller (`actions.perform_fetch._download_youtube` in the existing search path) ignores the return value — unaffected. `run_job(done_message)` accepts str or `Callable[[dict],str]`; existing search-path callers pass str (unchanged). `_profile_id_by_name(path, env_var, default_name)` used by both profile getters. New `mf.py` exports (`is_url`, `is_playlist_url`, `download_playlist`, `download_single`) match the names used in `server/app.py` and `server/actions.py`.

View File

@@ -19,6 +19,7 @@ from typing import Optional
import requests import requests
from requests.exceptions import RequestException from requests.exceptions import RequestException
from urllib.parse import urlparse, parse_qs
# Optional deps — degrade gracefully if missing. # Optional deps — degrade gracefully if missing.
try: try:
@@ -444,14 +445,30 @@ def get_existing_artist(name: str) -> Optional[dict]:
return None return None
def get_default_metadata_profile_id() -> int: def _profile_id_by_name(path: str, env_var: str, default_name: str) -> int:
"""Return the id of the profile whose name matches env_var (default
default_name, case-insensitive). Fall back to the first profile, then 1."""
name = os.environ.get(env_var, default_name)
try: try:
profiles = lidarr_get("/api/v1/metadataprofile", timeout=10) profiles = lidarr_get(path, timeout=10)
if profiles:
return profiles[0]["id"]
except RequestException as e: except RequestException as e:
dbg(f"metadataprofile fetch failed: {e}") dbg(f"{path} fetch failed: {e}")
return 1 return 1
if not profiles:
return 1
for p in profiles:
if p.get("name", "").casefold() == name.casefold():
return p["id"]
dbg(f"profile '{name}' not found at {path}; using first ('{profiles[0].get('name')}')")
return profiles[0]["id"]
def get_default_metadata_profile_id() -> int:
return _profile_id_by_name("/api/v1/metadataprofile", "LIDARR_METADATA_PROFILE", "Standard")
def get_quality_profile_id() -> int:
return _profile_id_by_name("/api/v1/qualityprofile", "LIDARR_QUALITY_PROFILE", "Any")
def add_artist(meta: dict, root: str, search_all: bool, dry_run: bool) -> Optional[dict]: def add_artist(meta: dict, root: str, search_all: bool, dry_run: bool) -> Optional[dict]:
@@ -463,7 +480,7 @@ def add_artist(meta: dict, root: str, search_all: bool, dry_run: bool) -> Option
payload = { payload = {
"foreignArtistId": foreign_id, "foreignArtistId": foreign_id,
"artistName": name, "artistName": name,
"qualityProfileId": 1, "qualityProfileId": get_quality_profile_id(),
"metadataProfileId": get_default_metadata_profile_id(), "metadataProfileId": get_default_metadata_profile_id(),
"rootFolderPath": root, "rootFolderPath": root,
"monitored": True, "monitored": True,
@@ -602,10 +619,10 @@ def yt_download(url_or_query: str, target_folder: str, quality: str, dry_run: bo
if dry_run: if dry_run:
print(f"[dry-run] mkdir -p {target_folder}") print(f"[dry-run] mkdir -p {target_folder}")
print(f"[dry-run] {' '.join(cmd)}") print(f"[dry-run] {' '.join(cmd)}")
return return True
os.makedirs(target_folder, exist_ok=True) os.makedirs(target_folder, exist_ok=True)
print(f"Downloading via yt-dlp -> {target_folder}") print(f"Downloading via yt-dlp -> {target_folder}")
subprocess.run(cmd) return subprocess.run(cmd).returncode == 0
def act_youtube(hit: Hit, root: str, quality: str, dry_run: bool): def act_youtube(hit: Hit, root: str, quality: str, dry_run: bool):
@@ -614,12 +631,90 @@ def act_youtube(hit: Hit, root: str, quality: str, dry_run: bool):
url = f"https://music.youtube.com/watch?v={vid}" if vid else f"ytsearch1:{hit.artist} {hit.title}" url = f"https://music.youtube.com/watch?v={vid}" if vid else f"ytsearch1:{hit.artist} {hit.title}"
artist_dir = hit.artist.split(",")[0].strip() or "Unknown Artist" artist_dir = hit.artist.split(",")[0].strip() or "Unknown Artist"
target = os.path.join(root, artist_dir, "youtube") target = os.path.join(root, artist_dir, "youtube")
yt_download(url, target, quality, dry_run, hit=hit) return yt_download(url, target, quality, dry_run, hit=hit)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# URL path # URL path
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _playlist_id(url: str) -> str:
return parse_qs(urlparse(url).query).get("list", [""])[0]
def is_playlist_url(url: str) -> bool:
"""True for a pure playlist URL (/playlist?list=… or list= without v=).
A watch?v=…&list=… URL is treated as a single track, not a batch."""
if not is_url(url):
return False
parsed = urlparse(url)
qs = parse_qs(parsed.query)
if "/playlist" in parsed.path:
return True
return "list" in qs and "v" not in qs
def expand_playlist(url: str) -> tuple[str, list[Hit]]:
"""Return (playlist_title, [track Hits]). Prefer ytmusicapi; fall back to
yt-dlp --flat-playlist. Returns ("", []) on failure."""
pid = _playlist_id(url)
if YTMusic is not None and pid:
try:
pl = YTMusic().get_playlist(pid, limit=None)
hits = []
for t in pl.get("tracks", []):
vid = t.get("videoId")
if not vid:
continue
alb = t.get("album")
album = alb.get("name", "") if isinstance(alb, dict) else (alb or "")
hits.append(Hit(source="youtube", kind="track", title=t.get("title", ""),
artist=_ytm_artists(t), album=album,
year=str(t.get("year") or ""), payload={"videoId": vid}))
if hits:
return pl.get("title", ""), hits
except Exception as e: # noqa: BLE001
dbg(f"ytmusicapi playlist expand failed: {e}")
try:
result = subprocess.run(["yt-dlp", "--flat-playlist", "-J", url],
capture_output=True, text=True, check=True)
data = json.loads(result.stdout)
except (subprocess.CalledProcessError, json.JSONDecodeError) as e:
err(f"yt-dlp playlist expand failed: {e}")
return "", []
hits = []
for entry in data.get("entries", []):
vid = entry.get("id")
if not vid:
continue
hits.append(Hit(source="youtube", kind="track", title=entry.get("title", ""),
artist=entry.get("uploader") or entry.get("channel") or "",
payload={"videoId": vid}))
return data.get("title", ""), hits
def download_playlist(url: str, root: str, quality: str, dry_run: bool) -> tuple[int, int, str]:
"""Download each playlist track via act_youtube. Returns (ok, total, title)."""
title, hits = expand_playlist(url)
ok = 0
for h in hits:
try:
if act_youtube(h, root, quality, dry_run):
ok += 1
except Exception as e: # noqa: BLE001 — one bad track shouldn't abort the batch
err(f"track failed ({h.title}): {e}")
return ok, len(hits), title
def download_single(url: str, root: str, quality: str, dry_run: bool) -> dict:
"""Download a single URL. Returns {title, artist, ok}."""
meta = run_yt_dlp_get_metadata(url)
artist = get_artist_from_metadata(meta) if meta else "Unknown Artist"
title = (meta or {}).get("title", "")
target = os.path.join(root, artist, "youtube")
ok = yt_download(url, target, quality, dry_run)
return {"title": title, "artist": artist, "ok": ok}
def run_yt_dlp_get_metadata(url: str) -> Optional[dict]: def run_yt_dlp_get_metadata(url: str) -> Optional[dict]:
try: try:
result = subprocess.run( result = subprocess.run(
@@ -642,10 +737,12 @@ def get_artist_from_metadata(meta: dict) -> str:
def handle_url(url: str, root: str, quality: str, dry_run: bool): def handle_url(url: str, root: str, quality: str, dry_run: bool):
meta = run_yt_dlp_get_metadata(url) if is_playlist_url(url):
artist = get_artist_from_metadata(meta) if meta else "Unknown Artist" ok, total, title = download_playlist(url, root, quality, dry_run)
target = os.path.join(root, artist, "youtube") label = f" from '{title}'" if title else ""
yt_download(url, target, quality, dry_run) print(f"Downloaded {ok}/{total} tracks{label}")
return
download_single(url, root, quality, dry_run)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------

View File

@@ -61,3 +61,38 @@ def perform_fetch(chosen, hits: list, quality: str, root: str) -> dict:
if not ok: if not ok:
raise RuntimeError("Failed to add artist to Lidarr.") raise RuntimeError("Failed to add artist to Lidarr.")
return {"path": None, "lidarr_album_id": None} return {"path": None, "lidarr_album_id": None}
def url_started_message(kind: str, title: str = "") -> str:
if kind == "playlist":
return (f"Fetching playlist '{title}'. Downloading tracks now."
if title else "Fetching playlist. Downloading tracks now.")
return f"Fetching '{title}'. Downloading now." if title else "Fetching track. Downloading now."
def playlist_done_message(result: dict) -> str:
ok, total = result.get("ok", 0), result.get("total", 0)
failed = total - ok
return f"Downloaded {ok}/{total} tracks" + (f" ({failed} failed)." if failed else ".")
def url_done_message(result: dict) -> str:
title = result.get("title", "")
return f"Downloaded '{title}'." if title else "Download complete."
def perform_url_fetch(url: str, quality: str, root: str) -> dict:
"""Download a URL (playlist -> batch, else single). Raises if nothing
downloaded so the job is marked failed."""
if mf.is_playlist_url(url):
ok, total, title = mf.download_playlist(url, root, quality, False)
if ok == 0:
raise RuntimeError(f"No tracks downloaded from playlist '{title}'." if title
else "No tracks downloaded from playlist.")
return {"kind": "playlist", "title": title, "ok": ok, "total": total,
"path": None, "lidarr_album_id": None}
info = mf.download_single(url, root, quality, False)
if not info.get("ok"):
raise RuntimeError("Download failed.")
return {"kind": "track", "title": info.get("title", ""), "artist": info.get("artist", ""),
"ok": 1, "total": 1, "path": None, "lidarr_album_id": None}

View File

@@ -51,6 +51,21 @@ def fetch(q: str = Query(..., min_length=1),
source: str = Query("auto")): source: str = Query("auto")):
if quality not in mf.QUALITY_CHOICES: if quality not in mf.QUALITY_CHOICES:
raise HTTPException(status_code=422, detail=f"Invalid quality '{quality}'.") raise HTTPException(status_code=422, detail=f"Invalid quality '{quality}'.")
if mf.is_url(q):
kind = "playlist" if mf.is_playlist_url(q) else "track"
syn = mf.Hit(source="youtube", kind=kind, title="", artist="")
job = jobs.create_job(hit=syn, message=actions.url_started_message(kind))
response = _job_public(job)
done_msg = actions.playlist_done_message if kind == "playlist" else actions.url_done_message
jobs.run_job(
job.id,
lambda: actions.perform_url_fetch(q, quality, ROOT),
done_message=done_msg,
fail_message="Download failed.",
)
return response
if source not in ("auto", "lidarr", "youtube"): if source not in ("auto", "lidarr", "youtube"):
raise HTTPException(status_code=422, detail=f"Invalid source '{source}'.") raise HTTPException(status_code=422, detail=f"Invalid source '{source}'.")

View File

@@ -48,7 +48,8 @@ def get_job(job_id: str) -> Optional["Job"]:
return JOBS.get(job_id) return JOBS.get(job_id)
def run_job(job_id: str, fn: Callable[[], dict], done_message: str, def run_job(job_id: str, fn: Callable[[], dict],
done_message: "str | Callable[[dict], str]",
fail_message: str = "Something went wrong while fetching.") -> None: fail_message: str = "Something went wrong while fetching.") -> None:
def _task(): def _task():
job = JOBS.get(job_id) job = JOBS.get(job_id)
@@ -57,7 +58,8 @@ def run_job(job_id: str, fn: Callable[[], dict], done_message: str,
_touch(job, status="running") _touch(job, status="running")
try: try:
result = fn() result = fn()
_touch(job, status="done", result=result, message=done_message) msg = done_message(result) if callable(done_message) else done_message
_touch(job, status="done", result=result, message=msg)
except Exception as e: # noqa: BLE001 — record any failure on the job except Exception as e: # noqa: BLE001 — record any failure on the job
_touch(job, status="failed", error=f"{type(e).__name__}: {e}", _touch(job, status="failed", error=f"{type(e).__name__}: {e}",
message=fail_message) message=fail_message)

View File

@@ -24,6 +24,11 @@ act_youtube = _mod.act_youtube
act_lidarr_album = _mod.act_lidarr_album act_lidarr_album = _mod.act_lidarr_album
act_lidarr_artist = _mod.act_lidarr_artist act_lidarr_artist = _mod.act_lidarr_artist
QUALITY_CHOICES = _mod.QUALITY_CHOICES QUALITY_CHOICES = _mod.QUALITY_CHOICES
is_url = _mod.is_url
is_playlist_url = _mod.is_playlist_url
download_playlist = _mod.download_playlist
download_single = _mod.download_single
__all__ = ["Hit", "build_combined_hits", "pick", "act_youtube", __all__ = ["Hit", "build_combined_hits", "pick", "act_youtube",
"act_lidarr_album", "act_lidarr_artist", "QUALITY_CHOICES"] "act_lidarr_album", "act_lidarr_artist", "QUALITY_CHOICES",
"is_url", "is_playlist_url", "download_playlist", "download_single"]

65
tests/test_api_url.py Normal file
View File

@@ -0,0 +1,65 @@
import time
import pytest
from server import jobs as jobs_mod
@pytest.fixture(autouse=True)
def _clear_jobs():
jobs_mod.JOBS.clear()
yield
jobs_mod.JOBS.clear()
def _wait_done(client, auth, job_id, timeout=2.0):
end = time.time() + timeout
while time.time() < end:
b = client.get(f"/jobs/{job_id}", headers=auth).json()
if b["status"] in ("done", "failed"):
return b
time.sleep(0.01)
raise AssertionError("job never finished")
def test_playlist_url_batch_job(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.download_playlist",
lambda url, root, quality, dry_run: (2, 3, "My Mix"))
r = client.post("/fetch", params={"q": "https://music.youtube.com/playlist?list=PLx"}, headers=auth)
assert r.status_code == 200
body = r.json()
assert body["status"] == "queued"
assert body["hit"]["kind"] == "playlist"
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "done"
assert "2/3" in done["message"]
assert done["result"]["ok"] == 2
def test_playlist_zero_success_fails(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.download_playlist",
lambda url, root, quality, dry_run: (0, 3, "Dead Mix"))
body = client.post("/fetch", params={"q": "https://www.youtube.com/playlist?list=PLy"}, headers=auth).json()
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "failed"
def test_single_video_url_download(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.download_single",
lambda url, root, quality, dry_run: {"title": "Song", "artist": "A", "ok": True})
body = client.post("/fetch", params={"q": "https://music.youtube.com/watch?v=abc"}, headers=auth).json()
assert body["hit"]["kind"] == "track"
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "done"
assert "Song" in done["message"]
def test_search_query_still_works(client, auth, monkeypatch):
from server import mf
hit = mf.Hit(source="youtube", kind="track", title="T", artist="A", payload={"videoId": "x"})
monkeypatch.setattr("server.app.mf.build_combined_hits",
lambda q, limit, yt_first, lidarr_only, yt_only: [hit])
monkeypatch.setattr("server.app.mf.pick", lambda hits, q, ni, yf: hits[0])
monkeypatch.setattr("server.app.actions.perform_fetch",
lambda chosen, hits, quality, root: {"path": "/x", "lidarr_album_id": None})
r = client.post("/fetch", params={"q": "Daft Punk - Discovery"}, headers=auth)
assert r.status_code == 200
assert r.json()["status"] == "queued"

View File

@@ -51,5 +51,13 @@ def test_eviction_keeps_within_cap():
jobs.JOBS.clear() jobs.JOBS.clear()
def test_run_job_callable_done_message():
job = jobs.create_job(hit={}, message="m")
jobs.run_job(job.id, lambda: {"ok": 2, "total": 3},
done_message=lambda res: f"{res['ok']}/{res['total']} done")
j = _wait(job.id, "done")
assert j.message == "2/3 done"
def teardown_module(): def teardown_module():
jobs.JOBS.clear() jobs.JOBS.clear()

View File

@@ -0,0 +1,8 @@
import server.mf as smf
def test_url_helpers_reexported():
assert callable(smf.is_url)
assert callable(smf.is_playlist_url)
assert callable(smf.download_playlist)
assert callable(smf.download_single)

74
tests/test_playlist.py Normal file
View File

@@ -0,0 +1,74 @@
import server.mf # noqa: F401
import musicfetch_core as mf
def test_pure_playlist_url_is_playlist():
assert mf.is_playlist_url("https://music.youtube.com/playlist?list=PLabc") is True
assert mf.is_playlist_url("https://www.youtube.com/playlist?list=PLabc") is True
def test_watch_with_list_is_not_playlist():
assert mf.is_playlist_url("https://www.youtube.com/watch?v=abc&list=PLx") is False
def test_plain_watch_is_not_playlist():
assert mf.is_playlist_url("https://www.youtube.com/watch?v=abc") is False
def test_non_url_is_not_playlist():
assert mf.is_playlist_url("Daft Punk - Discovery") is False
class _CP:
def __init__(self, stdout):
self.stdout = stdout
self.returncode = 0
def test_expand_playlist_ytdlp_fallback(monkeypatch):
import json as _json
monkeypatch.setattr(mf, "YTMusic", None)
payload = {"title": "My Mix", "entries": [
{"id": "v1", "title": "Song One", "uploader": "Artist A"},
{"id": "v2", "title": "Song Two", "channel": "Artist B"},
{"id": None, "title": "skip"},
]}
monkeypatch.setattr(mf.subprocess, "run", lambda *a, **k: _CP(_json.dumps(payload)))
title, hits = mf.expand_playlist("https://www.youtube.com/playlist?list=PLx")
assert title == "My Mix"
assert [h.payload["videoId"] for h in hits] == ["v1", "v2"]
assert hits[0].artist == "Artist A"
def test_download_playlist_counts_ok_and_total(monkeypatch):
h1 = mf.Hit(source="youtube", kind="track", title="A", artist="X", payload={"videoId": "1"})
h2 = mf.Hit(source="youtube", kind="track", title="B", artist="Y", payload={"videoId": "2"})
h3 = mf.Hit(source="youtube", kind="track", title="C", artist="Z", payload={"videoId": "3"})
monkeypatch.setattr(mf, "expand_playlist", lambda url: ("PL Title", [h1, h2, h3]))
monkeypatch.setattr(mf, "act_youtube", lambda hit, root, quality, dry_run: hit.title != "B")
ok, total, title = mf.download_playlist("u", "/tmp", "best", False)
assert (ok, total, title) == (2, 3, "PL Title")
def test_download_playlist_track_exception_counts_as_failure(monkeypatch):
h1 = mf.Hit(source="youtube", kind="track", title="A", artist="X", payload={"videoId": "1"})
h2 = mf.Hit(source="youtube", kind="track", title="B", artist="Y", payload={"videoId": "2"})
monkeypatch.setattr(mf, "expand_playlist", lambda url: ("T", [h1, h2]))
def fake_act(hit, root, quality, dry_run):
if hit.title == "B":
raise RuntimeError("boom")
return True
monkeypatch.setattr(mf, "act_youtube", fake_act)
ok, total, _ = mf.download_playlist("u", "/tmp", "best", False)
assert (ok, total) == (1, 2)
def test_yt_download_returns_true_on_zero_exit(monkeypatch):
monkeypatch.setattr(mf.os, "makedirs", lambda *a, **k: None)
monkeypatch.setattr(mf.subprocess, "run", lambda *a, **k: _CP(""))
assert mf.yt_download("u", "/tmp/x", "best", False) is True
def test_yt_download_dry_run_returns_true(monkeypatch):
assert mf.yt_download("u", "/tmp/x", "best", True) is True

43
tests/test_profiles.py Normal file
View File

@@ -0,0 +1,43 @@
import server.mf # noqa: F401
import musicfetch_core as mf
META = [{"id": 1, "name": "Standard"}, {"id": 2, "name": "None"}, {"id": 3, "name": "OST"}]
QUAL = [{"id": 1, "name": "Any"}, {"id": 2, "name": "Lossless"}]
def test_metadata_profile_default_standard_by_name(monkeypatch):
monkeypatch.delenv("LIDARR_METADATA_PROFILE", raising=False)
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: META)
assert mf.get_default_metadata_profile_id() == 1
def test_metadata_profile_env_override(monkeypatch):
monkeypatch.setenv("LIDARR_METADATA_PROFILE", "OST")
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: META)
assert mf.get_default_metadata_profile_id() == 3
def test_metadata_profile_unknown_name_falls_back_to_first(monkeypatch):
monkeypatch.setenv("LIDARR_METADATA_PROFILE", "Nonexistent")
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: META)
assert mf.get_default_metadata_profile_id() == 1
def test_quality_profile_default_any_by_name(monkeypatch):
monkeypatch.delenv("LIDARR_QUALITY_PROFILE", raising=False)
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: QUAL)
assert mf.get_quality_profile_id() == 1
def test_quality_profile_env_override(monkeypatch):
monkeypatch.setenv("LIDARR_QUALITY_PROFILE", "Lossless")
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: QUAL)
assert mf.get_quality_profile_id() == 2
def test_profile_fetch_error_returns_one(monkeypatch):
def boom(path, timeout=10):
raise mf.RequestException("down")
monkeypatch.setattr(mf, "lidarr_get", boom)
assert mf.get_default_metadata_profile_id() == 1
assert mf.get_quality_profile_id() == 1