danvics/ytdl-sub
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[FEATURE] Persisted subscription logs (#512)

Browse source
This commit is contained in:
Jesse Bannon 2023-03-08 00:17:11 -08:00 committed by GitHub
parent 022c24cc2c
commit 648027204f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 447 additions and 146 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
docs/config.rst
View file

@ -29,7 +29,25 @@ and subscriptions.
.. autoclass:: ytdl_sub.config.config_validator.ConfigOptions()
:members:
:member-order: bysource
:exclude-members: persist_logs
persist_logs
""""""""""""
Within ``configuration``, define whether logs from subscription downloads
should be persisted.
.. code-block:: yaml
configuration:
persist_logs:
logs_directory: "/path/to/log/directory"
Log files are stored as
``YYYY-mm-dd-HHMMSS.subscription_name.(success|error).log``.
.. autoclass:: ytdl_sub.config.config_validator.PersistLogsValidator()
:members:
:member-order: bysource
presets
^^^^^^^

79
src/ytdl_sub/cli/main.py
View file

@ -1,13 +1,18 @@
import argparse
import gc
import sys
from datetime import datetime
from pathlib import Path
from typing import List
from typing import Optional
from typing import Tuple
from yt_dlp.utils import sanitize_filename
from ytdl_sub.cli.download_args_parser import DownloadArgsParser
from ytdl_sub.cli.main_args_parser import parser
from ytdl_sub.config.config_file import ConfigFile
from ytdl_sub.subscriptions.subscription import Subscription
from ytdl_sub.utils.file_handler import FileHandler
from ytdl_sub.utils.file_handler import FileHandlerTransactionLog
from ytdl_sub.utils.file_lock import working_directory_lock
from ytdl_sub.utils.logger import Logger
@ -19,8 +24,41 @@ logger = Logger.get()
_VIEW_EXTRA_ARGS_FORMATTER = "--preset _view --overrides.url {}"
def _maybe_write_subscription_log_file(
config: ConfigFile,
subscription: Subscription,
dry_run: bool,
exception: Optional[Exception] = None,
) -> None:
success: bool = exception is None
# If dry-run, do nothing
if dry_run:
return
# If persisting logs is disabled, do nothing
if not config.config_options.persist_logs:
return
# If persisting successful logs is disabled, do nothing
if success and not config.config_options.persist_logs.keep_successful_logs:
return
log_time = datetime.now().strftime("%Y-%m-%d-%H%M%S")
log_subscription_name = sanitize_filename(subscription.name).lower().replace(" ", "_")
log_success = "success" if success else "error"
log_filename = f"{log_time}.{log_subscription_name}.{log_success}.log"
persist_log_path = Path(config.config_options.persist_logs.logs_directory) / log_filename
if not success:
Logger.log_exit_exception(exception=exception, log_filepath=persist_log_path)
FileHandler.copy(Logger.debug_log_filename(), persist_log_path)
def _download_subscriptions_from_yaml_files(
config: ConfigFile, args: argparse.Namespace
config: ConfigFile, subscription_paths: List[str], dry_run: bool
) -> List[Tuple[Subscription, FileHandlerTransactionLog]]:
"""
Downloads all subscriptions from one or many subscription yaml files.
@ -29,8 +67,10 @@ def _download_subscriptions_from_yaml_files(
----------
config
Configuration file
args
Arguments from argparse
subscription_paths
Path to subscription files to download
dry_run
Whether to dry run or not
Returns
-------
@ -38,9 +78,9 @@ def _download_subscriptions_from_yaml_files(
Raises
------
Validation exception if main arg is specified as a subscription path
Exception
Any exception during download
"""
subscription_paths: List[str] = args.subscription_paths
subscriptions: List[Subscription] = []
output: List[Tuple[Subscription, FileHandlerTransactionLog]] = []
@ -51,15 +91,26 @@ def _download_subscriptions_from_yaml_files(
for subscription in subscriptions:
logger.info(
"Beginning subscription %s for %s",
("dry run" if args.dry_run else "download"),
("dry run" if dry_run else "download"),
subscription.name,
)
logger.debug("Subscription full yaml:\n%s", subscription.as_yaml())
transaction_log = subscription.download(dry_run=args.dry_run)
output.append((subscription, transaction_log))
gc.collect() # Garbage collect after each subscription download
Logger.cleanup() # Cleanup logger after each successful subscription download
try:
transaction_log = subscription.download(dry_run=dry_run)
except Exception as exc: # pylint: disable=broad-except
_maybe_write_subscription_log_file(
config=config, subscription=subscription, dry_run=dry_run, exception=exc
)
raise
else:
output.append((subscription, transaction_log))
_maybe_write_subscription_log_file(
config=config, subscription=subscription, dry_run=dry_run
)
Logger.cleanup() # Cleanup logger after each successful subscription download
finally:
gc.collect() # Garbage collect after each subscription download
return output
@ -136,7 +187,11 @@ def main() -> List[Tuple[Subscription, FileHandlerTransactionLog]]:
with working_directory_lock(config=config):
if args.subparser == "sub":
transaction_logs = _download_subscriptions_from_yaml_files(config=config, args=args)
transaction_logs = _download_subscriptions_from_yaml_files(
config=config,
subscription_paths=args.subscription_paths,
dry_run=args.dry_run,
)
# One-off download
elif args.subparser == "dl":

9
src/ytdl_sub/config/config_file.py
View file

@ -1,5 +1,6 @@
import os
from typing import Any
from typing import Dict
from ytdl_sub.config.config_validator import ConfigValidator
from ytdl_sub.config.preset import Preset
@ -65,3 +66,11 @@ class ConfigFile(ConfigValidator):
"""
config_dict = load_yaml(file_path=config_path)
return ConfigFile.from_dict(config_dict)
def as_dict(self) -> Dict[str, Any]:
"""
Returns
-------
The config in its dict form
"""
return self._value

76
src/ytdl_sub/config/config_validator.py
View file

@ -3,12 +3,14 @@ from typing import Dict
from typing import Optional
from mergedeep import mergedeep
from yt_dlp.utils import datetime_from_str
from ytdl_sub.prebuilt_presets import PREBUILT_PRESETS
from ytdl_sub.utils.system import IS_WINDOWS
from ytdl_sub.validators.file_path_validators import FFmpegFileValidator
from ytdl_sub.validators.file_path_validators import FFprobeFileValidator
from ytdl_sub.validators.strict_dict_validator import StrictDictValidator
from ytdl_sub.validators.validators import BoolValidator
from ytdl_sub.validators.validators import LiteralDictValidator
from ytdl_sub.validators.validators import StringValidator
@ -22,9 +24,71 @@ else:
_DEFAULT_FFPROBE_PATH = "/usr/bin/ffprobe"
class PersistLogsValidator(StrictDictValidator):
_required_keys = {"logs_directory"}
_optional_keys = {"keep_logs_after", "keep_successful_logs"}
def __init__(self, name: str, value: Any):
super().__init__(name, value)
self._logs_directory = self._validate_key(key="logs_directory", validator=StringValidator)
self._keep_logs_after: Optional[str] = None
if keep_logs_validator := self._validate_key_if_present(
key="keep_logs_after", validator=StringValidator
):
try:
self._keep_logs_after = datetime_from_str(keep_logs_validator.value)
except Exception as exc:
raise self._validation_exception(f"Invalid datetime string: {str(exc)}")
self._keep_successful_logs = self._validate_key(
key="keep_successful_logs", validator=BoolValidator, default=True
)
@property
def logs_directory(self) -> str:
"""
Required. The directory to store the logs in.
"""
return self._logs_directory.value
# pylint: disable=line-too-long
# @property
# def keep_logs_after(self) -> Optional[str]:
# """
# Optional. Keep logs after this date, in yt-dlp datetime format.
#
# .. code-block:: Markdown
#
# A string in the format YYYYMMDD or
# (now|today|yesterday|date)[+-][0-9](microsecond|second|minute|hour|day|week|month|year)(s)
#
# For example, ``today-1week`` means keep 1 week's worth of logs. By default, ytdl-sub will
# keep all log files.
# """
# return self._keep_logs_after
# pylint: enable=line-too-long
@property
def keep_successful_logs(self) -> bool:
"""
Optional. Whether to store logs when downloading is successful. Defaults to True.
"""
return self._keep_successful_logs.value
class ConfigOptions(StrictDictValidator):
_required_keys = {"working_directory"}
_optional_keys = {"umask", "dl_aliases", "lock_directory", "ffmpeg_path", "ffprobe_path"}
_optional_keys = {
"umask",
"dl_aliases",
"persist_logs",
"lock_directory",
"ffmpeg_path",
"ffprobe_path",
}
def __init__(self, name: str, value: Any):
super().__init__(name, value)
@ -38,6 +102,9 @@ class ConfigOptions(StrictDictValidator):
self._dl_aliases = self._validate_key_if_present(
key="dl_aliases", validator=LiteralDictValidator
)
self._persist_logs = self._validate_key_if_present(
key="persist_logs", validator=PersistLogsValidator
)
self._lock_directory = self._validate_key(
key="lock_directory", validator=StringValidator, default=_DEFAULT_LOCK_DIRECTORY
)
@ -93,6 +160,13 @@ class ConfigOptions(StrictDictValidator):
return self._dl_aliases.dict
return {}
@property
def persist_logs(self) -> Optional[PersistLogsValidator]:
"""
Persist logs validator. readthedocs in the validator itself!
"""
return self._persist_logs
@property
def lock_directory(self) -> str:
"""

17
src/ytdl_sub/main.py
View file

@ -1,8 +1,6 @@
import sys
from ytdl_sub import __local_version__
from ytdl_sub.cli.main_args_parser import parser
from ytdl_sub.utils.exceptions import ValidationException
from ytdl_sub.utils.logger import Logger
@ -24,22 +22,11 @@ def main():
"""
Entrypoint for ytdl-sub
"""
logger = Logger.get()
try:
_main()
Logger.cleanup() # Ran successfully, so we can delete the debug file
except ValidationException as validation_exception:
logger.error(validation_exception)
sys.exit(1)
except Exception: # pylint: disable=broad-except
logger.exception("An uncaught error occurred:")
logger.error(
"Version %s\nPlease upload the error log file '%s' and make a Github "
"issue at https://github.com/jmbannon/ytdl-sub/issues with your config and "
"command/subscription yaml file to reproduce. Thanks for trying ytdl-sub!",
__local_version__,
Logger.debug_log_filename(),
)
except Exception as exc: # pylint: disable=broad-except
Logger.log_exit_exception(exception=exc)
sys.exit(1)
sys.exit(0)

53
src/ytdl_sub/utils/logger.py
View file

@ -4,9 +4,12 @@ import logging
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import List
from typing import Optional
from ytdl_sub import __local_version__
from ytdl_sub.utils.exceptions import ValidationException
from ytdl_sub.utils.file_handler import FileHandler
@ -90,6 +93,9 @@ class Logger:
_DEBUG_LOGGER_FILE = tempfile.NamedTemporaryFile(prefix="ytdl-sub.", delete=False)
# pylint: enable=R1732
# Whether the final exception lines were added to the debug log
_LOGGED_EXIT_EXCEPTION: bool = False
# Keep track of all Loggers created
_LOGGERS: List[logging.Logger] = []
@ -190,19 +196,48 @@ class Logger:
)
with StreamToLogger(logger=logger) as redirect_stream:
with contextlib.redirect_stdout(new_target=redirect_stream):
with contextlib.redirect_stderr(new_target=redirect_stream):
yield
try:
with contextlib.redirect_stdout(new_target=redirect_stream):
with contextlib.redirect_stderr(new_target=redirect_stream):
yield
finally:
redirect_stream.flush()
@classmethod
def cleanup(cls, delete_debug_file: bool = True):
def log_exit_exception(cls, exception: Exception, log_filepath: Optional[Path] = None):
"""
Cleans up any log files left behind
Performs the final log before exiting from an error
Parameters
----------
delete_debug_file
Whether to delete the debug log file. Defaults to True.
exception
The exception to log
log_filepath
Optional. The filepath to the debug logs
"""
if not cls._LOGGED_EXIT_EXCEPTION:
logger = cls.get()
# Log validation exceptions as-is
if isinstance(exception, ValidationException):
logger.error(exception)
# For other uncaught errors, log as bug:
else:
logger.exception("An uncaught error occurred:")
logger.error(
"Version %s\nPlease upload the error log file '%s' and make a Github "
"issue at https://github.com/jmbannon/ytdl-sub/issues with your config and "
"command/subscription yaml file to reproduce. Thanks for trying ytdl-sub!",
__local_version__,
log_filepath if log_filepath else Logger.debug_log_filename(),
)
cls._LOGGED_EXIT_EXCEPTION = True
@classmethod
def cleanup(cls):
"""
Cleans up debug log file left behind
"""
for logger in cls._LOGGERS:
for handler in logger.handlers:
@ -210,5 +245,5 @@ class Logger:
cls._DEBUG_LOGGER_FILE.close()
if delete_debug_file:
FileHandler.delete(cls.debug_log_filename())
FileHandler.delete(cls.debug_log_filename())
cls._LOGGED_EXIT_EXCEPTION = False

82
tests/conftest.py
View file

@ -2,6 +2,7 @@ import contextlib
import json
import logging
import os
import shutil
import tempfile
from pathlib import Path
from typing import Any
@ -11,15 +12,52 @@ from typing import List
from unittest.mock import patch
import pytest
from expected_download import _get_files_in_directory
from ytdl_sub.config.config_file import ConfigFile
from ytdl_sub.subscriptions.subscription_download import SubscriptionDownload
from ytdl_sub.utils.file_handler import FileHandler
from ytdl_sub.utils.logger import Logger
from ytdl_sub.utils.logger import LoggerLevels
from ytdl_sub.utils.yaml import load_yaml
@pytest.fixture(autouse=True)
def cleanup_debug_file():
"""
Clean logs after every test
"""
Logger.set_log_level(log_level_name=LoggerLevels.DEBUG.name)
yield
Logger.cleanup()
@pytest.fixture
def working_directory() -> Path:
def working_directory() -> str:
"""
Any time the working directory is used, ensure no files remain on cleaning it up
"""
logger = Logger.get("test")
with tempfile.TemporaryDirectory() as temp_dir:
yield temp_dir
def _assert_working_directory_empty(self, is_error: bool = False):
files = [str(file_path) for file_path in _get_files_in_directory(temp_dir)]
num_files = len(files)
if os.path.isdir(temp_dir):
shutil.rmtree(temp_dir)
if not is_error:
if num_files > 0:
logger.error("left-over files in working dir:\n%s", "\n".join(files))
assert num_files == 0
with patch.object(
SubscriptionDownload,
"_delete_working_directory",
new=_assert_working_directory_empty,
):
yield temp_dir
@pytest.fixture()
@ -95,3 +133,43 @@ def preset_dict_to_subscription_yaml_generator() -> Callable:
FileHandler.delete(tmp_file.name)
return _preset_dict_to_subscription_yaml_generator
###################################################################################################
# Example config fixtures
def _load_config(config_path: Path, working_directory: str) -> ConfigFile:
config_dict = load_yaml(file_path=config_path)
config_dict["configuration"]["working_directory"] = working_directory
return ConfigFile.from_dict(config_dict)
@pytest.fixture()
def music_video_config_path() -> Path:
return Path("examples/music_videos_config.yaml")
@pytest.fixture()
def music_video_config(music_video_config_path, working_directory) -> ConfigFile:
return _load_config(music_video_config_path, working_directory)
@pytest.fixture()
def music_video_subscription_path() -> Path:
return Path("examples/music_videos_subscriptions.yaml")
@pytest.fixture()
def channel_as_tv_show_config(working_directory) -> ConfigFile:
return _load_config(
config_path=Path("examples/tv_show_config.yaml"), working_directory=working_directory
)
@pytest.fixture()
def music_audio_config(working_directory) -> ConfigFile:
return _load_config(
config_path=Path("examples/music_audio_config.yaml"), working_directory=working_directory
)

67
tests/e2e/conftest.py
View file

@ -1,69 +1,16 @@
import json
import os
import shutil
import sys
import tempfile
from pathlib import Path
from typing import List
from typing import Tuple
from unittest.mock import patch
import pytest
from expected_download import _get_files_in_directory
from ytdl_sub.cli.main import main
from ytdl_sub.config.config_file import ConfigFile
from ytdl_sub.subscriptions.subscription import Subscription
from ytdl_sub.subscriptions.subscription_download import SubscriptionDownload
from ytdl_sub.utils.file_handler import FileHandler
from ytdl_sub.utils.file_handler import FileHandlerTransactionLog
from ytdl_sub.utils.logger import Logger
from ytdl_sub.utils.yaml import load_yaml
logger = Logger.get("test")
@pytest.fixture
def working_directory() -> str:
"""
Any time the working directory is used, ensure no files remain on cleaning it up
"""
with tempfile.TemporaryDirectory() as temp_dir:
def _assert_working_directory_empty(self, is_error: bool = False):
files = [str(file_path) for file_path in _get_files_in_directory(temp_dir)]
num_files = len(files)
if os.path.isdir(temp_dir):
shutil.rmtree(temp_dir)
if not is_error:
if num_files > 0:
logger.error("left-over files in working dir:\n%s", "\n".join(files))
assert num_files == 0
with patch.object(
SubscriptionDownload,
"_delete_working_directory",
new=_assert_working_directory_empty,
):
yield temp_dir
@pytest.fixture()
def music_video_config_path() -> Path:
return Path("examples/music_videos_config.yaml")
def _load_config(config_path: Path, working_directory: Path) -> ConfigFile:
config_dict = load_yaml(file_path=config_path)
config_dict["configuration"]["working_directory"] = working_directory
return ConfigFile.from_dict(config_dict)
@pytest.fixture()
def music_video_config(music_video_config_path, working_directory) -> ConfigFile:
return _load_config(music_video_config_path, working_directory)
@pytest.fixture()
@ -77,20 +24,6 @@ def music_video_config_for_cli(music_video_config) -> str:
FileHandler.delete(tmp_file.name)
@pytest.fixture()
def channel_as_tv_show_config(working_directory) -> ConfigFile:
return _load_config(
config_path=Path("examples/tv_show_config.yaml"), working_directory=working_directory
)
@pytest.fixture()
def music_audio_config(working_directory) -> ConfigFile:
return _load_config(
config_path=Path("examples/music_audio_config.yaml"), working_directory=working_directory
)
@pytest.fixture
def timestamps_file_path():
timestamps = [

132
tests/unit/cli/test_main.py
View file

@ -1,21 +1,125 @@
import sys
import re
import tempfile
import time
from pathlib import Path
from typing import Callable
from unittest.mock import patch
import mergedeep
import pytest
from ytdl_sub.cli.main import main
from ytdl_sub.utils.exceptions import ValidationException
from ytdl_sub.cli.main import _download_subscriptions_from_yaml_files
from ytdl_sub.config.config_file import ConfigFile
from ytdl_sub.subscriptions.subscription import Subscription
from ytdl_sub.utils.file_handler import FileHandlerTransactionLog
from ytdl_sub.utils.logger import Logger
def test_args_after_sub_work():
with patch.object(
sys,
"argv",
["ytdl-sub", "-c", "examples/tv_show_config.yaml", "sub", "--log-level", "debug"],
), patch("ytdl_sub.cli.main._download_subscriptions_from_yaml_files") as mock_sub:
main()
@pytest.fixture
def persist_logs_directory():
with tempfile.TemporaryDirectory() as temp_dir:
yield temp_dir
assert mock_sub.call_count == 1
assert mock_sub.call_args.kwargs["args"].config == "examples/tv_show_config.yaml"
assert mock_sub.call_args.kwargs["args"].subscription_paths == ["subscriptions.yaml"]
assert mock_sub.call_args.kwargs["args"].ytdl_sub_log_level == "debug"
@pytest.fixture
def persist_logs_config_factory(
music_video_config: ConfigFile, persist_logs_directory: str
) -> Callable:
def _persist_logs_config_factory(keep_successful_logs: bool) -> ConfigFile:
return ConfigFile.from_dict(
dict(
mergedeep.merge(
music_video_config.as_dict(),
{
"configuration": {
"persist_logs": {
"logs_directory": persist_logs_directory,
"keep_successful_logs": keep_successful_logs,
},
}
},
)
)
)
return _persist_logs_config_factory
@pytest.fixture
def mock_subscription_download_factory():
def _mock_subscription_download_factory(mock_success_output: bool) -> Callable:
def _mock_download(self: Subscription, dry_run: bool) -> FileHandlerTransactionLog:
Logger.get().info(
"name=%s success=%s dry_run=%s", self.name, mock_success_output, dry_run
)
time.sleep(1)
if not mock_success_output:
raise ValueError("error")
return FileHandlerTransactionLog()
return _mock_download
return _mock_subscription_download_factory
class TestPersistLogs:
@pytest.mark.parametrize("dry_run", [True, False])
@pytest.mark.parametrize("mock_success_output", [True, False])
@pytest.mark.parametrize("keep_successful_logs", [True, False])
def test_subscription_logs_write_to_file(
self,
persist_logs_directory: str,
persist_logs_config_factory: Callable,
mock_subscription_download_factory: Callable,
music_video_subscription_path: Path,
dry_run: bool,
mock_success_output: bool,
keep_successful_logs: bool,
):
num_subscriptions = 2
config = persist_logs_config_factory(keep_successful_logs=keep_successful_logs)
subscription_paths = [str(music_video_subscription_path)] * num_subscriptions
with patch.object(
Subscription,
"download",
new=mock_subscription_download_factory(mock_success_output=mock_success_output),
):
try:
_download_subscriptions_from_yaml_files(
config=config, subscription_paths=subscription_paths, dry_run=dry_run
)
except ValueError:
assert not mock_success_output
log_directory_files = list(Path(persist_logs_directory).rglob("*"))
# If dry run or success but success logging disabled, expect 0 log files
if dry_run or (mock_success_output and not keep_successful_logs):
assert len(log_directory_files) == 0
return
# If not success, expect 1 log file
elif not mock_success_output:
assert len(log_directory_files) == 1
log_path = log_directory_files[0]
assert bool(re.match(r"\d{4}-\d{2}-\d{2}-\d{6}\.john_smith\.error\.log", log_path.name))
with open(log_path, "r", encoding="utf-8") as log_file:
assert log_file.readlines()[-1] == (
f"Please upload the error log file '{str(log_path)}' and make a Github issue "
f"at https://github.com/jmbannon/ytdl-sub/issues with your config and "
f"command/subscription yaml file to reproduce. Thanks for trying ytdl-sub!\n"
)
# If success and success logging, expect 3 log files
else:
assert len(log_directory_files) == num_subscriptions
for log_file_path in log_directory_files:
assert bool(
re.match(
r"\d{4}-\d{2}-\d{2}-\d{6}\.john_smith\.success\.log", log_file_path.name
)
)
with open(log_file_path, "r", encoding="utf-8") as log_file:
assert (
log_file.readlines()[-1]
== "[ytdl-sub] name=john_smith success=True dry_run=False\n"
)

14
tests/unit/main/test_main.py
View file

@ -9,6 +9,7 @@ from src.ytdl_sub import __local_version__
from src.ytdl_sub.main import main
from ytdl_sub.utils.exceptions import ValidationException
from ytdl_sub.utils.logger import Logger
from ytdl_sub.utils.logger import LoggerLevels
@pytest.fixture
@ -66,3 +67,16 @@ def test_main_uncaught_error(capsys, mock_sys_exit, expected_uncaught_error_mess
assert mock_error.call_args.args[0] == expected_uncaught_error_message
assert mock_error.call_args.args[1] == __local_version__
assert mock_error.call_args.args[2] == Logger.debug_log_filename()
def test_args_after_sub_work(mock_sys_exit):
with mock_sys_exit(expected_exit_code=0), patch.object(
sys,
"argv",
["ytdl-sub", "-c", "examples/tv_show_config.yaml", "sub", "--log-level", "verbose"],
), patch("ytdl_sub.cli.main._download_subscriptions_from_yaml_files") as mock_sub:
main()
assert mock_sub.call_count == 1
assert mock_sub.call_args.kwargs["subscription_paths"] == ["subscriptions.yaml"]
assert Logger._LOGGER_LEVEL == LoggerLevels.VERBOSE

37
tests/unit/prebuilt_presets/conftest.py
View file

@ -56,6 +56,7 @@ def mock_entry_dict_factory(mock_downloaded_file_path) -> Callable:
playlist_index: int = 1,
playlist_count: int = 1,
is_youtube_channel: bool = False,
mock_download_to_working_dir: bool = True,
) -> Dict:
entry_dict = {
UID: uid,
@ -84,12 +85,14 @@ def mock_entry_dict_factory(mock_downloaded_file_path) -> Callable:
]
# Create mock video file
copy_file_fixture(
fixture_name="sample_vid.mp4", output_file_path=mock_downloaded_file_path(f"{uid}.mp4")
)
copy_file_fixture(
fixture_name="thumb.jpg", output_file_path=mock_downloaded_file_path(f"{uid}.jpg")
)
if mock_download_to_working_dir:
copy_file_fixture(
fixture_name="sample_vid.mp4",
output_file_path=mock_downloaded_file_path(f"{uid}.mp4"),
)
copy_file_fixture(
fixture_name="thumb.jpg", output_file_path=mock_downloaded_file_path(f"{uid}.jpg")
)
return entry_dict
return _mock_entry_dict_factory
@ -124,36 +127,33 @@ def mock_download_collection_entries(
):
@contextlib.contextmanager
def _mock_download_collection_entries_factory(is_youtube_channel: bool):
def _(**kwargs):
return mock_entry_dict_factory(**kwargs)
def _write_entries_to_working_dir(*args, **kwargs) -> List[Dict]:
if (len(args[0].collection.urls.list) == 1) or (
"season.2" in kwargs["url"] and len(args[0].download_options.urls.list) > 1
):
return [
_(
mock_entry_dict_factory(
uid="21-1",
upload_date="20210808",
playlist_index=1,
playlist_count=4,
is_youtube_channel=is_youtube_channel,
), # 1
_(
mock_entry_dict_factory(
uid="20-1",
upload_date="20200808",
playlist_index=2,
playlist_count=4,
is_youtube_channel=is_youtube_channel,
), # 2 98
_(
mock_entry_dict_factory(
uid="20-2",
upload_date="20200808",
playlist_index=3,
playlist_count=4,
is_youtube_channel=is_youtube_channel,
), # 1 99
_(
mock_entry_dict_factory(
uid="20-3",
upload_date="20200807",
playlist_index=4,
@ -163,35 +163,36 @@ def mock_download_collection_entries(
]
return [
# 20-3 should resolve to collection 1 (which is season 2)
_(
mock_entry_dict_factory(
uid="20-3",
upload_date="20200807",
playlist_index=1,
playlist_count=5,
is_youtube_channel=is_youtube_channel,
mock_download_to_working_dir=False,
),
_(
mock_entry_dict_factory(
uid="20-4",
upload_date="20200806",
playlist_index=2,
playlist_count=5,
is_youtube_channel=is_youtube_channel,
),
_(
mock_entry_dict_factory(
uid="20-5",
upload_date="20200706",
playlist_index=3,
playlist_count=5,
is_youtube_channel=is_youtube_channel,
),
_(
mock_entry_dict_factory(
uid="20-6",
upload_date="20200706",
playlist_index=4,
playlist_count=5,
is_youtube_channel=is_youtube_channel,
),
_(
mock_entry_dict_factory(
uid="20-7",
upload_date="20200606",
playlist_index=5,

9
tests/unit/utils/test_logger.py
View file

@ -6,13 +6,6 @@ from ytdl_sub.utils.logger import Logger
from ytdl_sub.utils.logger import LoggerLevels
@pytest.fixture(autouse=True)
def cleanup_debug_file():
Logger.set_log_level(log_level_name=LoggerLevels.DEBUG.name)
yield
Logger.cleanup()
class TestLogger:
@pytest.mark.parametrize(
"log_level",
@ -114,7 +107,7 @@ class TestLogger:
"[ytdl-sub:name_test] debug test\n",
]
Logger.cleanup(delete_debug_file=True)
Logger.cleanup()
assert not os.path.isfile(Logger._DEBUG_LOGGER_FILE.name)
@pytest.mark.parametrize(