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

[FEATURE] Process all subscriptions even if one or more error (#771)

Browse source 
Closes GH Issue #766, partially closes #520

Prior to this release, no other subscriptions would download if a subscription before it had an error. Now, subscriptions will continue to download even if one has an error, and will be reported in the output summary.
This commit is contained in:
Jesse Bannon 2023-10-21 12:25:38 -07:00 committed by GitHub
parent 82c9beb00a
commit 83ecd19c77
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
17 changed files with 518 additions and 361 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

89
src/ytdl_sub/cli/entrypoint.py
View file

@ -5,7 +5,6 @@ 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
@ -20,7 +19,6 @@ from ytdl_sub.subscriptions.subscription import Subscription
from ytdl_sub.utils.exceptions import ExperimentalFeatureNotEnabled
from ytdl_sub.utils.exceptions import ValidationException
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
@ -31,6 +29,10 @@ logger = Logger.get()
_VIEW_EXTRA_ARGS_FORMATTER = "--preset _view --overrides.url {}"
def _log_time() -> str:
return datetime.now().strftime("%Y-%m-%d-%H%M%S")
def _maybe_write_subscription_log_file(
config: ConfigFile,
subscription: Subscription,
@ -51,15 +53,14 @@ def _maybe_write_subscription_log_file(
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"
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)
Logger.log_exception(exception=exception, log_filepath=persist_log_path)
os.makedirs(os.path.dirname(persist_log_path), exist_ok=True)
FileHandler.copy(Logger.debug_log_filename(), persist_log_path)
@ -67,7 +68,7 @@ def _maybe_write_subscription_log_file(
def _download_subscriptions_from_yaml_files(
config: ConfigFile, subscription_paths: List[str], update_with_info_json: bool, dry_run: bool
) -> List[Tuple[Subscription, FileHandlerTransactionLog]]:
) -> List[Subscription]:
"""
Downloads all subscriptions from one or many subscription yaml files.
@ -84,7 +85,7 @@ def _download_subscriptions_from_yaml_files(
Returns
-------
List of (subscription, transaction_log)
List of subscriptions processed
Raises
------
@ -92,45 +93,41 @@ def _download_subscriptions_from_yaml_files(
Any exception during download
"""
subscriptions: List[Subscription] = []
output: List[Tuple[Subscription, FileHandlerTransactionLog]] = []
# Load all the subscriptions first to perform all validation before downloading
for path in subscription_paths:
subscriptions += Subscription.from_file_path(config=config, subscription_path=path)
for subscription in subscriptions:
logger.info(
"Beginning subscription %s for %s",
("dry run" if dry_run else "download"),
subscription.name,
)
logger.debug("Subscription full yaml:\n%s", subscription.as_yaml())
with subscription.exception_handling():
logger.info(
"Beginning subscription %s for %s",
("dry run" if dry_run else "download"),
subscription.name,
)
logger.debug("Subscription full yaml:\n%s", subscription.as_yaml())
try:
if update_with_info_json:
transaction_log = subscription.update_with_info_json(dry_run=dry_run)
subscription.update_with_info_json(dry_run=dry_run)
else:
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
subscription.download(dry_run=dry_run)
return output
_maybe_write_subscription_log_file(
config=config,
subscription=subscription,
dry_run=dry_run,
exception=subscription.exception,
)
Logger.cleanup(cleanup_error_log=False)
gc.collect() # Garbage collect after each subscription download
return subscriptions
def _download_subscription_from_cli(
config: ConfigFile, dry_run: bool, extra_args: List[str]
) -> Tuple[Subscription, FileHandlerTransactionLog]:
) -> Subscription:
"""
Downloads a one-off subscription using the CLI
@ -158,12 +155,12 @@ def _download_subscription_from_cli(
)
logger.info("Beginning CLI %s", ("dry run" if dry_run else "download"))
return subscription, subscription.download(dry_run=dry_run)
subscription.download(dry_run=dry_run)
return subscription
def _view_url_from_cli(
config: ConfigFile, url: str, split_chapters: bool
) -> Tuple[Subscription, FileHandlerTransactionLog]:
def _view_url_from_cli(config: ConfigFile, url: str, split_chapters: bool) -> Subscription:
"""
`ytdl-sub view` dry-runs a URL to print its source variables. Use the pre-built `_view` preset,
inject the URL argument, and dry-run.
@ -180,10 +177,12 @@ def _view_url_from_cli(
url,
" with split chapters" if split_chapters else "",
)
return subscription, subscription.download(dry_run=True)
subscription.download(dry_run=True)
return subscription
def main() -> List[Tuple[Subscription, FileHandlerTransactionLog]]:
def main() -> List[Subscription]:
"""
Entrypoint for ytdl-sub, without the error handling
"""
@ -203,7 +202,7 @@ def main() -> List[Tuple[Subscription, FileHandlerTransactionLog]]:
logger.info("No config specified, using defaults.")
config = ConfigFile(name="default_config", value={})
transaction_logs: List[Tuple[Subscription, FileHandlerTransactionLog]] = []
subscriptions: List[Subscription] = []
# If transaction log file is specified, make sure we can open it
_maybe_validate_transaction_log_file(transaction_log_file_path=args.transaction_log)
@ -222,7 +221,7 @@ def main() -> List[Tuple[Subscription, FileHandlerTransactionLog]]:
"full backup before usage. You have been warned!",
)
transaction_logs = _download_subscriptions_from_yaml_files(
subscriptions = _download_subscriptions_from_yaml_files(
config=config,
subscription_paths=args.subscription_paths,
update_with_info_json=args.update_with_info_json,
@ -231,13 +230,13 @@ def main() -> List[Tuple[Subscription, FileHandlerTransactionLog]]:
# One-off download
elif args.subparser == "dl":
transaction_logs.append(
subscriptions.append(
_download_subscription_from_cli(
config=config, dry_run=args.dry_run, extra_args=extra_args
)
)
elif args.subparser == "view":
transaction_logs.append(
subscriptions.append(
_view_url_from_cli(config=config, url=args.url, split_chapters=args.split_chapters)
)
else:
@ -245,10 +244,10 @@ def main() -> List[Tuple[Subscription, FileHandlerTransactionLog]]:
if not args.suppress_transaction_log:
output_transaction_log(
transaction_logs=transaction_logs,
subscriptions=subscriptions,
transaction_log_file_path=args.transaction_log,
)
output_summary(transaction_logs)
output_summary(subscriptions)
return transaction_logs
return subscriptions

77
src/ytdl_sub/cli/output_summary.py
View file

@ -1,10 +1,8 @@
from typing import List
from typing import Tuple
from colorama import Fore
from ytdl_sub.subscriptions.subscription import Subscription
from ytdl_sub.utils.file_handler import FileHandlerTransactionLog
from ytdl_sub.utils.logger import Logger
logger = Logger.get()
@ -37,51 +35,52 @@ def _color_int(value: int) -> str:
return _no_color(str_int)
def output_summary(transaction_logs: List[Tuple[Subscription, FileHandlerTransactionLog]]) -> str:
def output_summary(subscriptions: List[Subscription]) -> None:
"""
Parameters
----------
transaction_logs
Transaction logs from downloaded subscriptions
subscriptions
Processed subscriptions
Returns
-------
Output summary to print
"""
# many locals for proper output printing
# pylint: disable=too-many-locals
if len(subscriptions) == 0:
logger.info("No subscriptions ran")
return
summary: List[str] = []
# Initialize totals to 0
total_subs: int = len(subscriptions)
total_subs_str = f"Total: {total_subs}"
total_added: int = sum(sub.num_entries_added for sub in subscriptions)
total_modified: int = sum(sub.num_entries_modified for sub in subscriptions)
total_removed: int = sum(sub.num_entries_removed for sub in subscriptions)
total_entries: int = sum(sub.num_entries for sub in subscriptions)
total_errors: int = sum(sub.exception is not None for sub in subscriptions)
# Initialize widths to 0
width_sub_name: int = 0
width_num_entries_added: int = 0
width_num_entries_modified: int = 0
width_num_entries_removed: int = 0
width_num_entries: int = 0
# Calculate min width needed
for subscription, _ in transaction_logs:
width_sub_name = max(width_sub_name, len(subscription.name))
width_num_entries_added = max(
width_num_entries_added, len(_color_int(subscription.num_entries_added))
)
width_num_entries_modified = max(
width_num_entries_modified, len(_color_int(subscription.num_entries_modified))
)
width_num_entries_removed = max(
width_num_entries_removed, len(_color_int(subscription.num_entries_removed * -1))
)
width_num_entries = max(width_num_entries, len(str(subscription.num_entries)))
# Add spacing for aesthetics
width_sub_name += 4
width_num_entries += 4
width_sub_name: int = max(len(sub.name) for sub in subscriptions) + 4 # aesthetics
width_num_entries_added: int = len(_color_int(total_added))
width_num_entries_modified: int = len(_color_int(total_modified))
width_num_entries_removed: int = len(_color_int(total_removed))
width_num_entries: int = len(str(total_entries)) + 4 # aesthetics
# Build the summary
for subscription, _ in transaction_logs:
for subscription in subscriptions:
num_entries_added = _color_int(subscription.num_entries_added)
num_entries_modified = _color_int(subscription.num_entries_modified)
num_entries_removed = _color_int(subscription.num_entries_removed * -1)
num_entries = str(subscription.num_entries)
status = _green("success")
status = (
_red(subscription.exception.__class__.__name__)
if subscription.exception
else _green("✔")
)
summary.append(
f"{subscription.name:<{width_sub_name}} "
@ -92,5 +91,23 @@ def output_summary(transaction_logs: List[Tuple[Subscription, FileHandlerTransac
f"{status}"
)
total_errors_str = (
_green("Success") if total_errors == 0 else _red(f"Error{'s' if total_errors > 1 else ''}")
)
summary.append(
f"{total_subs_str:<{width_sub_name}} "
f"{_color_int(total_added):>{width_num_entries_added}} "
f"{_color_int(total_modified):>{width_num_entries_modified}} "
f"{_color_int(total_removed):>{width_num_entries_removed}} "
f"{total_entries:>{width_num_entries}} "
f"{total_errors_str}"
)
if total_errors > 0:
summary.append("")
summary.append(f"See `{Logger.error_log_filename()}` for details on errors.")
summary.append("Consider making a GitHub issue including the uploaded log file.")
# Hack to always show download summary, even if logs are set to quiet
logger.warning("Download Summary:\n%s", "\n".join(summary))

14
src/ytdl_sub/cli/output_transaction_log.py
View file

@ -1,10 +1,8 @@
from typing import List
from typing import Optional
from typing import Tuple
from ytdl_sub.subscriptions.subscription import Subscription
from ytdl_sub.utils.exceptions import ValidationException
from ytdl_sub.utils.file_handler import FileHandlerTransactionLog
from ytdl_sub.utils.logger import Logger
logger = Logger.get()
@ -23,7 +21,7 @@ def _maybe_validate_transaction_log_file(transaction_log_file_path: Optional[str
def output_transaction_log(
transaction_logs: List[Tuple[Subscription, FileHandlerTransactionLog]],
subscriptions: List[Subscription],
transaction_log_file_path: Optional[str],
) -> None:
"""
@ -31,19 +29,19 @@ def output_transaction_log(
Parameters
----------
transaction_logs
The transaction logs from downloaded subscriptions
subscriptions
Processed subscriptions
transaction_log_file_path
Optional file path to write to
"""
transaction_log_file_contents = ""
for subscription, transaction_log in transaction_logs:
if transaction_log.is_empty:
for subscription in subscriptions:
if subscription.transaction_log.is_empty:
transaction_log_contents = f"\nNo files changed for {subscription.name}"
else:
transaction_log_contents = (
f"Transaction log for {subscription.name}:\n"
f"{transaction_log.to_output_message(subscription.output_directory)}"
f"{subscription.transaction_log.to_output_message(subscription.output_directory)}"
)
if transaction_log_file_path:

16
src/ytdl_sub/main.py
View file

@ -4,7 +4,7 @@ from ytdl_sub.cli.parsers.main import parser
from ytdl_sub.utils.logger import Logger
def _main():
def _main() -> int:
# Set log level before any other ytdl-sub files are imported. That way, when loggers
# get initialized, they will see the set log level
args, _ = parser.parse_known_args()
@ -15,7 +15,10 @@ def _main():
# pylint: enable=import-outside-toplevel
ytdl_sub.cli.entrypoint.main()
subs = ytdl_sub.cli.entrypoint.main()
if any(sub.exception for sub in subs):
return 1 # Return error-code if any exceptions occurred
return 0
def main():
@ -23,14 +26,13 @@ def main():
Entrypoint for ytdl-sub
"""
try:
_main()
Logger.cleanup() # Ran successfully, so we can delete the debug file
return_code = _main()
Logger.cleanup(cleanup_error_log=return_code == 0)
sys.exit(return_code)
except Exception as exc: # pylint: disable=broad-except
Logger.log_exit_exception(exception=exc)
Logger.log_exception(exception=exc)
sys.exit(1)
sys.exit(0)
if __name__ == "__main__":
main()

21
src/ytdl_sub/subscriptions/base_subscription.py
View file

@ -9,6 +9,7 @@ from ytdl_sub.config.preset_options import OutputOptions
from ytdl_sub.config.preset_options import Overrides
from ytdl_sub.config.preset_options import YTDLOptions
from ytdl_sub.downloaders.url.validators import MultiUrlValidator
from ytdl_sub.utils.file_handler import FileHandlerTransactionLog
from ytdl_sub.utils.logger import Logger
from ytdl_sub.ytdl_additions.enhanced_download_archive import EnhancedDownloadArchive
@ -59,6 +60,8 @@ class BaseSubscription(ABC):
migrated_file_name=migrated_file_name,
)
self._exception: Optional[Exception] = None
@property
def downloader_options(self) -> MultiUrlValidator:
"""
@ -167,6 +170,24 @@ class BaseSubscription(ABC):
"""
return self._enhanced_download_archive.num_entries
@property
def transaction_log(self) -> FileHandlerTransactionLog:
"""
Returns
-------
Transaction log from the subscription
"""
return self._enhanced_download_archive.get_file_handler_transaction_log()
@property
def exception(self) -> Optional[Exception]:
"""
Returns
-------
An exception if one occurred while processing the subscription
"""
return self._exception
def as_yaml(self) -> str:
"""
Returns

17
src/ytdl_sub/subscriptions/subscription_download.py
View file

@ -326,7 +326,9 @@ class SubscriptionDownload(BaseSubscription, ABC):
If true, do not download any video/audio files or move anything to the output
directory.
"""
self._exception = None
self._enhanced_download_archive.reinitialize(dry_run=dry_run)
plugins = self._initialize_plugins()
subscription_ytdl_options = SubscriptionYTDLOptions(
@ -353,6 +355,19 @@ class SubscriptionDownload(BaseSubscription, ABC):
dry_run=dry_run,
)
@contextlib.contextmanager
def exception_handling(self) -> None:
"""
Try to perform something on the subscription.
Store the error if one occurs.
"""
try:
yield
except Exception as exc: # pylint: disable=broad-except
self._exception = exc
return self.transaction_log
def update_with_info_json(self, dry_run: bool = False) -> FileHandlerTransactionLog:
"""
Performs the subscription update using local info json files.
@ -362,7 +377,9 @@ class SubscriptionDownload(BaseSubscription, ABC):
dry_run
If true, do not modify any video/audio files or move anything to the output directory.
"""
self._exception = None
self._enhanced_download_archive.reinitialize(dry_run=dry_run)
plugins = self._initialize_plugins()
subscription_ytdl_options = SubscriptionYTDLOptions(

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

@ -91,11 +91,9 @@ class Logger:
# Ignore 'using with' warning since this will be cleaned up later
# pylint: disable=R1732
_DEBUG_LOGGER_FILE = tempfile.NamedTemporaryFile(prefix="ytdl-sub.", delete=False)
_ERROR_LOG_FILE = tempfile.NamedTemporaryFile(prefix="ytdl-sub.errors", 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] = []
@ -108,6 +106,15 @@ class Logger:
"""
return cls._DEBUG_LOGGER_FILE.name
@classmethod
def error_log_filename(cls) -> str:
"""
Returns
-------
File name of the error log file
"""
return cls._ERROR_LOG_FILE.name
@classmethod
def set_log_level(cls, log_level_name: str):
"""
@ -204,9 +211,12 @@ class Logger:
redirect_stream.flush()
@classmethod
def log_exit_exception(cls, exception: Exception, log_filepath: Optional[Path] = None):
def log_exception(cls, exception: Exception, log_filepath: Optional[Path] = None):
"""
Performs the final log before exiting from an error
Logs an exception based on the exception type. Will transfer all
debug logs into the error log file. This allows for subscriptions to only write to the
error log if an error occurred - successful subscriptions will clean the debug log file
w/out any write to the error log.
Parameters
----------
@ -215,34 +225,37 @@ class Logger:
log_filepath
Optional. The filepath to the debug logs
"""
if not cls._LOGGED_EXIT_EXCEPTION:
logger = cls.get()
logger = cls.get()
# Log validation exceptions as-is
if isinstance(exception, ValidationException):
logger.error(str(exception))
# Log permission errors explicitly
elif isinstance(exception, PermissionError):
logger.error(
"A permission error occurred:\n%s\n"
"The user running ytdl-sub must have permission to this file/directory.",
str(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(),
)
# Log validation exceptions as-is
if isinstance(exception, ValidationException):
logger.error(str(exception))
# Log permission errors explicitly
elif isinstance(exception, PermissionError):
logger.error(
"A permission error occurred:\n%s\n"
"The user running ytdl-sub must have permission to this file/directory.",
str(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.error_log_filename(),
)
cls._LOGGED_EXIT_EXCEPTION = True
# Any time an exception occurs, dump all debug logs into the error log
with open(cls.debug_log_filename(), mode="r", encoding="utf-8") as debug_logs, open(
cls.error_log_filename(), mode="a", encoding="utf-8"
) as error_logs:
error_logs.writelines(debug_logs.readlines())
@classmethod
def cleanup(cls):
def cleanup(cls, cleanup_error_log: bool = False):
"""
Cleans up debug log file left behind
"""
@ -251,6 +264,8 @@ class Logger:
handler.close()
cls._DEBUG_LOGGER_FILE.close()
FileHandler.delete(cls.debug_log_filename())
cls._LOGGED_EXIT_EXCEPTION = False
if cleanup_error_log:
cls._ERROR_LOG_FILE.close()
FileHandler.delete(cls.error_log_filename())

4
tests/e2e/conftest.py
View file

@ -2,7 +2,6 @@ import json
import sys
import tempfile
from typing import List
from typing import Tuple
from unittest.mock import patch
import pytest
@ -10,7 +9,6 @@ import pytest
from ytdl_sub.cli.entrypoint import main
from ytdl_sub.subscriptions.subscription import Subscription
from ytdl_sub.utils.file_handler import FileHandler
from ytdl_sub.utils.file_handler import FileHandlerTransactionLog
@pytest.fixture()
@ -46,7 +44,7 @@ def timestamps_file_path():
FileHandler.delete(tmp.name)
def mock_run_from_cli(args: str) -> List[Tuple[Subscription, FileHandlerTransactionLog]]:
def mock_run_from_cli(args: str) -> List[Subscription]:
args_list = ["ytdl-sub"] + args.split()
with patch.object(sys, "argv", args_list):
return main()

6
tests/e2e/plugins/internal/test_view.py
View file

@ -17,10 +17,10 @@ class TestView:
args = f"view "
args += "--split-chapters " if split_chapters else ""
args += f"https://www.youtube.com/playlist?list=PLBsm_SagFMmdWnCnrNtLjA9kzfrRkto4i"
subscription_transaction_log = mock_run_from_cli(args=args)
subscriptions = mock_run_from_cli(args=args)
assert len(subscription_transaction_log) == 1
transaction_log = subscription_transaction_log[0][1]
assert len(subscriptions) == 1
transaction_log = subscriptions[0].transaction_log
# Ensure the video and thumbnail are recognized
assert len(transaction_log.files_created) == 2

8
tests/e2e/youtube/test_playlist.py
View file

@ -163,10 +163,10 @@ class TestPlaylist:
) as subscription_path:
args = "--dry-run " if dry_run else ""
args += f"sub {subscription_path}"
subscription_transaction_log = mock_run_from_cli(args=args)
subscriptions = mock_run_from_cli(args=args)
assert len(subscription_transaction_log) == 1
transaction_log = subscription_transaction_log[0][1]
assert len(subscriptions) == 1
transaction_log = subscriptions[0].transaction_log
assert_transaction_log_matches(
output_directory=output_directory,
@ -186,7 +186,7 @@ class TestPlaylist:
expected_message="ExistingVideoReached, stopping additional downloads",
log_level="debug",
):
transaction_log = mock_run_from_cli(args=args)[0][1]
transaction_log = mock_run_from_cli(args=args)[0].transaction_log
assert transaction_log.is_empty
assert_expected_downloads(

8
tests/e2e/youtube/test_video.py
View file

@ -187,14 +187,12 @@ class TestYoutubeVideo:
args = "--dry-run " if dry_run else ""
args += f"--config {music_video_config_path} "
args += f"dl {single_video_preset_dict_dl_args}"
subscription_transaction_log = mock_run_from_cli(args=args)
assert len(subscription_transaction_log) == 1
transaction_log = subscription_transaction_log[0][1]
subscriptions = mock_run_from_cli(args=args)
assert len(subscriptions) == 1
assert_transaction_log_matches(
output_directory=output_directory,
transaction_log=transaction_log,
transaction_log=subscriptions[0].transaction_log,
transaction_log_summary_file_name="youtube/test_video_cli.txt",
)
assert_expected_downloads(

87
tests/unit/cli/conftest.py Normal file
View file

@ -0,0 +1,87 @@
import datetime
import os
import shutil
import tempfile
import time
from typing import Callable
from unittest.mock import patch
import mergedeep
import pytest
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.file_handler import FileMetadata
from ytdl_sub.utils.logger import Logger
@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
)
if not mock_success_output:
raise ValueError("error")
(
self._enhanced_download_archive.get_file_handler_transaction_log()
.log_created_file("created_file.txt", FileMetadata())
.log_modified_file("modified_file.txt", FileMetadata())
.log_removed_file("deleted_file.txt")
)
return self._enhanced_download_archive.get_file_handler_transaction_log()
return _mock_download
return _mock_subscription_download_factory
@pytest.fixture
def mock_subscription_download_success(mock_subscription_download_factory: Callable):
with patch.object(
Subscription,
"download",
new=mock_subscription_download_factory(mock_success_output=True),
):
yield
@pytest.fixture
def persist_logs_directory() -> str:
# Delete the temp_dir on creation
with tempfile.TemporaryDirectory() as temp_dir:
pass
yield temp_dir
if os.path.isdir(temp_dir):
shutil.rmtree(temp_dir)
@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

229
tests/unit/cli/test_entrypoint.py
View file

@ -1,110 +1,24 @@
import os.path
import re
import shutil
import sys
import tempfile
import time
from pathlib import Path
from typing import Callable
from typing import List
from typing import Optional
from typing import Tuple
from unittest.mock import MagicMock
from unittest.mock import Mock
from unittest.mock import patch
import mergedeep
import pytest
from conftest import assert_logs
from ytdl_sub.cli.entrypoint import _download_subscriptions_from_yaml_files
from ytdl_sub.cli.entrypoint import main
from ytdl_sub.cli.output_summary import output_summary
from ytdl_sub.cli.output_transaction_log import logger as transaction_logger
from ytdl_sub.config.config_file import ConfigFile
from ytdl_sub.subscriptions.subscription import Subscription
from ytdl_sub.utils.exceptions import ExperimentalFeatureNotEnabled
from ytdl_sub.utils.file_handler import FileHandler
from ytdl_sub.utils.file_handler import FileHandlerTransactionLog
from ytdl_sub.utils.file_handler import FileMetadata
from ytdl_sub.utils.logger import Logger
####################################################################################################
# SHARED FIXTURES
@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()
.log_created_file("created_file.txt", FileMetadata())
.log_modified_file("modified_file.txt", FileMetadata())
.log_removed_file("deleted_file.txt")
)
return _mock_download
return _mock_subscription_download_factory
@pytest.fixture
def mock_subscription_download_success(mock_subscription_download_factory: Callable):
with patch.object(
Subscription,
"download",
new=mock_subscription_download_factory(mock_success_output=True),
):
yield
####################################################################################################
# PERSIST LOGS FIXTURES + TESTS
@pytest.fixture
def persist_logs_directory() -> str:
# Delete the temp_dir on creation
with tempfile.TemporaryDirectory() as temp_dir:
pass
yield temp_dir
if os.path.isdir(temp_dir):
shutil.rmtree(temp_dir)
@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.mark.parametrize("dry_run", [True, False])
@pytest.mark.parametrize("mock_success_output", [True, False])
@pytest.mark.parametrize("keep_successful_logs", [True, False])
@ -125,7 +39,8 @@ def test_subscription_logs_write_to_file(
Subscription,
"download",
new=mock_subscription_download_factory(mock_success_output=mock_success_output),
):
# mock datetime to be an index to be able to run instantly
), patch("ytdl_sub.cli.entrypoint._log_time", side_effect=[str(idx) for idx in range(10)]):
try:
_download_subscriptions_from_yaml_files(
config=config,
@ -142,24 +57,22 @@ def test_subscription_logs_write_to_file(
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
# If not success, expect 2 log files for both sub errors
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"
)
assert len(log_directory_files) == 2
for log_path in log_directory_files:
assert bool(re.match(r"\d\.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)
)
assert bool(re.match(r"\d\.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]
@ -167,122 +80,6 @@ def test_subscription_logs_write_to_file(
)
####################################################################################################
# TRANSACTION LOGS FIXTURES + TESTS
@pytest.fixture
def transaction_log_file_path() -> str:
# Delete the temp_file on creation
with tempfile.NamedTemporaryFile() as temp_file:
pass
yield temp_file.name
if os.path.isfile(temp_file.name):
FileHandler.delete(temp_file.name)
@pytest.mark.parametrize("file_transaction_log", [None, "output.log"])
def test_suppress_transaction_log(
mock_subscription_download_success,
music_video_config_path: Path,
music_video_subscription_path: Path,
file_transaction_log: Optional[str],
) -> None:
with patch.object(
sys,
"argv",
[
"ytdl-sub",
"--config",
str(music_video_config_path),
"sub",
str(music_video_subscription_path),
"--suppress-transaction-log",
]
+ (["--transaction-log", file_transaction_log] if file_transaction_log else []),
), patch("ytdl_sub.cli.output_transaction_log.output_transaction_log") as mock_transaction_log:
transaction_logs = main()
assert transaction_logs
assert mock_transaction_log.call_count == 0
def test_transaction_log_to_file(
mock_subscription_download_success,
music_video_config_path: Path,
music_video_subscription_path: Path,
transaction_log_file_path: Path,
) -> None:
with patch.object(
sys,
"argv",
[
"ytdl-sub",
"--config",
str(music_video_config_path),
"sub",
str(music_video_subscription_path),
"--transaction-log",
str(transaction_log_file_path),
],
):
transaction_logs = main()
assert transaction_logs
with open(transaction_log_file_path, "r", encoding="utf-8") as transaction_log_file:
assert transaction_log_file.readlines()[0] == "Transaction log for john_smith:\n"
def test_transaction_log_to_logger(
mock_subscription_download_success,
music_video_config_path: Path,
music_video_subscription_path: Path,
) -> None:
with patch.object(
sys,
"argv",
[
"ytdl-sub",
"--config",
str(music_video_config_path),
"sub",
str(music_video_subscription_path),
],
), assert_logs(
logger=transaction_logger,
expected_message="Transaction log for john_smith:\n",
log_level="info",
):
transaction_logs = main()
assert transaction_logs
def test_output_summary():
subscription_values: List[Tuple[str, int, int, int, int]] = [
("long_name_but_lil_values", 0, 0, 0, 6),
("john_smith", 1, 0, 0, 52),
("david_gore", 0, 0, 0, 4),
("christopher_snoop", 50, 0, 3, 518),
("beyond funk", 0, 0, 0, 176),
]
mock_subscriptions: List[Tuple[MagicMock, FileHandlerTransactionLog]] = []
for values in subscription_values:
sub = Mock()
sub.name = values[0]
sub.num_entries_added = values[1]
sub.num_entries_modified = values[2]
sub.num_entries_removed = values[3]
sub.num_entries = values[4]
mock_subscriptions.append((sub, FileHandlerTransactionLog()))
_ = output_summary(transaction_logs=mock_subscriptions)
assert True # Test used for manual inspection - too hard to test ansi color codes
def test_update_with_info_json_requires_experimental_flag(
music_video_config_path: Path,
music_video_subscription_path: Path,

67
tests/unit/cli/test_output_summary.py Normal file
View file

@ -0,0 +1,67 @@
from typing import List
from typing import Optional
from typing import Tuple
from unittest.mock import MagicMock
from unittest.mock import Mock
from ytdl_sub.cli.output_summary import output_summary
def _to_mock_subscriptions(
subscription_values: List[Tuple[str, int, int, int, int, Optional[Exception]]]
) -> List[MagicMock]:
mock_subscriptions: List[MagicMock] = []
for values in subscription_values:
sub = Mock()
sub.name = values[0]
sub.num_entries_added = values[1]
sub.num_entries_modified = values[2]
sub.num_entries_removed = values[3]
sub.num_entries = values[4]
sub.exception = values[5]
mock_subscriptions.append(sub)
return mock_subscriptions
def test_output_summary_no_errors():
mock_subscriptions = _to_mock_subscriptions(
[
("long_name_but_lil_values", 0, 0, 0, 6, None),
("john_smith", 1, 0, 0, 52, None),
("david_gore", 0, 0, 0, 4, None),
("christopher_snoop", 50, 0, 3, 518, None),
("beyond funk", 352, 0, 0, 2342, None),
]
)
output_summary(subscriptions=mock_subscriptions)
def test_output_summary_one_error():
mock_subscriptions = _to_mock_subscriptions(
[
("long_name_but_lil_values", 0, 0, 0, 6, None),
("john_smith", 1, 0, 0, 52, None),
("david_gore", 0, 0, 0, 4, None),
("christopher_snoop", 50, 0, 3, 518, None),
("beyond funk", 0, 0, 0, 176, ValueError("lol")),
]
)
output_summary(subscriptions=mock_subscriptions)
def test_output_summary_multiple_errors():
mock_subscriptions = _to_mock_subscriptions(
[
("long_name_but_lil_values", 0, 0, 0, 6, None),
("john_smith", 1, 0, 0, 52, None),
("david_gore", 0, 0, 0, 4, PermissionError("ack")),
("christopher_snoop", 50, 0, 3, 518, None),
("beyond funk", 0, 0, 0, 176, ValueError("lol")),
]
)
output_summary(subscriptions=mock_subscriptions)

101
tests/unit/cli/test_output_transaction_log.py Normal file
View file

@ -0,0 +1,101 @@
import os
import sys
import tempfile
from pathlib import Path
from typing import Optional
from unittest.mock import patch
import pytest
from conftest import assert_logs
from ytdl_sub.cli.entrypoint import main
from ytdl_sub.cli.output_transaction_log import logger as transaction_logger
from ytdl_sub.utils.file_handler import FileHandler
@pytest.fixture
def transaction_log_file_path() -> str:
# Delete the temp_file on creation
with tempfile.NamedTemporaryFile() as temp_file:
pass
yield temp_file.name
if os.path.isfile(temp_file.name):
FileHandler.delete(temp_file.name)
@pytest.mark.parametrize("file_transaction_log", [None, "output.log"])
def test_suppress_transaction_log(
mock_subscription_download_success,
music_video_config_path: Path,
music_video_subscription_path: Path,
file_transaction_log: Optional[str],
) -> None:
with patch.object(
sys,
"argv",
[
"ytdl-sub",
"--config",
str(music_video_config_path),
"sub",
str(music_video_subscription_path),
"--suppress-transaction-log",
]
+ (["--transaction-log", file_transaction_log] if file_transaction_log else []),
), patch("ytdl_sub.cli.output_transaction_log.output_transaction_log") as mock_transaction_log:
subscriptions = main()
assert subscriptions
assert mock_transaction_log.call_count == 0
def test_transaction_log_to_file(
mock_subscription_download_success,
music_video_config_path: Path,
music_video_subscription_path: Path,
transaction_log_file_path: Path,
) -> None:
with patch.object(
sys,
"argv",
[
"ytdl-sub",
"--config",
str(music_video_config_path),
"sub",
str(music_video_subscription_path),
"--transaction-log",
str(transaction_log_file_path),
],
):
subscriptions = main()
assert subscriptions
with open(transaction_log_file_path, "r", encoding="utf-8") as transaction_log_file:
assert transaction_log_file.readlines()[0] == "Transaction log for john_smith:\n"
def test_transaction_log_to_logger(
mock_subscription_download_success,
music_video_config_path: Path,
music_video_subscription_path: Path,
) -> None:
with patch.object(
sys,
"argv",
[
"ytdl-sub",
"--config",
str(music_video_config_path),
"sub",
str(music_video_subscription_path),
],
), assert_logs(
logger=transaction_logger,
expected_message="Transaction log for john_smith:\n",
log_level="info",
):
subscriptions = main()
assert subscriptions

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

@ -38,10 +38,18 @@ def mock_sys_exit():
return _mock_sys_exit
def test_main_success(mock_sys_exit):
with mock_sys_exit(expected_exit_code=0):
with patch("src.ytdl_sub.main._main"):
main()
@pytest.mark.parametrize("return_code", [0, 1])
def test_main_exit_code(mock_sys_exit, return_code: int):
with mock_sys_exit(expected_exit_code=return_code), patch(
"src.ytdl_sub.main._main"
) as mock_inner_main, patch.object(Logger, "cleanup") as mock_logger_cleanup:
mock_inner_main.return_value = return_code
main()
assert mock_logger_cleanup.call_count == 1
assert mock_logger_cleanup.call_args.kwargs["cleanup_error_log"] == (
True if return_code == 0 else False
)
def test_main_validation_error(capsys, mock_sys_exit):
@ -70,7 +78,7 @@ def test_main_uncaught_error(capsys, mock_sys_exit, expected_uncaught_error_mess
assert mock_error.call_count == 1
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()
assert mock_error.call_args.args[2] == Logger.error_log_filename()
def test_main_permission_error(capsys, mock_sys_exit, expected_uncaught_error_message):

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

@ -111,11 +111,12 @@ class TestLogger:
Logger.cleanup()
assert not os.path.isfile(Logger._DEBUG_LOGGER_FILE.name)
def test_logger_can_be_cleaned_during_execution(self):
@pytest.mark.parametrize("clean_error_log", [True, False])
def test_logger_can_be_cleaned_during_execution(self, clean_error_log: bool):
Logger._LOGGER_LEVEL = LoggerLevels.INFO
logger = Logger.get(name="name_test")
for _ in range(2):
for iteration in range(2):
logger.info("info test")
logger.debug("debug test")
@ -127,8 +128,39 @@ class TestLogger:
"[ytdl-sub:name_test] debug test\n",
]
Logger.cleanup()
assert not os.path.isfile(Logger._DEBUG_LOGGER_FILE.name)
try:
raise ValueError("some error")
except ValueError as exc:
Logger.log_exception(exception=exc)
Logger.cleanup(cleanup_error_log=clean_error_log)
assert not os.path.isfile(Logger.debug_log_filename())
assert clean_error_log == (not os.path.isfile(Logger.error_log_filename()))
if not clean_error_log:
with open(Logger.error_log_filename(), mode="r", encoding="utf-8") as err_file:
err_logs = err_file.readlines()
expected = [
"[ytdl-sub:name_test] info test\n",
"[ytdl-sub:name_test] debug test\n",
"[ytdl-sub] An uncaught error occurred:\n",
"Traceback (most recent call last):\n",
' File "/home/j/workspace/ytdl-sub/tests/unit/utils/test_logger.py", line 132, in test_logger_can_be_cleaned_during_execution\n',
' raise ValueError("some error")\n',
"ValueError: some error\n",
"[ytdl-sub] Version 2023.03.24+14e4a4b\n",
f"Please upload the error log file '{Logger.error_log_filename()}' 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!\n",
]
assert err_logs[0:4] == expected[0:4]
assert err_logs[8] == expected[8]
if iteration == 1:
err_lines = len(expected)
# Two errors occurred, error log should contain 2
assert err_logs[err_lines : err_lines + 4] == expected[0:4]
assert err_logs[err_lines + 8] == expected[8]
@pytest.mark.parametrize(
"log_level, expected_stdout",