Refactor: improve linting and formatting
This commit is contained in:
arabcoders
parent
9595a69b50
commit
f32e323309
30 changed files with 301 additions and 552 deletions
4
.github/workflows/build-pr.yml
vendored
4
.github/workflows/build-pr.yml
vendored
|
|
@ -75,6 +75,10 @@ jobs:
|
|||
working-directory: ui
|
||||
run: pnpm run typecheck
|
||||
|
||||
- name: Run frontend tsc
|
||||
working-directory: ui
|
||||
run: pnpm run lint:tsc
|
||||
|
||||
- name: Run frontend tests
|
||||
working-directory: ui
|
||||
run: pnpm run test:ci
|
||||
|
|
|
|||
4
.github/workflows/main.yml
vendored
4
.github/workflows/main.yml
vendored
|
|
@ -109,6 +109,10 @@ jobs:
|
|||
working-directory: ui
|
||||
run: pnpm run typecheck
|
||||
|
||||
- name: Run frontend tsc
|
||||
working-directory: ui
|
||||
run: pnpm run lint:tsc
|
||||
|
||||
- name: Run frontend tests
|
||||
working-directory: ui
|
||||
run: pnpm run test:ci
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ from datetime import UTC, datetime, timedelta
|
|||
from functools import lru_cache, wraps
|
||||
from http.cookiejar import MozillaCookieJar
|
||||
from pathlib import Path
|
||||
from typing import Any, TypeVar
|
||||
from typing import Any
|
||||
|
||||
from Crypto.Cipher import AES
|
||||
from yt_dlp.utils import age_restricted
|
||||
|
|
@ -85,9 +85,6 @@ TAG_REGEX: re.Pattern[str] = re.compile(r"%{([^:}]+)(?::([^}]*))?}c")
|
|||
DT_PATTERN: re.Pattern[str] = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[+-]\d{2}:\d{2}))\s?")
|
||||
"Regex to match ISO 8601 datetime strings."
|
||||
|
||||
T = TypeVar("T")
|
||||
"Generic type variable."
|
||||
|
||||
|
||||
class StreamingError(Exception):
|
||||
"""Raised when an error occurs during streaming."""
|
||||
|
|
@ -1543,7 +1540,7 @@ def delete_dir(dir: Path) -> bool:
|
|||
return False
|
||||
|
||||
|
||||
def init_class(cls: type[T], data: dict, _fields=None) -> T:
|
||||
def init_class[T](cls: type[T], data: dict, _fields=None) -> T:
|
||||
"""
|
||||
Initialize a class instance with data from a dictionary, filtering out keys not present in the class fields.
|
||||
|
||||
|
|
|
|||
|
|
@ -237,7 +237,7 @@ class FFProbeResult:
|
|||
|
||||
|
||||
@timed_lru_cache(ttl_seconds=300, max_size=128)
|
||||
async def ffprobe(file: Path|str) -> FFProbeResult:
|
||||
async def ffprobe(file: Path | str) -> FFProbeResult:
|
||||
"""
|
||||
Run ffprobe on a file and return the parsed data as a dictionary.
|
||||
|
||||
|
|
|
|||
|
|
@ -1,4 +1,9 @@
|
|||
# flake8: noqa: S608
|
||||
"""
|
||||
This module is a fork of the Caribou library
|
||||
(https://pypi.org/project/caribou/) modified to work for our specific use case.
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import datetime
|
||||
import glob
|
||||
|
|
@ -18,6 +23,7 @@ UTC_LENGTH = 14
|
|||
|
||||
__version__ = "1.0.0"
|
||||
|
||||
|
||||
class Error(Exception):
|
||||
"""Base class for all Caribou errors."""
|
||||
|
||||
|
|
|
|||
|
|
@ -272,4 +272,3 @@ def find_all(items: list[dict], **conditions) -> list[dict]:
|
|||
|
||||
"""
|
||||
return filter_items(items, **conditions)
|
||||
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ if TYPE_CHECKING:
|
|||
LOG: logging.Logger = logging.getLogger(__name__)
|
||||
CACHE: Cache = Cache()
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class MatchRule:
|
||||
"""Represents a single URL matcher compiled to regex."""
|
||||
|
|
@ -720,12 +721,16 @@ class GenericTaskHandler(BaseHandler):
|
|||
)
|
||||
|
||||
if not info:
|
||||
LOG.error(f"[{definition.name}]: '{task.name}': Failed to extract info for URL '{url}' to generate archive ID. Skipping.")
|
||||
LOG.error(
|
||||
f"[{definition.name}]: '{task.name}': Failed to extract info for URL '{url}' to generate archive ID. Skipping."
|
||||
)
|
||||
CACHE.set(cache_key, None)
|
||||
continue
|
||||
|
||||
if not info.get("id") or not info.get("extractor_key"):
|
||||
LOG.error(f"[{definition.name}]: '{task.name}': Incomplete info extracted for URL '{url}' to generate archive ID. Skipping.")
|
||||
LOG.error(
|
||||
f"[{definition.name}]: '{task.name}': Incomplete info extracted for URL '{url}' to generate archive ID. Skipping."
|
||||
)
|
||||
CACHE.set(cache_key, None)
|
||||
continue
|
||||
|
||||
|
|
|
|||
|
|
@ -96,7 +96,7 @@ class TwitchHandler(BaseHandler):
|
|||
archive_id: str = entry.get("archive_id")
|
||||
task_items.append(TaskItem(url=url, title=entry.get("title"), archive_id=archive_id))
|
||||
|
||||
return TaskResult( items=task_items, metadata={ "feed_url": feed_url, "has_entries": has_items } )
|
||||
return TaskResult(items=task_items, metadata={"feed_url": feed_url, "has_entries": has_items})
|
||||
|
||||
@staticmethod
|
||||
def parse(url: str) -> str | None:
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ Migration Name: add_status_index
|
|||
Migration Version: 20251116173731
|
||||
"""
|
||||
|
||||
|
||||
async def upgrade(connection):
|
||||
"""
|
||||
Add index on json_extract(data, '$.status') for better query performance.
|
||||
|
|
@ -17,6 +18,7 @@ async def upgrade(connection):
|
|||
"""
|
||||
await connection.execute(sql)
|
||||
|
||||
|
||||
async def downgrade(connection):
|
||||
"""
|
||||
Remove the status index.
|
||||
|
|
|
|||
|
|
@ -27,9 +27,7 @@ async def add_url(queue: DownloadQueue, notify: EventBus, sid: str, data: dict):
|
|||
)
|
||||
except ValueError as e:
|
||||
LOG.exception(e)
|
||||
notify.emit(
|
||||
Events.LOG_ERROR, data={"preset": item.preset}, title="Error Adding URL", message=str(e), to=sid
|
||||
)
|
||||
notify.emit(Events.LOG_ERROR, data={"preset": item.preset}, title="Error Adding URL", message=str(e), to=sid)
|
||||
|
||||
|
||||
@route(RouteType.SOCKET, "item_cancel", "item_cancel")
|
||||
|
|
|
|||
|
|
@ -172,6 +172,7 @@ class TestCache:
|
|||
|
||||
def test_async_set(self):
|
||||
"""Test async set method using asyncio.run."""
|
||||
|
||||
async def async_test():
|
||||
await self.cache.aset("async_key", "async_value")
|
||||
assert self.cache.get("async_key") == "async_value"
|
||||
|
|
@ -180,6 +181,7 @@ class TestCache:
|
|||
|
||||
def test_async_set_with_ttl(self):
|
||||
"""Test async set with TTL using asyncio.run."""
|
||||
|
||||
async def async_test():
|
||||
await self.cache.aset("async_temp", "async_value", ttl=0.1)
|
||||
assert self.cache.get("async_temp") == "async_value"
|
||||
|
|
|
|||
|
|
@ -48,11 +48,7 @@ class TestCondition:
|
|||
extras = {"key": "value", "number": 42}
|
||||
|
||||
condition = Condition(
|
||||
id=test_id,
|
||||
name="full_test",
|
||||
filter="uploader = 'test'",
|
||||
cli="--format best",
|
||||
extras=extras
|
||||
id=test_id, name="full_test", filter="uploader = 'test'", cli="--format best", extras=extras
|
||||
)
|
||||
|
||||
assert condition.id == test_id
|
||||
|
|
@ -64,10 +60,7 @@ class TestCondition:
|
|||
def test_condition_serialize(self):
|
||||
"""Test condition serialization to dict."""
|
||||
condition = Condition(
|
||||
name="serialize_test",
|
||||
filter="title ~= 'test'",
|
||||
cli="--audio-quality 0",
|
||||
extras={"tag": "music"}
|
||||
name="serialize_test", filter="title ~= 'test'", cli="--audio-quality 0", extras={"tag": "music"}
|
||||
)
|
||||
|
||||
serialized = condition.serialize()
|
||||
|
|
@ -93,11 +86,7 @@ class TestCondition:
|
|||
|
||||
def test_condition_get_method(self):
|
||||
"""Test condition get method for accessing fields."""
|
||||
condition = Condition(
|
||||
name="get_test",
|
||||
filter="view_count > 1000",
|
||||
extras={"category": "popular"}
|
||||
)
|
||||
condition = Condition(name="get_test", filter="view_count > 1000", extras={"category": "popular"})
|
||||
|
||||
assert condition.get("name") == "get_test"
|
||||
assert condition.get("filter") == "view_count > 1000"
|
||||
|
|
@ -165,7 +154,7 @@ class TestConditions:
|
|||
# Add some test conditions
|
||||
conditions._items = [
|
||||
Condition(name="test1", filter="duration > 60"),
|
||||
Condition(name="test2", filter="uploader = 'test'")
|
||||
Condition(name="test2", filter="uploader = 'test'"),
|
||||
]
|
||||
|
||||
result = conditions.clear()
|
||||
|
|
@ -219,15 +208,15 @@ class TestConditions:
|
|||
"name": "short_videos",
|
||||
"filter": "duration < 300",
|
||||
"cli": "--format worst",
|
||||
"extras": {"category": "short"}
|
||||
"extras": {"category": "short"},
|
||||
},
|
||||
{
|
||||
"id": str(uuid.uuid4()),
|
||||
"name": "music_videos",
|
||||
"filter": "title ~= 'music'",
|
||||
"cli": "--audio-quality 0",
|
||||
"extras": {"type": "audio"}
|
||||
}
|
||||
"extras": {"type": "audio"},
|
||||
},
|
||||
]
|
||||
|
||||
file_path.write_text(json.dumps(test_data, indent=4))
|
||||
|
|
@ -254,12 +243,7 @@ class TestConditions:
|
|||
file_path = Path(temp_dir) / "no_id_conditions.json"
|
||||
|
||||
# Create test data without ID
|
||||
test_data = [
|
||||
{
|
||||
"name": "no_id_test",
|
||||
"filter": "duration > 120"
|
||||
}
|
||||
]
|
||||
test_data = [{"name": "no_id_test", "filter": "duration > 120"}]
|
||||
|
||||
file_path.write_text(json.dumps(test_data))
|
||||
|
||||
|
|
@ -280,13 +264,7 @@ class TestConditions:
|
|||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
file_path = Path(temp_dir) / "no_extras_conditions.json"
|
||||
|
||||
test_data = [
|
||||
{
|
||||
"id": str(uuid.uuid4()),
|
||||
"name": "no_extras_test",
|
||||
"filter": "uploader = 'test'"
|
||||
}
|
||||
]
|
||||
test_data = [{"id": str(uuid.uuid4()), "name": "no_extras_test", "filter": "uploader = 'test'"}]
|
||||
|
||||
file_path.write_text(json.dumps(test_data))
|
||||
|
||||
|
|
@ -321,7 +299,7 @@ class TestConditions:
|
|||
# Missing required fields
|
||||
test_data = [
|
||||
{"id": "valid", "name": "valid", "filter": "duration > 60"},
|
||||
{"invalid": "data"} # Missing required fields
|
||||
{"invalid": "data"}, # Missing required fields
|
||||
]
|
||||
|
||||
file_path.write_text(json.dumps(test_data))
|
||||
|
|
@ -345,7 +323,7 @@ class TestConditions:
|
|||
"name": "valid_test",
|
||||
"filter": "duration > 60",
|
||||
"cli": "--format best",
|
||||
"extras": {"key": "value"}
|
||||
"extras": {"key": "value"},
|
||||
}
|
||||
|
||||
result = conditions.validate(valid_condition)
|
||||
|
|
@ -357,10 +335,7 @@ class TestConditions:
|
|||
file_path = Path(temp_dir) / "validate_obj_test.json"
|
||||
conditions = Conditions(file=file_path)
|
||||
|
||||
valid_condition = Condition(
|
||||
name="valid_obj_test",
|
||||
filter="uploader = 'test'"
|
||||
)
|
||||
valid_condition = Condition(name="valid_obj_test", filter="uploader = 'test'")
|
||||
|
||||
result = conditions.validate(valid_condition)
|
||||
assert result is True
|
||||
|
|
@ -371,10 +346,7 @@ class TestConditions:
|
|||
file_path = Path(temp_dir) / "no_id_validate.json"
|
||||
conditions = Conditions(file=file_path)
|
||||
|
||||
invalid_condition = {
|
||||
"name": "no_id_test",
|
||||
"filter": "duration > 60"
|
||||
}
|
||||
invalid_condition = {"name": "no_id_test", "filter": "duration > 60"}
|
||||
|
||||
with pytest.raises(ValueError, match="No id found"):
|
||||
conditions.validate(invalid_condition)
|
||||
|
|
@ -385,10 +357,7 @@ class TestConditions:
|
|||
file_path = Path(temp_dir) / "no_name_validate.json"
|
||||
conditions = Conditions(file=file_path)
|
||||
|
||||
invalid_condition = {
|
||||
"id": str(uuid.uuid4()),
|
||||
"filter": "duration > 60"
|
||||
}
|
||||
invalid_condition = {"id": str(uuid.uuid4()), "filter": "duration > 60"}
|
||||
|
||||
with pytest.raises(ValueError, match="No name found"):
|
||||
conditions.validate(invalid_condition)
|
||||
|
|
@ -399,10 +368,7 @@ class TestConditions:
|
|||
file_path = Path(temp_dir) / "no_filter_validate.json"
|
||||
conditions = Conditions(file=file_path)
|
||||
|
||||
invalid_condition = {
|
||||
"id": str(uuid.uuid4()),
|
||||
"name": "no_filter_test"
|
||||
}
|
||||
invalid_condition = {"id": str(uuid.uuid4()), "name": "no_filter_test"}
|
||||
|
||||
with pytest.raises(ValueError, match="No filter found"):
|
||||
conditions.validate(invalid_condition)
|
||||
|
|
@ -419,7 +385,7 @@ class TestConditions:
|
|||
"name": "invalid_filter_test",
|
||||
"filter": "duration > & < 60", # Invalid syntax with consecutive operators
|
||||
"cli": "",
|
||||
"extras": {}
|
||||
"extras": {},
|
||||
}
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid filter"):
|
||||
|
|
@ -435,7 +401,7 @@ class TestConditions:
|
|||
"id": str(uuid.uuid4()),
|
||||
"name": "invalid_cli_test",
|
||||
"filter": "duration > 60",
|
||||
"cli": "--invalid-option-that-does-not-exist"
|
||||
"cli": "--invalid-option-that-does-not-exist",
|
||||
}
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid command options"):
|
||||
|
|
@ -451,7 +417,7 @@ class TestConditions:
|
|||
"id": str(uuid.uuid4()),
|
||||
"name": "invalid_extras_test",
|
||||
"filter": "duration > 60",
|
||||
"extras": "not a dict"
|
||||
"extras": "not a dict",
|
||||
}
|
||||
|
||||
with pytest.raises(ValueError, match="Extras must be a dictionary"):
|
||||
|
|
@ -474,7 +440,7 @@ class TestConditions:
|
|||
|
||||
test_conditions = [
|
||||
Condition(name="save_test1", filter="duration > 60"),
|
||||
Condition(name="save_test2", filter="uploader = 'test'")
|
||||
Condition(name="save_test2", filter="uploader = 'test'"),
|
||||
]
|
||||
|
||||
result = conditions.save(test_conditions)
|
||||
|
|
@ -495,13 +461,7 @@ class TestConditions:
|
|||
conditions = Conditions(file=file_path)
|
||||
|
||||
test_conditions = [
|
||||
{
|
||||
"id": str(uuid.uuid4()),
|
||||
"name": "dict_test",
|
||||
"filter": "duration < 300",
|
||||
"cli": "",
|
||||
"extras": {}
|
||||
}
|
||||
{"id": str(uuid.uuid4()), "name": "dict_test", "filter": "duration < 300", "cli": "", "extras": {}}
|
||||
]
|
||||
|
||||
conditions.save(test_conditions)
|
||||
|
|
|
|||
|
|
@ -77,6 +77,7 @@ class TestDownloadHooks:
|
|||
return Cfg
|
||||
|
||||
monkeypatch.setattr("app.library.Download.Config", Cfg)
|
||||
|
||||
# EventBus.get_instance is used during __init__ and start, we don't hit start here
|
||||
class EB:
|
||||
@staticmethod
|
||||
|
|
@ -113,7 +114,17 @@ class TestDownloadHooks:
|
|||
assert ev["action"] == "progress"
|
||||
# ensure only whitelisted keys included
|
||||
assert "other" not in ev
|
||||
for k in ("tmpfilename", "filename", "status", "msg", "total_bytes", "total_bytes_estimate", "downloaded_bytes", "speed", "eta"):
|
||||
for k in (
|
||||
"tmpfilename",
|
||||
"filename",
|
||||
"status",
|
||||
"msg",
|
||||
"total_bytes",
|
||||
"total_bytes_estimate",
|
||||
"downloaded_bytes",
|
||||
"speed",
|
||||
"eta",
|
||||
):
|
||||
assert k in ev
|
||||
|
||||
def test_postprocessor_hook_movefiles_sets_final_name(self, tmp_path: Path) -> None:
|
||||
|
|
|
|||
|
|
@ -54,6 +54,7 @@ class TestEncoder:
|
|||
|
||||
def test_object_with_serialize_method(self):
|
||||
"""Test that objects with serialize method use it."""
|
||||
|
||||
class CustomObject:
|
||||
def serialize(self):
|
||||
return {"custom": "data", "type": "test"}
|
||||
|
|
@ -64,6 +65,7 @@ class TestEncoder:
|
|||
|
||||
def test_object_with_dict_fallback(self):
|
||||
"""Test that objects without serialize method fall back to __dict__."""
|
||||
|
||||
class SimpleObject:
|
||||
def __init__(self):
|
||||
self.name = "test"
|
||||
|
|
@ -81,12 +83,7 @@ class TestEncoder:
|
|||
|
||||
def test_json_dumps_integration(self):
|
||||
"""Test full JSON serialization with various types."""
|
||||
data = {
|
||||
"path": Path("/tmp/test.txt"),
|
||||
"date": date(2024, 1, 1),
|
||||
"number": 42,
|
||||
"string": "test"
|
||||
}
|
||||
data = {"path": Path("/tmp/test.txt"), "date": date(2024, 1, 1), "number": 42, "string": "test"}
|
||||
|
||||
result = json.dumps(data, cls=Encoder)
|
||||
parsed = json.loads(result)
|
||||
|
|
@ -98,15 +95,13 @@ class TestEncoder:
|
|||
|
||||
def test_json_dumps_with_custom_object(self):
|
||||
"""Test JSON serialization with custom objects."""
|
||||
|
||||
class TestObject:
|
||||
def __init__(self):
|
||||
self.name = "test"
|
||||
self.items = [1, 2, 3]
|
||||
|
||||
data = {
|
||||
"object": TestObject(),
|
||||
"regular": "data"
|
||||
}
|
||||
data = {"object": TestObject(), "regular": "data"}
|
||||
|
||||
result = json.dumps(data, cls=Encoder)
|
||||
parsed = json.loads(result)
|
||||
|
|
@ -117,6 +112,7 @@ class TestEncoder:
|
|||
|
||||
def test_nested_serialization(self):
|
||||
"""Test serialization of nested structures with various types."""
|
||||
|
||||
class CustomObj:
|
||||
def serialize(self):
|
||||
return {"serialized": True}
|
||||
|
|
@ -125,10 +121,7 @@ class TestEncoder:
|
|||
"paths": [Path("/tmp/1.txt"), Path("/tmp/2.txt")],
|
||||
"dates": [date(2024, 1, 1), date(2024, 12, 31)],
|
||||
"custom": CustomObj(),
|
||||
"nested": {
|
||||
"path": Path("/nested/path"),
|
||||
"date": date(2024, 6, 15)
|
||||
}
|
||||
"nested": {"path": Path("/nested/path"), "date": date(2024, 6, 15)},
|
||||
}
|
||||
|
||||
result = json.dumps(data, cls=Encoder)
|
||||
|
|
@ -149,6 +142,7 @@ class TestEncoder:
|
|||
|
||||
# Mock isinstance to return True for DateRange
|
||||
import builtins
|
||||
|
||||
original_isinstance = builtins.isinstance
|
||||
|
||||
def mock_isinstance(obj, cls):
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ class TestFFProbe:
|
|||
def teardown_method(self):
|
||||
"""Clean up test files."""
|
||||
import shutil
|
||||
|
||||
shutil.rmtree(self.temp_dir, ignore_errors=True)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -64,6 +65,7 @@ class TestFFProbe:
|
|||
# Mock subprocess to avoid actual ffprobe execution
|
||||
call_count = 0
|
||||
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
||||
|
||||
def create_mock_process(*_args, **_kwargs):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
|
|
@ -130,20 +132,11 @@ class TestFFProbe:
|
|||
assert result.metadata == {}
|
||||
|
||||
# Test adding streams
|
||||
video_stream = FFStream({
|
||||
"index": 0,
|
||||
"codec_type": "video",
|
||||
"codec_name": "h264",
|
||||
"width": 1920,
|
||||
"height": 1080
|
||||
})
|
||||
video_stream = FFStream(
|
||||
{"index": 0, "codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080}
|
||||
)
|
||||
|
||||
audio_stream = FFStream({
|
||||
"index": 1,
|
||||
"codec_type": "audio",
|
||||
"codec_name": "aac",
|
||||
"channels": 2
|
||||
})
|
||||
audio_stream = FFStream({"index": 1, "codec_type": "audio", "codec_name": "aac", "channels": 2})
|
||||
|
||||
result.video.append(video_stream)
|
||||
result.audio.append(audio_stream)
|
||||
|
|
@ -158,13 +151,9 @@ class TestFFProbe:
|
|||
from app.library.ffprobe import FFStream
|
||||
|
||||
# Test video stream
|
||||
video_stream = FFStream({
|
||||
"codec_type": "video",
|
||||
"codec_name": "h264",
|
||||
"width": 1920,
|
||||
"height": 1080,
|
||||
"duration": "10.5"
|
||||
})
|
||||
video_stream = FFStream(
|
||||
{"codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080, "duration": "10.5"}
|
||||
)
|
||||
|
||||
assert video_stream.is_video()
|
||||
assert not video_stream.is_audio()
|
||||
|
|
@ -172,12 +161,7 @@ class TestFFProbe:
|
|||
assert not video_stream.is_attachment()
|
||||
|
||||
# Test audio stream
|
||||
audio_stream = FFStream({
|
||||
"codec_type": "audio",
|
||||
"codec_name": "aac",
|
||||
"channels": 2,
|
||||
"sample_rate": "44100"
|
||||
})
|
||||
audio_stream = FFStream({"codec_type": "audio", "codec_name": "aac", "channels": 2, "sample_rate": "44100"})
|
||||
|
||||
assert not audio_stream.is_video()
|
||||
assert audio_stream.is_audio()
|
||||
|
|
@ -185,10 +169,7 @@ class TestFFProbe:
|
|||
assert not audio_stream.is_attachment()
|
||||
|
||||
# Test subtitle stream
|
||||
subtitle_stream = FFStream({
|
||||
"codec_type": "subtitle",
|
||||
"codec_name": "subrip"
|
||||
})
|
||||
subtitle_stream = FFStream({"codec_type": "subtitle", "codec_name": "subrip"})
|
||||
|
||||
assert not subtitle_stream.is_video()
|
||||
assert not subtitle_stream.is_audio()
|
||||
|
|
@ -196,10 +177,7 @@ class TestFFProbe:
|
|||
assert not subtitle_stream.is_attachment()
|
||||
|
||||
# Test attachment stream
|
||||
attachment_stream = FFStream({
|
||||
"codec_type": "attachment",
|
||||
"codec_name": "ttf"
|
||||
})
|
||||
attachment_stream = FFStream({"codec_type": "attachment", "codec_name": "ttf"})
|
||||
|
||||
assert not attachment_stream.is_video()
|
||||
assert not attachment_stream.is_audio()
|
||||
|
|
|
|||
|
|
@ -252,7 +252,9 @@ class TestItemDTO:
|
|||
}
|
||||
|
||||
with (
|
||||
patch("app.library.ItemDTO.ItemDTO.get_file", autospec=True, return_value=Path("/downloads/video.mp4")) as mock_get_file,
|
||||
patch(
|
||||
"app.library.ItemDTO.ItemDTO.get_file", autospec=True, return_value=Path("/downloads/video.mp4")
|
||||
) as mock_get_file,
|
||||
patch("app.library.ItemDTO.get_file_sidecar", return_value=expected_sidecar) as mock_utils_sidecar,
|
||||
):
|
||||
result = dto.get_file_sidecar()
|
||||
|
|
|
|||
|
|
@ -63,11 +63,7 @@ class TestTargetRequest:
|
|||
|
||||
def test_target_request_creation_defaults(self):
|
||||
"""Test creating a TargetRequest with defaults."""
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
|
||||
assert request.type == "json"
|
||||
assert request.method == "POST"
|
||||
|
|
@ -79,14 +75,10 @@ class TestTargetRequest:
|
|||
"""Test creating a TargetRequest with headers."""
|
||||
headers = [
|
||||
TargetRequestHeader(key="Authorization", value="Bearer token"),
|
||||
TargetRequestHeader(key="Content-Type", value="application/json")
|
||||
TargetRequestHeader(key="Content-Type", value="application/json"),
|
||||
]
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook",
|
||||
headers=headers,
|
||||
data_key="payload"
|
||||
type="json", method="POST", url="https://example.com/webhook", headers=headers, data_key="payload"
|
||||
)
|
||||
|
||||
assert len(request.headers) == 2
|
||||
|
|
@ -97,11 +89,7 @@ class TestTargetRequest:
|
|||
"""Test serializing a TargetRequest."""
|
||||
headers = [TargetRequestHeader(key="X-Token", value="abc123")]
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="PUT",
|
||||
url="https://api.example.com/notify",
|
||||
headers=headers,
|
||||
data_key="content"
|
||||
type="json", method="PUT", url="https://api.example.com/notify", headers=headers, data_key="content"
|
||||
)
|
||||
|
||||
serialized = request.serialize()
|
||||
|
|
@ -110,17 +98,13 @@ class TestTargetRequest:
|
|||
"method": "PUT",
|
||||
"url": "https://api.example.com/notify",
|
||||
"data_key": "content",
|
||||
"headers": [{"key": "X-Token", "value": "abc123"}]
|
||||
"headers": [{"key": "X-Token", "value": "abc123"}],
|
||||
}
|
||||
assert serialized == expected
|
||||
|
||||
def test_target_request_json(self):
|
||||
"""Test JSON serialization of TargetRequest."""
|
||||
request = TargetRequest(
|
||||
type="form",
|
||||
method="POST",
|
||||
url="https://webhook.site/test"
|
||||
)
|
||||
request = TargetRequest(type="form", method="POST", url="https://webhook.site/test")
|
||||
json_str = request.json()
|
||||
|
||||
parsed = json.loads(json_str)
|
||||
|
|
@ -130,11 +114,7 @@ class TestTargetRequest:
|
|||
|
||||
def test_target_request_get(self):
|
||||
"""Test get method of TargetRequest."""
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
|
||||
assert request.get("type") == "json"
|
||||
assert request.get("method") == "POST"
|
||||
|
|
@ -146,16 +126,8 @@ class TestTarget:
|
|||
|
||||
def test_target_creation_minimal(self):
|
||||
"""Test creating a Target with minimal required fields."""
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
target = Target(
|
||||
id=str(uuid.uuid4()),
|
||||
name="Test Webhook",
|
||||
request=request
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(id=str(uuid.uuid4()), name="Test Webhook", request=request)
|
||||
|
||||
assert target.name == "Test Webhook"
|
||||
assert target.on == []
|
||||
|
|
@ -165,17 +137,13 @@ class TestTarget:
|
|||
def test_target_creation_full(self):
|
||||
"""Test creating a Target with all fields."""
|
||||
target_id = str(uuid.uuid4())
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(
|
||||
id=target_id,
|
||||
name="Full Test Webhook",
|
||||
on=["item_completed", "item_failed"],
|
||||
presets=["default", "audio_only"],
|
||||
request=request
|
||||
request=request,
|
||||
)
|
||||
|
||||
assert target.id == target_id
|
||||
|
|
@ -186,18 +154,8 @@ class TestTarget:
|
|||
def test_target_serialize(self):
|
||||
"""Test serializing a Target."""
|
||||
target_id = str(uuid.uuid4())
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
target = Target(
|
||||
id=target_id,
|
||||
name="Test Target",
|
||||
on=["item_completed"],
|
||||
presets=["default"],
|
||||
request=request
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(id=target_id, name="Test Target", on=["item_completed"], presets=["default"], request=request)
|
||||
|
||||
serialized = target.serialize()
|
||||
assert serialized["id"] == target_id
|
||||
|
|
@ -208,16 +166,8 @@ class TestTarget:
|
|||
|
||||
def test_target_json(self):
|
||||
"""Test JSON serialization of Target."""
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
target = Target(
|
||||
id=str(uuid.uuid4()),
|
||||
name="JSON Test",
|
||||
request=request
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(id=str(uuid.uuid4()), name="JSON Test", request=request)
|
||||
|
||||
json_str = target.json()
|
||||
parsed = json.loads(json_str)
|
||||
|
|
@ -225,16 +175,8 @@ class TestTarget:
|
|||
|
||||
def test_target_get(self):
|
||||
"""Test get method of Target."""
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
target = Target(
|
||||
id=str(uuid.uuid4()),
|
||||
name="Get Test",
|
||||
request=request
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(id=str(uuid.uuid4()), name="Get Test", request=request)
|
||||
|
||||
assert target.get("name") == "Get Test"
|
||||
assert target.get("nonexistent", "default") == "default"
|
||||
|
|
@ -299,11 +241,7 @@ class TestNotification:
|
|||
@patch("app.library.Notifications.BackgroundWorker")
|
||||
def test_notification_singleton(self, mock_background_worker, mock_config):
|
||||
"""Test that Notification follows singleton pattern."""
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False,
|
||||
config_path="/tmp",
|
||||
app_version="1.0.0"
|
||||
)
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_background_worker.get_instance.return_value = Mock()
|
||||
|
||||
instance1 = Notification.get_instance()
|
||||
|
|
@ -316,11 +254,7 @@ class TestNotification:
|
|||
@patch("app.library.Notifications.BackgroundWorker")
|
||||
def test_notification_init_default_params(self, mock_background_worker, mock_config):
|
||||
"""Test Notification initialization with default parameters."""
|
||||
mock_config_instance = Mock(
|
||||
debug=False,
|
||||
config_path="/tmp/test",
|
||||
app_version="1.0.0"
|
||||
)
|
||||
mock_config_instance = Mock(debug=False, config_path="/tmp/test", app_version="1.0.0")
|
||||
mock_config.get_instance.return_value = mock_config_instance
|
||||
mock_background_worker.get_instance.return_value = Mock()
|
||||
|
||||
|
|
@ -337,11 +271,7 @@ class TestNotification:
|
|||
def test_notification_init_custom_params(self, mock_background_worker, mock_config):
|
||||
"""Test Notification initialization with custom parameters."""
|
||||
_ = mock_background_worker, mock_config # Suppress unused variable warnings
|
||||
mock_config_instance = Mock(
|
||||
debug=True,
|
||||
config_path="/custom/path",
|
||||
app_version="2.0.0"
|
||||
)
|
||||
mock_config_instance = Mock(debug=True, config_path="/custom/path", app_version="2.0.0")
|
||||
mock_client = Mock(spec=httpx.AsyncClient)
|
||||
mock_encoder = Mock(spec=Encoder)
|
||||
mock_worker = Mock(spec=BackgroundWorker)
|
||||
|
|
@ -354,7 +284,7 @@ class TestNotification:
|
|||
client=mock_client,
|
||||
encoder=mock_encoder,
|
||||
config=mock_config_instance,
|
||||
background_worker=mock_worker
|
||||
background_worker=mock_worker,
|
||||
)
|
||||
|
||||
assert notification._debug is True
|
||||
|
|
@ -365,12 +295,11 @@ class TestNotification:
|
|||
|
||||
def test_get_targets_empty(self):
|
||||
"""Test get_targets when no targets are loaded."""
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
with (
|
||||
patch("app.library.Notifications.Config") as mock_config,
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
|
||||
):
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
|
@ -381,12 +310,11 @@ class TestNotification:
|
|||
|
||||
def test_clear_targets(self):
|
||||
"""Test clearing notification targets."""
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
with (
|
||||
patch("app.library.Notifications.Config") as mock_config,
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
|
||||
):
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
|
@ -405,25 +333,15 @@ class TestNotification:
|
|||
@patch("app.library.Notifications.BackgroundWorker")
|
||||
def test_save_targets(self, mock_worker, mock_config):
|
||||
"""Test saving targets to file."""
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file:
|
||||
temp_path = temp_file.name
|
||||
|
||||
# Create a test target
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
target = Target(
|
||||
id=str(uuid.uuid4()),
|
||||
name="Test Target",
|
||||
request=request
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(id=str(uuid.uuid4()), name="Test Target", request=request)
|
||||
|
||||
notification = Notification.get_instance(file=temp_path)
|
||||
result = notification.save([target])
|
||||
|
|
@ -445,9 +363,7 @@ class TestNotification:
|
|||
@patch("app.library.Notifications.BackgroundWorker")
|
||||
def test_load_targets_empty_file(self, mock_worker, mock_config):
|
||||
"""Test loading targets from empty file."""
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file:
|
||||
|
|
@ -468,9 +384,7 @@ class TestNotification:
|
|||
@patch("app.library.Notifications.Presets")
|
||||
def test_load_targets_valid_file(self, mock_presets, mock_worker, mock_config):
|
||||
"""Test loading targets from valid file."""
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
# Mock preset with name attribute
|
||||
|
|
@ -478,19 +392,21 @@ class TestNotification:
|
|||
mock_preset.name = "default"
|
||||
mock_presets.get_instance.return_value.get_all.return_value = [mock_preset]
|
||||
|
||||
target_data = [{
|
||||
"id": str(uuid.uuid4()),
|
||||
"name": "Test Webhook",
|
||||
"on": ["item_completed"],
|
||||
"presets": ["default"],
|
||||
"request": {
|
||||
"type": "json",
|
||||
"method": "POST",
|
||||
"url": "https://example.com/webhook",
|
||||
"data_key": "data",
|
||||
"headers": []
|
||||
target_data = [
|
||||
{
|
||||
"id": str(uuid.uuid4()),
|
||||
"name": "Test Webhook",
|
||||
"on": ["item_completed"],
|
||||
"presets": ["default"],
|
||||
"request": {
|
||||
"type": "json",
|
||||
"method": "POST",
|
||||
"url": "https://example.com/webhook",
|
||||
"data_key": "data",
|
||||
"headers": [],
|
||||
},
|
||||
}
|
||||
}]
|
||||
]
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file:
|
||||
json.dump(target_data, temp_file)
|
||||
|
|
@ -508,12 +424,11 @@ class TestNotification:
|
|||
|
||||
def test_make_target(self):
|
||||
"""Test make_target method."""
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
with (
|
||||
patch("app.library.Notifications.Config") as mock_config,
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
|
||||
):
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
|
@ -528,8 +443,8 @@ class TestNotification:
|
|||
"method": "POST",
|
||||
"url": "https://example.com/webhook",
|
||||
"data_key": "payload",
|
||||
"headers": [{"key": "Authorization", "value": "Bearer token"}]
|
||||
}
|
||||
"headers": [{"key": "Authorization", "value": "Bearer token"}],
|
||||
},
|
||||
}
|
||||
|
||||
target = notification.make_target(target_dict)
|
||||
|
|
@ -548,14 +463,13 @@ class TestNotification:
|
|||
target_dict = {
|
||||
"id": str(uuid.uuid4()),
|
||||
"name": "Valid Target",
|
||||
"request": {
|
||||
"url": "https://example.com/webhook"
|
||||
}
|
||||
"request": {"url": "https://example.com/webhook"},
|
||||
}
|
||||
|
||||
with patch("app.library.Notifications.NotificationEvents.get_events") as mock_events, \
|
||||
patch("app.library.Notifications.Presets") as mock_presets:
|
||||
|
||||
with (
|
||||
patch("app.library.Notifications.NotificationEvents.get_events") as mock_events,
|
||||
patch("app.library.Notifications.Presets") as mock_presets,
|
||||
):
|
||||
mock_events.return_value.values.return_value = ["item_completed"]
|
||||
mock_presets.get_instance.return_value.get_all.return_value = []
|
||||
|
||||
|
|
@ -564,12 +478,7 @@ class TestNotification:
|
|||
|
||||
def test_validate_target_missing_id(self):
|
||||
"""Test validate method with missing ID."""
|
||||
target_dict = {
|
||||
"name": "Missing ID Target",
|
||||
"request": {
|
||||
"url": "https://example.com/webhook"
|
||||
}
|
||||
}
|
||||
target_dict = {"name": "Missing ID Target", "request": {"url": "https://example.com/webhook"}}
|
||||
|
||||
with pytest.raises(ValueError, match=r"Invalid notification target\. No ID found\."):
|
||||
Notification.validate(target_dict)
|
||||
|
|
@ -579,9 +488,7 @@ class TestNotification:
|
|||
target_dict = {
|
||||
"id": "invalid-uuid",
|
||||
"name": "Invalid ID Target",
|
||||
"request": {
|
||||
"url": "https://example.com/webhook"
|
||||
}
|
||||
"request": {"url": "https://example.com/webhook"},
|
||||
}
|
||||
|
||||
with pytest.raises(ValueError, match=r"Invalid notification target\. No ID found\."):
|
||||
|
|
@ -589,33 +496,21 @@ class TestNotification:
|
|||
|
||||
def test_validate_target_missing_name(self):
|
||||
"""Test validate method with missing name."""
|
||||
target_dict = {
|
||||
"id": str(uuid.uuid4()),
|
||||
"request": {
|
||||
"url": "https://example.com/webhook"
|
||||
}
|
||||
}
|
||||
target_dict = {"id": str(uuid.uuid4()), "request": {"url": "https://example.com/webhook"}}
|
||||
|
||||
with pytest.raises(ValueError, match=r"Invalid notification target\. No name found\."):
|
||||
Notification.validate(target_dict)
|
||||
|
||||
def test_validate_target_missing_request(self):
|
||||
"""Test validate method with missing request."""
|
||||
target_dict = {
|
||||
"id": str(uuid.uuid4()),
|
||||
"name": "Missing Request Target"
|
||||
}
|
||||
target_dict = {"id": str(uuid.uuid4()), "name": "Missing Request Target"}
|
||||
|
||||
with pytest.raises(ValueError, match=r"Invalid notification target\. No request details found\."):
|
||||
Notification.validate(target_dict)
|
||||
|
||||
def test_validate_target_missing_url(self):
|
||||
"""Test validate method with missing URL."""
|
||||
target_dict = {
|
||||
"id": str(uuid.uuid4()),
|
||||
"name": "Missing URL Target",
|
||||
"request": {}
|
||||
}
|
||||
target_dict = {"id": str(uuid.uuid4()), "name": "Missing URL Target", "request": {}}
|
||||
|
||||
with pytest.raises(ValueError, match=r"Invalid notification target\. No URL found\."):
|
||||
Notification.validate(target_dict)
|
||||
|
|
@ -627,8 +522,8 @@ class TestNotification:
|
|||
"name": "Invalid Method Target",
|
||||
"request": {
|
||||
"url": "https://example.com/webhook",
|
||||
"method": "GET" # Only POST and PUT are allowed
|
||||
}
|
||||
"method": "GET", # Only POST and PUT are allowed
|
||||
},
|
||||
}
|
||||
|
||||
with pytest.raises(ValueError, match=r"Invalid notification target\. Invalid method found\."):
|
||||
|
|
@ -641,8 +536,8 @@ class TestNotification:
|
|||
"name": "Invalid Type Target",
|
||||
"request": {
|
||||
"url": "https://example.com/webhook",
|
||||
"type": "xml" # Only json and form are allowed
|
||||
}
|
||||
"type": "xml", # Only json and form are allowed
|
||||
},
|
||||
}
|
||||
|
||||
with pytest.raises(ValueError, match=r"Invalid notification target\. Invalid type found\."):
|
||||
|
|
@ -653,9 +548,7 @@ class TestNotification:
|
|||
@patch("app.library.Notifications.EventBus")
|
||||
def test_attach(self, mock_eventbus, mock_worker, mock_config):
|
||||
"""Test attach method."""
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
mock_eventbus_instance = Mock()
|
||||
mock_eventbus.get_instance.return_value = mock_eventbus_instance
|
||||
|
|
@ -674,16 +567,11 @@ class TestNotification:
|
|||
@patch("app.library.Notifications.BackgroundWorker")
|
||||
async def test_send_no_targets(self, mock_worker, mock_config):
|
||||
"""Test send method with no targets."""
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
event = Event(
|
||||
event="test",
|
||||
data={"test": "data"}
|
||||
)
|
||||
event = Event(event="test", data={"test": "data"})
|
||||
|
||||
result = await notification.send(event)
|
||||
assert result == []
|
||||
|
|
@ -693,28 +581,18 @@ class TestNotification:
|
|||
@patch("app.library.Notifications.BackgroundWorker")
|
||||
async def test_send_invalid_event_data(self, mock_worker, mock_config):
|
||||
"""Test send method with invalid event data."""
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
# Add a target
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
target = Target(
|
||||
id=str(uuid.uuid4()),
|
||||
name="Test Target",
|
||||
request=request
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(id=str(uuid.uuid4()), name="Test Target", request=request)
|
||||
notification._targets = [target]
|
||||
|
||||
event = Event(
|
||||
event="test",
|
||||
data="invalid_string_data" # Should be ItemDTO or dict
|
||||
data="invalid_string_data", # Should be ItemDTO or dict
|
||||
)
|
||||
|
||||
result = await notification.send(event)
|
||||
|
|
@ -725,9 +603,7 @@ class TestNotification:
|
|||
@patch("app.library.Notifications.BackgroundWorker")
|
||||
async def test_send_with_http_target(self, mock_worker, mock_config):
|
||||
"""Test send method with HTTP target."""
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
# Mock HTTP client
|
||||
|
|
@ -740,29 +616,15 @@ class TestNotification:
|
|||
notification = Notification.get_instance(client=mock_client)
|
||||
|
||||
# Add HTTP target
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
target = Target(
|
||||
id=str(uuid.uuid4()),
|
||||
name="Test HTTP Target",
|
||||
request=request
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(id=str(uuid.uuid4()), name="Test HTTP Target", request=request)
|
||||
notification._targets = [target]
|
||||
|
||||
# Create test event
|
||||
item_dto = ItemDTO(
|
||||
id="test_id",
|
||||
url="https://youtube.com/watch?v=test",
|
||||
title="Test Video",
|
||||
folder="/downloads"
|
||||
)
|
||||
event = Event(
|
||||
event="item_completed",
|
||||
data=item_dto
|
||||
id="test_id", url="https://youtube.com/watch?v=test", title="Test Video", folder="/downloads"
|
||||
)
|
||||
event = Event(event="item_completed", data=item_dto)
|
||||
|
||||
result = await notification.send(event)
|
||||
|
||||
|
|
@ -777,35 +639,21 @@ class TestNotification:
|
|||
async def test_send_with_apprise_target(self, mock_worker, mock_config):
|
||||
"""Test send method with Apprise target."""
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0",
|
||||
apprise_config="/tmp/apprise.conf"
|
||||
debug=False, config_path="/tmp", app_version="1.0.0", apprise_config="/tmp/apprise.conf"
|
||||
)
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
||||
# Add Apprise target (non-HTTP URL)
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="discord://webhook_id/webhook_token"
|
||||
)
|
||||
target = Target(
|
||||
id=str(uuid.uuid4()),
|
||||
name="Test Discord Target",
|
||||
request=request
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="discord://webhook_id/webhook_token")
|
||||
target = Target(id=str(uuid.uuid4()), name="Test Discord Target", request=request)
|
||||
notification._targets = [target]
|
||||
|
||||
# Create test event
|
||||
event = Event(
|
||||
event="item_completed",
|
||||
data={"test": "data"}
|
||||
)
|
||||
|
||||
with patch("app.library.Notifications.Path") as mock_path, \
|
||||
patch("builtins.__import__") as mock_import:
|
||||
event = Event(event="item_completed", data={"test": "data"})
|
||||
|
||||
with patch("app.library.Notifications.Path") as mock_path, patch("builtins.__import__") as mock_import:
|
||||
# Mock apprise config file doesn't exist
|
||||
mock_path.return_value.exists.return_value = False
|
||||
|
||||
|
|
@ -825,103 +673,70 @@ class TestNotification:
|
|||
|
||||
def test_check_preset_no_presets(self):
|
||||
"""Test _check_preset method with target having no preset filters."""
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
with (
|
||||
patch("app.library.Notifications.Config") as mock_config,
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
|
||||
):
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(
|
||||
id=str(uuid.uuid4()),
|
||||
name="No Preset Filter",
|
||||
presets=[], # No preset filter
|
||||
request=request
|
||||
request=request,
|
||||
)
|
||||
|
||||
event = Event(
|
||||
event="item_completed",
|
||||
data={"preset": "default"}
|
||||
)
|
||||
event = Event(event="item_completed", data={"preset": "default"})
|
||||
|
||||
result = notification._check_preset(target, event)
|
||||
assert result is True # Should pass when no preset filter
|
||||
|
||||
def test_check_preset_with_matching_preset(self):
|
||||
"""Test _check_preset method with matching preset."""
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
with (
|
||||
patch("app.library.Notifications.Config") as mock_config,
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
|
||||
):
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(
|
||||
id=str(uuid.uuid4()),
|
||||
name="Preset Filter",
|
||||
presets=["default", "audio_only"],
|
||||
request=request
|
||||
id=str(uuid.uuid4()), name="Preset Filter", presets=["default", "audio_only"], request=request
|
||||
)
|
||||
|
||||
# Test with ItemDTO
|
||||
item_dto = ItemDTO(
|
||||
id="test_id",
|
||||
url="https://youtube.com/test",
|
||||
title="Test Video",
|
||||
folder="/downloads",
|
||||
preset="default"
|
||||
)
|
||||
event = Event(
|
||||
event="item_completed",
|
||||
data=item_dto
|
||||
id="test_id", url="https://youtube.com/test", title="Test Video", folder="/downloads", preset="default"
|
||||
)
|
||||
event = Event(event="item_completed", data=item_dto)
|
||||
|
||||
result = notification._check_preset(target, event)
|
||||
assert result is True
|
||||
|
||||
def test_check_preset_with_non_matching_preset(self):
|
||||
"""Test _check_preset method with non-matching preset."""
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
with (
|
||||
patch("app.library.Notifications.Config") as mock_config,
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
|
||||
):
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
target = Target(
|
||||
id=str(uuid.uuid4()),
|
||||
name="Preset Filter",
|
||||
presets=["audio_only"],
|
||||
request=request
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(id=str(uuid.uuid4()), name="Preset Filter", presets=["audio_only"], request=request)
|
||||
|
||||
event = Event(
|
||||
event="item_completed",
|
||||
data={"preset": "video_only"} # Different preset
|
||||
data={"preset": "video_only"}, # Different preset
|
||||
)
|
||||
|
||||
result = notification._check_preset(target, event)
|
||||
|
|
@ -929,35 +744,23 @@ class TestNotification:
|
|||
|
||||
def test_emit_invalid_event(self):
|
||||
"""Test emit method with invalid event."""
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
with (
|
||||
patch("app.library.Notifications.Config") as mock_config,
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
|
||||
):
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker_instance = Mock()
|
||||
mock_worker.get_instance.return_value = mock_worker_instance
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
||||
# Add a target
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
target = Target(
|
||||
id=str(uuid.uuid4()),
|
||||
name="Test Target",
|
||||
request=request
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(id=str(uuid.uuid4()), name="Test Target", request=request)
|
||||
notification._targets = [target]
|
||||
|
||||
# Create invalid event
|
||||
event = Event(
|
||||
event="invalid_event",
|
||||
data={"test": "data"}
|
||||
)
|
||||
event = Event(event="invalid_event", data={"test": "data"})
|
||||
|
||||
with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=False):
|
||||
result = notification.emit(event, None)
|
||||
|
|
@ -968,35 +771,23 @@ class TestNotification:
|
|||
|
||||
def test_emit_valid_event(self):
|
||||
"""Test emit method with valid event."""
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
with (
|
||||
patch("app.library.Notifications.Config") as mock_config,
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
|
||||
):
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker_instance = Mock()
|
||||
mock_worker.get_instance.return_value = mock_worker_instance
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
||||
# Add a target
|
||||
request = TargetRequest(
|
||||
type="json",
|
||||
method="POST",
|
||||
url="https://example.com/webhook"
|
||||
)
|
||||
target = Target(
|
||||
id=str(uuid.uuid4()),
|
||||
name="Test Target",
|
||||
request=request
|
||||
)
|
||||
request = TargetRequest(type="json", method="POST", url="https://example.com/webhook")
|
||||
target = Target(id=str(uuid.uuid4()), name="Test Target", request=request)
|
||||
notification._targets = [target]
|
||||
|
||||
# Create valid event
|
||||
event = Event(
|
||||
event="item_completed",
|
||||
data={"test": "data"}
|
||||
)
|
||||
event = Event(event="item_completed", data={"test": "data"})
|
||||
|
||||
with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=True):
|
||||
result = notification.emit(event, None)
|
||||
|
|
@ -1008,12 +799,11 @@ class TestNotification:
|
|||
@pytest.mark.asyncio
|
||||
async def test_noop(self):
|
||||
"""Test noop method."""
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
with (
|
||||
patch("app.library.Notifications.Config") as mock_config,
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker,
|
||||
):
|
||||
mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0")
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
"""Tests for the generic operations module."""
|
||||
|
||||
|
||||
from app.library.operations import (
|
||||
Operation,
|
||||
filter_items,
|
||||
|
|
|
|||
|
|
@ -216,7 +216,9 @@ class TestActionAndCheck:
|
|||
@patch.object(PackageInstaller, "_install_pkg")
|
||||
@patch.object(PackageInstaller, "_get_installed_version")
|
||||
@patch.object(PackageInstaller, "_get_latest_version")
|
||||
def test_action_upgrade_skip_when_latest(self, mock_get_latest, mock_get_installed, mock_install, tmp_path: Path) -> None:
|
||||
def test_action_upgrade_skip_when_latest(
|
||||
self, mock_get_latest, mock_get_installed, mock_install, tmp_path: Path
|
||||
) -> None:
|
||||
inst = PackageInstaller(pkg_path=tmp_path)
|
||||
mock_get_installed.return_value = "2.0.0"
|
||||
mock_get_latest.return_value = "2.0.0"
|
||||
|
|
@ -226,7 +228,9 @@ class TestActionAndCheck:
|
|||
@patch.object(PackageInstaller, "_install_pkg")
|
||||
@patch.object(PackageInstaller, "_get_installed_version")
|
||||
@patch.object(PackageInstaller, "_get_latest_version")
|
||||
def test_action_upgrade_runs_when_newer_available(self, mock_get_latest, mock_get_installed, mock_install, tmp_path: Path) -> None:
|
||||
def test_action_upgrade_runs_when_newer_available(
|
||||
self, mock_get_latest, mock_get_installed, mock_install, tmp_path: Path
|
||||
) -> None:
|
||||
inst = PackageInstaller(pkg_path=tmp_path)
|
||||
mock_get_installed.return_value = "1.0.0"
|
||||
mock_get_latest.return_value = "1.1.0"
|
||||
|
|
|
|||
|
|
@ -27,6 +27,7 @@ async def test_make_playlist_no_subtitles(tmp_path: Path, monkeypatch: pytest.Mo
|
|||
|
||||
# Patch module-level quote to be robust for Path objects
|
||||
from urllib.parse import quote as _std_quote
|
||||
|
||||
monkeypatch.setattr("app.library.Playlist.quote", lambda v: _std_quote(str(v)))
|
||||
|
||||
pl = Playlist(download_path=base, url="http://localhost/")
|
||||
|
|
@ -67,6 +68,7 @@ async def test_make_playlist_with_subtitles(tmp_path: Path, monkeypatch: pytest.
|
|||
monkeypatch.setattr("app.library.Playlist.get_file_sidecar", lambda _f: sidecar)
|
||||
|
||||
from urllib.parse import quote as _std_quote
|
||||
|
||||
monkeypatch.setattr("app.library.Playlist.quote", lambda v: _std_quote(str(v)))
|
||||
|
||||
pl = Playlist(download_path=base, url="https://server/")
|
||||
|
|
|
|||
|
|
@ -228,7 +228,17 @@ class TestScheduler:
|
|||
def test_add_executes_function_when_cron_runs(self, cron_patch) -> None:
|
||||
# Cron stub that auto-runs the function on creation when start=True
|
||||
class AutoRunCron(DummyCron):
|
||||
def __init__(self, *_, spec: str, func: callable, args: tuple = (), kwargs: dict | None = None, uuid: str | None = None, start: bool | None = None, loop: asyncio.AbstractEventLoop | None = None):
|
||||
def __init__(
|
||||
self,
|
||||
*_,
|
||||
spec: str,
|
||||
func: callable,
|
||||
args: tuple = (),
|
||||
kwargs: dict | None = None,
|
||||
uuid: str | None = None,
|
||||
start: bool | None = None,
|
||||
loop: asyncio.AbstractEventLoop | None = None,
|
||||
):
|
||||
super().__init__(spec=spec, func=func, args=args, kwargs=kwargs, uuid=uuid, start=start, loop=loop)
|
||||
if start:
|
||||
# Simulate immediate execution
|
||||
|
|
@ -253,7 +263,17 @@ class TestScheduler:
|
|||
async def test_event_schedule_runs_function(self, cron_patch) -> None:
|
||||
# Cron stub that auto-runs the function on creation
|
||||
class AutoRunCron(DummyCron):
|
||||
def __init__(self, *_, spec: str, func: callable, args: tuple = (), kwargs: dict | None = None, uuid: str | None = None, start: bool | None = None, loop: asyncio.AbstractEventLoop | None = None):
|
||||
def __init__(
|
||||
self,
|
||||
*_,
|
||||
spec: str,
|
||||
func: callable,
|
||||
args: tuple = (),
|
||||
kwargs: dict | None = None,
|
||||
uuid: str | None = None,
|
||||
start: bool | None = None,
|
||||
loop: asyncio.AbstractEventLoop | None = None,
|
||||
):
|
||||
super().__init__(spec=spec, func=func, args=args, kwargs=kwargs, uuid=uuid, start=start, loop=loop)
|
||||
if start:
|
||||
self.func(*self.args, **self.kwargs)
|
||||
|
|
|
|||
|
|
@ -178,9 +178,7 @@ async def test_build_ffmpeg_args_no_dri_falls_back_to_software(tmp_path: Path, m
|
|||
monkeypatch.setattr("app.library.Segments.ffprobe", fake_ffprobe)
|
||||
# Simulate no /dev/dri present but GPU encoders otherwise available
|
||||
monkeypatch.setattr("app.library.SegmentEncoders.has_dri_devices", lambda: False)
|
||||
monkeypatch.setattr(
|
||||
"app.library.SegmentEncoders.ffmpeg_encoders", lambda: {"h264_nvenc", "h264_qsv", "h264_amf"}
|
||||
)
|
||||
monkeypatch.setattr("app.library.SegmentEncoders.ffmpeg_encoders", lambda: {"h264_nvenc", "h264_qsv", "h264_amf"})
|
||||
|
||||
# reset encoder cache to ensure clean selection in this test
|
||||
from app.library.Segments import Segments as _Seg
|
||||
|
|
@ -257,8 +255,7 @@ async def test_stream_gpu_failure_falls_back_to_software(
|
|||
assert len(resp.data) > 0
|
||||
# Ensure we logged the reason for GPU failure (message text may vary)
|
||||
assert any(
|
||||
("Hardware encoder failed" in r.message) or ("transcoding has failed" in r.message)
|
||||
for r in caplog.records
|
||||
("Hardware encoder failed" in r.message) or ("transcoding has failed" in r.message) for r in caplog.records
|
||||
)
|
||||
assert any("nvenc failure" in r.message for r in caplog.records)
|
||||
# Verify second invocation switched codec to a safe fallback (software)
|
||||
|
|
@ -269,9 +266,7 @@ async def test_stream_gpu_failure_falls_back_to_software(
|
|||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_gpu_fallback_switches_codec(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
async def test_stream_gpu_fallback_switches_codec(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
async def fake_ffprobe(_file: Path):
|
||||
return DummyFF(v=True, a=True)
|
||||
|
||||
|
|
|
|||
|
|
@ -77,11 +77,7 @@ class TestServices:
|
|||
def test_add_all_services(self):
|
||||
"""Test adding multiple services at once."""
|
||||
services = Services()
|
||||
services_dict = {
|
||||
"service1": "value1",
|
||||
"service2": "value2",
|
||||
"service3": "value3"
|
||||
}
|
||||
services_dict = {"service1": "value1", "service2": "value2", "service3": "value3"}
|
||||
|
||||
services.add_all(services_dict)
|
||||
|
||||
|
|
@ -394,6 +390,7 @@ class TestServices:
|
|||
# Lambda function replacement
|
||||
def lambda_handler(x):
|
||||
return f"Lambda: {x}"
|
||||
|
||||
services.add("x", "lambda_value")
|
||||
result = services.handle_sync(lambda_handler)
|
||||
assert result == "Lambda: lambda_value"
|
||||
|
|
@ -448,10 +445,7 @@ class TestServices:
|
|||
services = Services()
|
||||
services.add("existing", "original")
|
||||
|
||||
new_services = {
|
||||
"existing": "overwritten",
|
||||
"new": "value"
|
||||
}
|
||||
new_services = {"existing": "overwritten", "new": "value"}
|
||||
|
||||
services.add_all(new_services)
|
||||
|
||||
|
|
@ -531,7 +525,9 @@ class TestServices:
|
|||
assert result == "anon: value"
|
||||
|
||||
# Function with minimal signature info
|
||||
def minimal(param): return param
|
||||
def minimal(param):
|
||||
return param
|
||||
|
||||
result = services.handle_sync(minimal)
|
||||
assert result == "value"
|
||||
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ class TestSingleton:
|
|||
|
||||
def test_singleton_same_instance(self):
|
||||
"""Test that Singleton returns same instance for same class."""
|
||||
|
||||
class TestClass(metaclass=Singleton):
|
||||
def __init__(self, value=None):
|
||||
self.value = value
|
||||
|
|
@ -44,6 +45,7 @@ class TestSingleton:
|
|||
|
||||
def test_singleton_different_classes(self):
|
||||
"""Test that Singleton creates different instances for different classes."""
|
||||
|
||||
class ClassA(metaclass=Singleton):
|
||||
def __init__(self):
|
||||
self.name = "A"
|
||||
|
|
@ -68,6 +70,7 @@ class TestSingleton:
|
|||
|
||||
def test_threadsafe_same_instance(self):
|
||||
"""Test that ThreadSafe returns same instance for same class."""
|
||||
|
||||
class TestClass(metaclass=ThreadSafe):
|
||||
def __init__(self, value=None):
|
||||
self.value = value
|
||||
|
|
@ -83,6 +86,7 @@ class TestSingleton:
|
|||
|
||||
def test_threadsafe_different_classes(self):
|
||||
"""Test that ThreadSafe creates different instances for different classes."""
|
||||
|
||||
class ClassA(metaclass=ThreadSafe):
|
||||
def __init__(self):
|
||||
self.name = "A"
|
||||
|
|
@ -153,6 +157,7 @@ class TestSingleton:
|
|||
|
||||
def test_singleton_inheritance(self):
|
||||
"""Test singleton behavior with inheritance."""
|
||||
|
||||
class BaseClass(metaclass=Singleton):
|
||||
def __init__(self):
|
||||
self.base_value = "base"
|
||||
|
|
@ -182,6 +187,7 @@ class TestSingleton:
|
|||
|
||||
def test_threadsafe_inheritance(self):
|
||||
"""Test threadsafe singleton behavior with inheritance."""
|
||||
|
||||
class BaseClass(metaclass=ThreadSafe):
|
||||
def __init__(self):
|
||||
self.base_value = "base"
|
||||
|
|
@ -211,6 +217,7 @@ class TestSingleton:
|
|||
|
||||
def test_singleton_with_args_and_kwargs(self):
|
||||
"""Test singleton behavior with various constructor arguments."""
|
||||
|
||||
class ConfigClass(metaclass=Singleton):
|
||||
def __init__(self, name, value=None, **kwargs):
|
||||
self.name = name
|
||||
|
|
@ -231,6 +238,7 @@ class TestSingleton:
|
|||
|
||||
def test_threadsafe_with_args_and_kwargs(self):
|
||||
"""Test threadsafe singleton behavior with various constructor arguments."""
|
||||
|
||||
class ConfigClass(metaclass=ThreadSafe):
|
||||
def __init__(self, name, value=None, **kwargs):
|
||||
self.name = name
|
||||
|
|
@ -252,4 +260,5 @@ class TestSingleton:
|
|||
|
||||
if __name__ == "__main__":
|
||||
import pytest
|
||||
|
||||
pytest.main([__file__])
|
||||
|
|
|
|||
|
|
@ -103,6 +103,7 @@ async def test_paginate_order_and_bounds():
|
|||
assert items_desc[0][0] == _list[14]._id
|
||||
await store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_by_id():
|
||||
store = await make_store()
|
||||
|
|
@ -112,6 +113,7 @@ async def test_get_by_id():
|
|||
assert item._id == (await store.get_by_id("queue", item._id))._id
|
||||
await store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shutdown_closes_worker_queue():
|
||||
store = await make_store()
|
||||
|
|
@ -213,6 +215,7 @@ async def test_get_item_matches_conditions():
|
|||
assert no_match is None
|
||||
await store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_on_shutdown_closes_connection():
|
||||
store = await make_store()
|
||||
|
|
|
|||
|
|
@ -888,9 +888,7 @@ class TestTasks:
|
|||
@patch("app.library.Tasks.EventBus")
|
||||
@patch("app.library.Tasks.Scheduler")
|
||||
@patch("app.library.Tasks.DownloadQueue")
|
||||
async def test_tasks_runner_disabled_task(
|
||||
self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config
|
||||
):
|
||||
async def test_tasks_runner_disabled_task(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config):
|
||||
"""Test that disabled tasks are skipped by the runner."""
|
||||
mock_config_instance = Mock(
|
||||
debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *"
|
||||
|
|
|
|||
|
|
@ -130,7 +130,10 @@ async def test_tver_handler_extract(monkeypatch):
|
|||
assert "木村拓哉がタクシー 運転手！目黒蓮が木村と二人旅" == result.items[0].title
|
||||
|
||||
assert result.items[1].url == "https://tver.jp/episodes/epejwb9mvx"
|
||||
assert "木村拓哉がタクシー運転手！蒼井優＆上戸彩高校の同級生コンビがディズニーリゾートを大満喫！" == result.items[1].title
|
||||
assert (
|
||||
"木村拓哉がタクシー運転手！蒼井優＆上戸彩高校の同級生コンビがディズニーリゾートを大満喫！"
|
||||
== result.items[1].title
|
||||
)
|
||||
|
||||
assert result.metadata.get("has_entries") is True
|
||||
assert "callSeriesEpisodes" in result.metadata.get("feed_url", "")
|
||||
|
|
|
|||
|
|
@ -1,4 +1,3 @@
|
|||
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
from app.library.Utils import REMOVE_KEYS
|
||||
|
|
|
|||
|
|
@ -75,40 +75,33 @@ exclude = [
|
|||
"venv",
|
||||
]
|
||||
|
||||
# Same as Black.
|
||||
line-length = 120
|
||||
indent-width = 4
|
||||
|
||||
# Assume Python 3.11
|
||||
target-version = "py311"
|
||||
target-version = "py313"
|
||||
|
||||
[project.optional-dependencies]
|
||||
installer = ["pyinstaller"]
|
||||
|
||||
[tool.ruff.lint]
|
||||
# Enable Pyflakes (`F`) and a subset of the pycodestyle (`E`) codes by default.
|
||||
# Unlike Flake8, Ruff doesn't enable pycodestyle warnings (`W`) or
|
||||
# McCabe complexity (`C901`) by default.
|
||||
#select = ["E4", "E7", "E9", "F", "G", "W"]
|
||||
#ignore = ["PTH", "G0", "G1", "G201"]
|
||||
select = [
|
||||
"ALL", # include all the rules, including new ones
|
||||
]
|
||||
# Ignore some rules in test files.
|
||||
per-file-ignores = { "test_*.py" = ["D"] }
|
||||
select = ["ALL"]
|
||||
per-file-ignores = { "test_*.py" = [
|
||||
"ALL",
|
||||
], "migrations/versions/*.py" = [
|
||||
"ALL",
|
||||
] }
|
||||
|
||||
ignore = [
|
||||
#### modules
|
||||
"ANN", # flake8-annotations
|
||||
"COM", # flake8-commas
|
||||
"C90", # mccabe complexity
|
||||
"DJ", # django
|
||||
"EXE", # flake8-executable
|
||||
"T10", # debugger
|
||||
"TID", # flake8-tidy-imports
|
||||
"ANN",
|
||||
"COM",
|
||||
"C90",
|
||||
"DJ",
|
||||
"EXE",
|
||||
"T10",
|
||||
"TID",
|
||||
"PTH",
|
||||
|
||||
#### specific rules
|
||||
"D100", # ignore missing docs
|
||||
"D100",
|
||||
"D101",
|
||||
"D102",
|
||||
"D103",
|
||||
|
|
@ -122,14 +115,13 @@ ignore = [
|
|||
"D400",
|
||||
"D401",
|
||||
"D415",
|
||||
"E402", # false positives for local imports
|
||||
"E501", # line too long
|
||||
"TRY003", # external messages in exceptions are too verbose
|
||||
"E402",
|
||||
"E501",
|
||||
"TRY003",
|
||||
"TD002",
|
||||
"TD003",
|
||||
"FIX002", # too verbose descriptions of todos,
|
||||
"N806", # variable names should be snake_case
|
||||
"PTH", # prefer using absolute imports
|
||||
"FIX002",
|
||||
"N806",
|
||||
"PGH003",
|
||||
"D404",
|
||||
"PLR0913",
|
||||
|
|
@ -165,49 +157,25 @@ ignore = [
|
|||
"PLC0415",
|
||||
"S603",
|
||||
"D203",
|
||||
"TRY004", # Like it's our choice to use ValuesError :|
|
||||
"TRY004",
|
||||
"PT011",
|
||||
"RUF001", # We like unicode chars.
|
||||
"S311", # Not used for cryptography
|
||||
"RUF001",
|
||||
"S311",
|
||||
]
|
||||
|
||||
# Allow fix for all enabled rules (when `--fix`) is provided.
|
||||
fixable = ["ALL"]
|
||||
unfixable = []
|
||||
|
||||
# Allow unused variables when underscore-prefixed.
|
||||
dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
|
||||
|
||||
[tool.ruff.format]
|
||||
# Like Black, use double quotes for strings.
|
||||
quote-style = "double"
|
||||
|
||||
# Like Black, indent with spaces, rather than tabs.
|
||||
indent-style = "space"
|
||||
|
||||
# Like Black, respect magic trailing commas.
|
||||
skip-magic-trailing-comma = false
|
||||
|
||||
# Like Black, automatically detect the appropriate line ending.
|
||||
line-ending = "auto"
|
||||
|
||||
# Enable auto-formatting of code examples in docstrings. Markdown,
|
||||
# reStructuredText code/literal blocks and doctests are all supported.
|
||||
#
|
||||
# This is currently disabled by default, but it is planned for this
|
||||
# to be opt-out in the future.
|
||||
docstring-code-format = false
|
||||
|
||||
# Set the line length limit used when formatting code snippets in
|
||||
# docstrings.
|
||||
#
|
||||
# This only has an effect when the `docstring-code-format` setting is
|
||||
# enabled.
|
||||
docstring-code-line-length = "dynamic"
|
||||
|
||||
[tool.autopep8]
|
||||
--max-line-length = 120
|
||||
|
||||
[tool.uv]
|
||||
link-mode = "copy"
|
||||
|
||||
|
|
|
|||
|
|
@ -14,7 +14,8 @@
|
|||
"lint:fix": "eslint . --fix",
|
||||
"test": "vitest run",
|
||||
"test:watch": "vitest",
|
||||
"test:ci": "vitest run --silent --reporter=dot"
|
||||
"test:ci": "vitest run --silent --reporter=dot",
|
||||
"lint:tsc": "vue-tsc --noEmit"
|
||||
},
|
||||
"web-types": "./web-types.json",
|
||||
"dependencies": {
|
||||
|
|
|
|||
Loading…
Reference in a new issue