danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor: update test functions to use tmp_path for archive file handling

Browse source
This commit is contained in:
arabcoders 2026-05-08 21:56:41 +03:00
parent a0b50ac351
commit 56dbf6fc2d
9 changed files with 112 additions and 108 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
app/features/tasks/definitions/tests/test_rss_handler.py
View file

@ -1,3 +1,5 @@
from pathlib import Path
import pytest
from app.features.tasks.definitions.handlers.rss import RssGenericHandler
@ -21,6 +23,10 @@ class DummyOpts:
return self._data
def _opts(tmp_path: Path) -> DummyOpts:
return DummyOpts({"download_archive": str(tmp_path / "archive.txt")})
class TestRssHandlerParsing:
"""Test URL parsing for RSS/Atom feeds using the tests() method."""
@ -47,7 +53,7 @@ class TestRssHandlerExtraction:
"""Test RSS feed extraction and parsing."""
@pytest.mark.asyncio
async def test_rss_atom_feed_extraction(self, monkeypatch):
async def test_rss_atom_feed_extraction(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Test extraction from Atom feed."""
atom_feed = """<?xml version="1.0" encoding="UTF-8"?>
<feed xmlns="http://www.w3.org/2005/Atom">
@ -69,7 +75,7 @@ class TestRssHandlerExtraction:
return DummyResponse(atom_feed)
monkeypatch.setattr(RssGenericHandler, "request", staticmethod(fake_request))
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: _opts(tmp_path)) # noqa: ARG005
task = HandleTask(
id=1,
@ -89,7 +95,7 @@ class TestRssHandlerExtraction:
assert result.metadata["entry_count"] == 2
@pytest.mark.asyncio
async def test_rss_feed_extraction(self, monkeypatch):
async def test_rss_feed_extraction(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Test extraction from RSS feed."""
rss_feed = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
@ -113,7 +119,7 @@ class TestRssHandlerExtraction:
return DummyResponse(rss_feed)
monkeypatch.setattr(RssGenericHandler, "request", staticmethod(fake_request))
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: _opts(tmp_path)) # noqa: ARG005
task = HandleTask(
id=1,
@ -133,7 +139,7 @@ class TestRssHandlerExtraction:
assert result.metadata["entry_count"] == 2
@pytest.mark.asyncio
async def test_can_handle(self, monkeypatch):
async def test_can_handle(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Test can_handle method."""
atom_feed = """<?xml version="1.0" encoding="UTF-8"?>
<feed xmlns="http://www.w3.org/2005/Atom">
@ -147,7 +153,7 @@ class TestRssHandlerExtraction:
return DummyResponse(atom_feed)
monkeypatch.setattr(RssGenericHandler, "request", staticmethod(fake_request))
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: _opts(tmp_path)) # noqa: ARG005
task = HandleTask(
id=1,
@ -172,7 +178,7 @@ class TestRssHandlerEdgeCases:
"""Test edge cases in RSS handling."""
@pytest.mark.asyncio
async def test_empty_feed(self, monkeypatch):
async def test_empty_feed(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Test handling of empty feed."""
empty_feed = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
@ -186,7 +192,7 @@ class TestRssHandlerEdgeCases:
return DummyResponse(empty_feed)
monkeypatch.setattr(RssGenericHandler, "request", staticmethod(fake_request))
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: _opts(tmp_path)) # noqa: ARG005
task = HandleTask(
id=1,
@ -202,7 +208,7 @@ class TestRssHandlerEdgeCases:
assert result.metadata["entry_count"] == 0
@pytest.mark.asyncio
async def test_invalid_feed_url(self, monkeypatch):
async def test_invalid_feed_url(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Test handling of invalid feed URL."""
from app.features.tasks.definitions.results import TaskFailure
@ -211,7 +217,7 @@ class TestRssHandlerEdgeCases:
raise Exception(msg)
monkeypatch.setattr(RssGenericHandler, "request", staticmethod(fake_request))
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: _opts(tmp_path)) # noqa: ARG005
task = HandleTask(
id=1,
@ -225,7 +231,7 @@ class TestRssHandlerEdgeCases:
assert isinstance(result, TaskFailure)
@pytest.mark.asyncio
async def test_missing_urls_in_feed(self, monkeypatch):
async def test_missing_urls_in_feed(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Test handling of entries missing URLs."""
feed = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
@ -245,7 +251,7 @@ class TestRssHandlerEdgeCases:
return DummyResponse(feed)
monkeypatch.setattr(RssGenericHandler, "request", staticmethod(fake_request))
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: _opts(tmp_path)) # noqa: ARG005
task = HandleTask(
id=1,

12
app/features/tasks/definitions/tests/test_tver_handler.py
View file

@ -1,3 +1,5 @@
from pathlib import Path
import pytest
from app.features.tasks.definitions.handlers.tver import TverHandler
@ -25,7 +27,7 @@ class DummyOpts:
@pytest.mark.asyncio
async def test_tver_handler_extract(monkeypatch):
async def test_tver_handler_extract(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Test tver handler extraction of episodes from series."""
session_response = {
"result": {
@ -118,8 +120,10 @@ async def test_tver_handler_extract(monkeypatch):
msg = f"Unexpected URL: {url}"
raise RuntimeError(msg)
archive_path = tmp_path / "archive.txt"
monkeypatch.setattr(TverHandler, "request", staticmethod(fake_request))
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda _: DummyOpts({"download_archive": "/tmp/archive"}))
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda _: DummyOpts({"download_archive": str(archive_path)}))
task = HandleTask(id=1, name="Test Tver Series", url="https://tver.jp/series/sr8sb9pnhc", preset="default")
@ -142,7 +146,7 @@ async def test_tver_handler_extract(monkeypatch):
@pytest.mark.parametrize(("url", "should_match"), TverHandler.tests())
def test_parse(url: str, should_match: bool):
def test_parse(url: str, should_match: bool) -> None:
"""Test tver URL parsing."""
result = TverHandler.parse(url)
if should_match:
@ -153,7 +157,7 @@ def test_parse(url: str, should_match: bool):
@pytest.mark.asyncio
async def test_can_handle():
async def test_can_handle() -> None:
"""Test tver handler can_handle method."""
task_valid = HandleTask(id=1, name="Test", url="https://tver.jp/series/sr8sb9pnhc", preset="default")
task_invalid = HandleTask(id=2, name="Test", url="https://youtube.com/watch?v=123", preset="default")

8
app/features/tasks/definitions/tests/test_twitch_handler.py
View file

@ -1,3 +1,5 @@
from pathlib import Path
import pytest
from app.features.tasks.definitions.handlers.twitch import TwitchHandler
@ -22,7 +24,7 @@ class DummyOpts:
@pytest.mark.asyncio
async def test_twitch_handler_inspect(monkeypatch):
async def test_twitch_handler_inspect(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
feed = """
<rss><channel>
<item>
@ -39,8 +41,10 @@ async def test_twitch_handler_inspect(monkeypatch):
async def fake_request(**kwargs): # noqa: ARG001
return DummyResponse(feed)
archive_path = tmp_path / "archive.txt"
monkeypatch.setattr(TwitchHandler, "request", staticmethod(fake_request))
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": str(archive_path)})) # noqa: ARG005
task = HandleTask(
id=1,

8
app/features/tasks/definitions/tests/test_youtube_handler.py
View file

@ -1,3 +1,5 @@
from pathlib import Path
import pytest
from app.features.tasks.definitions.handlers.youtube import YoutubeHandler
@ -22,7 +24,7 @@ class DummyOpts:
@pytest.mark.asyncio
async def test_youtube_handler_inspect(monkeypatch):
async def test_youtube_handler_inspect(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
feed = """
<feed xmlns='http://www.w3.org/2005/Atom' xmlns:yt='http://www.youtube.com/xml/schemas/2015'>
<entry>
@ -41,8 +43,10 @@ async def test_youtube_handler_inspect(monkeypatch):
async def fake_request(**kwargs): # noqa: ARG001
return DummyResponse(feed)
archive_path = tmp_path / "archive.txt"
monkeypatch.setattr(YoutubeHandler, "request", staticmethod(fake_request))
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005
monkeypatch.setattr(HandleTask, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": str(archive_path)})) # noqa: ARG005
task = HandleTask(
id=1,

46
app/features/ytdlp/tests/test_ytdlp_module.py
View file

@ -1,4 +1,5 @@
import importlib
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
import pytest
@ -10,8 +11,12 @@ from app.features.ytdlp.utils import _DATA
from app.features.ytdlp.ytdlp import YTDLP, _ArchiveProxy, ytdlp_options
def _archive_path(tmp_path: Path) -> str:
return str(tmp_path / "archive.txt")
class TestArchiveProxy:
def test_bool_and_falsey_cases(self) -> None:
def test_bool_and_falsey_cases(self, tmp_path: Path) -> None:
# No file path means proxy is falsey and operations return False
p = _ArchiveProxy(file=None)
assert bool(p) is False
@ -19,22 +24,24 @@ class TestArchiveProxy:
assert p.add("id") is False
# Empty item also returns False
p2 = _ArchiveProxy(file="/tmp/archive.txt")
p2 = _ArchiveProxy(file=_archive_path(tmp_path))
assert bool(p2) is True
assert ("" in p2) is False
assert p2.add("") is False
@patch("app.features.ytdlp.archiver.Archiver.get_instance")
def test_delegates_to_archiver(self, mock_get_instance) -> None:
def test_delegates_to_archiver(self, mock_get_instance, tmp_path: Path) -> None:
arch = MagicMock()
mock_get_instance.return_value = arch
p = _ArchiveProxy(file="/tmp/archive.txt")
file = _archive_path(tmp_path)
p = _ArchiveProxy(file=file)
# contains -> read(file, [item]) and check membership
arch.read.return_value = ["abc"]
assert ("abc" in p) is True
arch.read.assert_called_with("/tmp/archive.txt", ["abc"])
arch.read.assert_called_with(file, ["abc"])
arch.read.return_value = []
assert ("xyz" in p) is False
@ -42,7 +49,7 @@ class TestArchiveProxy:
# add -> add(file, [item]) returns boolean
arch.add.return_value = True
assert p.add("abc") is True
arch.add.assert_called_with("/tmp/archive.txt", ["abc"])
arch.add.assert_called_with(file, ["abc"])
arch.add.return_value = False
assert p.add("xyz") is False
@ -96,11 +103,12 @@ class TestYTDLP:
return ytdlp
@patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL.__init__")
def test_init_archive_param(self, mock_super_init) -> None:
def test_init_archive_param(self, mock_super_init, tmp_path: Path) -> None:
"""Test that __init__ removes download_archive before calling super, then restores it."""
mock_super_init.return_value = None
params = {"download_archive": "/tmp/archive.txt", "quiet": True}
file = _archive_path(tmp_path)
params = {"download_archive": file, "quiet": True}
ytdlp = YTDLP(params=params, auto_init=True)
# Set params as it would be after super().__init__
@ -114,13 +122,13 @@ class TestYTDLP:
assert call_kwargs["auto_init"] is True
# Our __init__ code manually sets these after super()
ytdlp.params["download_archive"] = "/tmp/archive.txt"
ytdlp.params["download_archive"] = file
assert ytdlp.params["download_archive"] == "/tmp/archive.txt", "Verify download_archive was restored to params"
assert ytdlp.params["download_archive"] == file, "Verify download_archive was restored to params"
# Verify archive proxy was set up
assert isinstance(ytdlp.archive, _ArchiveProxy)
assert ytdlp.archive._file == "/tmp/archive.txt"
assert ytdlp.archive._file == file
@patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL.__init__")
def test_init_no_archive(self, mock_super_init) -> None:
@ -251,9 +259,9 @@ class TestYTDLP:
# Should not interact with archive
ytdlp.archive.add.assert_not_called()
def test_record_archive_adds_id(self) -> None:
def test_record_archive_adds_id(self, tmp_path: Path) -> None:
"""Test record_download_archive adds the archive ID."""
ytdlp = self._create_ytdlp(params={"download_archive": "/tmp/archive.txt"})
ytdlp = self._create_ytdlp(params={"download_archive": _archive_path(tmp_path)})
ytdlp.write_debug = Mock()
ytdlp.archive = Mock()
ytdlp._make_archive_id = Mock(return_value="youtube test123")
@ -269,9 +277,9 @@ class TestYTDLP:
ytdlp.archive.add.assert_called_once_with("youtube test123")
ytdlp.write_debug.assert_called_with("Adding to archive: youtube test123")
def test_record_archive_old_ids(self) -> None:
def test_record_archive_old_ids(self, tmp_path: Path) -> None:
"""Test record_download_archive adds _old_archive_ids when present."""
ytdlp = self._create_ytdlp(params={"download_archive": "/tmp/archive.txt"})
ytdlp = self._create_ytdlp(params={"download_archive": _archive_path(tmp_path)})
ytdlp.write_debug = Mock()
ytdlp.archive = Mock()
ytdlp._make_archive_id = Mock(return_value="youtube new123")
@ -296,9 +304,9 @@ class TestYTDLP:
# Should not add duplicate (youtube new123 appears only once)
assert calls.count("youtube new123") == 1
def test_record_archive_empty_old_ids(self) -> None:
def test_record_archive_empty_old_ids(self, tmp_path: Path) -> None:
"""Test record_download_archive handles empty or invalid _old_archive_ids."""
ytdlp = self._create_ytdlp(params={"download_archive": "/tmp/archive.txt"})
ytdlp = self._create_ytdlp(params={"download_archive": _archive_path(tmp_path)})
ytdlp.write_debug = Mock()
ytdlp.archive = Mock()
ytdlp._make_archive_id = Mock(return_value="youtube test123")
@ -322,9 +330,9 @@ class TestYTDLP:
ytdlp.record_download_archive(info_dict)
assert ytdlp.archive.add.call_count == 1 # Only main ID
def test_record_archive_empty_id(self) -> None:
def test_record_archive_empty_id(self, tmp_path: Path) -> None:
"""Test record_download_archive returns early when _make_archive_id returns empty."""
ytdlp = self._create_ytdlp(params={"download_archive": "/tmp/archive.txt"})
ytdlp = self._create_ytdlp(params={"download_archive": _archive_path(tmp_path)})
ytdlp.archive = Mock()
ytdlp._make_archive_id = Mock(return_value=None)

12
app/features/ytdlp/tests/test_ytdlp_opts.py
View file

@ -1,5 +1,6 @@
"""Tests for YTDLPOpts class."""
from pathlib import Path
from unittest.mock import Mock, patch
import pytest
@ -912,7 +913,7 @@ class TestYTDLPCli:
@patch("app.features.presets.service.Presets")
@patch("app.features.ytdlp.ytdlp_opts.create_cookies_file")
@patch("app.features.ytdlp.ytdlp_opts.Config")
def test_build_with_cookies_from_user(self, mock_config, mock_create_cookies, mock_presets):
def test_build_with_cookies_from_user(self, mock_config, mock_create_cookies, mock_presets, tmp_path: Path):
"""Test build with cookies from user."""
from app.library.ItemDTO import Item
from app.features.ytdlp.ytdlp_opts import YTDLPCli
@ -920,11 +921,13 @@ class TestYTDLPCli:
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
mock_config_instance.output_template = "%(title)s.%(ext)s"
mock_config_instance.temp_path = str(tmp_path)
mock_config_instance.get_replacers.return_value = {}
mock_config.get_instance.return_value = mock_config_instance
mock_cookie_path = Mock()
mock_cookie_path.__str__ = Mock(return_value="/tmp/cookies.txt")
cookie_path = tmp_path / "cookies.txt"
mock_cookie_path.__str__ = Mock(return_value=str(cookie_path))
mock_create_cookies.return_value = mock_cookie_path
mock_presets.get_instance.return_value.get.return_value = None
@ -936,8 +939,8 @@ class TestYTDLPCli:
mock_create_cookies.assert_called_once_with("session_id=abc123")
assert "--cookies" in command
assert "/tmp/cookies.txt" in command
assert info["merged"]["cookie_file"] == "/tmp/cookies.txt"
assert str(cookie_path) in command
assert info["merged"]["cookie_file"] == str(cookie_path)
@patch("app.features.presets.service.Presets")
@patch("app.features.ytdlp.ytdlp_opts.Config")
@ -950,6 +953,7 @@ class TestYTDLPCli:
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
mock_config_instance.output_template = "%(title)s.%(ext)s"
mock_config_instance.temp_path = "/tmp"
mock_config_instance.get_replacers.return_value = {}
mock_config.get_instance.return_value = mock_config_instance

9
app/tests/test_condition_ignore.py
View file

@ -1,3 +1,4 @@
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock, patch
@ -67,9 +68,9 @@ class TestConditionIgnoreMatching:
class TestConditionIgnorePropagation:
@pytest.mark.asyncio
async def test_add_passes_ignore(self) -> None:
async def test_add_passes_ignore(self, tmp_path: Path) -> None:
queue = Mock()
queue.config = Mock(temp_path="/tmp", ignore_archived_items=False, ytdlp_debug=False)
queue.config = Mock(temp_path=str(tmp_path), ignore_archived_items=False, ytdlp_debug=False)
queue._notify = Mock()
item = Item(
@ -102,9 +103,9 @@ class TestConditionIgnorePropagation:
assert result == {"status": "ok"}
@pytest.mark.asyncio
async def test_add_coerces_ignore(self) -> None:
async def test_add_coerces_ignore(self, tmp_path: Path) -> None:
queue = Mock()
queue.config = Mock(temp_path="/tmp", ignore_archived_items=False, ytdlp_debug=False)
queue.config = Mock(temp_path=str(tmp_path), ignore_archived_items=False, ytdlp_debug=False)
queue._notify = Mock()
item = Item(

27
app/tests/test_itemdto.py
View file

@ -7,6 +7,10 @@ import pytest
from app.library.ItemDTO import Item, ItemDTO
def _archive_path(tmp_path: Path) -> str:
return str(tmp_path / "archive.txt")
class TestItemFormatAndBasics:
@patch("app.features.presets.service.Presets.get_instance")
def test_format_validates_and_normalizes(self, mock_presets_get):
@ -71,8 +75,9 @@ class TestItemFormatAndBasics:
assert json.loads(item.json())["url"] == "https://example.com"
@patch("app.library.ItemDTO.get_archive_id")
def test_item_archive_id_and_is_archived(self, mock_get_id):
def test_item_archive_id_and_is_archived(self, mock_get_id, tmp_path: Path):
mock_get_id.return_value = {"archive_id": "x", "id": "x", "ie_key": "k"}
file = _archive_path(tmp_path)
# get_archive_id
item = Item(url="https://example.com")
@ -85,7 +90,7 @@ class TestItemFormatAndBasics:
):
mock_opts.get_instance.return_value.preset.return_value = mock_opts.get_instance.return_value
mock_opts.get_instance.return_value.add_cli.return_value = mock_opts.get_instance.return_value
mock_opts.get_instance.return_value.get_all.return_value = {"download_archive": "/tmp/archive.txt"}
mock_opts.get_instance.return_value.get_all.return_value = {"download_archive": file}
mock_read.return_value = ["x"]
assert item.is_archived() is True
@ -95,30 +100,32 @@ class TestItemDTO:
@patch("app.library.ItemDTO.get_archive_id")
@patch("app.library.ItemDTO.YTDLPOpts")
@patch("app.library.ItemDTO.archive_read")
def test_post_init_sets_archive_flags(self, mock_read, mock_opts, mock_get_id):
def test_post_init_sets_archive_flags(self, mock_read, mock_opts, mock_get_id, tmp_path: Path):
# Setup archive id and archive file
mock_get_id.return_value = {"archive_id": "arch", "id": "arch", "ie_key": "YT"}
mock_opts.get_instance.return_value.preset.return_value = mock_opts.get_instance.return_value
mock_opts.get_instance.return_value.add_cli.return_value = mock_opts.get_instance.return_value
mock_opts.get_instance.return_value.get_all.return_value = {"download_archive": "/tmp/a.txt"}
file = _archive_path(tmp_path)
mock_opts.get_instance.return_value.get_all.return_value = {"download_archive": file}
mock_read.return_value = ["arch"]
dto = ItemDTO(id="vid", title="t", url="u", folder="f")
assert dto.archive_id == "arch"
assert dto._archive_file == "/tmp/a.txt"
assert dto._archive_file == file
assert dto.is_archivable is True
assert dto.is_archived is True
@patch("app.library.ItemDTO.get_archive_id")
@patch("app.library.ItemDTO.YTDLPOpts")
@patch("app.library.ItemDTO.archive_read")
def test_post_init_keeps_skipped(self, mock_read, mock_opts, mock_get_id):
def test_post_init_keeps_skipped(self, mock_read, mock_opts, mock_get_id, tmp_path: Path):
mock_get_id.return_value = {"archive_id": "arch", "id": "arch", "ie_key": "YT"}
mock_opts.get_instance.return_value.preset.return_value = mock_opts.get_instance.return_value
mock_opts.get_instance.return_value.add_cli.return_value = mock_opts.get_instance.return_value
file = _archive_path(tmp_path)
mock_opts.get_instance.return_value.get_all.return_value = {
"download_archive": "/tmp/a.txt",
"download_archive": file,
"skip_download": True,
}
mock_read.return_value = ["arch"]
@ -128,11 +135,11 @@ class TestItemDTO:
assert dto.download_skipped is False
@patch("app.library.ItemDTO.archive_read")
def test_serialize_archives_finished(self, mock_read):
def test_serialize_archives_finished(self, mock_read, tmp_path: Path):
# Given a finished item with archive info
dto = ItemDTO(id="vid", title="t", url="u", folder="f")
dto.archive_id = "arch"
dto._archive_file = "/tmp/a.txt"
dto._archive_file = _archive_path(tmp_path)
dto.status = "finished"
mock_read.return_value = ["arch"]
@ -186,7 +193,7 @@ class TestItemDTO:
# Set up to allow add
dto.archive_id = "arch"
dto._archive_file = "/tmp/a.txt"
dto._archive_file = "var/tests/archive.txt"
dto.is_archivable = True
dto.is_archived = False

68
app/tests/test_nfo_maker.py
View file

@ -4,18 +4,8 @@ from pathlib import Path
from app.yt_dlp_plugins.postprocessor.nfo_maker import NFOMakerPP
TEST_DIR = Path("var/tmp/tests_nfo")
TEST_DIR.mkdir(parents=True, exist_ok=True)
def _clean_file(path: Path) -> None:
try:
path.unlink()
except FileNotFoundError:
pass
def sample_info_tv() -> dict:
def sample_info_tv(path: Path) -> dict:
return {
"id": "abc123",
"title": "Test Show - S01E02",
@ -25,11 +15,11 @@ def sample_info_tv() -> dict:
"episode_number": 2,
"description": "First line.\n#tag\n00:01:23 Intro",
"extractor": "youtube",
"filename": str(TEST_DIR / "test_show_s01e02.mkv"),
"filename": str(path),
}
def sample_info_movie() -> dict:
def sample_info_movie(path: Path) -> dict:
return {
"id": "mov001",
"title": "Test Movie",
@ -38,60 +28,46 @@ def sample_info_movie() -> dict:
"duration": 7200,
"thumbnail": "http://example.com/thumb.jpg",
"webpage_url": "http://example.com/trailer",
"filename": str(TEST_DIR / "test_movie.mp4"),
"filename": str(path),
}
def test_generate_nfo_tv_mode(tmp_path):
# arrange
TEST_DIR.mkdir(parents=True, exist_ok=True)
info = sample_info_tv()
media_file = Path(info["filename"])
media_file.write_text("dummy")
def test_generate_nfo_tv_mode(tmp_path: Path) -> None:
media_file = tmp_path / "test_show_s01e02.mkv"
info = sample_info_tv(media_file)
media_file.write_text("dummy", encoding="utf-8")
nfo_path = media_file.with_suffix(".nfo")
_clean_file(nfo_path)
nfo_path.unlink(missing_ok=True)
# act
res = NFOMakerPP.generate_nfo(info_dict=info, filepath=media_file, mode="tv", overwrite=True, prefix=True)
# assert
assert res["success"] is True
assert nfo_path.exists()
content = nfo_path.read_text(encoding="utf-8")
assert "<episodedetails>" in content
assert "<title>Test Show - S01E02</title>" in content
# description cleaned (no timestamps / hashtags)
assert "Intro" not in content
assert "#tag" not in content
# also ensure PostProcessor.run wrapper produces the same result and syncs mtime
_clean_file(nfo_path)
nfo_path.unlink()
pp = NFOMakerPP(None, mode="tv", prefix=True)
media_mtime = media_file.stat().st_mtime
_, info_out = pp.run(info.copy())
pp.run(info.copy())
assert nfo_path.exists()
nfo_mtime = nfo_path.stat().st_mtime
assert abs(nfo_mtime - media_mtime) < 2.0
# cleanup
media_file.unlink()
nfo_path.unlink()
def test_generate_nfo_movie_mode_and_run_wrapper(tmp_path):
# arrange
TEST_DIR.mkdir(parents=True, exist_ok=True)
info = sample_info_movie()
media_file = Path(info["filename"])
media_file.write_text("dummy-movie")
def test_generate_nfo_movie_mode_and_run_wrapper(tmp_path: Path) -> None:
media_file = tmp_path / "test_movie.mp4"
info = sample_info_movie(media_file)
media_file.write_text("dummy-movie", encoding="utf-8")
nfo_path = media_file.with_suffix(".nfo")
_clean_file(nfo_path)
nfo_path.unlink(missing_ok=True)
# act: generate_nfo static
res = NFOMakerPP.generate_nfo(info_dict=info, filepath=media_file, mode="movie", overwrite=True, prefix=False)
# assert static helper
assert res["success"] is True
assert nfo_path.exists()
content = nfo_path.read_text(encoding="utf-8")
@ -99,26 +75,16 @@ def test_generate_nfo_movie_mode_and_run_wrapper(tmp_path):
assert "<title>Test Movie</title>" in content
assert "<runtime>7200</runtime>" in content or "<runtime>7200" in content
# cleanup nfo and re-run via PP.run wrapper
nfo_path.unlink()
# call through instance run() to ensure wrapper works (mtime sync path exercised)
pp = NFOMakerPP(None, mode="movie", prefix=False)
# ensure file exists
assert media_file.exists()
# capture media file mtime before run
media_mtime = media_file.stat().st_mtime
_, info_out = pp.run(info.copy())
pp.run(info.copy())
# run() should not raise and should produce nfo file
assert nfo_path.exists()
# verify mtime was synced (allow small rounding differences)
nfo_mtime = nfo_path.stat().st_mtime
assert abs(nfo_mtime - media_mtime) < 2.0
# cleanup
media_file.unlink()
nfo_path.unlink()