diff --git a/core/automation/blocks.py b/core/automation/blocks.py index dbbfb2f4..60de924b 100644 --- a/core/automation/blocks.py +++ b/core/automation/blocks.py @@ -273,9 +273,9 @@ ACTIONS: list[dict] = [ {"key": "backfill_count", "type": "number", "label": "Always keep the last N videos from each channel", "default": 10, "min": 0} ]}, {"type": "video_process_youtube_wishlist", "label": "Download YouTube Wishlist", "icon": "download", "scope": "video", - "description": "Grab wished YouTube videos: pushes a batch into the download queue (yt-dlp), organised as a Plex 'TV by date' show (channel/year/date). A completed download leaves the wishlist. Needs the YouTube library folder set on Settings → Downloads. Pair with a schedule to drain continuously.", "available": True, + "description": "Download wished YouTube videos (yt-dlp), organised as a Plex 'TV by date' show (channel/year/date). Queues the WHOLE wishlist — the setting only limits how many download at the same time; each finished one starts the next, so it all drains. A completed download leaves the wishlist. Needs the YouTube library folder set on Settings → Downloads.", "available": True, "config_fields": [ - {"key": "batch_size", "type": "number", "label": "Max downloads to queue per run", "default": 3, "min": 0} + {"key": "max_concurrent", "type": "number", "label": "Max simultaneous downloads", "default": 3, "min": 1} ]}, # Video twins of the music maintenance actions. Distinct action_type (the # system seeder keys on action_type, so a shared key would collide with the diff --git a/core/automation/handlers/video_process_youtube_wishlist.py b/core/automation/handlers/video_process_youtube_wishlist.py index 9ab79730..0eccd501 100644 --- a/core/automation/handlers/video_process_youtube_wishlist.py +++ b/core/automation/handlers/video_process_youtube_wishlist.py @@ -1,20 +1,23 @@ """Automation handler: ``video_process_youtube_wishlist`` action. The drain side of the YouTube fulfillment lane. The watchlist-channels scan keeps the -wishlist fed with new uploads; THIS takes wished YouTube videos and pushes them into the -shared ``video_downloads`` queue, spawning the yt-dlp worker per video. The worker -(``core.video.youtube_download``) downloads → organises (channel/year/date) → archives to -history → removes the video from the wishlist, so a completed grab leaves the wishlist and -won't be re-queued. +wishlist fed; THIS queues wished videos for download and keeps a few flowing at a time. -Polite by default: only a small BATCH is enqueued per run (a big first-time backlog drains -over several scheduled runs rather than spawning hundreds of yt-dlp processes at once), and -videos already in flight (an active ``source='youtube'`` download) are skipped so re-runs +There is NO cap on how much of the wishlist gets processed — it queues the WHOLE thing. +The only limit is how many download SIMULTANEOUSLY (``max_concurrent``, default 3): every +wished video becomes a ``queued`` row in the shared ``video_downloads`` table, the handler +starts up to the limit, and each finished download starts the next (one-out-one-in, in the +worker) so the entire queue drains in a controlled stream. This mirrors the music side's +download worker — a concurrency cap plus a small inter-download delay (handled in the +worker) to avoid yt-dlp 429s — but stays on the isolated video side. + +The worker (``core.video.youtube_download``) downloads → organises (channel/year/date) → +archives to history → removes the video from the wishlist, so a completed grab leaves the +wishlist and won't be re-queued; videos already queued or downloading are skipped so re-runs never double-grab. Shared automation side (may import ``core.video`` / ``api.video``); owns its own progress. -All I/O is injected as seams, so the selection + batching is a pure unit-testable function; -production lazily binds the real DB + the worker spawn. +All I/O is injected as seams, so selection + the pump are pure unit-testable functions. """ from __future__ import annotations @@ -24,21 +27,23 @@ from typing import Any, Callable, Dict, Iterable, List, Optional from core.automation.deps import AutomationDeps -def select_to_enqueue(wanted: List[Dict[str, Any]], active_ids: Iterable, batch_size: int) -> List[Dict[str, Any]]: - """Which wished videos to enqueue now: skip ones already in flight, cap at the batch - size (0 = no cap). Pure.""" - active = {str(x) for x in (active_ids or ()) if x} +def videos_to_enqueue(wanted: List[Dict[str, Any]], already_ids: Iterable) -> List[Dict[str, Any]]: + """Wished videos not already queued or downloading. NO cap — the whole backlog is + queued; concurrency is bounded at start time, not here. Pure.""" + already = {str(x) for x in (already_ids or ()) if x} out: List[Dict[str, Any]] = [] for v in wanted or []: vid = v.get("video_id") - if not vid or str(vid) in active: - continue - out.append(v) - if batch_size and len(out) >= batch_size: - break + if vid and str(vid) not in already: + out.append(v) return out +def slots_free(running: int, max_concurrent: int) -> int: + """How many new downloads may start now given how many are already fetching. Pure.""" + return max(0, int(max_concurrent) - max(0, int(running))) + + # ── production seams ────────────────────────────────────────────────────────── def _default_youtube_root() -> str: from api.video import get_video_db @@ -56,27 +61,34 @@ def _default_active_ids() -> List[Any]: if d.get("source") == "youtube" and d.get("media_id")] +def _default_running_count() -> int: + from api.video import get_video_db + return get_video_db().count_active_youtube_downloads() + + def _default_enqueue(video: Dict[str, Any], root: str) -> Any: - """Create the download row + spawn the yt-dlp worker thread. Returns the row id.""" + """Create a QUEUED download row (no thread spawned here — the pump starts it). Returns + the row id.""" import json - import threading from api.video import get_video_db from core.video.sources import resolve_video_server - from core.video.youtube_download import run_youtube_download - db = get_video_db() - dl_id = db.add_video_download({ + return get_video_db().add_video_download({ "kind": "youtube", "source": "youtube", "media_source": "youtube", "title": video.get("video_title") or video.get("channel_title"), - "media_id": video.get("video_id"), "target_dir": root, "status": "downloading", + "media_id": video.get("video_id"), "target_dir": root, "status": "queued", "year": video.get("published_at"), "poster_url": video.get("thumbnail_url"), "search_ctx": json.dumps({"channel": video.get("channel_title"), "video_title": video.get("video_title"), "published_at": video.get("published_at"), "server_source": resolve_video_server()}), }) - threading.Thread(target=run_youtube_download, args=(dl_id, get_video_db), - daemon=True, name="yt-dl-%s" % dl_id).start() - return dl_id + + +def _default_start_next() -> Any: + """Claim + start the next queued YouTube download (or None). The worker chains the rest.""" + from api.video import get_video_db + from core.video.youtube_download import start_next_queued + return start_next_queued(get_video_db) def auto_video_process_youtube_wishlist( @@ -86,17 +98,21 @@ def auto_video_process_youtube_wishlist( youtube_root: Optional[Callable[[], str]] = None, fetch_wanted: Optional[Callable[[], List[Dict[str, Any]]]] = None, active_ids: Optional[Callable[[], Iterable]] = None, + running_count: Optional[Callable[[], int]] = None, enqueue: Optional[Callable[[Dict[str, Any], str], Any]] = None, + start_next: Optional[Callable[[], Any]] = None, ) -> Dict[str, Any]: - """Enqueue a batch of wished YouTube videos for download. + """Queue the whole YouTube wishlist for download and start up to ``max_concurrent`` now. - Returns ``{'status': 'completed', 'queued': int, 'remaining': int, ...}``.""" + Returns ``{'status': 'completed', 'queued': int, 'started': int, 'running': int, ...}``.""" youtube_root = youtube_root or _default_youtube_root fetch_wanted = fetch_wanted or _default_fetch_wanted active_ids = active_ids or _default_active_ids + running_count = running_count or _default_running_count enqueue = enqueue or _default_enqueue + start_next = start_next or _default_start_next automation_id = config.get('_automation_id') - batch_size = max(0, int(config.get('batch_size', 3) or 0)) + max_concurrent = max(1, int(config.get('max_concurrent', 3) or 3)) try: root = youtube_root() @@ -106,39 +122,44 @@ def auto_video_process_youtube_wishlist( log_line=msg, log_type='error') return {'status': 'error', 'error': msg, '_manages_own_progress': True} - deps.update_progress(automation_id, phase='Checking the YouTube wishlist…', progress=10, - log_line='Looking for new videos to download', log_type='info') + deps.update_progress(automation_id, phase='Checking the YouTube wishlist…', progress=15, + log_line='Queueing new videos for download', log_type='info') wanted = fetch_wanted() or [] - active = list(active_ids() or []) - picks = select_to_enqueue(wanted, active, batch_size) - if not picks: - note = ('All %d wished video(s) are already downloading' % len(active)) if active \ - else 'No wished YouTube videos to download' - deps.update_progress(automation_id, status='finished', progress=100, phase='Complete', - log_line=note, log_type='info') - return {'status': 'completed', 'queued': 0, 'remaining': max(0, len(wanted) - len(active)), - '_manages_own_progress': True} + already = list(active_ids() or []) + new = videos_to_enqueue(wanted, already) queued = 0 - for i, v in enumerate(picks): - deps.update_progress(automation_id, phase='Queueing downloads…', - progress=20 + int(70 * i / max(len(picks), 1)), - log_line="Queued '%s'" % (v.get('video_title') or v.get('video_id')), - log_type='info') + for v in new: try: if enqueue(v, root) is not None: queued += 1 - except Exception: # noqa: BLE001 - one bad enqueue shouldn't stop the batch + except Exception: # noqa: BLE001 - one bad enqueue shouldn't stop the rest deps.update_progress(automation_id, log_type='warning', log_line="Couldn't queue '%s'" % (v.get('video_title') or v.get('video_id'))) - remaining = max(0, len(wanted) - len(active) - queued) - done = 'Queued %d YouTube download(s)' % queued - if remaining: - done += ' · %d more waiting (drains over the next runs)' % remaining + # Fill the concurrency slots now; each finished download starts the next, so the + # whole queue drains on its own from here. + deps.update_progress(automation_id, phase='Starting downloads…', progress=70, + log_line='Queued %d new video(s)' % queued, log_type='info') + started = 0 + for _ in range(slots_free(running_count() or 0, max_concurrent)): + if start_next() is None: + break + started += 1 + + running = (running_count() or 0) + if queued or started: + done = 'Queued %d new · %d downloading now (the rest drain automatically)' % (queued, running) + log_type = 'success' + elif running: + done = '%d already downloading; nothing new to queue' % running + log_type = 'info' + else: + done = 'No wished YouTube videos to download' + log_type = 'info' deps.update_progress(automation_id, status='finished', progress=100, phase='Complete', - log_line=done, log_type='success') - return {'status': 'completed', 'queued': queued, 'remaining': remaining, + log_line=done, log_type=log_type) + return {'status': 'completed', 'queued': queued, 'started': started, 'running': running, '_manages_own_progress': True} except Exception as e: # noqa: BLE001 deps.update_progress(automation_id, status='error', phase='Error', log_line=str(e), log_type='error') diff --git a/core/video/youtube_download.py b/core/video/youtube_download.py index 4438b1cc..446f9b92 100644 --- a/core/video/youtube_download.py +++ b/core/video/youtube_download.py @@ -25,6 +25,8 @@ from __future__ import annotations import json import logging import os +import threading +import time from typing import Any, Callable, Dict, Optional from core.video import organization, youtube_quality @@ -172,13 +174,52 @@ def process_youtube_download( return {"status": "failed", "error": err} +# ── concurrency + pacing (the music side's lesson: cap concurrency AND space starts) ── +# yt-dlp 429s if hammered, so fetch STARTS are spaced ≥ this far apart across all workers. +_DELAY_SECONDS = 3.0 +_pace_lock = threading.Lock() +_last_start = [0.0] + + +def _pace(delay: float) -> None: + """Block until at least ``delay`` seconds after the previous fetch start, then reserve + this start slot. Reserving under the lock (sleeping outside it) staggers concurrent + workers without serialising them past the delay.""" + if delay <= 0: + return + with _pace_lock: + now = time.monotonic() + start_at = max(now, _last_start[0] + delay) if _last_start[0] else now + _last_start[0] = start_at + wait = start_at - time.monotonic() + if wait > 0: + time.sleep(wait) + + +def _spawn_worker(dl_id: Any, db_provider: Callable) -> None: + threading.Thread(target=run_youtube_download, args=(dl_id, db_provider), + daemon=True, name="yt-dl-%s" % dl_id).start() + + +def start_next_queued(db_provider: Callable) -> Any: + """Claim the next queued YouTube download and start it. Returns its id, or None if the + queue is empty. The wishlist pump uses this to fill slots; each finished worker calls it + once (one-out-one-in) so the queue drains continuously at the established concurrency.""" + row = db_provider().claim_next_youtube_queued() + if not row: + return None + _spawn_worker(row["id"], db_provider) + return row["id"] + + # ── production wiring ───────────────────────────────────────────────────────── def run_youtube_download(dl_id: Any, db_provider: Callable) -> None: """Production entry: fetch the row, bind real seams, fulfil it. Called on a worker - thread by the enqueue path; swallows nothing silently into the DB.""" + thread by the pump; on finish it starts the next queued download (one-out-one-in).""" db = db_provider() dl = db.get_video_download(dl_id) if not dl: + start_next_queued(db_provider) # keep the queue moving even on a stale id return profile = youtube_quality.load(db) settings = organization.load(db) @@ -211,14 +252,19 @@ def run_youtube_download(dl_id: Any, db_provider: Callable) -> None: except Exception: # noqa: BLE001 - cookies are optional cookie_opts = None - process_youtube_download( - dl, profile=profile, settings=settings, - update_row=db.update_video_download, archive=_archive, - clear_wishlist=lambda vid: db.remove_youtube_from_wishlist("video", vid), - progress_hook=_progress, cookie_opts=cookie_opts, now=_now) + try: + _pace(_DELAY_SECONDS) # space fetch starts to avoid yt-dlp 429s + process_youtube_download( + dl, profile=profile, settings=settings, + update_row=db.update_video_download, archive=_archive, + clear_wishlist=lambda vid: db.remove_youtube_from_wishlist("video", vid), + progress_hook=_progress, cookie_opts=cookie_opts, now=_now) + finally: + start_next_queued(db_provider) # one out, one in — drain the queue __all__ = [ "youtube_fields_from_download", "plan_destination", "ydl_download_opts", "download_one", "process_youtube_download", "run_youtube_download", + "start_next_queued", ] diff --git a/database/video_database.py b/database/video_database.py index 0d0382aa..dce1afe4 100644 --- a/database/video_database.py +++ b/database/video_database.py @@ -1428,6 +1428,37 @@ class VideoDatabase: finally: conn.close() + def count_active_youtube_downloads(self) -> int: + """How many YouTube downloads are ACTIVELY fetching right now (not queued) — + the concurrency gauge the wishlist pump checks before starting more.""" + conn = self._get_connection() + try: + r = conn.execute("SELECT COUNT(*) AS n FROM video_downloads " + "WHERE source='youtube' AND status='downloading'").fetchone() + return int(r["n"]) if r else 0 + finally: + conn.close() + + def claim_next_youtube_queued(self) -> dict | None: + """Atomically take the oldest QUEUED YouTube download and flip it to + 'downloading', returning the row (or None if none are queued). The + ``WHERE status='queued'`` guard makes the claim race-safe — two workers finishing + at once can't grab the same row.""" + conn = self._get_connection() + try: + row = conn.execute( + "SELECT * FROM video_downloads WHERE source='youtube' AND status='queued' " + "ORDER BY id LIMIT 1").fetchone() + if not row: + return None + cur = conn.execute( + "UPDATE video_downloads SET status='downloading', updated_at=datetime('now') " + "WHERE id=? AND status='queued'", (row["id"],)) + conn.commit() + return dict(row) if cur.rowcount else None # rowcount 0 = lost the race + finally: + conn.close() + def media_tmdb_id(self, kind: str, media_id) -> tuple: """(tmdb_id, imdb_id) for a library movie/show row — used to resolve sidecar / subtitle metadata for an owned re-grab (whose media_id is the library id, not a diff --git a/tests/test_video_process_youtube_wishlist.py b/tests/test_video_process_youtube_wishlist.py index edecdc12..6796ec61 100644 --- a/tests/test_video_process_youtube_wishlist.py +++ b/tests/test_video_process_youtube_wishlist.py @@ -1,6 +1,6 @@ -"""Drain side of the YouTube fulfillment lane: enqueue wished videos into the download -queue in polite batches, skipping in-flight ones. Pure selection + handler with all I/O -injected, plus the DB query that feeds it. +"""Drain side of the YouTube fulfillment lane: queue the WHOLE wishlist (no total cap), +start up to a concurrency limit, and let finished downloads start the next. Pure selection + +pump with all I/O injected, plus the DB queue methods that back it. """ from __future__ import annotations @@ -9,7 +9,8 @@ import pytest from core.automation.handlers.video_process_youtube_wishlist import ( auto_video_process_youtube_wishlist, - select_to_enqueue, + slots_free, + videos_to_enqueue, ) @@ -26,74 +27,108 @@ def _v(vid, title="T", date="2024-01-01"): "thumbnail_url": "/t.jpg", "published_at": date} -# ── pure batching ───────────────────────────────────────────────────────────── -def test_select_skips_inflight_and_caps_batch(): +# ── pure ────────────────────────────────────────────────────────────────────── +def test_videos_to_enqueue_skips_already_queued_no_cap(): wanted = [_v("a"), _v("b"), _v("c"), _v("d")] - picks = select_to_enqueue(wanted, active_ids=["b"], batch_size=2) - assert [p["video_id"] for p in picks] == ["a", "c"] # b skipped (in flight), capped at 2 + out = videos_to_enqueue(wanted, already_ids=["b"]) + assert [v["video_id"] for v in out] == ["a", "c", "d"] # b in flight; rest ALL kept -def test_select_zero_batch_means_no_cap(): - wanted = [_v("a"), _v("b"), _v("c")] - assert len(select_to_enqueue(wanted, [], 0)) == 3 +def test_videos_to_enqueue_drops_idless(): + assert videos_to_enqueue([{"video_title": "no id"}], []) == [] -def test_select_drops_idless(): - assert select_to_enqueue([{"video_title": "no id"}], [], 5) == [] +def test_slots_free(): + assert slots_free(running=0, max_concurrent=3) == 3 + assert slots_free(running=2, max_concurrent=3) == 1 + assert slots_free(running=5, max_concurrent=3) == 0 # over the limit → no new starts -# ── handler ─────────────────────────────────────────────────────────────────── -def _run(wanted, *, active=None, root="/yt", batch=3): - enq = [] +# ── handler: queue everything, start up to the limit ────────────────────────── +def _run(wanted, *, active=None, running=0, root="/yt", max_concurrent=3, start_results=None): + enq, starts = [], {"n": 0} def enqueue(video, r): enq.append((video["video_id"], r)) return len(enq) + # start_next returns an id until the (simulated) queue is exhausted + seq = list(start_results) if start_results is not None else [1] * 999 + + def start_next(): + if starts["n"] < len(seq) and seq[starts["n"]] is not None: + starts["n"] += 1 + return seq[starts["n"] - 1] + return None + deps = _Deps() res = auto_video_process_youtube_wishlist( - {"_automation_id": "a", "batch_size": batch}, deps, + {"_automation_id": "a", "max_concurrent": max_concurrent}, deps, youtube_root=lambda: root, fetch_wanted=lambda: wanted, - active_ids=lambda: list(active or []), enqueue=enqueue) - return res, enq, deps + active_ids=lambda: list(active or []), running_count=lambda: running, + enqueue=enqueue, start_next=start_next) + return res, enq, starts["n"], deps -def test_enqueues_a_batch_and_reports_remaining(): - wanted = [_v("a"), _v("b"), _v("c"), _v("d"), _v("e")] - res, enq, _ = _run(wanted, batch=2) - assert res["status"] == "completed" and res["queued"] == 2 - assert [vid for vid, _ in enq] == ["a", "b"] - assert res["remaining"] == 3 # 5 wanted - 2 queued - assert enq[0][1] == "/yt" # root passed through +def test_queues_entire_wishlist_and_starts_up_to_the_limit(): + wanted = [_v(c) for c in "abcdefgh"] # 8 wished + res, enq, started, _ = _run(wanted, running=0, max_concurrent=3) + assert res["status"] == "completed" + assert res["queued"] == 8 # the WHOLE wishlist is queued (no cap) + assert [vid for vid, _ in enq] == list("abcdefgh") + assert started == 3 and res["started"] == 3 # only 3 start now; rest drain via workers + assert enq[0][1] == "/yt" -def test_inflight_videos_are_not_requeued(): - wanted = [_v("a"), _v("b")] - res, enq, _ = _run(wanted, active=["a", "b"], batch=5) - assert enq == [] and res["queued"] == 0 +def test_does_not_exceed_concurrency_when_some_already_running(): + wanted = [_v(c) for c in "abcde"] + res, enq, started, _ = _run(wanted, running=2, max_concurrent=3) + assert res["queued"] == 5 # still queues everything + assert started == 1 # only 1 free slot (3 - 2 already running) + + +def test_full_pipeline_starts_nothing_new(): + wanted = [_v("a")] + res, enq, started, _ = _run(wanted, active=["a"], running=3, max_concurrent=3) + assert enq == [] # 'a' already in flight → not re-queued + assert started == 0 + + +def test_stops_starting_when_queue_drains_early(): + # only 2 things can actually start even though 3 slots are free + res, enq, started, _ = _run([_v("a"), _v("b")], running=0, max_concurrent=3, + start_results=[10, 11, None]) + assert started == 2 def test_missing_youtube_folder_is_an_error(): - res, enq, deps = _run([_v("a")], root="") + res, enq, started, deps = _run([_v("a")], root="") assert res["status"] == "error" and "library folder" in res["error"] assert enq == [] and any(p.get("status") == "error" for p in deps.progress) -def test_nothing_wanted_is_a_clean_noop(): - res, enq, _ = _run([]) - assert res["status"] == "completed" and res["queued"] == 0 and enq == [] +def test_nothing_wanted_and_empty_queue_is_a_clean_noop(): + res, enq, started, _ = _run([], start_results=[None]) # nothing wanted, queue empty + assert res["status"] == "completed" and res["queued"] == 0 and enq == [] and started == 0 -def test_one_bad_enqueue_does_not_stop_the_batch(): +def test_drains_leftover_queue_even_with_nothing_new_wanted(): + # a prior run queued items; this run adds nothing new but still fills the slots + res, enq, started, _ = _run([], running=0, max_concurrent=3, start_results=[1, 2, 3]) + assert res["queued"] == 0 and enq == [] and started == 3 + + +def test_one_bad_enqueue_does_not_stop_the_rest(): def enqueue(video, r): if video["video_id"] == "a": - raise RuntimeError("queue full") + raise RuntimeError("disk full") return 1 res = auto_video_process_youtube_wishlist( - {"_automation_id": "x", "batch_size": 5}, _Deps(), + {"_automation_id": "x", "max_concurrent": 5}, _Deps(), youtube_root=lambda: "/yt", fetch_wanted=lambda: [_v("a"), _v("b")], - active_ids=lambda: [], enqueue=enqueue) + active_ids=lambda: [], running_count=lambda: 0, enqueue=enqueue, + start_next=lambda: None) assert res["status"] == "completed" and res["queued"] == 1 # b still queued @@ -105,7 +140,7 @@ def test_top_level_error_is_caught(): assert res["status"] == "error" and "db down" in res["error"] -# ── the DB query that feeds it ──────────────────────────────────────────────── +# ── the DB queue methods ────────────────────────────────────────────────────── from database.video_database import VideoDatabase # noqa: E402 @@ -124,12 +159,26 @@ def test_youtube_wishlist_to_download_shape(db): top = rows[0] assert top["channel_id"] == "UC1" and top["channel_title"] == "Cool Channel" assert top["video_title"] == "Second" and top["published_at"] == "2024-05-01" - assert top["thumbnail_url"] == "/2.jpg" -def test_youtube_wishlist_to_download_excludes_movies(db): - db.add_movie_to_wishlist(99, "A Movie") - db.add_videos_to_wishlist({"youtube_id": "UC1", "title": "Ch"}, - [{"youtube_id": "v1", "title": "Vid", "published_at": "2024-01-01"}]) - rows = db.youtube_wishlist_to_download() - assert [r["video_id"] for r in rows] == ["v1"] # the movie isn't a youtube video +def test_count_and_claim_queue(db): + a = db.add_video_download({"kind": "youtube", "source": "youtube", "media_id": "v1", + "title": "A", "status": "queued"}) + db.add_video_download({"kind": "youtube", "source": "youtube", "media_id": "v2", + "title": "B", "status": "queued"}) + assert db.count_active_youtube_downloads() == 0 # nothing fetching yet + + claimed = db.claim_next_youtube_queued() + assert claimed["id"] == a and claimed["media_id"] == "v1" # oldest first + assert db.count_active_youtube_downloads() == 1 # now one is 'downloading' + + db.claim_next_youtube_queued() + assert db.count_active_youtube_downloads() == 2 + assert db.claim_next_youtube_queued() is None # queue empty + + +def test_claim_ignores_non_youtube_and_terminal(db): + db.add_video_download({"kind": "movie", "source": "soulseek", "media_id": "m1", + "title": "Movie", "status": "queued"}) + assert db.claim_next_youtube_queued() is None # soulseek queue isn't ours + assert db.count_active_youtube_downloads() == 0