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-mails" 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
|