import logging import os import signal import time from multiprocessing.reduction import ForkingPickler from pathlib import Path from types import SimpleNamespace from typing import Any, cast from unittest.mock import AsyncMock, MagicMock, Mock, patch import pytest from app.library.Events import EventBus, Events from app.library.downloads import Download, NestedLogger, Terminator from app.library.downloads.hooks import HookHandlers from app.library.downloads.pool_manager import PoolManager from app.library.downloads.process_manager import ProcessManager from app.library.downloads.queue_manager import DownloadQueue from app.library.downloads.status_tracker import StatusTracker from app.library.downloads.temp_manager import TempManager from app.library.downloads.video_processor import add_video from app.library.ItemDTO import ItemDTO class CaptureHandler(logging.Handler): def __init__(self) -> None: super().__init__() self.records: list[logging.LogRecord] = [] def emit(self, record: logging.LogRecord) -> None: self.records.append(record) def make_item(id: str = "id1", title: str = "T", url: str = "http://u", folder: str = "f") -> ItemDTO: return ItemDTO(id=id, title=title, url=url, folder=folder) class DummyQueue: def __init__(self) -> None: self.items: list[Any] = [] def put(self, obj: Any) -> None: self.items.append(obj) def get(self, timeout: float | None = None) -> Any: if not self.items: return None return self.items.pop(0) class TestNestedLogger: def test_debug_maps_levels_and_strips_prefix(self) -> None: logger = logging.getLogger("nl_test") logger.setLevel(logging.DEBUG) cap = CaptureHandler() for h in list(logger.handlers): logger.removeHandler(h) logger.addHandler(cap) nl = NestedLogger(logger) nl.debug("[debug] detail") nl.debug("[download] progress") nl.debug("[info] info message") levels = [r.levelno for r in cap.records] assert 2 == levels.count(logging.DEBUG), "Should have 2 debug messages" assert 1 == levels.count(logging.INFO), "Should have 1 info message" msgs = [r.getMessage() for r in cap.records] assert "[debug]" not in msgs[0], "[debug] prefix should be stripped" assert msgs[1] == "[download] progress", "[download] prefix is not stripped by NestedLogger" assert msgs[2] == "info message", "info message should have [info] prefix stripped" class TestDownloadHooks: @pytest.fixture(autouse=True) def cfg_and_bus(self, monkeypatch: pytest.MonkeyPatch): class Cfg: debug = False ytdlp_debug = False max_workers = 1 temp_keep = False temp_disabled = True download_info_expires = 0 @staticmethod def get_instance(): return Cfg monkeypatch.setattr("app.library.downloads.core.Config", Cfg) class EB: @staticmethod def get_instance(): return EB @staticmethod def emit(*_args, **_kwargs): return None monkeypatch.setattr("app.library.downloads.core.EventBus", EB) def test_progress_hook_filters_fields(self) -> None: d = Download(make_item()) q = DummyQueue() hooks = HookHandlers(d.id, cast(Any, q), d.logger, d.debug) payload = { "tmpfilename": "t", "filename": "f", "status": "downloading", "msg": "m", "total_bytes": 10, "total_bytes_estimate": 12, "downloaded_bytes": 5, "speed": 1, "eta": 2, "other": "x", } hooks.progress_hook(payload) assert 1 == len(q.items), "Should have 1 item in queue" ev = q.items[0] assert ev["id"] == d.id, "Event should have correct download ID" assert ev["action"] == "progress", "Action should be 'progress'" assert "other" not in ev, "Non-whitelisted keys should not be included in event" for k in ( "tmpfilename", "filename", "status", "msg", "total_bytes", "total_bytes_estimate", "downloaded_bytes", "speed", "eta", ): assert k in ev, f"Key '{k}' should be in event" def test_post_hooks_pushes_filename(self) -> None: d = Download(make_item()) q = DummyQueue() hooks = HookHandlers(d.id, cast(Any, q), d.logger, d.debug) hooks.post_hook("name.ext") assert 1 == len(q.items), "Should have 1 item when filename is provided" assert q.items[0]["final_name"] == "name.ext", "Filename should match" class TestDownloadStale: @pytest.fixture(autouse=True) def cfg_and_bus(self, monkeypatch: pytest.MonkeyPatch): class Cfg: debug = False ytdlp_debug = False max_workers = 1 temp_keep = False temp_disabled = True download_info_expires = 0 @staticmethod def get_instance(): return Cfg @staticmethod def get_manager(): # Return a mock manager with Queue method mock_manager = MagicMock() mock_manager.Queue = MagicMock(return_value=DummyQueue()) return mock_manager monkeypatch.setattr("app.library.downloads.core.Config", Cfg) class EB: @staticmethod def get_instance(): return EB @staticmethod def emit(*_args, **_kwargs): return None monkeypatch.setattr("app.library.downloads.core.EventBus", EB) def test_is_stale_conditions(self, monkeypatch: pytest.MonkeyPatch) -> None: d = Download(make_item()) d.info.auto_start = False assert d.is_stale() is False, "Download with auto_start disabled should not be stale" d.info.auto_start = True d.started_time = 0 assert d.is_stale() is False, "Download that has not been started should not be stale" d.started_time = 1000 monkeypatch.setattr("time.time", lambda: 1200) assert d.is_stale() is False, "Download running for less than 300 seconds should not be stale" monkeypatch.setattr("time.time", lambda: 1401) d.info.status = "finished" assert d.is_stale() is False, "Download with status 'finished' should not be stale regardless of process state" d.info.status = "error" assert d.is_stale() is False, "Download with status 'error' should not be stale regardless of process state" d.info.status = "cancelled" assert d.is_stale() is False, "Download with status 'cancelled' should not be stale regardless of process state" d.info.status = "downloading" assert d.is_stale() is False, ( "Download with status 'downloading' should not be stale regardless of process state" ) d.info.status = "postprocessing" assert d.is_stale() is False, ( "Download with status 'postprocessing' should not be stale regardless of process state" ) d.info.status = "preparing" assert d.is_stale() is True, ( "Download with status 'preparing' and no running process after 300s should be stale" ) d.info.status = "queued" assert d.is_stale() is True, "Download with status 'queued' and no running process after 300s should be stale" d.info.status = None assert d.is_stale() is True, "Download with no status and no running process after 300s should be stale" @pytest.mark.asyncio async def test_started_time_is_set_in_main_process(self, monkeypatch: pytest.MonkeyPatch) -> None: d = Download(make_item()) # Create a mock process mock_proc = MagicMock() mock_proc.join = MagicMock(return_value=0) # Create a proper mock for create_task that consumes the coroutine created_tasks = [] def mock_create_task(coro, **kwargs): # Close the coroutine to avoid warning coro.close() task = MagicMock() created_tasks.append(task) return task # Mock process manager to avoid actually starting a subprocess with ( patch.object(d._process_manager, "create_process", return_value=mock_proc) as mock_create, patch.object(d._process_manager, "start") as mock_start, patch("asyncio.create_task", side_effect=mock_create_task) as mock_create_task_fn, patch("asyncio.get_running_loop") as mock_loop, ): # Set the mock proc on the process manager d._process_manager.proc = mock_proc # Mock the join to return immediately async def mock_executor(*args): return 0 mock_loop.return_value.run_in_executor = mock_executor # Mock status tracker to prevent actual status updates d._status_tracker = MagicMock() d._status_tracker.final_update = True assert d.started_time == 0, "started_time should be 0 before start() is called" # Call start() - this should set started_time in the main process await d.start() # Verify started_time was set to a non-zero value assert d.started_time > 0, "started_time should be set in main process after start() is called" # Verify process was actually started mock_create.assert_called_once() mock_start.assert_called_once() class TestDownloadFlow: def test_download_pushes_download_skipped_flag(self, monkeypatch: pytest.MonkeyPatch) -> None: class Cfg: debug = False ytdlp_debug = False max_workers = 1 temp_keep = False temp_disabled = True download_info_expires = 3600 @staticmethod def get_instance(): return Cfg monkeypatch.setattr("app.library.downloads.core.Config", Cfg) download = Download( info=make_item(), info_dict={ "id": "test-id", "url": "http://u", "formats": [{"format_id": "18"}], "epoch": int(time.time()), }, ) download.status_queue = cast(Any, DummyQueue()) download._hook_handlers = Mock( progress_hook=Mock(), postprocessor_hook=Mock(), post_hook=Mock(), ) download.info.get_ytdlp_opts = Mock( return_value=Mock( add=Mock( return_value=Mock( get_all=Mock(return_value={"skip_download": True}), ) ) ) ) class FakeYTDLP: def __init__(self, params, enable_custom_outtmpl=False): self.params = params self.enable_custom_outtmpl = enable_custom_outtmpl self._download_retcode = 0 self._interrupted = False def process_ie_result(self, ie_result, download): return ie_result, download monkeypatch.setattr("app.library.downloads.core.YTDLP", FakeYTDLP) download._download() queue = cast(DummyQueue, download.status_queue) assert queue.items[0]["download_skipped"] is True assert queue.items[1]["status"] == "finished" assert queue.items[1]["download_skipped"] is True def test_playlist_extras(self, monkeypatch: pytest.MonkeyPatch) -> None: class Cfg: debug = False ytdlp_debug = False max_workers = 1 temp_keep = False temp_disabled = True download_info_expires = 0 @staticmethod def get_instance(): return Cfg monkeypatch.setattr("app.library.downloads.core.Config", Cfg) item = make_item() item.extras = { "playlist": "Internet Dating Slang", "playlist_title": "Internet Dating Slang", "playlist_index": 1, "playlist_autonumber": 1, "n_entries": 2, } download = Download( info=item, info_dict={ "id": "test-id", "url": "http://u", "title": "Video Title", "formats": [{"format_id": "18"}], "playlist": "NA", "playlist_title": None, "playlist_index": "NA", "playlist_autonumber": "", "n_entries": None, }, ) download.status_queue = cast(Any, DummyQueue()) download._hook_handlers = Mock( progress_hook=Mock(), postprocessor_hook=Mock(), post_hook=Mock(), ) download.info.get_ytdlp_opts = Mock( return_value=Mock(add=Mock(return_value=Mock(get_all=Mock(return_value={})))) ) captured: dict[str, Any] = {} class FakeYTDLP: def __init__(self, params): self.params = params self._download_retcode = 0 self._interrupted = False def process_ie_result(self, ie_result, download): captured["ie_result"] = ie_result return ie_result, download monkeypatch.setattr("app.library.downloads.core.YTDLP", FakeYTDLP) download._download() ie_result = captured["ie_result"] assert ie_result["playlist"] == "Internet Dating Slang" assert ie_result["playlist_title"] == "Internet Dating Slang" assert ie_result["playlist_index"] == 1 assert ie_result["playlist_autonumber"] == 1 assert ie_result["n_entries"] == 2 @pytest.mark.asyncio async def test_download_flow_inline_process(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: class Cfg: debug = False ytdlp_debug = False max_workers = 1 temp_keep = False temp_disabled = True download_info_expires = 3600 @staticmethod def get_instance(): return Cfg @staticmethod def get_manager(): class DummyManager: def Queue(self): return DummyQueue() return DummyManager() monkeypatch.setattr("app.library.downloads.core.Config", Cfg) class EB: @staticmethod def get_instance(): return EB @staticmethod def emit(*_args, **_kwargs): return None monkeypatch.setattr("app.library.downloads.core.EventBus", EB) item = ItemDTO( id="id1", title="T", url="http://u", folder="f", download_dir=str(tmp_path), temp_dir=str(tmp_path), ) download = Download(info=item) def fake_download(): queue = download.status_queue assert queue is not None queue = cast(Any, queue) queue.put( { "id": download.id, "status": "downloading", "downloaded_bytes": 10, "total_bytes": 10, } ) download._status_tracker = StatusTracker( info=download.info, download_id=download.id, download_dir=str(tmp_path), temp_path=None, status_queue=queue, logger=download.logger, debug=False, ) queue.put( { "id": download.id, "status": "finished", "final_name": str(tmp_path / "video.mp4"), } ) queue.put(Terminator()) download._download = fake_download class InlineProcess: def __init__(self, target): self._target = target self.pid = 12345 self.ident = 12345 def start(self): self._target() def join(self): return 0 def is_alive(self): return False def terminate(self): return None def kill(self): return None def close(self): return None def create_process(target): inline_proc = InlineProcess(target) download._process_manager.proc = cast(Any, inline_proc) return download._process_manager.proc def start_process(): assert download._process_manager.proc is not None download._process_manager.proc.start() monkeypatch.setattr(download._process_manager, "create_process", create_process) monkeypatch.setattr(download._process_manager, "start", start_process) await download.start() assert download.info.status == "finished", "Download should finish via inline process" assert download.info.filename == "video.mp4", "Final filename should be set from status update" @pytest.mark.asyncio async def test_live_cancel_drains_final(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: class Cfg: debug = False ytdlp_debug = False max_workers = 1 temp_keep = False temp_disabled = True download_info_expires = 3600 @staticmethod def get_instance(): return Cfg @staticmethod def get_manager(): class DummyManager: def Queue(self): return DummyQueue() return DummyManager() monkeypatch.setattr("app.library.downloads.core.Config", Cfg) class EB: @staticmethod def get_instance(): return EB @staticmethod def emit(*_args, **_kwargs): return None monkeypatch.setattr("app.library.downloads.core.EventBus", EB) item = ItemDTO( id="id-live", title="Live", url="http://u", folder="f", download_dir=str(tmp_path), temp_dir=str(tmp_path), is_live=True, ) download = Download(info=item) final_file = tmp_path / "live.mp4" final_file.write_text("test content") def fake_download(): queue = cast(Any, download.status_queue) queue.put({"id": download.id, "status": "downloading", "downloaded_bytes": 10}) queue.put({"id": download.id, "status": "finished", "final_name": str(final_file)}) queue.put(Terminator()) download._download = fake_download class InlineProcess: def __init__(self, target): self._target = target self.pid = 12345 self.ident = 12345 def start(self): self._target() def join(self): return 0 def is_alive(self): return False def terminate(self): return None def kill(self): return None def close(self): return None def create_process(target): inline_proc = InlineProcess(target) download._process_manager.proc = cast(Any, inline_proc) return download._process_manager.proc def start_process(): assert download._process_manager.proc is not None download._process_manager.proc.start() download._process_manager.cancelled = True def mock_create_task(coro, **_kwargs): coro.close() return MagicMock() monkeypatch.setattr(download._process_manager, "create_process", create_process) monkeypatch.setattr(download._process_manager, "start", start_process) monkeypatch.setattr("asyncio.create_task", mock_create_task) await download.start() assert download.info.status == "finished", "Final live file should win over cancel state" assert download.info.filename == "live.mp4", "Finalized live filename should be preserved" @pytest.mark.asyncio async def test_regular_cancel_skips_drain(self, monkeypatch: pytest.MonkeyPatch) -> None: class Cfg: debug = False ytdlp_debug = False max_workers = 1 temp_keep = False temp_disabled = True download_info_expires = 3600 @staticmethod def get_instance(): return Cfg @staticmethod def get_manager(): class DummyManager: def Queue(self): return DummyQueue() return DummyManager() monkeypatch.setattr("app.library.downloads.core.Config", Cfg) class EB: @staticmethod def get_instance(): return EB @staticmethod def emit(*_args, **_kwargs): return None monkeypatch.setattr("app.library.downloads.core.EventBus", EB) tracker = Mock() tracker.final_update = False tracker.drain_queue = AsyncMock() async def progress_update(): return None tracker.progress_update = progress_update monkeypatch.setattr("app.library.downloads.core.StatusTracker", Mock(return_value=tracker)) monkeypatch.setattr("app.library.downloads.core.HookHandlers", Mock()) download = Download(make_item(id="regular-id")) mock_proc = Mock() mock_proc.join = Mock(return_value=0) def start_process(): download._process_manager.cancelled = True def mock_create_task(coro, **_kwargs): coro.close() return MagicMock() monkeypatch.setattr(download._process_manager, "create_process", Mock(return_value=mock_proc)) monkeypatch.setattr(download._process_manager, "start", start_process) monkeypatch.setattr("asyncio.create_task", mock_create_task) download._process_manager.proc = mock_proc await download.start() tracker.drain_queue.assert_not_awaited() assert download.info.status == "cancelled", "Regular cancels should keep the fast cancel path" class TestDownloadSpawnPickling: def setup_method(self): EventBus._reset_singleton() def test_spawn_pickling_ignores_listener(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: class Cfg: debug = False ytdlp_debug = False max_workers = 1 temp_keep = False temp_disabled = True download_info_expires = 3600 @staticmethod def get_instance(): return Cfg monkeypatch.setattr("app.library.downloads.core.Config", Cfg) bus = EventBus.get_instance() def local_event_handler(_event, _name, **_kwargs): return None bus.subscribe(Events.LOG_INFO, local_event_handler, "local-event-handler") item = ItemDTO( id="id1", title="T", url="http://u", folder="f", download_dir=str(tmp_path), temp_dir=str(tmp_path), ) download = Download(info=item) download.status_queue = cast(Any, DummyQueue()) assert download.status_queue is not None download._status_tracker = StatusTracker( info=item, download_id=download.id, download_dir=str(tmp_path), temp_path=None, status_queue=cast(Any, download.status_queue), logger=download.logger, debug=False, ) state = download.__getstate__() assert state.get("_status_tracker") is None, "StatusTracker should be excluded from pickled state" ForkingPickler.dumps(download._download) class TestTempManager: def test_create_temp_path_disabled(self) -> None: info = make_item() logger = logging.getLogger("test") tm = TempManager(info, "/tmp", temp_disabled=True, temp_keep=False, logger=logger) result = tm.create_temp_path() assert result is None, "Should return None when temp_disabled is True" assert tm.temp_path is None, "temp_path should remain None when disabled" def test_create_temp_path_no_dir(self) -> None: info = make_item() logger = logging.getLogger("test") tm = TempManager(info, None, temp_disabled=False, temp_keep=False, logger=logger) result = tm.create_temp_path() assert result is None, "Should return None when temp_dir is None" assert tm.temp_path is None, "temp_path should remain None when no temp_dir" def test_create_temp_path_creates_directory(self, tmp_path: Path) -> None: info = make_item(id="test123") logger = logging.getLogger("test") tm = TempManager(info, str(tmp_path), temp_disabled=False, temp_keep=False, logger=logger) result = tm.create_temp_path() assert result is not None, "Should return Path when enabled" assert result.exists(), "Temporary directory should be created" assert result.parent == tmp_path, "Temp directory should be created in temp_dir" assert tm.temp_path == result, "temp_path should be set to created path" def test_create_temp_path_uses_consistent_hash(self, tmp_path: Path) -> None: info = make_item(id="test123") logger = logging.getLogger("test") tm1 = TempManager(info, str(tmp_path), temp_disabled=False, temp_keep=False, logger=logger) tm2 = TempManager(info, str(tmp_path), temp_disabled=False, temp_keep=False, logger=logger) path1 = tm1.create_temp_path() path2 = tm2.create_temp_path() assert path1 == path2, "Same download ID should produce same temp path" def test_delete_temp_disabled(self, tmp_path: Path) -> None: info = make_item() logger = logging.getLogger("test") tm = TempManager(info, str(tmp_path), temp_disabled=True, temp_keep=False, logger=logger) tm.temp_path = tmp_path / "test" tm.temp_path.mkdir() tm.delete_temp() assert tm.temp_path.exists(), "Should not delete when temp_disabled is True" def test_delete_temp_keep(self, tmp_path: Path) -> None: info = make_item() logger = logging.getLogger("test") tm = TempManager(info, str(tmp_path), temp_disabled=False, temp_keep=True, logger=logger) tm.temp_path = tmp_path / "test" tm.temp_path.mkdir() tm.delete_temp() assert tm.temp_path.exists(), "Should not delete when temp_keep is True" def test_delete_temp_no_path(self) -> None: info = make_item() logger = logging.getLogger("test") tm = TempManager(info, "/tmp", temp_disabled=False, temp_keep=False, logger=logger) tm.delete_temp() assert tm.temp_path is None, "temp_path should stay unset" def test_delete_temp_keeps_partial_download(self, tmp_path: Path) -> None: info = make_item() info.status = "downloading" info.downloaded_bytes = 1000 logger = logging.getLogger("test") tm = TempManager(info, str(tmp_path), temp_disabled=False, temp_keep=False, logger=logger) tm.temp_path = tmp_path / "test" tm.temp_path.mkdir() tm.delete_temp() assert tm.temp_path.exists(), "Should keep temp dir for partial download" def test_delete_temp_with_bypass(self, tmp_path: Path) -> None: info = make_item() info.status = "downloading" info.downloaded_bytes = 1000 logger = logging.getLogger("test") tm = TempManager(info, str(tmp_path), temp_disabled=False, temp_keep=False, logger=logger) tm.temp_path = tmp_path / "test" tm.temp_path.mkdir() (tm.temp_path / "file.txt").write_text("test") tm.delete_temp(by_pass=True) assert tm.temp_path.exists(), "Directory should still exist with bypass" assert not (tm.temp_path / "file.txt").exists(), "Contents should be deleted with bypass" def test_delete_temp_finished_download(self, tmp_path: Path) -> None: info = make_item() info.status = "finished" logger = logging.getLogger("test") tm = TempManager(info, str(tmp_path), temp_disabled=False, temp_keep=False, logger=logger) tm.temp_path = tmp_path / "test" tm.temp_path.mkdir() tm.delete_temp() assert not tm.temp_path.exists(), "Should delete temp dir for finished download" def test_delete_temp_refuses_to_delete_temp_root(self, tmp_path: Path) -> None: info = make_item() info.status = "finished" logger = logging.getLogger("test") tm = TempManager(info, str(tmp_path), temp_disabled=False, temp_keep=False, logger=logger) tm.temp_path = tmp_path tm.delete_temp() assert tm.temp_path.exists(), "Should refuse to delete temp root directory" class TestProcessManager: def test_create_process(self) -> None: logger = logging.getLogger("test") pm = ProcessManager("test-id", is_live=False, logger=logger) pm.cancel_event.set() def dummy_target(): pass proc = pm.create_process(dummy_target) assert proc is not None, "Should create a process" assert pm.proc is proc, "Should store process reference" assert pm.cancel_event.is_set() is False, "Should clear stale cancel events before starting" assert "download-test-id" == proc.name, "Process name should include download ID" def test_started(self) -> None: logger = logging.getLogger("test") pm = ProcessManager("test-id", is_live=False, logger=logger) assert pm.started() is False, "Should return False when no process created" pm.create_process(lambda: None) assert pm.started() is True, "Should return True after process created" def test_running_no_process(self) -> None: logger = logging.getLogger("test") pm = ProcessManager("test-id", is_live=False, logger=logger) assert pm.running() is False, "Should return False when no process" def test_is_cancelled_default(self) -> None: logger = logging.getLogger("test") pm = ProcessManager("test-id", is_live=False, logger=logger) assert pm.is_cancelled() is False, "Should return False by default" def test_cancel_marks_as_cancelled(self) -> None: logger = logging.getLogger("test") pm = ProcessManager("test-id", is_live=False, logger=logger) pm.proc = Mock() pm.proc.is_alive = Mock(return_value=False) result = pm.cancel() assert pm.is_cancelled() is True, "Should mark as cancelled" def test_cancel_not_started(self) -> None: logger = logging.getLogger("test") pm = ProcessManager("test-id", is_live=False, logger=logger) result = pm.cancel() assert result is False, "Should return False when process not started" assert pm.is_cancelled() is False, "Should not mark as cancelled when not started" def test_kill_not_running(self) -> None: logger = logging.getLogger("test") pm = ProcessManager("test-id", is_live=False, logger=logger) result = pm.kill() assert result is False, "Should return False when process not running" def test_kill_sends_sigusr1_on_posix(self) -> None: if "posix" != os.name: pytest.skip("Test only runs on POSIX systems") logger = logging.getLogger("test") pm = ProcessManager("test-id", is_live=False, logger=logger) pm.proc = Mock() pm.proc.pid = 12345 pm.proc.ident = 67890 pm.proc.is_alive = Mock(side_effect=[True, False, False]) with patch("app.library.downloads.process_manager.os.kill") as mock_kill: result = pm.kill() mock_kill.assert_called_once_with(12345, signal.SIGUSR1) assert result is True, "Should return True when process killed successfully" def test_kill_live_uses_event(self) -> None: logger = logging.getLogger("test") pm_live = ProcessManager("test-id", is_live=True, logger=logger) pm_regular = ProcessManager("test-id", is_live=False, logger=logger) pm_live.proc = Mock() pm_live.proc.pid = 12345 pm_live.proc.ident = 67890 pm_live.proc.is_alive = Mock(return_value=True) pm_regular.proc = Mock() pm_regular.proc.pid = 12346 pm_regular.proc.ident = 67891 pm_regular.proc.is_alive = Mock(return_value=True) with ( patch("app.library.downloads.process_manager.os.kill") as mock_kill, patch( "app.library.downloads.process_manager.wait_for_process_with_timeout", return_value=True ) as mock_wait, ): assert pm_live.kill() is True, "Live downloads should stop via the shared cancel event" assert pm_live.cancel_event.is_set() is True, "Live kill should signal the worker cancel event" mock_kill.assert_not_called() mock_wait.assert_called_once_with(pm_live.proc, 10) if "posix" != os.name: pytest.skip("Regular SIGUSR1 path only runs on POSIX systems") with ( patch("app.library.downloads.process_manager.os.kill") as mock_kill, patch( "app.library.downloads.process_manager.wait_for_process_with_timeout", return_value=True ) as mock_wait, ): assert pm_regular.kill() is True, "Regular downloads should keep SIGUSR1 behavior" mock_kill.assert_called_once_with(12346, signal.SIGUSR1) mock_wait.assert_called_once_with(pm_regular.proc, 5) @pytest.mark.asyncio async def test_close_not_started(self) -> None: logger = logging.getLogger("test") pm = ProcessManager("test-id", is_live=False, logger=logger) result = await pm.close() assert result is False, "Should return False when process not started" @pytest.mark.asyncio async def test_close_during_cancel(self) -> None: logger = logging.getLogger("test") pm = ProcessManager("test-id", is_live=False, logger=logger) pm.proc = Mock() pm.cancel_in_progress = True result = await pm.close() assert result is False, "Should return False when cancellation already in progress" @pytest.mark.asyncio async def test_close_kills_and_joins_process(self) -> None: logger = logging.getLogger("test") pm = ProcessManager("test-id", is_live=False, logger=logger) pm.proc = Mock() pm.proc.ident = 12345 pm.proc.is_alive = Mock(side_effect=[False, False]) pm.proc.join = Mock() pm.proc.close = Mock() result = await pm.close() assert result is True, "Should return True on successful close" assert pm.proc is None, "Process reference should be cleared" class TestStatusTracker: @pytest.fixture def mock_config(self): return { "info": make_item(id="test-id"), "download_id": "test-id", "download_dir": "/downloads", "temp_path": None, "status_queue": DummyQueue(), "logger": logging.getLogger("test"), "debug": False, } def test_init_sets_attributes(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) assert st.id == "test-id", "Should set download ID" assert st.info == mock_config["info"], "Should set info reference" assert st.tmpfilename is None, "Should initialize tmpfilename as None" assert st.final_update is False, "Should initialize final_update as False" @pytest.mark.asyncio async def test_status_ignores_bad_id(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) status = {"id": "wrong-id", "status": "downloading"} await st.process_status_update(status) assert st.info.status != "downloading", "Should not update status for wrong ID" @pytest.mark.asyncio async def test_status_ignores_short(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) status = {"id": "test-id"} await st.process_status_update(status) @pytest.mark.asyncio async def test_process_status_update_sets_status(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) status = {"id": "test-id", "status": "downloading", "downloaded_bytes": 1000} await st.process_status_update(status) assert st.info.status == "downloading", "Should update info status" @pytest.mark.asyncio async def test_status_sets_skipped(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) status = {"id": "test-id", "status": "downloading", "download_skipped": True} await st.process_status_update(status) assert st.info.download_skipped is True, "Should update download_skipped from status queue" @pytest.mark.asyncio async def test_process_status_update_sets_tmpfilename(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) status = {"id": "test-id", "status": "downloading", "tmpfilename": "/tmp/file.part"} await st.process_status_update(status) assert st.tmpfilename == "/tmp/file.part", "Should update tmpfilename" @pytest.mark.asyncio async def test_status_sets_percent(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) status = { "id": "test-id", "status": "downloading", "downloaded_bytes": 50, "total_bytes": 100, } await st.process_status_update(status) assert st.info.downloaded_bytes == 50, "Should set downloaded_bytes" assert st.info.total_bytes == 100, "Should set total_bytes" assert st.info.percent == 50.0, "Should calculate percent correctly" @pytest.mark.asyncio async def test_status_uses_estimate(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) status = { "id": "test-id", "status": "downloading", "downloaded_bytes": 30, "total_bytes_estimate": 100, } await st.process_status_update(status) assert st.info.total_bytes == 100, "Should use total_bytes_estimate when total_bytes not available" assert st.info.percent == 30.0, "Should calculate percent from estimate" @pytest.mark.asyncio async def test_status_percent(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) status = { "id": "test-id", "status": "downloading", "downloaded_bytes": 50, "total_bytes": 100, } await st.process_status_update(status) assert st.info.percent == 50.0, "Should calculate percent correctly with valid total" @pytest.mark.asyncio async def test_status_sets_speed_eta(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) status = {"id": "test-id", "status": "downloading", "speed": 1024000, "eta": 60} await st.process_status_update(status) assert st.info.speed == 1024000, "Should set speed" assert st.info.eta == 60, "Should set eta" @pytest.mark.asyncio async def test_process_status_update_sets_error(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) status = {"id": "test-id", "status": "error", "error": "Download failed"} await st.process_status_update(status) assert st.info.status == "error", "Should set status to error" assert st.info.error == "Download failed", "Should set error message" @pytest.mark.asyncio async def test_process_status_update_sets_final_update(self, tmp_path: Path, mock_config: dict) -> None: test_file = tmp_path / "test.mp4" test_file.write_text("test content") st = StatusTracker(**mock_config) st.download_dir = str(tmp_path) status = {"id": "test-id", "status": "finished", "final_name": str(test_file)} await st.process_status_update(status) assert st.final_update is True, "Should set final_update when final file exists" assert st.info.filename == "test.mp4", "Should set relative filename" @pytest.mark.asyncio async def test_drain_queue_processes_remaining_updates(self, mock_config: dict) -> None: queue = DummyQueue() queue.put({"id": "test-id", "status": "downloading", "downloaded_bytes": 100}) queue.put({"id": "test-id", "status": "downloading", "downloaded_bytes": 200}) queue.put(Terminator()) config = {**mock_config, "status_queue": queue} st = StatusTracker(**config) await st.drain_queue(max_iterations=10) assert st.info.downloaded_bytes == 200, "Should process all queued updates" @pytest.mark.asyncio async def test_drain_queue_stops_on_final_update(self, tmp_path: Path, mock_config: dict) -> None: test_file = tmp_path / "test.mp4" test_file.write_text("test content") queue = DummyQueue() queue.put({"id": "test-id", "status": "finished", "final_name": str(test_file)}) queue.put({"id": "test-id", "status": "downloading", "downloaded_bytes": 999}) config = {**mock_config, "status_queue": queue, "download_dir": str(tmp_path)} st = StatusTracker(**config) await st.drain_queue(max_iterations=10) assert st.final_update is True, "Should stop draining after final update" @pytest.mark.asyncio async def test_drain_queue_skips_invalid(self, mock_config: dict) -> None: queue = DummyQueue() queue.put({"id": "test-id", "status": "downloading"}) queue.put(None) config = {**mock_config, "status_queue": queue} st = StatusTracker(**config) await st.drain_queue(max_iterations=5) assert st.info.status == "downloading", "valid updates should still be processed" def test_cancel_update_task(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) st.update_task = Mock() st.update_task.done = Mock(return_value=False) st.update_task.cancel = Mock() st.cancel_update_task() st.update_task.cancel.assert_called_once() def test_cancel_update_task_noop(self, mock_config: dict) -> None: st = StatusTracker(**mock_config) st.cancel_update_task() assert st.update_task is None, "missing tasks should be ignored" def test_put_terminator_adds_to_queue(self, mock_config: dict) -> None: queue = DummyQueue() config = {**mock_config, "status_queue": queue} st = StatusTracker(**config) st.put_terminator() assert 1 == len(queue.items), "Should add terminator to queue" assert isinstance(queue.items[0], Terminator), "Should add Terminator instance" class TestQueueManager: @staticmethod def _video_queue() -> Mock: async def put(item): return item queue_manager = Mock() queue_manager.config = Mock( download_path="/tmp", temp_path="/tmp", output_template="%(title)s.%(ext)s", output_template_chapter="%(title)s.%(ext)s", prevent_live_premiere=False, ) queue_manager.done.get = AsyncMock(side_effect=KeyError) queue_manager.queue.get = AsyncMock(side_effect=KeyError) queue_manager.done.put = AsyncMock(side_effect=put) queue_manager.queue.put = AsyncMock(side_effect=put) queue_manager.pool.trigger_download = Mock() queue_manager._notify.emit = Mock() return queue_manager @staticmethod def _video_item() -> SimpleNamespace: return SimpleNamespace( extras={}, folder="", preset="default", cookies=None, template=None, cli=[], auto_start=True, ) @pytest.mark.asyncio async def test_live_reextracts(self, monkeypatch: pytest.MonkeyPatch) -> None: seen: list[dict | None] = [] def fake_download(*, info, info_dict, logs): seen.append(info_dict) return SimpleNamespace(info=info, info_dict=info_dict, logs=logs) monkeypatch.setattr("app.library.downloads.video_processor.Download", fake_download) result = await add_video( queue=self._video_queue(), item=self._video_item(), entry={ "id": "live-id", "title": "Live stream", "webpage_url": "https://example.test/live", "is_live": True, "live_status": "is_live", "_ytptube_reextract": True, }, ) assert result == {"status": "ok"} assert seen == [None] @pytest.mark.asyncio async def test_regular_reuses_info(self, monkeypatch: pytest.MonkeyPatch) -> None: seen: list[dict | None] = [] def fake_download(*, info, info_dict, logs): seen.append(info_dict) return SimpleNamespace(info=info, info_dict=info_dict, logs=logs) monkeypatch.setattr("app.library.downloads.video_processor.Download", fake_download) entry = { "id": "video-id", "title": "Video", "webpage_url": "https://example.test/video", "live_status": "not_live", "formats": [{"format_id": "18"}], } result = await add_video(queue=self._video_queue(), item=self._video_item(), entry=entry) assert result == {"status": "ok"} assert seen == [entry] @pytest.mark.asyncio async def test_cleanup_thumbnails(self, tmp_path: Path) -> None: from app.library.downloads.monitors import cleanup_thumbnails queue_manager = object.__new__(DownloadQueue) queue_manager.config = SimpleNamespace(temp_path=str(tmp_path), thumb_sidecar=False) queue_manager.done = Mock() cache_root = tmp_path / "thumbnails" cache_root.mkdir(parents=True, exist_ok=True) keep = cache_root / "keep-id.jpg" drop = cache_root / "drop-id.jpg" keep.write_text("keep") drop.write_text("drop") queue_manager.done.get_by_id = AsyncMock(side_effect=lambda item_id: item_id == "keep-id") await cleanup_thumbnails(queue_manager) assert keep.exists() assert not drop.exists() @pytest.mark.asyncio async def test_cancel_running_live_item_defers_close(self) -> None: queue_manager = object.__new__(DownloadQueue) queue_manager.queue = Mock() queue_manager.done = Mock() queue_manager._notify = Mock() item = Mock() item.info = make_item(id="queued-id") item.is_live = True item.running.return_value = True item.cancel.return_value = True item.close = AsyncMock() queue_manager.queue.get = AsyncMock(return_value=item) status = await DownloadQueue.cancel(queue_manager, [item.info._id]) item.cancel.assert_called_once() item.close.assert_not_awaited() assert status[item.info._id] == "ok", "Running cancel should still report success" @pytest.mark.asyncio async def test_cancel_regular_closes(self) -> None: queue_manager = object.__new__(DownloadQueue) queue_manager.queue = Mock() queue_manager.done = Mock() queue_manager._notify = Mock() item = Mock() item.info = make_item(id="queued-id") item.is_live = False item.running.return_value = True item.cancel.return_value = True item.close = AsyncMock() queue_manager.queue.get = AsyncMock(return_value=item) status = await DownloadQueue.cancel(queue_manager, [item.info._id]) item.cancel.assert_called_once() item.close.assert_awaited_once() assert status[item.info._id] == "ok", "Regular running cancel should still report success" @pytest.mark.asyncio async def test_clear_flushes_history(self) -> None: queue_manager = object.__new__(DownloadQueue) queue_manager.config = Mock(remove_files=False, download_path="/tmp") queue_manager._notify = Mock() item = Mock() item.info = make_item(id="done-id", title="Finished clip") item.info._id = "done-id" item.info.status = "finished" item.info.filename = "clip.mp4" item.info.folder = "" done_store = Mock() done_store.get = AsyncMock(return_value=item) done_store.delete = AsyncMock() done_store._connection = Mock() done_store._connection.flush = AsyncMock() queue_manager.done = done_store status = await DownloadQueue.clear(queue_manager, [item.info._id], remove_file=False) done_store.delete.assert_awaited_once_with(item.info._id) done_store._connection.flush.assert_awaited_once() assert status[item.info._id] == "ok", "Clear should still report success after flushing deletes" @pytest.mark.asyncio async def test_clear_bulk_notifies(self) -> None: queue_manager = object.__new__(DownloadQueue) queue_manager.config = Mock(remove_files=False, download_path="/tmp") queue_manager._notify = Mock() item_one = Mock() item_one.info = make_item(id="done-id-1", title="Finished clip 1") item_one.info._id = "done-id-1" item_one.info.status = "finished" item_two = Mock() item_two.info = make_item(id="done-id-2", title="Finished clip 2") item_two.info._id = "done-id-2" item_two.info.status = "finished" done_store = Mock() done_store.get_many_by_ids = AsyncMock(return_value=[("done-id-1", item_one), ("done-id-2", item_two)]) done_store.bulk_delete = AsyncMock(return_value=2) queue_manager.done = done_store result = await DownloadQueue.clear_bulk(queue_manager, ["done-id-1", "done-id-2"], remove_file=False) assert result == {"deleted": 2} done_store.get_many_by_ids.assert_awaited_once_with(["done-id-1", "done-id-2"]) done_store.bulk_delete.assert_awaited_once_with(["done-id-1", "done-id-2"]) queue_manager._notify.emit.assert_called_once() assert queue_manager._notify.emit.call_args.args[0] == Events.ITEM_BULK_DELETED assert queue_manager._notify.emit.call_args.kwargs["data"] == {"ids": ["done-id-1", "done-id-2"], "count": 2} @pytest.mark.asyncio async def test_clear_status_fetches(self) -> None: queue_manager = object.__new__(DownloadQueue) queue_manager.config = Mock(remove_files=False, download_path="/tmp") queue_manager._notify = Mock() done_store = Mock() done_store.bulk_delete_by_status = AsyncMock(return_value=1) done_store.get_many_by_status = AsyncMock() queue_manager.done = done_store result = await DownloadQueue.clear_by_status(queue_manager, "finished", remove_file=False) assert result == {"deleted": 1} done_store.bulk_delete_by_status.assert_awaited_once_with("finished") done_store.get_many_by_status.assert_not_called() queue_manager._notify.emit.assert_called_once() assert queue_manager._notify.emit.call_args.args[0] == Events.ITEM_BULK_DELETED assert queue_manager._notify.emit.call_args.kwargs["data"] == {"count": 1, "status": "finished"} @pytest.mark.asyncio async def test_clear_status_files_fetch(self) -> None: queue_manager = object.__new__(DownloadQueue) queue_manager.config = Mock(remove_files=True, download_path="/tmp") queue_manager._notify = Mock() item = Mock() item.info = make_item(id="done-id", title="Finished clip") item.info._id = "done-id" item.info.status = "finished" done_store = Mock() done_store.get_many_by_status = AsyncMock(return_value=[("done-id", item)]) queue_manager.done = done_store queue_manager.clear_bulk = AsyncMock(return_value={"deleted": 1}) result = await DownloadQueue.clear_by_status(queue_manager, "finished", remove_file=True) assert result == {"deleted": 1} done_store.get_many_by_status.assert_awaited_once_with("finished") queue_manager.clear_bulk.assert_awaited_once_with(["done-id"], remove_file=True) class TestPoolManager: @pytest.mark.asyncio async def test_cancelled_file_stays_finished(self, monkeypatch: pytest.MonkeyPatch) -> None: emitted_events: list[str] = [] class EB: @staticmethod def get_instance(): return EB @staticmethod def emit(event, **_kwargs): emitted_events.append(event) monkeypatch.setattr("app.library.downloads.pool_manager.EventBus", EB) queue_store = Mock() queue_store.exists = AsyncMock(return_value=True) queue_store.delete = AsyncMock() done_store = Mock() done_store.put = AsyncMock() queue = Mock(queue=queue_store, done=done_store) config = Mock(max_workers=1, max_workers_per_extractor=1, download_path="/tmp") pool = PoolManager(queue=queue, config=config) info = make_item(id="done-id", title="Live clip") info.status = "finished" info.filename = "live.mp4" info.is_archivable = False info.is_archived = False entry = Mock() entry.id = info._id entry.is_live = True entry.info = info entry.start = AsyncMock() entry.close = AsyncMock() entry.is_cancelled.return_value = True await pool._download_file(info._id, entry) assert info.status == "finished", "Finished live downloads should not be rewritten to cancelled" assert Events.ITEM_COMPLETED in emitted_events, "Completed event should be emitted for finalized file" assert Events.ITEM_CANCELLED not in emitted_events, "Cancelled event should not be emitted once finalized" done_store.put.assert_awaited_once_with(entry) @pytest.mark.asyncio async def test_cancelled_regular_file_stays(self, monkeypatch: pytest.MonkeyPatch) -> None: emitted_events: list[str] = [] class EB: @staticmethod def get_instance(): return EB @staticmethod def emit(event, **_kwargs): emitted_events.append(event) monkeypatch.setattr("app.library.downloads.pool_manager.EventBus", EB) queue_store = Mock() queue_store.exists = AsyncMock(return_value=True) queue_store.delete = AsyncMock() done_store = Mock() done_store.put = AsyncMock() queue = Mock(queue=queue_store, done=done_store) config = Mock(max_workers=1, max_workers_per_extractor=1, download_path="/tmp") pool = PoolManager(queue=queue, config=config) info = make_item(id="done-id", title="Regular clip") info.status = "finished" info.filename = "video.mp4" info.is_archivable = False info.is_archived = False entry = Mock() entry.id = info._id entry.is_live = False entry.info = info entry.start = AsyncMock() entry.close = AsyncMock() entry.is_cancelled.return_value = True await pool._download_file(info._id, entry) assert info.status == "cancelled", "Regular cancelled downloads should keep cancelled status" assert Events.ITEM_CANCELLED in emitted_events, "Cancelled event should be emitted for regular downloads" assert Events.ITEM_COMPLETED not in emitted_events, ( "Completed event should remain live-only for cancel finalization" ) done_store.put.assert_awaited_once_with(entry)