maurane
150eb5145d
add `from-mails` command that connect to the mailbox using configurations in .env file, parse, classify and report all not analyzed files.
433 lines
13 KiB
Python
433 lines
13 KiB
Python
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from logwatcher import __version__
|
|
from logwatcher.cli import _get_files_from_dir, _is_log_file, app
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
@pytest.fixture(name="output_path")
|
|
def get_output_path_fixture(tmp_path: Path):
|
|
"""
|
|
Fixture that return output directory path where all logs and output files
|
|
will be located during tests.
|
|
|
|
Args:
|
|
tmp_path: Temporary root path for testing results
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
return tmp_path / "output"
|
|
|
|
|
|
def get_argument_from_files(log_dir: Path, output_path: Path) -> list[str]:
|
|
"""
|
|
Utility function that implement 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 file.suffix in ("", ".txt", ".log")
|
|
]
|
|
|
|
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
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
results = runner.invoke(app, ["--help"])
|
|
assert results.exit_code == 0
|
|
assert "--verbose" in results.output
|
|
assert "--version" in results.output
|
|
assert "from-mail" in results.output
|
|
assert "from-files" in results.output
|
|
|
|
|
|
def test_version_cli():
|
|
"""
|
|
Test the version option. The version must be printed
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
result = runner.invoke(app, ["--version"])
|
|
assert result.exit_code == 0
|
|
assert f"logwatcher version: {__version__}" in result.output
|
|
|
|
|
|
def test_is_log_file(tmp_path: Path):
|
|
"""
|
|
Assert only .log, .txt or file without extensions are valid
|
|
|
|
Args:
|
|
tmp_path: Temporary root path for testing results
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
# set up files
|
|
Path(tmp_path / "a.log").touch()
|
|
Path(tmp_path / "a.txt").touch()
|
|
Path(tmp_path / "a.csv").touch()
|
|
Path(tmp_path / "a.py").touch()
|
|
Path(tmp_path / "a").touch()
|
|
Path(tmp_path / "a.pdf").touch()
|
|
|
|
# test
|
|
assert _is_log_file(Path(tmp_path / "a.log"))
|
|
assert _is_log_file(Path(tmp_path / "a.txt"))
|
|
assert _is_log_file(Path(tmp_path / "a")) # without extensions
|
|
assert not _is_log_file(Path(tmp_path / "a.csv"))
|
|
assert not _is_log_file(Path(tmp_path / "a.py"))
|
|
assert not _is_log_file(Path(tmp_path / "a.pdf"))
|
|
|
|
|
|
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
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
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
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
# 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_input_files_and_input_dir(output_path: Path):
|
|
"""
|
|
Assert that input files and input dir options cannot be provided in the same command.
|
|
|
|
Args:
|
|
tmp_path: Temporary root path for testing results
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
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_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)
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
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_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 directory of invalid logs
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
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_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 directory of invalid logs
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
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_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 directory of initial logs received by n2 technicians
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
# 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_input_files_one_file(valid_log_dir: Path, output_path: Path):
|
|
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_empty_input_directory(tmp_path: Path, output_path: Path):
|
|
"""
|
|
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
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
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_invalid_logs(output_path: Path, invalid_log_dir: Path):
|
|
"""
|
|
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 directory of invalid logs
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
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_valid_logs(output_path: Path, valid_log_dir: Path):
|
|
"""
|
|
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 directory of valid logs
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
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
|