♻️refactor(project): adapt to new naming of parsing methods

adapt to new naming of parsing methods, remove unused paths in parser module, all module import logwatcher.config instead of importing logwatcher, rename test functions
This commit is contained in:
2026-09-10 10:13:12 +02:00
parent cf83a06cfe
commit c5466a4171
10 changed files with 59 additions and 71 deletions
-2
View File
@@ -2,7 +2,5 @@ from importlib.metadata import version
from dotenv import load_dotenv from dotenv import load_dotenv
from logwatcher.config import *
__version__ = version("logwatcher") __version__ = version("logwatcher")
load_dotenv() load_dotenv()
+2 -2
View File
@@ -3,12 +3,12 @@ from enum import Enum
from pathlib import Path from pathlib import Path
### GENERAL DATA ### GENERAL DATA
# output path # output paths
OUTPUT_PATH = Path("output") OUTPUT_PATH = Path("output")
RESULT_PATH = OUTPUT_PATH / "results" RESULT_PATH = OUTPUT_PATH / "results"
LOGGING_PATH = OUTPUT_PATH / "logs" LOGGING_PATH = OUTPUT_PATH / "logs"
# test path # test paths
TEST_PATH = Path("tests") TEST_PATH = Path("tests")
FIXTURE_PATH = TEST_PATH / "fixtures" FIXTURE_PATH = TEST_PATH / "fixtures"
+3 -21
View File
@@ -1,10 +1,12 @@
import logging import logging
from pathlib import Path from pathlib import Path
from logwatcher.config import DATETIME_FORMAT
def _setup_formatter( def _setup_formatter(
format: str = "[%(asctime)s] - %(levelname)s: %(message)s", format: str = "[%(asctime)s] - %(levelname)s: %(message)s",
datefmt: str = "%d/%m/%Y %H:%M:%S", datefmt: str = DATETIME_FORMAT,
) -> logging.Formatter: ) -> logging.Formatter:
return logging.Formatter(fmt=format, datefmt=datefmt) return logging.Formatter(fmt=format, datefmt=datefmt)
@@ -22,26 +24,6 @@ def _setup_handler(
return handler return handler
def _convert_string_to_log_level(log_level: str) -> int:
"""
Get a log_level that contains one or three 'v' characters.
'v' = ERROR = 40
'vv' = WARNING = 30
'vvv' = DEBUG = 10
Args:
log_level:
Returns:
"""
if log_level == "v":
return logging.ERROR
elif log_level == "vv":
return logging.WARNING
else:
return logging.DEBUG
def setup_logging( def setup_logging(
formatter: logging.Formatter | None = None, formatter: logging.Formatter | None = None,
file_path: Path = Path("output/logs/logwatcher.log"), file_path: Path = Path("output/logs/logwatcher.log"),
+5 -2
View File
@@ -1,7 +1,7 @@
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from logwatcher import DATETIME_FORMAT from logwatcher.config import DATETIME_FORMAT
@dataclass() @dataclass()
@@ -9,6 +9,7 @@ class LogEntry:
""" """
Represents a line in log file. Represents a line in log file.
""" """
server_ip: str server_ip: str
mdc_server_name: str mdc_server_name: str
start_time: datetime start_time: datetime
@@ -36,4 +37,6 @@ class LogEntry:
""" """
# error time cannot be earlier than start time # error time cannot be earlier than start time
if self.start_time > self.error_time: if self.start_time > self.error_time:
raise ValueError(f"Error in date-times : error_time '{self.error_time}' cannot be earlier than start_time '{self.start_time}") raise ValueError(
f"Error in date-times : error_time '{self.error_time}' cannot be earlier than start_time '{self.start_time}"
)
+1 -7
View File
@@ -4,7 +4,7 @@ from collections.abc import Iterable
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from logwatcher import DATETIME_FORMAT, FRENCH_TIMEZONE from logwatcher.config import DATETIME_FORMAT, FRENCH_TIMEZONE
from logwatcher.models import LogEntry from logwatcher.models import LogEntry
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -29,12 +29,6 @@ LOG_PATTERN = re.compile(
rf"(?P<error_message>{ERROR_MESSAGE_PATTERN})" rf"(?P<error_message>{ERROR_MESSAGE_PATTERN})"
) )
# misc log file
misc_file_path = Path.cwd() / "output/misc.log"
# effective logs
effective_file_path = Path.cwd() / "output/effective.log"
def parse_line(log_line: str) -> LogEntry | None: def parse_line(log_line: str) -> LogEntry | None:
""" """
+15 -13
View File
@@ -6,7 +6,7 @@ import pytest
from logwatcher.classifier import N2_PATTERNS, classify_log_entries from logwatcher.classifier import N2_PATTERNS, classify_log_entries
from logwatcher.config import FIXTURE_PATH from logwatcher.config import FIXTURE_PATH
from logwatcher.parser import parse_log_file from logwatcher.parser import parse_file
def _pattern_name_to_filename(name: str) -> str: def _pattern_name_to_filename(name: str) -> str:
@@ -15,6 +15,7 @@ def _pattern_name_to_filename(name: str) -> str:
""" """
return f"CR_{name.lower()}.txt" return f"CR_{name.lower()}.txt"
def _read_fixture(path: Path) -> str: def _read_fixture(path: Path) -> str:
""" """
Read a file with fallback Windows-1252. Read a file with fallback Windows-1252.
@@ -27,6 +28,7 @@ def _read_fixture(path: Path) -> str:
return content return content
return path.read_text(encoding="windows-1252") # TODO: improve coverage score return path.read_text(encoding="windows-1252") # TODO: improve coverage score
@pytest.mark.parametrize("name, pattern", N2_PATTERNS.items()) @pytest.mark.parametrize("name, pattern", N2_PATTERNS.items())
def test_pattern_matches_fixture(valid_log_dir, name, pattern): def test_pattern_matches_fixture(valid_log_dir, name, pattern):
""" """
@@ -39,10 +41,7 @@ def test_pattern_matches_fixture(valid_log_dir, name, pattern):
warnings.warn(msg) warnings.warn(msg)
pytest.skip(msg) pytest.skip(msg)
lignes = [ lignes = [l for l in _read_fixture(fixture).splitlines() if l.strip()]
l for l in _read_fixture(fixture).splitlines()
if l.strip()
]
assert lignes, f"empty fixture file : {fixture}" assert lignes, f"empty fixture file : {fixture}"
@@ -54,8 +53,7 @@ def test_pattern_matches_fixture(valid_log_dir, name, pattern):
match_count += 1 match_count += 1
assert match_count > 0, ( assert match_count > 0, (
f"{name} : no match in file {fixture.name}\n" f"{name} : no match in file {fixture.name}\npattern: {pattern}"
f"pattern: {pattern}"
) )
@@ -65,10 +63,13 @@ def test_patterns_do_not_overlap():
""" """
fixtures_dir = FIXTURE_PATH fixtures_dir = FIXTURE_PATH
for fixture in fixtures_dir.glob("CR_*.txt"): for fixture in fixtures_dir.glob("CR_*.txt"):
line_list = [line for line in _read_fixture(fixture).splitlines() if line.strip()] # TODO: improve coverage score line_list = [
line for line in _read_fixture(fixture).splitlines() if line.strip()
] # TODO: improve coverage score
for line in line_list: # TODO: improve coverage score for line in line_list: # TODO: improve coverage score
matching = [ # TODO: improve coverage score matching = [ # TODO: improve coverage score
name for name, raw in N2_PATTERNS.items() name
for name, raw in N2_PATTERNS.items()
if re.compile(raw).search(line) if re.compile(raw).search(line)
] ]
assert len(matching) <= 1, ( # TODO: improve coverage score assert len(matching) <= 1, ( # TODO: improve coverage score
@@ -76,6 +77,7 @@ def test_patterns_do_not_overlap():
f"{matching} match line : {line[:80]}" f"{matching} match line : {line[:80]}"
) )
def test_no_orphan_fixtures(): def test_no_orphan_fixtures():
""" """
Assert that a fixture exist only if its associated pattern exists. Assert that a fixture exist only if its associated pattern exists.
@@ -92,7 +94,7 @@ def test_classify_log_entries_all_relevant(valid_log_dir: Path):
Must return an empty irrelevant log entry list Must return an empty irrelevant log entry list
""" """
log_file = valid_log_dir / "only_relevant_logs.txt" log_file = valid_log_dir / "only_relevant_logs.txt"
log_entries = parse_log_file(log_file) log_entries = parse_file(log_file)
# all entries are relevant # all entries are relevant
relevant, irrelevant = classify_log_entries(log_entries) relevant, irrelevant = classify_log_entries(log_entries)
@@ -105,7 +107,7 @@ def test_classify_log_entries_none_relevant(valid_log_dir: Path):
Must return an empty relevant log entry list Must return an empty relevant log entry list
""" """
log_file = valid_log_dir / "no_relevant_logs.txt" log_file = valid_log_dir / "no_relevant_logs.txt"
log_entries = parse_log_file(log_file) log_entries = parse_file(log_file)
# all entries are relevant # all entries are relevant
relevant, irrelevant = classify_log_entries(log_entries) relevant, irrelevant = classify_log_entries(log_entries)
@@ -118,7 +120,7 @@ def test_classify_log_entries_mixed(valid_log_dir: Path):
Test classification on relevant and irrelevant log entry list Test classification on relevant and irrelevant log entry list
""" """
log_file = valid_log_dir / "mixed_logs.txt" log_file = valid_log_dir / "mixed_logs.txt"
log_entries = parse_log_file(log_file) log_entries = parse_file(log_file)
relevant, irrelevant = classify_log_entries(log_entries) relevant, irrelevant = classify_log_entries(log_entries)
@@ -131,7 +133,7 @@ def test_classify_log_entries_empty_log_entries(invalid_log_dir: Path):
Test classification on an empty log entry list Test classification on an empty log entry list
""" """
empty_file = invalid_log_dir / "empty_file.txt" empty_file = invalid_log_dir / "empty_file.txt"
log_entries = parse_log_file(empty_file) log_entries = parse_file(empty_file)
relevant, irrelevant = classify_log_entries(log_entries) relevant, irrelevant = classify_log_entries(log_entries)
+2 -1
View File
@@ -15,6 +15,7 @@ from logging import DEBUG, ERROR, INFO, WARNING, FileHandler, StreamHandler, get
import pytest import pytest
from logwatcher.config import DATETIME_FORMAT
from logwatcher.logging_config import _setup_formatter, _setup_handler, setup_logging from logwatcher.logging_config import _setup_formatter, _setup_handler, setup_logging
@@ -54,7 +55,7 @@ def test_type_of_handlers():
else: else:
pytest.fail("handlers must be either a file or a stream handler") pytest.fail("handlers must be either a file or a stream handler")
assert handler.formatter._fmt == "[%(asctime)s] - %(levelname)s: %(message)s" assert handler.formatter._fmt == "[%(asctime)s] - %(levelname)s: %(message)s"
assert handler.formatter.datefmt == "%d/%m/%Y %H:%M:%S" assert handler.formatter.datefmt == DATETIME_FORMAT
def test_modified_format(): def test_modified_format():
+1 -1
View File
@@ -2,7 +2,7 @@ from datetime import datetime
import pytest import pytest
from logwatcher import DATETIME_FORMAT, FRENCH_TIMEZONE from logwatcher.config import DATETIME_FORMAT, FRENCH_TIMEZONE
from logwatcher.models import LogEntry from logwatcher.models import LogEntry
+12 -4
View File
@@ -57,7 +57,7 @@ def test_parse_line(specific_logs: list[str]):
assert log_entry.raw_line != "" assert log_entry.raw_line != ""
def test_parse_empty_log_line(): def test_parse_empty_line():
""" """
An empty or structurally invalid line returns None. An empty or structurally invalid line returns None.
Covered cases: Covered cases:
@@ -71,7 +71,15 @@ def test_parse_empty_log_line():
assert not log_entry assert not log_entry
def test_parse_valid_log_file(original_log_dir: Path): def test_parse_lines():
pass
def test_parse_empty_lines():
pass
def test_parse_valid_file(original_log_dir: Path):
""" """
Each real log file produces at least one valid LogEntry. Each real log file produces at least one valid LogEntry.
Integration test: files provided by technicians contain Integration test: files provided by technicians contain
@@ -86,7 +94,7 @@ def test_parse_valid_log_file(original_log_dir: Path):
assert isinstance(log_entry, LogEntry) assert isinstance(log_entry, LogEntry)
def test_parse_empty_log_file(empty_log_file): def test_parse_empty_file(empty_log_file):
""" """
An empty log file can be parsed and must returns an An empty log file can be parsed and must returns an
empty LogEntry list. empty LogEntry list.
@@ -95,7 +103,7 @@ def test_parse_empty_log_file(empty_log_file):
assert len(log_list) == 0 assert len(log_list) == 0
def test_parse_bad_log_file(bad_log_file): def test_parse_bad_file(bad_log_file):
""" """
A file containing only invalid lines produces nothing. A file containing only invalid lines produces nothing.
Any line that does not match the LAME MDC structure Any line that does not match the LAME MDC structure
+3 -3
View File
@@ -6,7 +6,7 @@ import pytest
from logwatcher.classifier import classify_log_entries from logwatcher.classifier import classify_log_entries
from logwatcher.config import DATETIME_FORMAT, FIXTURE_PATH, SourceType from logwatcher.config import DATETIME_FORMAT, FIXTURE_PATH, SourceType
from logwatcher.models import LogEntry from logwatcher.models import LogEntry
from logwatcher.parser import parse_log_file from logwatcher.parser import parse_file
from logwatcher.reporter import ( from logwatcher.reporter import (
N2_SUPPORT_TEMPLATE, N2_SUPPORT_TEMPLATE,
OTHER_TEMPLATE, OTHER_TEMPLATE,
@@ -25,7 +25,7 @@ def _get_classified_log_entries(
log_file: Path, log_file: Path,
) -> tuple[list[LogEntry], list[LogEntry]]: ) -> tuple[list[LogEntry], list[LogEntry]]:
"""Parse a log file and split its entries into relevant and irrelevant lists.""" """Parse a log file and split its entries into relevant and irrelevant lists."""
log_entries = parse_log_file(log_file) log_entries = parse_file(log_file)
return classify_log_entries(log_entries) return classify_log_entries(log_entries)
@@ -40,7 +40,7 @@ def get_all_log_entries_fixture() -> tuple[list[LogEntry], list[LogEntry]]:
"""Parse and classify every CR_* fixture, returning (relevant, irrelevant).""" """Parse and classify every CR_* fixture, returning (relevant, irrelevant)."""
all_entries = [] all_entries = []
for log_file in CR_LOG_FILES: for log_file in CR_LOG_FILES:
all_entries.extend(parse_log_file(log_file)) all_entries.extend(parse_file(log_file))
relevant_list, irrelevant_list = classify_log_entries(all_entries) relevant_list, irrelevant_list = classify_log_entries(all_entries)
return relevant_list, irrelevant_list return relevant_list, irrelevant_list