From 49d8dcf012607d2da2e017a165ece971ea01fe21 Mon Sep 17 00:00:00 2001 From: DARKZOUL5 Date: Sat, 16 May 2026 18:39:33 +0300 Subject: [PATCH] tests: add skips on youtube bot check --- tests/test_integration_audio.py | 21 +++++++++++++++++++-- tests/test_integration_both.py | 24 +++++++++++++++++++++--- tests/test_integration_video.py | 21 +++++++++++++++++++-- 3 files changed, 59 insertions(+), 7 deletions(-) diff --git a/tests/test_integration_audio.py b/tests/test_integration_audio.py index 5572a6d..2eff3c7 100644 --- a/tests/test_integration_audio.py +++ b/tests/test_integration_audio.py @@ -6,6 +6,7 @@ import pytest from app.core.database.db import Database from app.core.sync.executor import ActionExecutor +from app.core.events.event_bus import EventBus from app.core.sync.service import SyncService @@ -19,6 +20,12 @@ def _require_integration(): pytest.skip("Set TEST_PLAYLIST_URL to enable real download tests") +def _skip_if_bot_check(errors: list[str]) -> None: + msg = "\n".join(errors).lower() + if "sign in to confirm you’re not a bot" in msg or "sign in to confirm you're not a bot" in msg: + pytest.skip("YouTube bot-check blocked download; provide cookies to yt-dlp to run this test reliably") + + @pytest.mark.integration def test_integration_download_audio(tmp_path): _require_integration() @@ -53,11 +60,21 @@ def test_integration_download_audio(tmp_path): subset = [a for a in subset if (a.to_name or "").endswith(".mp3")] assert subset - executor = ActionExecutor(db, concurrency=1) + errors: list[str] = [] + bus = EventBus() + + async def on_failed(payload): + errors.append(str(payload.get("error", ""))) + + bus.subscribe("DownloadFailed", on_failed) + + executor = ActionExecutor(db, concurrency=1, event_bus=bus) import asyncio asyncio.run(executor.execute(subset, cfg)) audio_dir = save_path / "audio" assert audio_dir.exists() - assert any(p.suffix.lower() == ".mp3" for p in audio_dir.glob("*.mp3")) + if not any(p.suffix.lower() == ".mp3" for p in audio_dir.glob("*.mp3")): + _skip_if_bot_check(errors) + assert False diff --git a/tests/test_integration_both.py b/tests/test_integration_both.py index ec4ca42..352a8e7 100644 --- a/tests/test_integration_both.py +++ b/tests/test_integration_both.py @@ -6,6 +6,7 @@ import pytest from app.core.database.db import Database from app.core.sync.executor import ActionExecutor +from app.core.events.event_bus import EventBus from app.core.sync.service import SyncService @@ -19,6 +20,12 @@ def _require_integration(): pytest.skip("Set TEST_PLAYLIST_URL to enable real download tests") +def _skip_if_bot_check(errors: list[str]) -> None: + msg = "\n".join(errors).lower() + if "sign in to confirm you’re not a bot" in msg or "sign in to confirm you're not a bot" in msg: + pytest.skip("YouTube bot-check blocked download; provide cookies to yt-dlp to run this test reliably") + + @pytest.mark.integration def test_integration_download_both(tmp_path): _require_integration() @@ -54,7 +61,15 @@ def test_integration_download_both(tmp_path): mp3 = [a for a in subset if (a.to_name or "").endswith(".mp3")] assert mp4 and mp3 - executor = ActionExecutor(db, concurrency=1) + errors: list[str] = [] + bus = EventBus() + + async def on_failed(payload): + errors.append(str(payload.get("error", ""))) + + bus.subscribe("DownloadFailed", on_failed) + + executor = ActionExecutor(db, concurrency=1, event_bus=bus) import asyncio asyncio.run(executor.execute(mp4 + mp3, cfg)) @@ -62,5 +77,8 @@ def test_integration_download_both(tmp_path): audio_dir = save_path / "audio" video_dir = save_path / "video" assert audio_dir.exists() and video_dir.exists() - assert any(p.suffix.lower() == ".mp3" for p in audio_dir.glob("*.mp3")) - assert any(p.suffix.lower() == ".mp4" for p in video_dir.glob("*.mp4")) + has_mp3 = any(p.suffix.lower() == ".mp3" for p in audio_dir.glob("*.mp3")) + has_mp4 = any(p.suffix.lower() == ".mp4" for p in video_dir.glob("*.mp4")) + if not (has_mp3 and has_mp4): + _skip_if_bot_check(errors) + assert False diff --git a/tests/test_integration_video.py b/tests/test_integration_video.py index d372dc4..fb1dd25 100644 --- a/tests/test_integration_video.py +++ b/tests/test_integration_video.py @@ -6,6 +6,7 @@ import pytest from app.core.database.db import Database from app.core.sync.executor import ActionExecutor +from app.core.events.event_bus import EventBus from app.core.sync.service import SyncService @@ -19,6 +20,12 @@ def _require_integration(): pytest.skip("Set TEST_PLAYLIST_URL to enable real download tests") +def _skip_if_bot_check(errors: list[str]) -> None: + msg = "\n".join(errors).lower() + if "sign in to confirm you’re not a bot" in msg or "sign in to confirm you're not a bot" in msg: + pytest.skip("YouTube bot-check blocked download; provide cookies to yt-dlp to run this test reliably") + + @pytest.mark.integration def test_integration_download_video(tmp_path): _require_integration() @@ -49,11 +56,21 @@ def test_integration_download_video(tmp_path): subset = [a for a in subset if (a.to_name or "").endswith(".mp4")] assert subset - executor = ActionExecutor(db, concurrency=1) + errors: list[str] = [] + bus = EventBus() + + async def on_failed(payload): + errors.append(str(payload.get("error", ""))) + + bus.subscribe("DownloadFailed", on_failed) + + executor = ActionExecutor(db, concurrency=1, event_bus=bus) import asyncio asyncio.run(executor.execute(subset, cfg)) video_dir = save_path / "video" assert video_dir.exists() - assert any(p.suffix.lower() == ".mp4" for p in video_dir.glob("*.mp4")) + if not any(p.suffix.lower() == ".mp4" for p in video_dir.glob("*.mp4")): + _skip_if_bot_check(errors) + assert False