chore: update tests assertion messages

This commit is contained in:
arabcoders 2026-01-09 18:57:51 +03:00
parent 345337a1a2
commit b912186502
14 changed files with 224 additions and 290 deletions

View file

@ -30,16 +30,14 @@ class TestCondition:
"""Test creating a condition with default values.""" """Test creating a condition with default values."""
condition = Condition(name="test", filter="duration > 60") condition = Condition(name="test", filter="duration > 60")
# Check that ID is generated assert condition.id, "Check that ID is generated"
assert condition.id
assert isinstance(condition.id, str) assert isinstance(condition.id, str)
# Check required fields # Check required fields
assert condition.name == "test" assert condition.name == "test"
assert condition.filter == "duration > 60" assert condition.filter == "duration > 60"
# Check defaults assert condition.cli == "", "Check defaults"
assert condition.cli == ""
assert condition.extras == {} assert condition.extras == {}
assert condition.enabled is True assert condition.enabled is True
@ -235,8 +233,7 @@ class TestConditions:
assert result is conditions assert result is conditions
assert len(conditions._items) == 2 assert len(conditions._items) == 2
# Check first condition assert conditions._items[0].name == "short_videos", "Check first condition"
assert conditions._items[0].name == "short_videos"
assert conditions._items[0].filter == "duration < 300" assert conditions._items[0].filter == "duration < 300"
assert conditions._items[0].cli == "--format worst" assert conditions._items[0].cli == "--format worst"
assert conditions._items[0].extras == {"category": "short"} assert conditions._items[0].extras == {"category": "short"}
@ -244,8 +241,7 @@ class TestConditions:
assert conditions._items[0].priority == 0 assert conditions._items[0].priority == 0
assert conditions._items[0].description == "Download short videos" assert conditions._items[0].description == "Download short videos"
# Check second condition assert conditions._items[1].name == "music_videos", "Check second condition"
assert conditions._items[1].name == "music_videos"
assert conditions._items[1].filter == "title ~= 'music'" assert conditions._items[1].filter == "title ~= 'music'"
assert conditions._items[1].enabled is False assert conditions._items[1].enabled is False
assert conditions._items[1].priority == 5 assert conditions._items[1].priority == 5
@ -265,8 +261,7 @@ class TestConditions:
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
conditions.load() conditions.load()
# Should have generated ID assert len(conditions._items) == 1, "Should have generated ID"
assert len(conditions._items) == 1
assert conditions._items[0].id assert conditions._items[0].id
assert conditions._items[0].name == "no_id_test" assert conditions._items[0].name == "no_id_test"
@ -286,8 +281,7 @@ class TestConditions:
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
conditions.load() conditions.load()
# Should have generated empty extras assert len(conditions._items) == 1, "Should have generated empty extras"
assert len(conditions._items) == 1
assert conditions._items[0].extras == {} assert conditions._items[0].extras == {}
# Should call save due to changes # Should call save due to changes
@ -306,8 +300,7 @@ class TestConditions:
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
conditions.load() conditions.load()
# Should have generated enabled=True assert len(conditions._items) == 1, "Should have generated enabled=True"
assert len(conditions._items) == 1
assert conditions._items[0].enabled is True assert conditions._items[0].enabled is True
# Should call save due to changes # Should call save due to changes
@ -326,8 +319,7 @@ class TestConditions:
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
conditions.load() conditions.load()
# Should have generated priority=0 assert len(conditions._items) == 1, "Should have generated priority=0"
assert len(conditions._items) == 1
assert conditions._items[0].priority == 0 assert conditions._items[0].priority == 0
# Should call save due to changes # Should call save due to changes
@ -348,8 +340,7 @@ class TestConditions:
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
conditions.load() conditions.load()
# Should have generated description='' assert len(conditions._items) == 1, "Should have generated description=''"
assert len(conditions._items) == 1
assert conditions._items[0].description == "" assert conditions._items[0].description == ""
# Should call save due to changes # Should call save due to changes
@ -383,8 +374,7 @@ class TestConditions:
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
result = conditions.load() result = conditions.load()
# Should load only valid conditions assert result is conditions, "Should load only valid conditions"
assert result is conditions
assert len(conditions._items) == 1 assert len(conditions._items) == 1
assert conditions._items[0].name == "valid" assert conditions._items[0].name == "valid"
@ -731,14 +721,11 @@ class TestConditions:
test_condition = Condition(name="test", filter="duration > 60") test_condition = Condition(name="test", filter="duration > 60")
conditions._items = [test_condition] conditions._items = [test_condition]
# Test with None assert conditions.match(None) is None, "Test with None"
assert conditions.match(None) is None
# Test with empty dict assert conditions.match({}) is None, "Test with empty dict"
assert conditions.match({}) is None
# Test with non-dict assert conditions.match("not a dict") is None, "Test with non-dict"
assert conditions.match("not a dict") is None
@patch("app.library.conditions.match_str") @patch("app.library.conditions.match_str")
def test_match_filter_evaluation_error(self, mock_match_str): def test_match_filter_evaluation_error(self, mock_match_str):
@ -774,8 +761,7 @@ class TestConditions:
info_dict = {"duration": 150} info_dict = {"duration": 150}
result = conditions.match(info_dict) result = conditions.match(info_dict)
# Should skip disabled condition and match enabled one assert result is enabled_condition, "Should skip disabled condition and match enabled one"
assert result is enabled_condition
# Should only call match_str once for enabled condition # Should only call match_str once for enabled condition
mock_match_str.assert_called_once_with("duration > 120", info_dict) mock_match_str.assert_called_once_with("duration > 120", info_dict)
@ -800,8 +786,7 @@ class TestConditions:
info_dict = {"duration": 120} info_dict = {"duration": 120}
result = conditions.match(info_dict) result = conditions.match(info_dict)
# Should match high_priority first (priority=10) assert result is high_priority, "Should match high_priority first (priority=10)"
assert result is high_priority
# Should only call match_str once for highest priority condition # Should only call match_str once for highest priority condition
mock_match_str.assert_called_once_with("duration > 60", info_dict) mock_match_str.assert_called_once_with("duration > 60", info_dict)
@ -891,8 +876,7 @@ class TestConditions:
info_dict = {"duration": 120} info_dict = {"duration": 120}
result = conditions.single_match("disabled_single", info_dict) result = conditions.single_match("disabled_single", info_dict)
# Should return None because condition is disabled assert result is None, "Should return None because condition is disabled"
assert result is None
def test_single_match_invalid_inputs(self): def test_single_match_invalid_inputs(self):
"""Test single matching with invalid inputs.""" """Test single matching with invalid inputs."""
@ -900,14 +884,10 @@ class TestConditions:
file_path = Path(temp_dir) / "invalid_single_test.json" file_path = Path(temp_dir) / "invalid_single_test.json"
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
# Test with empty conditions assert conditions.single_match("test", {"duration": 120}) is None, "Test with empty conditions"
assert conditions.single_match("test", {"duration": 120}) is None
# Test with None info assert conditions.single_match("test", None) is None, "Test with None info"
assert conditions.single_match("test", None) is None
# Test with empty info dict assert conditions.single_match("test", {}) is None, "Test with empty info dict"
assert conditions.single_match("test", {}) is None
# Test with non-dict info assert conditions.single_match("test", "not a dict") is None, "Test with non-dict info"
assert conditions.single_match("test", "not a dict") is None

View file

@ -523,8 +523,7 @@ class TestDataStore:
assert "id1" in ids assert "id1" in ids
assert "id2" in ids assert "id2" in ids
# Verify order is maintained (OrderedDict) assert result[0][0] == "id1", "Verify order is maintained (OrderedDict)"
assert result[0][0] == "id1"
assert result[1][0] == "id2" assert result[1][0] == "id2"
await db.close() await db.close()

View file

@ -71,8 +71,7 @@ class TestDLField:
"""Test creating a DLField with default values.""" """Test creating a DLField with default values."""
field = DLField(name="test_field", description="Test description", field="--test-option") field = DLField(name="test_field", description="Test description", field="--test-option")
# Check that ID is generated assert field.id, "Check that ID is generated"
assert field.id
assert isinstance(field.id, str) assert isinstance(field.id, str)
# Check required fields # Check required fields
@ -80,8 +79,7 @@ class TestDLField:
assert field.description == "Test description" assert field.description == "Test description"
assert field.field == "--test-option" assert field.field == "--test-option"
# Check defaults assert field.kind == FieldType.TEXT, "Check defaults"
assert field.kind == FieldType.TEXT
assert field.icon == "" assert field.icon == ""
assert field.order == 0 assert field.order == 0
assert field.value == "" assert field.value == ""
@ -272,8 +270,7 @@ class TestDLFields:
fields = DLFields(file=str(temp_file)) fields = DLFields(file=str(temp_file))
fields.load() fields.load()
# Should handle error gracefully and return empty list assert fields.get_all() == [], "Should handle error gracefully and return empty list"
assert fields.get_all() == []
@patch("app.library.dl_fields.Config") @patch("app.library.dl_fields.Config")
def test_dl_fields_load_missing_id_auto_generation(self, mock_config, temp_file): def test_dl_fields_load_missing_id_auto_generation(self, mock_config, temp_file):
@ -490,8 +487,7 @@ class TestDLFields:
fields.save(test_fields) fields.save(test_fields)
# Verify file was written assert temp_file.exists(), "Verify file was written"
assert temp_file.exists()
# Verify content # Verify content
saved_data = json.loads(temp_file.read_text()) saved_data = json.loads(temp_file.read_text())

View file

@ -55,8 +55,7 @@ class TestNestedLogger:
assert levels.count(logging.INFO) == 1 assert levels.count(logging.INFO) == 1
msgs = [r.getMessage() for r in cap.records] msgs = [r.getMessage() for r in cap.records]
assert "[debug]" not in msgs[0] assert "[debug]" not in msgs[0]
# [download] prefix is not stripped by NestedLogger assert msgs[1] == "[download] progress", "[download] prefix is not stripped by NestedLogger"
assert msgs[1] == "[download] progress"
assert msgs[2] == "info message" assert msgs[2] == "info message"
@ -112,8 +111,7 @@ class TestDownloadHooks:
ev = q.items[0] ev = q.items[0]
assert ev["id"] == d.id assert ev["id"] == d.id
assert ev["action"] == "progress" assert ev["action"] == "progress"
# ensure only whitelisted keys included assert "other" not in ev, "ensure only whitelisted keys included"
assert "other" not in ev
for k in ( for k in (
"tmpfilename", "tmpfilename",
"filename", "filename",

View file

@ -55,8 +55,9 @@ class TestFFProbe:
"""Test that ffprobe results are cached with enhanced async timed_lru_cache.""" """Test that ffprobe results are cached with enhanced async timed_lru_cache."""
from app.library.ffprobe import ffprobe from app.library.ffprobe import ffprobe
# Test that the function has been decorated with caching assert hasattr(ffprobe, "cache_clear"), (
assert hasattr(ffprobe, "cache_clear"), "ffprobe should have cache_clear method from timed_lru_cache" "Test that the function has been decorated with caching - ffprobe should have cache_clear method from timed_lru_cache"
)
assert hasattr(ffprobe, "cache_info"), "ffprobe should have cache_info method from timed_lru_cache" assert hasattr(ffprobe, "cache_info"), "ffprobe should have cache_info method from timed_lru_cache"
# Clear cache to start fresh # Clear cache to start fresh
@ -91,12 +92,13 @@ class TestFFProbe:
assert result2 is not None assert result2 is not None
assert isinstance(result2.metadata, dict) assert isinstance(result2.metadata, dict)
# The subprocess should not be called again for the actual ffprobe execution assert call_count == first_call_count, (
# (it may be called for the -h check, but the main execution should be cached) "The subprocess should not be called again for the actual ffprobe execution (it may be called for the -h check, but the main execution should be cached) - Second call should use cached result"
assert call_count == first_call_count, "Second call should use cached result" )
# Results should be equivalent (same data, may not be same object due to async nature) assert result1.metadata == result2.metadata, (
assert result1.metadata == result2.metadata "Results should be equivalent (same data, may not be same object due to async nature)"
)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_ffprobe_with_path_object(self): async def test_ffprobe_with_path_object(self):
@ -124,8 +126,7 @@ class TestFFProbe:
result = FFProbeResult() result = FFProbeResult()
# Test empty result assert result.video == [], "Test empty result"
assert result.video == []
assert result.audio == [] assert result.audio == []
assert result.subtitle == [] assert result.subtitle == []
assert result.attachment == [] assert result.attachment == []

View file

@ -32,8 +32,7 @@ class TestItemFormatAndBasics:
item = Item.format(data) item = Item.format(data)
assert isinstance(item, Item) assert isinstance(item, Item)
# URL normalized to full YouTube URL assert item.url.startswith("https://www.youtube.com/watch?v="), "URL normalized to full YouTube URL"
assert item.url.startswith("https://www.youtube.com/watch?v=")
assert item.preset == "custom" assert item.preset == "custom"
assert item.folder == "media" assert item.folder == "media"
assert item.cookies == "abc" assert item.cookies == "abc"
@ -146,8 +145,7 @@ class TestItemDTO:
def test_archive_add_and_delete_paths(self): def test_archive_add_and_delete_paths(self):
dto = ItemDTO(id="id", title="t", url="u", folder="f") dto = ItemDTO(id="id", title="t", url="u", folder="f")
# Precondition not met yet assert dto.archive_add() is False, "Precondition not met yet"
assert dto.archive_add() is False
# Set up to allow add # Set up to allow add
dto.archive_id = "arch" dto.archive_id = "arch"

View file

@ -216,13 +216,11 @@ class TestNotificationEvents:
def test_is_valid(self): def test_is_valid(self):
"""Test is_valid static method.""" """Test is_valid static method."""
# Valid events assert NotificationEvents.is_valid("test"), "Valid events"
assert NotificationEvents.is_valid("test")
assert NotificationEvents.is_valid("item_added") assert NotificationEvents.is_valid("item_added")
assert NotificationEvents.is_valid("item_completed") assert NotificationEvents.is_valid("item_completed")
# Invalid events assert not NotificationEvents.is_valid("invalid_event"), "Invalid events"
assert not NotificationEvents.is_valid("invalid_event")
assert not NotificationEvents.is_valid("") assert not NotificationEvents.is_valid("")
assert not NotificationEvents.is_valid(None) assert not NotificationEvents.is_valid(None)
@ -417,8 +415,7 @@ class TestNotification:
# Verify save was called due to schema update # Verify save was called due to schema update
mock_save.assert_called_once() mock_save.assert_called_once()
# Verify the target has enabled=True by default assert len(notification._targets) == 1, "Verify the target has enabled=True by default"
assert len(notification._targets) == 1
assert notification._targets[0].enabled is True assert notification._targets[0].enabled is True
# Clean up # Clean up
@ -744,8 +741,7 @@ class TestNotification:
result = await notification.send(event) result = await notification.send(event)
# Only enabled target should be called assert len(result) == 1, "Only enabled target should be called"
assert len(result) == 1
assert result[0]["status"] == 200 assert result[0]["status"] == 200
mock_client.request.assert_called_once() mock_client.request.assert_called_once()
@ -783,8 +779,7 @@ class TestNotification:
result = await notification.send(event) result = await notification.send(event)
# Should return empty dict from _apprise method assert len(result) == 1, "Should return empty dict from _apprise method"
assert len(result) == 1
assert result[0] == {} assert result[0] == {}
def test_check_preset_no_presets(self): def test_check_preset_no_presets(self):
@ -881,8 +876,7 @@ class TestNotification:
with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=False): with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=False):
result = notification.emit(event, None) result = notification.emit(event, None)
# Should return None and not submit to background worker assert result is None, "Should return None and not submit to background worker"
assert result is None
mock_worker_instance.submit.assert_not_called() mock_worker_instance.submit.assert_not_called()
def test_emit_valid_event(self): def test_emit_valid_event(self):
@ -908,8 +902,7 @@ class TestNotification:
with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=True): with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=True):
result = notification.emit(event, None) result = notification.emit(event, None)
# Should return None but submit to background worker assert result is None, "Should return None but submit to background worker"
assert result is None
mock_worker_instance.submit.assert_called_once_with(notification.send, event) mock_worker_instance.submit.assert_called_once_with(notification.send, event)
@pytest.mark.asyncio @pytest.mark.asyncio

View file

@ -23,8 +23,7 @@ class TestPreset:
assert preset.cookies == "" assert preset.cookies == ""
assert preset.cli == "" assert preset.cli == ""
assert preset.default is False assert preset.default is False
# ID should be auto-generated UUID assert len(preset.id) == 36, "ID should be auto-generated UUID" # UUID4 string length
assert len(preset.id) == 36 # UUID4 string length
assert "-" in preset.id assert "-" in preset.id
def test_preset_creation_with_all_fields(self): def test_preset_creation_with_all_fields(self):
@ -228,8 +227,7 @@ class TestPresets:
assert len(presets._items) == 1 assert len(presets._items) == 1
loaded_preset = presets._items[0] loaded_preset = presets._items[0]
assert loaded_preset.name == "old_preset" assert loaded_preset.name == "old_preset"
# Should have migrated format to cli assert "best[height<=720]" in loaded_preset.cli, "Should have migrated format to cli"
assert "best[height<=720]" in loaded_preset.cli
assert "--format" in loaded_preset.cli assert "--format" in loaded_preset.cli
# Should have generated ID # Should have generated ID
assert loaded_preset.id is not None assert loaded_preset.id is not None
@ -322,8 +320,7 @@ class TestPresets:
all_presets = presets.get_all() all_presets = presets.get_all()
# Should include both default and custom presets assert len(all_presets) > 1, "Should include both default and custom presets"
assert len(all_presets) > 1
assert any(p.name == "custom" for p in all_presets) assert any(p.name == "custom" for p in all_presets)
assert any(p.default is True for p in all_presets) assert any(p.default is True for p in all_presets)
@ -462,7 +459,7 @@ class TestPresets:
# Note: The actual chmod might fail in test environment, # Note: The actual chmod might fail in test environment,
# but we're testing that the code attempts it # but we're testing that the code attempts it
assert presets is not None assert presets is not None, "but we're testing that the code attempts it"
def test_default_presets_validation(self): def test_default_presets_validation(self):
"""Test that all default presets are valid.""" """Test that all default presets are valid."""
@ -519,8 +516,7 @@ class TestPresets:
with patch.object(presets, "get", return_value=None): with patch.object(presets, "get", return_value=None):
# The actual config instance stored in presets should be checked # The actual config instance stored in presets should be checked
presets.attach(mock_app) presets.attach(mock_app)
# Should have reset default_preset to "default" assert presets._config.default_preset == "default", "Should have reset default_preset to 'default'"
assert presets._config.default_preset == "default"
@pytest.mark.asyncio @pytest.mark.asyncio
@patch("app.library.Presets.Config") @patch("app.library.Presets.Config")
@ -567,8 +563,7 @@ class TestPresets:
presets = Presets(file=presets_file, config=mock_config_instance) presets = Presets(file=presets_file, config=mock_config_instance)
presets.load() presets.load()
# Should only load the valid preset assert len(presets._items) == 1, "Should only load the valid preset"
assert len(presets._items) == 1
assert presets._items[0].name == "valid_preset", f"Expected 'valid_preset', got '{presets._items}'" assert presets._items[0].name == "valid_preset", f"Expected 'valid_preset', got '{presets._items}'"
# Should have logged an error for the invalid preset # Should have logged an error for the invalid preset
@ -658,8 +653,7 @@ class TestPresets:
all_presets = presets.get_all() all_presets = presets.get_all()
non_default = [p for p in all_presets if not p.default] non_default = [p for p in all_presets if not p.default]
# Should be sorted by priority descending assert non_default[0].name == "high", "Should be sorted by priority descending"
assert non_default[0].name == "high"
assert non_default[0].priority == 10 assert non_default[0].priority == 10
assert non_default[1].name == "medium" assert non_default[1].name == "medium"
assert non_default[1].priority == 5 assert non_default[1].priority == 5

View file

@ -130,8 +130,7 @@ class TestScheduler:
assert sched.has("job1") is True assert sched.has("job1") is True
new_job = sched.get("job1") new_job = sched.get("job1")
assert new_job is not old assert new_job is not old
# Old job should have been stopped via remove() assert old.stopped is True, "Old job should have been stopped via remove()"
assert old.stopped is True
@patch("app.library.Scheduler.Cron", new=DummyCron) @patch("app.library.Scheduler.Cron", new=DummyCron)
def test_remove_single_job_success(self) -> None: def test_remove_single_job_success(self) -> None:
@ -154,8 +153,7 @@ class TestScheduler:
result = sched.remove("jobB") result = sched.remove("jobB")
assert result is False assert result is False
# Job should remain since stop failed assert sched.has("jobB") is True, "Job should remain since stop failed"
assert sched.has("jobB") is True
@patch("app.library.Scheduler.Cron", new=DummyCron) @patch("app.library.Scheduler.Cron", new=DummyCron)
def test_remove_list_of_jobs(self) -> None: def test_remove_list_of_jobs(self) -> None:
@ -193,8 +191,9 @@ class TestScheduler:
sched = Scheduler() sched = Scheduler()
sched.attach(app) sched.attach(app)
# on_shutdown handler should be registered assert Scheduler.on_shutdown in [cb.__func__ if hasattr(cb, "__func__") else cb for cb in app.on_shutdown], (
assert Scheduler.on_shutdown in [cb.__func__ if hasattr(cb, "__func__") else cb for cb in app.on_shutdown] "on_shutdown handler should be registered"
)
# Patch add to verify it is called from event handler # Patch add to verify it is called from event handler
add_spy = MagicMock(wraps=sched.add) add_spy = MagicMock(wraps=sched.add)

View file

@ -13,12 +13,13 @@ class TestMsToTimestamp:
assert ms_to_timestamp(9) == "0:00:00.00" assert ms_to_timestamp(9) == "0:00:00.00"
assert ms_to_timestamp(10) == "0:00:00.01" assert ms_to_timestamp(10) == "0:00:00.01"
assert ms_to_timestamp(12345) == "0:00:12.34" assert ms_to_timestamp(12345) == "0:00:12.34"
# 1 hour, 2 minutes, 3 seconds assert ms_to_timestamp(3600000 + 120000 + 3000) == "1:02:03.00", "1 hour, 2 minutes, 3 seconds"
assert ms_to_timestamp(3600000 + 120000 + 3000) == "1:02:03.00" assert ms_to_timestamp(12 * 3600000 + 34 * 60000 + 56 * 1000 + 780) == "12:34:56.78", (
# Over 10 hours (SubStation limit is < 10h, our override must exceed) "Over 10 hours (SubStation limit is < 10h, our override must exceed)"
assert ms_to_timestamp(12 * 3600000 + 34 * 60000 + 56 * 1000 + 780) == "12:34:56.78" )
# Well over 36 hours to ensure no clamping at 9:59:59.99 assert ms_to_timestamp(37 * 3600000 + 12 * 60000 + 34 * 1000 + 560) == "37:12:34.56", (
assert ms_to_timestamp(37 * 3600000 + 12 * 60000 + 34 * 1000 + 560) == "37:12:34.56" "Well over 36 hours to ensure no clamping at 9:59:59.99"
)
@pytest.mark.asyncio @pytest.mark.asyncio
@ -77,8 +78,7 @@ async def test_make_single_event_returns_vtt(tmp_path: Path) -> None:
sub = Subtitle() sub = Subtitle()
out = await sub.make(srt) out = await sub.make(srt)
assert out == "OUT" assert out == "OUT"
# Snapshot should contain the single event assert d.snapshot == [1000], "Snapshot should contain the single event"
assert d.snapshot == [1000]
@pytest.mark.asyncio @pytest.mark.asyncio
@ -94,8 +94,7 @@ async def test_make_two_events_pop_first_when_ends_equal(tmp_path: Path) -> None
sub = Subtitle() sub = Subtitle()
out = await sub.make(srt) out = await sub.make(srt)
assert out == "OUT" assert out == "OUT"
# Since ends are equal, first should be popped => only last remains assert d.snapshot == [5000], "Since ends are equal, first should be popped => only last remains"
assert d.snapshot == [5000]
@pytest.mark.asyncio @pytest.mark.asyncio
@ -111,5 +110,4 @@ async def test_make_two_events_no_pop_when_different(tmp_path: Path) -> None:
sub = Subtitle() sub = Subtitle()
out = await sub.make(srt) out = await sub.make(srt)
assert out == "OUT" assert out == "OUT"
# Both remain since ends differ assert d.snapshot == [5000, 6000], "Both remain since ends differ"
assert d.snapshot == [5000, 6000]

View file

@ -321,8 +321,7 @@ class TestTasks:
tasks_file = Path(temp_dir) / "tasks.json" tasks_file = Path(temp_dir) / "tasks.json"
tasks = Tasks(file=tasks_file, config=mock_config_instance) tasks = Tasks(file=tasks_file, config=mock_config_instance)
# Check initialization assert tasks._debug is True, "Check initialization"
assert tasks._debug is True
assert tasks._default_preset == "test_preset" assert tasks._default_preset == "test_preset"
assert tasks._file == tasks_file assert tasks._file == tasks_file
assert tasks._tasks == [] assert tasks._tasks == []
@ -505,8 +504,7 @@ class TestTasks:
assert result == tasks assert result == tasks
assert tasks_file.exists() assert tasks_file.exists()
# Verify file content was written assert tasks_file.stat().st_size > 0, "Verify file content was written"
assert tasks_file.stat().st_size > 0
@patch("app.library.Tasks.Config") @patch("app.library.Tasks.Config")
@patch("app.library.Tasks.EventBus") @patch("app.library.Tasks.EventBus")
@ -788,8 +786,7 @@ class TestTasks:
} }
) )
# Verify events were emitted assert mock_eventbus_instance.emit.call_count == 2, "Verify events were emitted"
assert mock_eventbus_instance.emit.call_count == 2
@pytest.mark.asyncio @pytest.mark.asyncio
@patch("app.library.Tasks.Config") @patch("app.library.Tasks.Config")
@ -963,8 +960,7 @@ class TestHandleTaskInspect:
url="https://example.com/feed", preset="default", handler_name=None, static_only=True url="https://example.com/feed", preset="default", handler_name=None, static_only=True
) )
# Verify result structure assert hasattr(result, "items"), "Verify result structure"
assert hasattr(result, "items")
assert hasattr(result, "metadata") assert hasattr(result, "metadata")
assert result.items == [] assert result.items == []
assert result.metadata["matched"] is True assert result.metadata["matched"] is True
@ -1018,8 +1014,7 @@ class TestHandleTaskInspect:
# Call inspect with static_only=False (default) # Call inspect with static_only=False (default)
result = await handler.inspect(url="https://example.com/feed", preset="default", handler_name=None) result = await handler.inspect(url="https://example.com/feed", preset="default", handler_name=None)
# Verify result structure includes extracted items assert hasattr(result, "items"), "Verify result structure includes extracted items"
assert hasattr(result, "items")
assert hasattr(result, "metadata") assert hasattr(result, "metadata")
assert len(result.items) > 0 assert len(result.items) > 0
assert result.metadata["matched"] is True assert result.metadata["matched"] is True

View file

@ -34,7 +34,7 @@ from app.library.Utils import (
get_files, get_files,
get_mime_type, get_mime_type,
get_possible_images, get_possible_images,
get_static_ytdlp, get_ytdlp,
init_class, init_class,
is_private_address, is_private_address,
list_folders, list_folders,
@ -137,9 +137,8 @@ class TestTimedLruCache:
def test_function(x): def test_function(x):
return x * 2 return x * 2
# Test that methods exist assert hasattr(test_function, "cache_clear"), "Cached function should have cache_clear method"
assert hasattr(test_function, "cache_clear") assert hasattr(test_function, "cache_info"), "Cached function should have cache_info method"
assert hasattr(test_function, "cache_info")
# Call function to populate cache # Call function to populate cache
test_function(5) test_function(5)
@ -256,9 +255,8 @@ class TestAsyncTimedLruCache:
async def async_method_test(x): async def async_method_test(x):
return x + 1 return x + 1
# Test that cache methods exist assert hasattr(async_method_test, "cache_clear"), "Async cached function should have cache_clear method"
assert hasattr(async_method_test, "cache_clear") assert hasattr(async_method_test, "cache_info"), "Async cached function should have cache_info method"
assert hasattr(async_method_test, "cache_info")
# Test cache_info # Test cache_info
info = async_method_test.cache_info() info = async_method_test.cache_info()
@ -286,12 +284,11 @@ class TestAsyncTimedLruCache:
# Fill cache beyond max_size # Fill cache beyond max_size
result1 = await async_limited_func(1) result1 = await async_limited_func(1)
result2 = await async_limited_func(2) result2 = await async_limited_func(2)
result3 = await async_limited_func(3) # Should evict oldest entry result3 = await async_limited_func(3)
# Verify results assert result1 == 4, "async_limited_func(1) should return 4"
assert result1 == 4 assert result2 == 8, "async_limited_func(2) should return 8"
assert result2 == 8 assert result3 == 12, "async_limited_func(3) should return 12 (should evict oldest entry)"
assert result3 == 12
# Check cache size is limited # Check cache size is limited
info = async_limited_func.cache_info() info = async_limited_func.cache_info()
@ -664,9 +661,8 @@ class TestMergeDict:
source = {"nested": {"a": 1}} source = {"nested": {"a": 1}}
destination = {"nested": {"b": 2}, "other": 3} destination = {"nested": {"b": 2}, "other": 3}
result = merge_dict(source, destination) result = merge_dict(source, destination)
# Should merge nested dictionaries assert "nested" in result, "Should merge nested dictionaries"
assert "nested" in result assert "other" in result, "Should preserve other keys"
assert "other" in result
def test_merge_dict_empty_source(self): def test_merge_dict_empty_source(self):
"""Test merging with empty source.""" """Test merging with empty source."""
@ -695,7 +691,7 @@ class TestMergeDict:
destination = {"existing": "data"} destination = {"existing": "data"}
result = merge_dict(source, destination) result = merge_dict(source, destination)
assert "__class__" not in result, "__class__ should be filtered out" assert "__class__" not in result, "__class__ attribute pollution should be blocked"
assert result["safe"] == "value", "Safe values should be preserved" assert result["safe"] == "value", "Safe values should be preserved"
assert result["existing"] == "data", "Existing data should be preserved" assert result["existing"] == "data", "Existing data should be preserved"
@ -742,11 +738,10 @@ class TestMergeDict:
# All dangerous attributes should be filtered out # All dangerous attributes should be filtered out
dangerous_keys = ["__class__", "__dict__", "__globals__", "__builtins__"] dangerous_keys = ["__class__", "__dict__", "__globals__", "__builtins__"]
for key in dangerous_keys: for key in dangerous_keys:
assert key not in result, f"{key} should be filtered out" assert key not in result, f"{key} should be filtered out (all dangerous attributes)"
# Safe data should be preserved assert result["safe_key"] == "safe_value", "Safe data should be preserved"
assert result["safe_key"] == "safe_value" assert result["existing"] == "data", "Existing data should be preserved"
assert result["existing"] == "data"
def test_merge_dict_nested_dunder_pollution(self): def test_merge_dict_nested_dunder_pollution(self):
"""Test that nested dangerous attributes are handled correctly.""" """Test that nested dangerous attributes are handled correctly."""
@ -754,11 +749,9 @@ class TestMergeDict:
destination = {"nested": {"existing_nested": "original"}} destination = {"nested": {"existing_nested": "original"}}
result = merge_dict(source, destination) result = merge_dict(source, destination)
# Nested dangerous attributes should be filtered out assert "__class__" not in result["nested"], "Nested dangerous attributes should be filtered out"
assert "__class__" not in result["nested"], "Nested __class__ should be filtered" assert result["nested"]["safe_nested"] == "value", "Safe nested data should be preserved"
# Safe nested data should be preserved assert result["nested"]["existing_nested"] == "original", "Existing nested data should be preserved"
assert result["nested"]["safe_nested"] == "value"
assert result["nested"]["existing_nested"] == "original"
def test_merge_dict_prototype_pollution_attempt(self): def test_merge_dict_prototype_pollution_attempt(self):
"""Test protection against prototype pollution attempts.""" """Test protection against prototype pollution attempts."""
@ -766,10 +759,10 @@ class TestMergeDict:
destination = {"existing": "value"} destination = {"existing": "value"}
result = merge_dict(source, destination) result = merge_dict(source, destination)
# These should be treated as regular keys (not filtered unless explicitly in the filter list) assert result["safe"] == "data", (
# The function filters Python-specific dangerous attributes, not JavaScript ones "Function filters Python-specific dangerous attributes, not JS ones like __proto__"
assert result["safe"] == "data" )
assert result["existing"] == "value" assert result["existing"] == "value", "Existing data should be preserved"
def test_merge_dict_special_method_pollution(self): def test_merge_dict_special_method_pollution(self):
"""Test with various Python special methods.""" """Test with various Python special methods."""
@ -784,10 +777,10 @@ class TestMergeDict:
destination = {"target": "data"} destination = {"target": "data"}
result = merge_dict(source, destination) result = merge_dict(source, destination)
# These are not in the current filter list, so they should pass through assert result["safe"] == "value", (
# This test documents current behavior - may need updating if more filters are added "Safe data should be preserved (special methods not in filter list, documents current behavior)"
assert result["safe"] == "value" )
assert result["target"] == "data" assert result["target"] == "data", "Target data should be preserved"
def test_merge_dict_list_pollution_safe(self): def test_merge_dict_list_pollution_safe(self):
"""Test that list merging doesn't allow dangerous manipulation.""" """Test that list merging doesn't allow dangerous manipulation."""
@ -795,8 +788,9 @@ class TestMergeDict:
destination = {"items": ["old1", "old2"]} destination = {"items": ["old1", "old2"]}
result = merge_dict(source, destination) result = merge_dict(source, destination)
# Lists should be concatenated safely (destination + source) assert result["items"] == ["old1", "old2", "new1", "new2"], (
assert result["items"] == ["old1", "old2", "new1", "new2"] "Lists should be concatenated safely (destination + source)"
)
def test_merge_dict_deep_nested_pollution(self): def test_merge_dict_deep_nested_pollution(self):
"""Test with deeply nested dangerous attributes.""" """Test with deeply nested dangerous attributes."""
@ -812,13 +806,13 @@ class TestMergeDict:
destination = {"level1": {"level2": {"existing": "data"}}} destination = {"level1": {"level2": {"existing": "data"}}}
result = merge_dict(source, destination) result = merge_dict(source, destination)
# The function should now properly filter all dangerous keys recursively assert "__class__" not in result["level1"]["level2"], (
assert "__class__" not in result["level1"]["level2"], "Deep __class__ should be filtered" "Function should properly filter all dangerous keys recursively (deep __class__)"
assert "__globals__" not in result["level1"]["level2"]["level3"], "Very deep __globals__ should be filtered" )
assert "__globals__" not in result["level1"]["level2"]["level3"], "Function should filter very deep __globals__"
# Safe data should be preserved assert result["level1"]["level2"]["safe_deep"] == "value", "Safe nested data should be preserved"
assert result["level1"]["level2"]["safe_deep"] == "value" assert result["level1"]["level2"]["existing"] == "data", "Existing nested data should be preserved"
assert result["level1"]["level2"]["existing"] == "data"
def test_merge_dict_type_validation(self): def test_merge_dict_type_validation(self):
"""Test that non-dict parameters are properly rejected.""" """Test that non-dict parameters are properly rejected."""
@ -845,13 +839,13 @@ class TestMergeDict:
result = merge_dict(original_source, original_destination) result = merge_dict(original_source, original_destination)
# Original dictionaries should be unchanged assert original_source == source_copy, "Original source dictionary should be unchanged (immutability)"
assert original_source == source_copy, "Source should not be modified" assert original_destination == destination_copy, (
assert original_destination == destination_copy, "Destination should not be modified" "Original destination dictionary should be unchanged (immutability)"
)
# Result should be different from both originals assert result != original_source, "Result should be different from source original"
assert result != original_source assert result != original_destination, "Result should be different from destination original"
assert result != original_destination
def test_merge_dict_custom_max_depth(self): def test_merge_dict_custom_max_depth(self):
"""Test custom max_depth parameter.""" """Test custom max_depth parameter."""
@ -900,16 +894,14 @@ class TestMergeDict:
source = {"items": list(range(3000))} source = {"items": list(range(3000))}
destination = {"items": list(range(2000, 5000))} # 3000 items destination = {"items": list(range(2000, 5000))} # 3000 items
# Total would be 6000 items, but limit is 4000
result = merge_dict(source, destination, max_list_size=4000) result = merge_dict(source, destination, max_list_size=4000)
# Should have original destination (3000) + truncated source (1000) = 4000 assert len(result["items"]) == 4000, (
assert len(result["items"]) == 4000 "Total would be 6000 items, but limit is 4000: destination (3000) + truncated source (1000)"
)
# First 3000 should be from destination assert result["items"][:3000] == list(range(2000, 5000)), "First 3000 items should be from destination"
assert result["items"][:3000] == list(range(2000, 5000)) assert result["items"][3000:] == list(range(1000)), "Next 1000 items should be from source (truncated)"
# Next 1000 should be from source (truncated)
assert result["items"][3000:] == list(range(1000))
def test_merge_dict_nested_with_limits(self): def test_merge_dict_nested_with_limits(self):
"""Test nested merging with both depth and size limits.""" """Test nested merging with both depth and size limits."""
@ -2018,8 +2010,7 @@ class TestDecryptDataCornerCases:
encrypted = encrypt_data("test data", correct_key) encrypted = encrypt_data("test data", correct_key)
result = decrypt_data(encrypted, wrong_key) result = decrypt_data(encrypted, wrong_key)
# Should return None on decryption failure assert result is None, "Should return None on decryption failure with wrong key"
assert result is None
def test_decrypt_data_truncated(self): def test_decrypt_data_truncated(self):
"""Test decrypting truncated encrypted data.""" """Test decrypting truncated encrypted data."""
@ -2175,11 +2166,10 @@ class TestRenameFile:
# Rename file # Rename file
new_path, sidecars = rename_file(test_file, "renamed_video.mp4") new_path, sidecars = rename_file(test_file, "renamed_video.mp4")
# Assertions assert new_path.exists(), "Renamed file should exist"
assert new_path.exists() assert "renamed_video.mp4" == new_path.name, "File should have new name"
assert "renamed_video.mp4" == new_path.name assert not test_file.exists(), "Original file should not exist"
assert not test_file.exists() assert 0 == len(sidecars), "Should have no sidecar files"
assert 0 == len(sidecars)
def test_rename_file_with_subtitle_sidecar(self, tmp_path: Path): def test_rename_file_with_subtitle_sidecar(self, tmp_path: Path):
"""Test renaming a file with subtitle sidecar.""" """Test renaming a file with subtitle sidecar."""
@ -2193,12 +2183,11 @@ class TestRenameFile:
# Rename file # Rename file
new_path, sidecars = rename_file(test_file, "renamed_video.mp4") new_path, sidecars = rename_file(test_file, "renamed_video.mp4")
# Assertions assert new_path.exists(), "Renamed file should exist"
assert new_path.exists() assert "renamed_video.mp4" == new_path.name, "File should have new name"
assert "renamed_video.mp4" == new_path.name assert not test_file.exists(), "Original file should not exist after rename"
assert not test_file.exists()
assert 1 == len(sidecars) assert 1 == len(sidecars), "Should have renamed 1 sidecar file"
old_sidecar, new_sidecar = sidecars[0] old_sidecar, new_sidecar = sidecars[0]
assert new_sidecar.exists() assert new_sidecar.exists()
assert "renamed_video.en.srt" == new_sidecar.name assert "renamed_video.en.srt" == new_sidecar.name
@ -2223,12 +2212,11 @@ class TestRenameFile:
# Rename file # Rename file
new_path, sidecars = rename_file(test_file, "renamed_video.mp4") new_path, sidecars = rename_file(test_file, "renamed_video.mp4")
# Assertions assert new_path.exists(), "Renamed file should exist"
assert new_path.exists() assert "renamed_video.mp4" == new_path.name, "File should have new name"
assert "renamed_video.mp4" == new_path.name assert not test_file.exists(), "Original file should not exist after rename"
assert not test_file.exists()
assert 3 == len(sidecars) assert 3 == len(sidecars), "Should have renamed 3 sidecar files"
# Check all sidecars were renamed # Check all sidecars were renamed
sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars} sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars}
@ -2236,10 +2224,9 @@ class TestRenameFile:
assert "renamed_video.fr.srt" in sidecar_names assert "renamed_video.fr.srt" in sidecar_names
assert "renamed_video.info.json" in sidecar_names assert "renamed_video.info.json" in sidecar_names
# Check old files don't exist assert not subtitle_en.exists(), "Old subtitle file should not exist after rename"
assert not subtitle_en.exists() assert not subtitle_fr.exists(), "Old subtitle file should not exist after rename"
assert not subtitle_fr.exists() assert not info_file.exists(), "Old info file should not exist after rename"
assert not info_file.exists()
def test_rename_file_destination_exists(self, tmp_path: Path): def test_rename_file_destination_exists(self, tmp_path: Path):
"""Test renaming a file when destination already exists.""" """Test renaming a file when destination already exists."""
@ -2254,9 +2241,8 @@ class TestRenameFile:
with pytest.raises(ValueError, match="already exists"): with pytest.raises(ValueError, match="already exists"):
rename_file(test_file, "renamed_video.mp4") rename_file(test_file, "renamed_video.mp4")
# Original files should still exist assert test_file.exists(), "Original file should still exist when rename fails"
assert test_file.exists() assert existing_file.exists(), "Existing file should still exist when rename fails"
assert existing_file.exists()
def test_rename_file_sidecar_destination_exists(self, tmp_path: Path): def test_rename_file_sidecar_destination_exists(self, tmp_path: Path):
"""Test renaming when sidecar destination already exists.""" """Test renaming when sidecar destination already exists."""
@ -2295,11 +2281,10 @@ class TestRenameFile:
# Rename file # Rename file
new_path, sidecars = rename_file(test_file, "renamed.mp4") new_path, sidecars = rename_file(test_file, "renamed.mp4")
# Assertions assert new_path.exists(), "Renamed file should exist"
assert new_path.exists() assert "renamed.mp4" == new_path.name, "File should have new name"
assert "renamed.mp4" == new_path.name
assert 2 == len(sidecars) assert 2 == len(sidecars), "Should have renamed 2 sidecar files"
sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars} sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars}
assert "renamed.en-US.ass" in sidecar_names assert "renamed.en-US.ass" in sidecar_names
assert "renamed.thumb.jpg" in sidecar_names assert "renamed.thumb.jpg" in sidecar_names
@ -2322,12 +2307,11 @@ class TestMoveFile:
# Move file # Move file
new_path, sidecars = move_file(test_file, target_dir) new_path, sidecars = move_file(test_file, target_dir)
# Assertions assert new_path.exists(), "Moved file should exist at destination"
assert new_path.exists() assert "video.mp4" == new_path.name, "File should keep same name"
assert "video.mp4" == new_path.name assert new_path.parent == target_dir, "File should be in target directory"
assert new_path.parent == target_dir assert not test_file.exists(), "Original file should not exist after move"
assert not test_file.exists() assert 0 == len(sidecars), "Should have no sidecar files"
assert 0 == len(sidecars)
def test_move_file_with_subtitle_sidecar(self, tmp_path: Path): def test_move_file_with_subtitle_sidecar(self, tmp_path: Path):
"""Test moving a file with subtitle sidecar.""" """Test moving a file with subtitle sidecar."""
@ -2346,13 +2330,12 @@ class TestMoveFile:
# Move file # Move file
new_path, sidecars = move_file(test_file, target_dir) new_path, sidecars = move_file(test_file, target_dir)
# Assertions assert new_path.exists(), "Moved file should exist at destination"
assert new_path.exists() assert "video.mp4" == new_path.name, "File should keep same name"
assert "video.mp4" == new_path.name assert new_path.parent == target_dir, "File should be in target directory"
assert new_path.parent == target_dir assert not test_file.exists(), "Original file should not exist after move"
assert not test_file.exists()
assert 1 == len(sidecars) assert 1 == len(sidecars), "Should have moved 1 sidecar file"
old_sidecar, new_sidecar = sidecars[0] old_sidecar, new_sidecar = sidecars[0]
assert new_sidecar.exists() assert new_sidecar.exists()
assert "video.en.srt" == new_sidecar.name assert "video.en.srt" == new_sidecar.name
@ -2383,13 +2366,12 @@ class TestMoveFile:
# Move file # Move file
new_path, sidecars = move_file(test_file, target_dir) new_path, sidecars = move_file(test_file, target_dir)
# Assertions assert new_path.exists(), "Moved file should exist at destination"
assert new_path.exists() assert "video.mp4" == new_path.name, "File should keep same name"
assert "video.mp4" == new_path.name assert new_path.parent == target_dir, "File should be in target directory"
assert new_path.parent == target_dir assert not test_file.exists(), "Original file should not exist after move"
assert not test_file.exists()
assert 3 == len(sidecars) assert 3 == len(sidecars), "Should have moved 3 sidecar files"
# Check all sidecars were moved # Check all sidecars were moved
sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars} sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars}
@ -2399,12 +2381,11 @@ class TestMoveFile:
# Check all are in target directory # Check all are in target directory
for _old_sidecar, new_sidecar in sidecars: for _old_sidecar, new_sidecar in sidecars:
assert new_sidecar.parent == target_dir assert new_sidecar.parent == target_dir, "All sidecars should be in target directory"
# Check old files don't exist assert not subtitle_en.exists(), "Old subtitle file should not exist after move"
assert not subtitle_en.exists() assert not subtitle_fr.exists(), "Old subtitle file should not exist after move"
assert not subtitle_fr.exists() assert not info_file.exists(), "Old info file should not exist after move"
assert not info_file.exists()
def test_move_file_destination_exists(self, tmp_path: Path): def test_move_file_destination_exists(self, tmp_path: Path):
"""Test moving a file when destination already exists.""" """Test moving a file when destination already exists."""
@ -2423,9 +2404,8 @@ class TestMoveFile:
with pytest.raises(ValueError, match="already exists"): with pytest.raises(ValueError, match="already exists"):
move_file(test_file, target_dir) move_file(test_file, target_dir)
# Original files should still exist assert test_file.exists(), "Original file should still exist when move fails"
assert test_file.exists() assert existing_file.exists(), "Existing file should still exist when move fails"
assert existing_file.exists()
def test_move_file_sidecar_destination_exists(self, tmp_path: Path): def test_move_file_sidecar_destination_exists(self, tmp_path: Path):
"""Test moving when sidecar destination already exists.""" """Test moving when sidecar destination already exists."""
@ -2682,41 +2662,43 @@ class TestGetExtras:
class TestGetStaticYtdlp: class TestGetStaticYtdlp:
"""Test the get_static_ytdlp function.""" """Test the get_static_ytdlp function."""
def setup_method(self):
"""Reset YTDLP singleton state before each test."""
import app.library.Utils as Utils
Utils.YTDLP_INFO_CLS = None
def test_get_static_ytdlp_returns_instance(self): def test_get_static_ytdlp_returns_instance(self):
"""Test that get_static_ytdlp returns a YTDLP instance.""" """Test that get_static_ytdlp returns a YTDLP instance."""
from app.library.Utils import get_static_ytdlp
from app.library.ytdlp import YTDLP from app.library.ytdlp import YTDLP
# Force reload to ensure we get a real instance, not a mock # Get the cached instance
instance = get_static_ytdlp(reload=True) instance = get_ytdlp()
assert instance is not None assert instance is not None
assert isinstance(instance, YTDLP) assert isinstance(instance, YTDLP)
def test_get_static_ytdlp_returns_same_instance(self): def test_get_static_ytdlp_returns_same_instance(self):
"""Test that get_static_ytdlp returns the same cached instance.""" """Test that get_static_ytdlp returns the same cached instance."""
from app.library.Utils import get_static_ytdlp
instance1 = get_static_ytdlp() instance1 = get_ytdlp()
instance2 = get_static_ytdlp() instance2 = get_ytdlp()
assert instance1 is instance2 assert instance1 is instance2
def test_get_static_ytdlp_reload(self): def test_get_static_ytdlp_with_params(self):
"""Test that get_static_ytdlp can reload and return a new instance.""" """Test that get_static_ytdlp returns a new instance when params are provided."""
from app.library.Utils import get_static_ytdlp
instance1 = get_static_ytdlp() instance1 = get_ytdlp()
instance2 = get_static_ytdlp(reload=True) instance2 = get_ytdlp(params={"quiet": False})
assert instance1 is not instance2 assert instance1 is not instance2
assert instance2 is not None assert instance2 is not None
def test_get_static_ytdlp_has_correct_params(self): def test_get_static_ytdlp_has_correct_params(self):
"""Test that get_static_ytdlp initializes with correct parameters.""" """Test that get_static_ytdlp initializes with correct parameters."""
from app.library.Utils import get_static_ytdlp
instance = get_static_ytdlp(reload=True) instance = get_ytdlp()
# Access the internal params # Access the internal params
params = instance.params params = instance.params
@ -2732,9 +2714,14 @@ class TestGetStaticYtdlp:
class TestParseOuttmpl: class TestParseOuttmpl:
"""Test the parse_outtmpl function.""" """Test the parse_outtmpl function."""
def setup_method(self):
"""Reset YTDLP singleton state before each test."""
import app.library.Utils as Utils
Utils.YTDLP_INFO_CLS = None
def test_parse_outtmpl_basic(self): def test_parse_outtmpl_basic(self):
"""Test basic template parsing with simple placeholders.""" """Test basic template parsing with simple placeholders."""
from app.library.Utils import parse_outtmpl
template = "%(title)s.%(ext)s" template = "%(title)s.%(ext)s"
info_dict = { info_dict = {
@ -2748,7 +2735,6 @@ class TestParseOuttmpl:
def test_parse_outtmpl_with_id(self): def test_parse_outtmpl_with_id(self):
"""Test template parsing with video ID.""" """Test template parsing with video ID."""
from app.library.Utils import parse_outtmpl
template = "[%(id)s] %(title)s.%(ext)s" template = "[%(id)s] %(title)s.%(ext)s"
info_dict = { info_dict = {
@ -2763,7 +2749,6 @@ class TestParseOuttmpl:
def test_parse_outtmpl_with_uploader(self): def test_parse_outtmpl_with_uploader(self):
"""Test template parsing with uploader information.""" """Test template parsing with uploader information."""
from app.library.Utils import parse_outtmpl
template = "%(uploader)s - %(title)s.%(ext)s" template = "%(uploader)s - %(title)s.%(ext)s"
info_dict = { info_dict = {
@ -2778,7 +2763,6 @@ class TestParseOuttmpl:
def test_parse_outtmpl_with_nested_path(self): def test_parse_outtmpl_with_nested_path(self):
"""Test template parsing with nested directory structure.""" """Test template parsing with nested directory structure."""
from app.library.Utils import parse_outtmpl
template = "%(uploader)s/%(title)s.%(ext)s" template = "%(uploader)s/%(title)s.%(ext)s"
info_dict = { info_dict = {
@ -2793,22 +2777,19 @@ class TestParseOuttmpl:
def test_parse_outtmpl_with_missing_field(self): def test_parse_outtmpl_with_missing_field(self):
"""Test template parsing with missing field defaults to NA.""" """Test template parsing with missing field defaults to NA."""
from app.library.Utils import parse_outtmpl
template = "%(title)s - %(upload_date)s.%(ext)s" template = "%(title)s - %(upload_date)s.%(ext)s"
info_dict = { info_dict = {
"title": "Test Video", "title": "Test Video",
"ext": "mp4", "ext": "mp4",
# upload_date is missing
} }
result = parse_outtmpl(template, info_dict) result = parse_outtmpl(template, info_dict)
assert result == "Test Video - NA.mp4" assert result == "Test Video - NA.mp4", "Missing field upload_date should default to NA"
def test_parse_outtmpl_complex(self): def test_parse_outtmpl_complex(self):
"""Test complex template with multiple fields.""" """Test complex template with multiple fields."""
from app.library.Utils import parse_outtmpl
template = "%(uploader)s/%(playlist_title)s/%(playlist_index)03d - %(title)s [%(id)s].%(ext)s" template = "%(uploader)s/%(playlist_title)s/%(playlist_index)03d - %(title)s [%(id)s].%(ext)s"
info_dict = { info_dict = {
@ -2826,7 +2807,6 @@ class TestParseOuttmpl:
def test_parse_outtmpl_with_special_characters(self): def test_parse_outtmpl_with_special_characters(self):
"""Test template parsing handles special characters in values.""" """Test template parsing handles special characters in values."""
from app.library.Utils import parse_outtmpl
template = "%(title)s.%(ext)s" template = "%(title)s.%(ext)s"
info_dict = { info_dict = {
@ -2836,13 +2816,11 @@ class TestParseOuttmpl:
result = parse_outtmpl(template, info_dict) result = parse_outtmpl(template, info_dict)
# yt-dlp sanitizes special characters in filenames assert ".mp4" in result, "yt-dlp should sanitize special characters but preserve extension"
assert ".mp4" in result assert "Test" in result, "yt-dlp should preserve safe parts of title"
assert "Test" in result
def test_parse_outtmpl_with_playlist_info(self): def test_parse_outtmpl_with_playlist_info(self):
"""Test template parsing with playlist information.""" """Test template parsing with playlist information."""
from app.library.Utils import parse_outtmpl
template = "%(playlist)s/%(title)s.%(ext)s" template = "%(playlist)s/%(title)s.%(ext)s"
info_dict = { info_dict = {
@ -2854,3 +2832,19 @@ class TestParseOuttmpl:
result = parse_outtmpl(template, info_dict) result = parse_outtmpl(template, info_dict)
assert result == "My Playlist/Video Title.webm" assert result == "My Playlist/Video Title.webm"
def test_parse_outtmpl_with_restrict_filename(self):
"""Test template parsing with restrict_filename parameter."""
template = "%(uploader)s/%(title)s.%(ext)s"
info_dict = {
"uploader": "Foobar's Workshop",
"title": "Test Video",
"ext": "mp4",
}
result_unrestricted: str = parse_outtmpl(template, info_dict)
assert result_unrestricted == "Foobar's Workshop/Test Video.mp4"
result_restricted: str = parse_outtmpl(template, info_dict, params={"restrictfilenames": True})
assert result_restricted == "Foobar_s_Workshop/Test_Video.mp4"

View file

@ -75,8 +75,7 @@ class TestYtDlpOptions:
# For any ignored flag that actually exists in yt-dlp parser, ensure it is marked ignored # For any ignored flag that actually exists in yt-dlp parser, ensure it is marked ignored
present_ignored_flags = [f for f in ignored_flags if f in flag_to_ignored] present_ignored_flags = [f for f in ignored_flags if f in flag_to_ignored]
# We expect at least one to be present (e.g., -P / --paths, etc.) assert len(present_ignored_flags) > 0, "We expect at least one to be present (e.g., -P / --paths, etc.)"
assert len(present_ignored_flags) > 0
assert all(flag_to_ignored[f] is True for f in present_ignored_flags) assert all(flag_to_ignored[f] is True for f in present_ignored_flags)
@ -111,8 +110,7 @@ class TestYTDLP:
# Our __init__ code manually sets these after super() # Our __init__ code manually sets these after super()
ytdlp.params["download_archive"] = "/tmp/archive.txt" ytdlp.params["download_archive"] = "/tmp/archive.txt"
# Verify download_archive was restored to params assert ytdlp.params["download_archive"] == "/tmp/archive.txt", "Verify download_archive was restored to params"
assert ytdlp.params["download_archive"] == "/tmp/archive.txt"
# Verify archive proxy was set up # Verify archive proxy was set up
assert isinstance(ytdlp.archive, _ArchiveProxy) assert isinstance(ytdlp.archive, _ArchiveProxy)
@ -133,8 +131,7 @@ class TestYTDLP:
assert call_kwargs["params"]["quiet"] is True assert call_kwargs["params"]["quiet"] is True
assert call_kwargs["auto_init"] is False assert call_kwargs["auto_init"] is False
# Verify archive proxy is falsey assert isinstance(ytdlp.archive, _ArchiveProxy), "Verify archive proxy is falsey"
assert isinstance(ytdlp.archive, _ArchiveProxy)
assert not ytdlp.archive assert not ytdlp.archive
@patch("app.library.ytdlp.yt_dlp.YoutubeDL.__init__") @patch("app.library.ytdlp.yt_dlp.YoutubeDL.__init__")
@ -230,8 +227,7 @@ class TestYTDLP:
assert ytdlp.archive.add.call_count == 3 assert ytdlp.archive.add.call_count == 3
calls = [call[0][0] for call in ytdlp.archive.add.call_args_list] calls = [call[0][0] for call in ytdlp.archive.add.call_args_list]
# First call is main archive_id assert calls[0] == "youtube new123", "First call is main archive_id"
assert calls[0] == "youtube new123"
# Should add old IDs except the duplicate # Should add old IDs except the duplicate
assert "youtube old123" in calls assert "youtube old123" in calls

View file

@ -474,8 +474,7 @@ class TestYTDLPOpts:
assert "Failed to load" in error_args assert "Failed to load" in error_args
assert "Cookie Preset" in error_args assert "Cookie Preset" in error_args
# cookiefile should not be set assert "cookiefile" not in opts._preset_opts, "cookiefile should not be set"
assert "cookiefile" not in opts._preset_opts
def test_replacer_substitution_in_cli(self): def test_replacer_substitution_in_cli(self):
"""Test that CLI arguments get replacer substitution.""" """Test that CLI arguments get replacer substitution."""
@ -540,13 +539,11 @@ class TestARGSMerger:
merger.add(cli_with_comments) merger.add(cli_with_comments)
# Comments should be filtered out assert "#" not in merger.as_string(), "Comments should be filtered out"
assert "#" not in merger.as_string()
assert "This is a comment" not in merger.as_string() assert "This is a comment" not in merger.as_string()
assert "Another comment" not in merger.as_string() assert "Another comment" not in merger.as_string()
# Valid options should remain assert "--format" in merger.args, "Valid options should remain"
assert "--format" in merger.args
assert "best" in merger.args assert "best" in merger.args
assert "--output" in merger.args assert "--output" in merger.args
assert "test.mp4" in merger.args assert "test.mp4" in merger.args
@ -566,14 +563,12 @@ class TestARGSMerger:
merger.add(cli_with_indented_comments) merger.add(cli_with_indented_comments)
# Comments should be filtered out
result = merger.as_string() result = merger.as_string()
assert "# Indented comment with spaces" not in result assert "# Indented comment with spaces" not in result, "Comments should be filtered out"
assert "# Indented comment with tabs" not in result assert "# Indented comment with tabs" not in result
assert "# Another indented comment" not in result assert "# Another indented comment" not in result
# Valid options should remain assert "--format" in merger.args, "Valid options should remain"
assert "--format" in merger.args
assert "--output" in merger.args assert "--output" in merger.args
assert "--socket-timeout" in merger.args assert "--socket-timeout" in merger.args
@ -591,13 +586,14 @@ class TestARGSMerger:
result = merger.as_string() result = merger.as_string()
# Commented lines should be filtered out completely assert "player_js_version=actual" not in result, "Commented lines should be filtered out completely"
assert "player_js_version=actual" not in result assert "mweb,web_safari;formats=incomplete" not in result, (
# Check the specific commented variant (with comma before web_safari, not dash) "Check the specific commented variant (with comma before web_safari, not dash)"
assert "mweb,web_safari;formats=incomplete" not in result )
# Valid extractor-args should remain (with -web_safari, note the dash) assert "youtube:player-client=default,tv,mweb,-web_safari;formats=incomplete" in result, (
assert "youtube:player-client=default,tv,mweb,-web_safari;formats=incomplete" in result "Valid extractor-args should remain (with -web_safari, note the dash)"
)
assert "--socket-timeout" in merger.args assert "--socket-timeout" in merger.args
assert "60" in merger.args assert "60" in merger.args
@ -839,8 +835,7 @@ class TestYTDLPCli:
cli = YTDLPCli(item=item) cli = YTDLPCli(item=item)
command, info = cli.build() command, info = cli.build()
# Should use preset values assert info["merged"]["template"] == "%(channel)s/%(title)s.%(ext)s", "Should use preset values"
assert info["merged"]["template"] == "%(channel)s/%(title)s.%(ext)s"
assert info["merged"]["save_path"] == "/downloads/preset_folder" assert info["merged"]["save_path"] == "/downloads/preset_folder"
assert "--format 720p" in command assert "--format 720p" in command
@ -879,11 +874,9 @@ class TestYTDLPCli:
cli = YTDLPCli(item=item) cli = YTDLPCli(item=item)
command, info = cli.build() command, info = cli.build()
# Should use user values, not preset assert info["merged"]["template"] == "%(id)s.%(ext)s", "Should use user values, not preset"
assert info["merged"]["template"] == "%(id)s.%(ext)s"
assert info["merged"]["save_path"] == "/downloads/user_folder" assert info["merged"]["save_path"] == "/downloads/user_folder"
# User CLI should appear after preset CLI in command assert "--format best" in command, "User CLI should appear after preset CLI in command"
assert "--format best" in command
@patch("app.library.Presets.Presets") @patch("app.library.Presets.Presets")
@patch("app.library.YTDLPOpts.Config") @patch("app.library.YTDLPOpts.Config")
@ -998,8 +991,9 @@ class TestYTDLPCli:
cli = YTDLPCli(item=item) cli = YTDLPCli(item=item)
command, info = cli.build() command, info = cli.build()
# The implementation strips leading slash and joins with download_path assert info["merged"]["save_path"] == "/downloads/absolute/path", (
assert info["merged"]["save_path"] == "/downloads/absolute/path" "The implementation strips leading slash and joins with download_path"
)
assert "--paths" in command assert "--paths" in command
@patch("app.library.Presets.Presets") @patch("app.library.Presets.Presets")
@ -1053,5 +1047,4 @@ class TestYTDLPCli:
preset_format_idx = args_list.index("720p") if "720p" in args_list else -1 preset_format_idx = args_list.index("720p") if "720p" in args_list else -1
user_format_idx = args_list.index("best") if "best" in args_list else -1 user_format_idx = args_list.index("best") if "best" in args_list else -1
# User's 'best' should appear after preset's '720p' assert user_format_idx > preset_format_idx, "User's 'best' should appear after preset's '720p'"
assert user_format_idx > preset_format_idx