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.
197 lines
6.8 KiB
Python
197 lines
6.8 KiB
Python
import os.path
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from ytdl_sub.utils.logger import Logger
|
|
from ytdl_sub.utils.logger import LoggerLevels
|
|
|
|
|
|
class TestLogger:
|
|
@pytest.mark.parametrize(
|
|
"log_level",
|
|
[
|
|
LoggerLevels.QUIET,
|
|
LoggerLevels.INFO,
|
|
LoggerLevels.VERBOSE,
|
|
LoggerLevels.DEBUG,
|
|
],
|
|
)
|
|
def test_logger_level_set(self, log_level):
|
|
Logger.set_log_level(log_level_name=log_level.name)
|
|
assert Logger._LOGGER_LEVEL == log_level
|
|
|
|
def test_logger_level_invalid_name(self):
|
|
with pytest.raises(ValueError):
|
|
Logger.set_log_level("nope")
|
|
|
|
@pytest.mark.parametrize(
|
|
"log_level",
|
|
[
|
|
LoggerLevels.QUIET,
|
|
LoggerLevels.INFO,
|
|
LoggerLevels.VERBOSE,
|
|
LoggerLevels.DEBUG,
|
|
],
|
|
)
|
|
def test_logger_warning_stdout(self, capsys, log_level):
|
|
Logger._LOGGER_LEVEL = log_level
|
|
logger = Logger.get(name="name_test")
|
|
|
|
logger.warning("test")
|
|
captured = capsys.readouterr()
|
|
|
|
# Warning should always be captured
|
|
assert captured.out == "[ytdl-sub:name_test] test\n"
|
|
|
|
@pytest.mark.parametrize(
|
|
"log_level, outputs_to_stdout",
|
|
[
|
|
(LoggerLevels.QUIET, False),
|
|
(LoggerLevels.INFO, True),
|
|
(LoggerLevels.VERBOSE, True),
|
|
(LoggerLevels.DEBUG, True),
|
|
],
|
|
)
|
|
def test_logger_info_stdout(self, capsys, log_level, outputs_to_stdout):
|
|
Logger._LOGGER_LEVEL = log_level
|
|
logger = Logger.get(name="name_test")
|
|
|
|
logger.info("test")
|
|
captured = capsys.readouterr()
|
|
|
|
if outputs_to_stdout:
|
|
assert captured.out == "[ytdl-sub:name_test] test\n"
|
|
else:
|
|
assert not captured.out
|
|
|
|
@pytest.mark.parametrize(
|
|
"log_level, outputs_to_stdout",
|
|
[
|
|
(LoggerLevels.QUIET, False),
|
|
(LoggerLevels.INFO, False),
|
|
(LoggerLevels.VERBOSE, False),
|
|
(LoggerLevels.DEBUG, True),
|
|
],
|
|
)
|
|
def test_logger_debug_stdout(self, capsys, log_level, outputs_to_stdout):
|
|
Logger._LOGGER_LEVEL = log_level
|
|
logger = Logger.get(name="name_test")
|
|
|
|
logger.debug("test")
|
|
captured = capsys.readouterr()
|
|
|
|
if outputs_to_stdout:
|
|
assert captured.out == "[ytdl-sub:name_test] test\n"
|
|
else:
|
|
assert not captured.out
|
|
|
|
@pytest.mark.parametrize(
|
|
"log_level",
|
|
[
|
|
LoggerLevels.QUIET,
|
|
LoggerLevels.INFO,
|
|
LoggerLevels.VERBOSE,
|
|
LoggerLevels.DEBUG,
|
|
],
|
|
)
|
|
def test_logger_always_outputs_to_debug_file(self, log_level):
|
|
Logger._LOGGER_LEVEL = log_level
|
|
logger = Logger.get(name="name_test")
|
|
|
|
logger.info("info test")
|
|
logger.debug("debug test")
|
|
|
|
with open(Logger._DEBUG_LOGGER_FILE.name, "r", encoding="utf-8") as log_file:
|
|
lines = log_file.readlines()
|
|
|
|
assert lines == ["[ytdl-sub:name_test] info test\n", "[ytdl-sub:name_test] debug test\n"]
|
|
|
|
# Ensure the file cleans up too
|
|
Logger.cleanup()
|
|
assert not os.path.isfile(Logger._DEBUG_LOGGER_FILE.name)
|
|
|
|
@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 iteration in range(2):
|
|
logger.info("info test")
|
|
logger.debug("debug test")
|
|
|
|
with open(Logger._DEBUG_LOGGER_FILE.name, "r", encoding="utf-8") as log_file:
|
|
lines = log_file.readlines()
|
|
|
|
assert lines == [
|
|
"[ytdl-sub:name_test] info test\n",
|
|
"[ytdl-sub:name_test] debug test\n",
|
|
]
|
|
|
|
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",
|
|
[
|
|
(LoggerLevels.QUIET, False),
|
|
(LoggerLevels.INFO, False),
|
|
(LoggerLevels.VERBOSE, True),
|
|
(LoggerLevels.DEBUG, True),
|
|
],
|
|
)
|
|
def test_handle_external_logs(self, capsys, log_level, expected_stdout):
|
|
Logger._LOGGER_LEVEL = log_level
|
|
expected_lines = [
|
|
"[ytdl-sub:name_test] test line 1\n",
|
|
"[ytdl-sub:name_test] test line 2\n",
|
|
]
|
|
with Logger.handle_external_logs(name="name_test"):
|
|
print("test line 1")
|
|
print("test line 2")
|
|
|
|
# Wait for flushed logs
|
|
time.sleep(0.1)
|
|
|
|
# Ensure it goes to stdout only if it is expected to
|
|
captured = capsys.readouterr()
|
|
if expected_stdout:
|
|
assert captured.out == "".join(expected_lines)
|
|
else:
|
|
assert not captured.out
|
|
|
|
# Ensure it always go to the debug file
|
|
with open(Logger._DEBUG_LOGGER_FILE.name, "r", encoding="utf-8") as log_file:
|
|
lines = log_file.readlines()
|
|
assert lines == expected_lines
|