maurane/logwatcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(cleaner): ✨ add command prune

Browse Source 
this new command will be called by cron-like process to clean purge old logs and old mails. All files were formatted and mail_utils was renamed in utils since it now defines utility function not only for mail functions.
This commit is contained in:
maurane
2026-09-25 16:30:45 +02:00
parent 63c4d1387e
commit 02e85ddbfd
20 changed files with 471 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Entry point to all package tests."""
tests/conftest.py
+8 -3
View File
@@ -6,7 +6,7 @@ from unittest.mock import MagicMock
import pytest
from exchangelib import Account, Message
from logwatcher.config import FIXTURE_PATH
from logwatcher.config import FIXTURE_PATH, OUTPUT_PATH
from logwatcher.logging_config import setup_logging
# original log directory
@@ -19,18 +19,21 @@ VALID_LOG_DIR = FIXTURE_PATH / "valid_logs"
INVALID_LOG_DIR = FIXTURE_PATH / "invalid_logs"
# define temp directory only for testing
os.environ["PYTEST_DEBUG_TEMPROOT"] = str(Path("output/test").absolute())
os.environ["PYTEST_DEBUG_TEMPROOT"] = str(OUTPUT_PATH)
@pytest.fixture
def tmp_log_file(tmp_path):
"""Return path of logs logwatcher related."""
return tmp_path / "logwatcher_test.log"
@pytest.fixture(autouse=True)
def tmp_setup_logging(tmp_log_file):
"""Redirige les logs vers un dossier temporaire pendant les tests."""
setup_logging(file_path=tmp_log_file)
@pytest.fixture(name="original_log_dir")
def original_log_dir_fixture() -> Path:
"""Return log directory containing real logs.
@@ -41,6 +44,7 @@ def original_log_dir_fixture() -> Path:
"""
return FIXTURE_PATH / "original_logs"
@pytest.fixture(name="valid_log_dir")
def valid_log_dir_fixture() -> Path:
"""Return log directory containing only valid logs.
@@ -51,6 +55,7 @@ def valid_log_dir_fixture() -> Path:
"""
return FIXTURE_PATH / "valid_logs"
@pytest.fixture(name="invalid_log_dir")
def invalid_log_dir_fixture() -> Path:
"""Return log directory containing invalid logs.
@@ -78,4 +83,4 @@ def make_mock_account():
logs_folder.glob.return_value = [analyzed_folder] if analyzed_folder else []
return account
return _make_mock_account
return _make_mock_account
tests/test_cli.py
+28 -44
View File
@@ -8,7 +8,7 @@ from unittest.mock import MagicMock
from logwatcher import __version__
from logwatcher.cli import _get_files_from_dir, _is_log_file, app
from logwatcher.cli import _get_files_from_dir, app
runner = CliRunner()
@@ -62,33 +62,9 @@ def test_version_cli():
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
"""
# 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:
@@ -154,7 +130,7 @@ def test_cli_from_files_input_files_and_input_dir(output_path: Path):
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)
"""
@@ -163,7 +139,9 @@ def test_cli_from_files_neither_input_files_nor_input_dir():
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):
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
@@ -277,7 +255,7 @@ def test_cli_from_files_input_files_or_input_dir(
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.
@@ -405,22 +383,22 @@ def test_cli_from_files_valid_logs(output_path: Path, valid_log_dir: Path):
def test_cli_from_mails_no_output_dir(monkeypatch, make_mock_account):
"""Assert everything works when no output_path is provided.
Must use "output/" directory, located in project root.
"""
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")
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.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"]
results = runner.invoke(app, arguments)
assert results.exit_code == 0
@@ -428,7 +406,7 @@ def test_cli_from_mails_no_output_dir(monkeypatch, make_mock_account):
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
@@ -442,9 +420,12 @@ def test_cli_from_mails_invalid_output_dir(tmp_path, monkeypatch):
]
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.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
@@ -452,7 +433,7 @@ def test_cli_from_mails_invalid_output_dir(tmp_path, monkeypatch):
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
@@ -464,16 +445,19 @@ def test_cli_from_mails_valid_output_dir(tmp_path, monkeypatch, make_mock_accoun
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")
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.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"]
tests/test_logging.py
+2 -2
View File
@@ -20,7 +20,7 @@ def test_setup_handler(tmp_log_file):
def test_count_of_handler(tmp_log_file):
"""Assert `setup_handler` instanciate exactly one StreamHandler and one FileHandler.
Must have exactly two handlers in total.
"""
logger = getLogger()
@@ -82,7 +82,7 @@ def test_modified_format():
def test_logging_in_correct_path(tmp_log_file):
"""Assert log file is correctly written in output directory.
Check content of the file. It must contain logs from CRITICAL to DEBUG level.
"""
logger = logging.getLogger("logwatcher")
tests/test_models.py
+30 -12
View File
@@ -11,46 +11,64 @@ def test_valid_log_entry():
LogEntry(
server_ip="192.168.13.27",
mdc_server_name="MDC_720",
start_time=datetime.strptime("18/07/2026 11:34:41", DATETIME_FORMAT).astimezone(FRENCH_TIMEZONE),
start_time=datetime.strptime("18/07/2026 11:34:41", DATETIME_FORMAT).astimezone(
FRENCH_TIMEZONE
),
store_name="PECCOUD",
error_time=datetime.strptime("18/07/2026 11:35:58", DATETIME_FORMAT).astimezone(FRENCH_TIMEZONE),
error_time=datetime.strptime("18/07/2026 11:35:58", DATETIME_FORMAT).astimezone(
FRENCH_TIMEZONE
),
error_message="Erreur FTP SDUpdatePrixISF : Requested action not taken",
raw_line="\\\\192.168.13.27\\e\\MDC_720\\Logs\\26\\07\\18\\20260718113428.txt [18/07/2026 11:34:41] DOSSIER EN COURS : PECCOUD [18/07/2026 11:35:58] Erreur FTP SDUpdatePrixISF : Requested action not taken"
raw_line="\\\\192.168.13.27\\e\\MDC_720\\Logs\\26\\07\\18\\20260718113428.txt [18/07/2026 11:34:41] DOSSIER EN COURS : PECCOUD [18/07/2026 11:35:58] Erreur FTP SDUpdatePrixISF : Requested action not taken",
)
def test_log_entry_invalid_dates():
"""Assert a log entry cannot be created if start time is not later than error time."""
with pytest.raises(ValueError, match="Error in date-times"):
LogEntry(
server_ip="192.168.13.27",
mdc_server_name="MDC_720",
start_time=datetime.strptime("18/07/2026 11:35:58", DATETIME_FORMAT).astimezone(FRENCH_TIMEZONE),
start_time=datetime.strptime(
"18/07/2026 11:35:58", DATETIME_FORMAT
).astimezone(FRENCH_TIMEZONE),
store_name="PECCOUD",
error_time=datetime.strptime("18/07/2026 11:34:41", DATETIME_FORMAT).astimezone(FRENCH_TIMEZONE),
error_time=datetime.strptime(
"18/07/2026 11:34:41", DATETIME_FORMAT
).astimezone(FRENCH_TIMEZONE),
error_message="Erreur FTP SDUpdatePrixISF : Requested action not taken",
raw_line="\\\\192.168.13.27\\e\\MDC_720\\Logs\\26\\07\\18\\20260718113428.txt [18/07/2026 11:34:41] DOSSIER EN COURS : PECCOUD [18/07/2026 11:35:58] Erreur FTP SDUpdatePrixISF : Requested action not taken"
raw_line="\\\\192.168.13.27\\e\\MDC_720\\Logs\\26\\07\\18\\20260718113428.txt [18/07/2026 11:34:41] DOSSIER EN COURS : PECCOUD [18/07/2026 11:35:58] Erreur FTP SDUpdatePrixISF : Requested action not taken",
)
def test_equals_models():
"""Assert two models are equals if they have the same value as attributes."""
log_entry_1 = LogEntry(
server_ip="192.168.13.27",
mdc_server_name="MDC_720",
start_time=datetime.strptime("18/07/2026 11:34:41", DATETIME_FORMAT).astimezone(FRENCH_TIMEZONE),
start_time=datetime.strptime("18/07/2026 11:34:41", DATETIME_FORMAT).astimezone(
FRENCH_TIMEZONE
),
store_name="PECCOUD",
error_time=datetime.strptime("18/07/2026 11:35:58", DATETIME_FORMAT).astimezone(FRENCH_TIMEZONE),
error_time=datetime.strptime("18/07/2026 11:35:58", DATETIME_FORMAT).astimezone(
FRENCH_TIMEZONE
),
error_message="Erreur FTP SDUpdatePrixISF : Requested action not taken",
raw_line="\\\\192.168.13.27\\e\\MDC_720\\Logs\\26\\07\\18\\20260718113428.txt [18/07/2026 11:34:41] DOSSIER EN COURS : PECCOUD [18/07/2026 11:35:58] Erreur FTP SDUpdatePrixISF : Requested action not taken"
raw_line="\\\\192.168.13.27\\e\\MDC_720\\Logs\\26\\07\\18\\20260718113428.txt [18/07/2026 11:34:41] DOSSIER EN COURS : PECCOUD [18/07/2026 11:35:58] Erreur FTP SDUpdatePrixISF : Requested action not taken",
)
log_entry_2 = LogEntry(
server_ip="192.168.13.27",
mdc_server_name="MDC_720",
start_time=datetime.strptime("18/07/2026 11:34:41", DATETIME_FORMAT).astimezone(FRENCH_TIMEZONE),
start_time=datetime.strptime("18/07/2026 11:34:41", DATETIME_FORMAT).astimezone(
FRENCH_TIMEZONE
),
store_name="PECCOUD",
error_time=datetime.strptime("18/07/2026 11:35:58", DATETIME_FORMAT).astimezone(FRENCH_TIMEZONE),
error_time=datetime.strptime("18/07/2026 11:35:58", DATETIME_FORMAT).astimezone(
FRENCH_TIMEZONE
),
error_message="Erreur FTP SDUpdatePrixISF : Requested action not taken",
raw_line="\\\\192.168.13.27\\e\\MDC_720\\Logs\\26\\07\\18\\20260718113428.txt [18/07/2026 11:34:41] DOSSIER EN COURS : PECCOUD [18/07/2026 11:35:58] Erreur FTP SDUpdatePrixISF : Requested action not taken"
raw_line="\\\\192.168.13.27\\e\\MDC_720\\Logs\\26\\07\\18\\20260718113428.txt [18/07/2026 11:34:41] DOSSIER EN COURS : PECCOUD [18/07/2026 11:35:58] Erreur FTP SDUpdatePrixISF : Requested action not taken",
)
assert log_entry_1 == log_entry_2
tests/test_notifier.py
+26 -8
View File
@@ -11,23 +11,37 @@ def mock_get_or_create_folder(account, folder_name): # noqa: D103
return folder_name
@pytest.mark.parametrize("vars", [("N2_REPORT_SUBJECT", "N2_REPORT_RECIPIENTS"), ("N2_REPORT_RECIPIENTS", "N2_REPORT_SUBJECT")])
def test_send_n2_report_missing_recipients_env(vars, monkeypatch, ):
@pytest.mark.parametrize(
"vars",
[
("N2_REPORT_SUBJECT", "N2_REPORT_RECIPIENTS"),
("N2_REPORT_RECIPIENTS", "N2_REPORT_SUBJECT"),
],
)
def test_send_n2_report_missing_recipients_env(vars, monkeypatch):
"""Assert that missing values in environmnent file raises KeyError."""
monkeypatch.delenv("N2_REPORT_RECIPIENTS")
monkeypatch.delenv("N2_REPORT_SUBJECT")
monkeypatch.setenv(vars[0], "test")
with pytest.raises(KeyError) as err:
send_n2_report(account=MagicMock(), summary="summary", n2_log_file=MagicMock())
assert vars[1] in err.value.args
@pytest.mark.parametrize("recipients", ["mail@client.fr", "mail@ client.fr ", "mail1@client.fr,mail2@client.fr,mail3@client.fr", "mail1@client.fr , mail2@client.fr , mail3@client.fr"])
@pytest.mark.parametrize(
"recipients",
[
"mail@client.fr",
"mail@ client.fr ",
"mail1@client.fr,mail2@client.fr,mail3@client.fr",
"mail1@client.fr , mail2@client.fr , mail3@client.fr",
],
)
def test_send_n2_report_multiple_recipients(recipients, make_mock_account, monkeypatch):
"""Assert gathering multiple account as recipients from env file is valid.
Tests with value having space characters.
Mails must be seperated by a comma.
@@ -35,12 +49,16 @@ def test_send_n2_report_multiple_recipients(recipients, make_mock_account, monke
recipients: e-mail recipients in one string
make_mock_account: return a fake account used for tests
monkeypatch: used to change environment variables and functions accross the script
"""
monkeypatch.setenv("N2_REPORT_SUBJECT", "test")
monkeypatch.setenv("N2_REPORT_RECIPIENTS", "mail@client.fr")
with monkeypatch.context() as m:
m.setattr("logwatcher.notifier.get_or_create_folder", mock_get_or_create_folder)
m.setattr("logwatcher.notifier.Message.send", MagicMock(id="1", return_value=None))
send_n2_report(account=make_mock_account(), summary="summary", n2_log_file=MagicMock())
m.setattr(
"logwatcher.notifier.Message.send", MagicMock(id="1", return_value=None)
)
send_n2_report(
account=make_mock_account(), summary="summary", n2_log_file=MagicMock()
)
tests/test_parser.py
+2 -2
View File
@@ -19,7 +19,7 @@ def empty_log_file_fixture(invalid_log_dir: Path) -> Path:
@pytest.fixture(name="bad_log_file")
def bad_log_file_fixture(invalid_log_dir: Path) -> Path:
"""Path of a file containing only incorrectly formatted lines.
Each line fails at a different point in LOG_PATTERN
(invalid IP, missing timestamp, missing DOSSIER EN COURS, etc.).
No LogEntry should be produced.
@@ -93,7 +93,7 @@ def test_parse_lines():
def test_parse_valid_lines(specific_logs: Path):
"""Assert parse_lines works for all valid lines."""
"""Assert parse_lines works for all valid lines."""
# selected valid logs
with open(specific_logs, "r") as file:
log_lines = file.readlines()
tests/test_mail_utils.py → tests/test_utils.py
+30 -5
View File
@@ -1,6 +1,31 @@
from pathlib import Path
from unittest.mock import MagicMock
from logwatcher.mail_utils import get_or_create_folder
from logwatcher.utils import get_or_create_folder, is_log_file
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
"""
# 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_or_create_analyzed_folder_exists(make_mock_account, monkeypatch):
@@ -9,14 +34,14 @@ def test_get_or_create_analyzed_folder_exists(make_mock_account, monkeypatch):
Args:
make_mock_account: Fake account owning the Logs folder
monkeypatch: MonkeyPatch to generate test context
"""
existing_folder = MagicMock()
account = make_mock_account(analyzed_folder=existing_folder)
fake_folder_cls = MagicMock()
with monkeypatch.context() as m:
m.setattr("logwatcher.mail_utils.Folder", fake_folder_cls)
m.setattr("logwatcher.utils.Folder", fake_folder_cls)
result = get_or_create_folder(account, "Analyzed")
assert result is existing_folder
@@ -30,8 +55,8 @@ def test_get_or_create_analyzed_folder_creates(make_mock_account, monkeypatch):
fake_folder_instance = MagicMock()
fake_folder_cls = MagicMock(return_value=fake_folder_instance)
with monkeypatch.context() as m:
m.setattr("logwatcher.mail_utils.Folder", fake_folder_cls)
m.setattr("logwatcher.utils.Folder", fake_folder_cls)
result = get_or_create_folder(account, "Analyzed")
assert result is fake_folder_instance
fake_folder_instance.save.assert_called_once()
fake_folder_instance.save.assert_called_once()