541 lines
18 KiB
Python
541 lines
18 KiB
Python
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from exchangelib import Message
|
|
from typer.testing import CliRunner
|
|
from unittest.mock import MagicMock
|
|
|
|
|
|
from logwatcher import __version__
|
|
from logwatcher.cli import _get_files_from_dir, app
|
|
from logwatcher.config import RESULT_PATH
|
|
from logwatcher.utils import is_log_file
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
@pytest.fixture(name="output_path")
|
|
def get_output_path_fixture(tmp_path: Path):
|
|
"""Return output directory path of all logs and output files will be located during tests.
|
|
|
|
Args:
|
|
tmp_path: Temporary root path for testing results
|
|
|
|
"""
|
|
return tmp_path / "output"
|
|
|
|
|
|
def get_argument_from_files(log_dir: Path, output_path: Path) -> list[str]:
|
|
"""Return an argument list to call cli.
|
|
|
|
Args:
|
|
log_dir: Path to the logwatcher directory
|
|
output_path: Temporary path for all output generated during testing
|
|
|
|
Returns:
|
|
List of arguments in string format
|
|
|
|
"""
|
|
files = [str(file) for file in log_dir.iterdir() if is_log_file(file)]
|
|
|
|
arguments = ["from-files", "--output-dir", str(output_path)]
|
|
for file in files:
|
|
arguments += ["--input-files", file]
|
|
return arguments
|
|
|
|
|
|
def test_cli_help():
|
|
"""Test the help command and assert that all options are listed."""
|
|
results = runner.invoke(app, ["--help"])
|
|
assert results.exit_code == 0
|
|
assert "--verbose" in results.output
|
|
assert "--version" in results.output
|
|
assert "from-mails" in results.output
|
|
assert "from-files" in results.output
|
|
|
|
|
|
def test_version_cli():
|
|
"""Test the version option. The version must be printed."""
|
|
result = runner.invoke(app, ["--version"])
|
|
assert result.exit_code == 0
|
|
assert f"logwatcher version: {__version__}" in result.output
|
|
|
|
|
|
def test_get_files_from_dir_empty_dir(tmp_path: Path):
|
|
"""Assert that a non-existing directory (or wrong path) cannot be given as a working input directory.
|
|
|
|
Assert that everything works even if the directory is empty.
|
|
|
|
Args:
|
|
tmp_path: Temporary root path for testing results
|
|
|
|
"""
|
|
with pytest.raises(FileNotFoundError):
|
|
_get_files_from_dir(tmp_path / "input_test")
|
|
input_test_dir = tmp_path / "input_test"
|
|
input_test_dir.mkdir()
|
|
assert input_test_dir.is_dir()
|
|
assert _get_files_from_dir(input_test_dir) == []
|
|
|
|
|
|
def test_get_files_from_dir_not_empty_dir(tmp_path: Path):
|
|
"""Assert the right files are retrieved if the directory is not empty.
|
|
|
|
Args:
|
|
tmp_path: Temporary root path for testing results
|
|
|
|
"""
|
|
# create directory
|
|
input_test_dir = tmp_path / "input_test"
|
|
input_test_dir.mkdir()
|
|
# populate it with files
|
|
Path(input_test_dir / "a.log").touch()
|
|
Path(input_test_dir / "a.txt").touch()
|
|
Path(input_test_dir / "a.csv").touch()
|
|
Path(input_test_dir / "a.py").touch()
|
|
Path(input_test_dir / "a").touch()
|
|
Path(input_test_dir / "a.pdf").touch()
|
|
|
|
assert input_test_dir.is_dir()
|
|
file_list = _get_files_from_dir(input_test_dir)
|
|
assert file_list != []
|
|
assert len(file_list) == 3
|
|
assert {file.name for file in file_list} == {"a.log", "a.txt", "a"}
|
|
|
|
|
|
def test_cli_from_files_input_files_and_input_dir(output_path: Path):
|
|
"""Assert that input files and input dir options cannot be provided in the same command.
|
|
|
|
Args:
|
|
output_path: directory containing all files generated by the process
|
|
|
|
"""
|
|
results = runner.invoke(
|
|
app,
|
|
[
|
|
"from-files",
|
|
"--input-files",
|
|
"a.log",
|
|
"--input-dir",
|
|
"logs",
|
|
"--output-dir",
|
|
str(output_path),
|
|
],
|
|
)
|
|
assert results.exit_code == 2
|
|
assert "Only one of input_files or input_dir must be provided." in results.output
|
|
assert not output_path.exists()
|
|
|
|
|
|
def test_cli_from_files_neither_input_files_nor_input_dir():
|
|
"""Test calling cli without providing input files or input dir options.
|
|
|
|
Must return an error on provided parameters (BadParameter = exit code 2)
|
|
|
|
"""
|
|
results = runner.invoke(app, ["from-files"])
|
|
assert results.exit_code == 2
|
|
assert "Either input_files or input_dir must be provided" in results.output
|
|
|
|
|
|
def test_cli_from_files_bad_input_file(
|
|
tmp_path: Path, output_path: Path, invalid_log_dir: Path
|
|
):
|
|
"""Test calling cli with bad input file (directory instead of file).
|
|
|
|
Must return an exit code at 1 and the ValueError exception raised
|
|
while `_files_to_log_entries()` was running.
|
|
|
|
Args:
|
|
tmp_path: Temporary root path for testing results
|
|
output_path: Temporary path for all output generated during testing
|
|
invalid_log_dir: Fixture of directory of invalid logs
|
|
|
|
"""
|
|
directory = str(tmp_path)
|
|
results = runner.invoke(
|
|
app,
|
|
["from-files", "--input-files", directory, "--output-dir", str(output_path)],
|
|
)
|
|
assert results.exit_code == 1
|
|
assert isinstance(results.exception, ValueError)
|
|
assert f"'{directory}' is not a file" in results.exception.args[0]
|
|
assert (
|
|
not output_path.exists()
|
|
) # would be created if `directory` was a valid parameter
|
|
|
|
|
|
def test_cli_from_files_empty_input_files(output_path: Path, invalid_log_dir: Path):
|
|
"""Test calling cli with empty input files.
|
|
|
|
Must pass and generate reports that mention reading a file only.
|
|
All other parameters must be at 0.
|
|
|
|
Args:
|
|
output_path: Temporary path for all output generated during testing
|
|
invalid_log_dir: Fixture of directory of invalid logs
|
|
|
|
"""
|
|
empty_file = str(next(invalid_log_dir.glob("empty_file.txt")))
|
|
results = runner.invoke(
|
|
app,
|
|
["from-files", "--input-files", empty_file, "--output-dir", str(output_path)],
|
|
)
|
|
assert results.exit_code == 0
|
|
# check files
|
|
nb_files = 1 # only empty_file.txt should have been read
|
|
file_list = list(output_path.iterdir())
|
|
assert len(file_list) == 3
|
|
for file in file_list:
|
|
content = file.read_text(encoding="windows-1252")
|
|
assert file.name in ("all.log", "n2.log", "other.log")
|
|
assert f"Fichier(s) lu(s)\t : {nb_files}" in content
|
|
assert "Nombre total d'erreur(s)\t: 0" in content
|
|
|
|
|
|
def test_cli_from_files_input_files_or_input_dir(
|
|
tmp_path: Path, output_path: Path, original_log_dir: Path
|
|
):
|
|
"""Assert that calling the CLI with either input files or input dir options is a valid process.
|
|
|
|
Args:
|
|
tmp_path: Temporary root path for testing results
|
|
output_path: Temporary path for all output generated during testing
|
|
original_log_dir: Fixture of directory of initial logs received by n2 technicians
|
|
|
|
"""
|
|
# output for input-files option
|
|
input_files_output = output_path / "input_files"
|
|
# save number of files from original log directory for future assertion test
|
|
nb_files = len(list(original_log_dir.iterdir()))
|
|
|
|
# test with `input files` option provided only
|
|
result = runner.invoke(
|
|
app, get_argument_from_files(original_log_dir, input_files_output)
|
|
)
|
|
assert result.exit_code == 0
|
|
assert result.exception is None
|
|
|
|
output_file_list = list(input_files_output.iterdir())
|
|
assert len(output_file_list) == 3
|
|
# check files
|
|
for file in output_file_list:
|
|
assert file.name in ("all.log", "n2.log", "other.log")
|
|
assert file.stat().st_size != 0 # must contain basic execution information
|
|
assert f"Fichier(s) lu(s)\t : {nb_files}" in file.read_text(
|
|
encoding="windows-1252"
|
|
)
|
|
|
|
# test with `input dir` option provided only
|
|
# output for input-dir option
|
|
input_dir_output = output_path / "input_dir"
|
|
result = runner.invoke(
|
|
app,
|
|
[
|
|
"from-files",
|
|
"--input-dir",
|
|
str(original_log_dir),
|
|
"--output-dir",
|
|
str(input_dir_output),
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert result.exception is None
|
|
output_file_list = list(input_dir_output.iterdir())
|
|
assert len(output_file_list) == 3
|
|
# check files
|
|
for file in output_file_list:
|
|
assert file.name in ("all.log", "n2.log", "other.log")
|
|
assert file.stat().st_size != 0 # must contain basic execution information
|
|
assert f"Fichier(s) lu(s)\t : {nb_files}" in file.read_text(
|
|
encoding="windows-1252"
|
|
)
|
|
|
|
|
|
def test_cli_from_files_input_files_one_file(valid_log_dir: Path, output_path: Path):
|
|
"""Assert that CLI can analyze a single valid file.
|
|
|
|
Args:
|
|
valid_log_dir: Fixture of directory of valid logs
|
|
output_path: Fixture of directory receiving all file generated during process.
|
|
|
|
"""
|
|
mixed_logs_file = valid_log_dir / "mixed_logs.txt"
|
|
result = runner.invoke(
|
|
app,
|
|
[
|
|
"from-files",
|
|
"--input-files",
|
|
str(mixed_logs_file),
|
|
"--output-dir",
|
|
str(output_path),
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert result.exception is None
|
|
output_file_list = list(output_path.iterdir())
|
|
assert len(output_file_list) == 3
|
|
# check file
|
|
date_pattern = r"\d{1,2}/\d{1,2}/\d{4} \d{2}:\d{2}:\d{2}"
|
|
for file in output_file_list:
|
|
content = file.read_text(encoding="windows-1252")
|
|
assert re.search(f"Période\t : {date_pattern} -> {date_pattern}\n", content)
|
|
assert file.name in ("all.log", "n2.log", "other.log")
|
|
assert "Fichier(s) lu(s)\t : 1" in file.read_text(encoding="windows-1252")
|
|
assert "Nombre total d'erreur(s)\t: 0" not in content
|
|
|
|
|
|
def test_cli_from_files_empty_input_directory(tmp_path: Path, output_path: Path):
|
|
"""Assert CLI can run on empty directory.
|
|
|
|
Assert that CLI can run on empty directory without error and
|
|
generates reports that indicates no file read.
|
|
|
|
Args:
|
|
tmp_path: Temporary root path for testing results
|
|
output_path: Temporary path for all output generated during testing
|
|
|
|
"""
|
|
input_dir = Path(tmp_path / "input_test")
|
|
input_dir.mkdir(parents=True)
|
|
|
|
result = runner.invoke(
|
|
app,
|
|
[
|
|
"from-files",
|
|
"--input-dir",
|
|
str(input_dir),
|
|
"--output-dir",
|
|
str(output_path),
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
output_file_list = list(output_path.iterdir())
|
|
assert len(output_file_list) == 3
|
|
# check files
|
|
for file in output_file_list:
|
|
content = file.read_text(encoding="windows-1252")
|
|
assert file.name in ("all.log", "n2.log", "other.log")
|
|
assert file.stat().st_size != 0 # must contain basic execution information
|
|
assert "Période : -> " in content
|
|
assert "Fichier(s) lu(s)\t : 0" in content
|
|
assert "Nombre total d'erreur(s)\t: 0" in content
|
|
|
|
|
|
def test_cli_from_files_invalid_logs(output_path: Path, invalid_log_dir: Path):
|
|
"""Assert CLI return empty values if logs are invalid.
|
|
|
|
On invalid logs (empty files, log from different sources than MDC LAME...),
|
|
CLI should run without error and creates report files with metrics at 0.
|
|
|
|
Args:
|
|
output_path: Temporary path for all output generated during testing
|
|
invalid_log_dir: Fixture of directory of invalid logs
|
|
|
|
"""
|
|
arguments = get_argument_from_files(invalid_log_dir, output_path)
|
|
results = runner.invoke(app, arguments)
|
|
|
|
assert results.exit_code == 0
|
|
assert output_path.exists()
|
|
|
|
nb_files = len(list(invalid_log_dir.iterdir()))
|
|
output_file_list = list(output_path.iterdir())
|
|
|
|
assert len(output_file_list) == 3
|
|
for file in output_file_list:
|
|
content = file.read_text(encoding="windows-1252")
|
|
assert "Période : -> " in content
|
|
assert f"Fichier(s) lu(s)\t : {nb_files}" in content
|
|
assert "Nombre total d'erreur(s)\t: 0" in content
|
|
|
|
|
|
def test_cli_from_files_valid_logs(output_path: Path, valid_log_dir: Path):
|
|
"""Assert everything works when logs are valid.
|
|
|
|
On valid test, everything should work successfully. Metrics
|
|
should not be at 0.
|
|
|
|
Args:
|
|
output_path: Temporary path for all output generated during testing
|
|
valid_log_dir: Fixture of directory of valid logs
|
|
|
|
"""
|
|
arguments = get_argument_from_files(valid_log_dir, output_path)
|
|
results = runner.invoke(app, arguments)
|
|
|
|
assert results.exit_code == 0
|
|
assert output_path.exists()
|
|
|
|
nb_files = len(list(valid_log_dir.iterdir()))
|
|
date_pattern = r"\d{1,2}/\d{1,2}/\d{4} \d{2}:\d{2}:\d{2}"
|
|
output_file_list = list(output_path.iterdir())
|
|
|
|
assert len(output_file_list) == 3
|
|
for file in output_file_list:
|
|
content = file.read_text(encoding="windows-1252")
|
|
assert re.search(f"Période\t : {date_pattern} -> {date_pattern}\n", content)
|
|
assert f"Fichier(s) lu(s)\t : {nb_files}" in content
|
|
assert "Nombre total d'erreur(s)\t: 0" not in content
|
|
|
|
|
|
def test_cli_from_mails_no_output_dir(monkeypatch, make_mock_account):
|
|
"""Assert everything works when no output_path is provided.
|
|
|
|
Must use `RESULT_PATH` as the default output directory.
|
|
"""
|
|
messages = [
|
|
Message(subject="subject test 1", text_body="body test 1"),
|
|
Message(subject="subject test 2", text_body="body test 2"),
|
|
Message(subject="subject test 3", text_body="body test 3"),
|
|
]
|
|
write_log_report_mock = MagicMock()
|
|
|
|
with monkeypatch.context() as m:
|
|
m.setattr(
|
|
"logwatcher.cli.connect_to_mailbox",
|
|
lambda: make_mock_account(messages=messages, analyzed_folder="Analyzed"),
|
|
)
|
|
m.setattr("logwatcher.cli.write_log_report", write_log_report_mock)
|
|
m.setattr("logwatcher.cli.send_n2_report", MagicMock())
|
|
|
|
arguments = ["from-mails"]
|
|
results = runner.invoke(app, arguments)
|
|
|
|
assert results.exit_code == 0
|
|
assert write_log_report_mock.call_args.kwargs["output_dir"] == RESULT_PATH
|
|
|
|
|
|
def test_cli_from_mails_invalid_output_dir(tmp_path, monkeypatch):
|
|
"""Assert that if output_dir is a file, raise an error.
|
|
|
|
Args:
|
|
tmp_path: path where to store temporarily output files
|
|
monkeypatch: used to fake a context and method calls
|
|
make_mock_account: initialize a fake account object
|
|
|
|
"""
|
|
bad_output_dir = tmp_path.touch("test_file.txt")
|
|
|
|
messages = [
|
|
Message(subject="subject test 1", text_body="body test 1"),
|
|
]
|
|
|
|
with monkeypatch.context() as m:
|
|
m.setattr(
|
|
"logwatcher.cli.connect_to_mailbox",
|
|
lambda: make_mock_account(messages=messages, analyzed_folder="Analyzed"),
|
|
)
|
|
m.setattr("logwatcher.notifier.Message.send", MagicMock(return_value=None))
|
|
|
|
arguments = ["from-mails", "--output-dir", bad_output_dir]
|
|
results = runner.invoke(app, arguments)
|
|
assert results.exit_code != 0
|
|
|
|
|
|
def test_cli_from_mails_valid_output_dir(tmp_path, monkeypatch, make_mock_account):
|
|
"""Assert everything works when a valid output_path is provided.
|
|
|
|
Args:
|
|
tmp_path: path where to store temporarily output files
|
|
monkeypatch: used to fake a context and method calls
|
|
make_mock_account: initialize a fake account object
|
|
|
|
"""
|
|
output_path = tmp_path / "output_test"
|
|
|
|
messages = [
|
|
Message(subject="subject test 1", text_body="body test 1"),
|
|
Message(subject="subject test 2", text_body="body test 2"),
|
|
Message(subject="subject test 3", text_body="body test 3"),
|
|
]
|
|
|
|
with monkeypatch.context() as m:
|
|
m.setattr(
|
|
"logwatcher.cli.connect_to_mailbox",
|
|
lambda: make_mock_account(messages=messages, analyzed_folder="Analyzed"),
|
|
)
|
|
m.setattr("logwatcher.notifier.Message.send", MagicMock(return_value=None))
|
|
|
|
arguments = ["from-mails", "--output-dir", output_path]
|
|
results = runner.invoke(app, arguments)
|
|
|
|
assert results.exit_code == 0
|
|
output_children = list(output_path.iterdir())
|
|
expected = ["n2.log", "all.log", "other.log"]
|
|
file_names = [file.name for file in output_children]
|
|
assert len(output_children) == len(expected)
|
|
for name in expected:
|
|
assert name in file_names
|
|
|
|
|
|
def test_prune_help_lists_targets():
|
|
"""Assert the prune command lists both of its targets."""
|
|
results = runner.invoke(app, ["prune", "--help"])
|
|
|
|
assert results.exit_code == 0
|
|
assert "--mails" in results.output
|
|
assert "--logs" in results.output
|
|
|
|
|
|
def test_prune_without_target_raises():
|
|
"""Assert that prune without any target is rejected."""
|
|
results = runner.invoke(app, ["prune"])
|
|
|
|
assert results.exit_code == 2
|
|
assert "At least one of --mails or --logs must be provided." in results.output
|
|
|
|
|
|
def test_prune_logs_only(monkeypatch):
|
|
"""Assert `--logs` prunes logs only, without connecting to the mailbox."""
|
|
connect_mock = MagicMock()
|
|
prune_mails_mock = MagicMock()
|
|
get_logs_mock = MagicMock(return_value=[MagicMock()])
|
|
archive_logs_mock = MagicMock()
|
|
prune_logs_mock = MagicMock()
|
|
|
|
with monkeypatch.context() as m:
|
|
m.setattr("logwatcher.cli.connect_to_mailbox", connect_mock)
|
|
m.setattr("logwatcher.cli.prune_mails", prune_mails_mock)
|
|
m.setattr("logwatcher.cli.get_logs_to_archive", get_logs_mock)
|
|
m.setattr("logwatcher.cli.archive_logs", archive_logs_mock)
|
|
m.setattr("logwatcher.cli.prune_logs", prune_logs_mock)
|
|
|
|
results = runner.invoke(app, ["prune", "--logs"])
|
|
|
|
assert results.exit_code == 0
|
|
get_logs_mock.assert_called_once()
|
|
archive_logs_mock.assert_called_once()
|
|
prune_logs_mock.assert_called_once()
|
|
connect_mock.assert_not_called()
|
|
prune_mails_mock.assert_not_called()
|
|
|
|
|
|
def test_prune_mails_only(monkeypatch):
|
|
"""Assert `--mails` prunes mails only, without touching log archives."""
|
|
fake_account = MagicMock()
|
|
connect_mock = MagicMock(return_value=fake_account)
|
|
prune_mails_mock = MagicMock()
|
|
get_logs_mock = MagicMock()
|
|
archive_logs_mock = MagicMock()
|
|
prune_logs_mock = MagicMock()
|
|
|
|
with monkeypatch.context() as m:
|
|
m.setattr("logwatcher.cli.connect_to_mailbox", connect_mock)
|
|
m.setattr("logwatcher.cli.prune_mails", prune_mails_mock)
|
|
m.setattr("logwatcher.cli.get_logs_to_archive", get_logs_mock)
|
|
m.setattr("logwatcher.cli.archive_logs", archive_logs_mock)
|
|
m.setattr("logwatcher.cli.prune_logs", prune_logs_mock)
|
|
|
|
results = runner.invoke(app, ["prune", "--mails"])
|
|
|
|
assert results.exit_code == 0
|
|
connect_mock.assert_called_once()
|
|
prune_mails_mock.assert_called_once()
|
|
assert prune_mails_mock.call_args.args[1] is fake_account
|
|
get_logs_mock.assert_not_called()
|
|
archive_logs_mock.assert_not_called()
|
|
prune_logs_mock.assert_not_called()
|