diff --git a/.env.example b/.env.example index 8e40fdc..f7516a5 100644 --- a/.env.example +++ b/.env.example @@ -7,4 +7,16 @@ EWS_URL=https://mail.client.fr/EWS/Exchange.asmx # Default mail server ## Report email N2_REPORT_RECIPIENTS=n2-support@client.fr -N2_REPORT_SUBJECT=[Logwatcher] Rapport erreurs N2 \ No newline at end of file +N2_REPORT_SUBJECT=[Logwatcher] Rapport erreurs N2 + +## Log retention +### number of days before archiving logs +LOG_RETENTION_DAYS=30 + +### number of days before deleting archive logs +ARCHIVED_LOG_RETENTION_DAYS=60 + + +## Mail retention +### number of days before deleting mails +MAIL_RETENTION_DAYS=30 \ No newline at end of file diff --git a/src/logwatcher/classifier.py b/src/logwatcher/classifier.py index 244c4b0..5256152 100644 --- a/src/logwatcher/classifier.py +++ b/src/logwatcher/classifier.py @@ -82,7 +82,7 @@ def classify_log_entries( irrelevant_log_entries = [] previous_log_entry: LogEntry | None = None - logger.info("\tclassification job started.") + logger.info("classification job started.") for log_entry in log_entries: error_name = _match_n2_pattern(log_entry) @@ -107,5 +107,5 @@ def classify_log_entries( logger.info( f"\t\tn2 logs: {len(relevant_log_entries)}, other logs: {len(irrelevant_log_entries)}" ) - logger.info("\tclassification job completed.") + logger.info("classification job completed.") return relevant_log_entries, irrelevant_log_entries diff --git a/src/logwatcher/cleaner.py b/src/logwatcher/cleaner.py new file mode 100644 index 0000000..7512f53 --- /dev/null +++ b/src/logwatcher/cleaner.py @@ -0,0 +1,153 @@ +import datetime +import logging +import os + +from pathlib import Path +from zipfile import ZipFile + +from exchangelib import Account + +from logwatcher.config import ( + ARCHIVE_DIR, + DATE_FILE_FORMAT, + FRENCH_TIMEZONE, + LOGGING_PATH, +) +from logwatcher.utils import ANALYZED_FOLDER, LOG_FOLDER, SENT_FOLDER, is_log_file + + +logger = logging.getLogger(__name__) + + +def _get_retention_days(key: str) -> int: + """Get specific environment variable defined by a number of days. + + Args: + key: name of the environment variable + + Raises: + KeyError + ValueError + + Returns: + number of days found in environment variable `key`. + + """ + try: + value = int(os.environ[key]) + except KeyError: + logger.error(f"'{key}' is not defined in environment variables.") + raise + except ValueError: + logger.error(f"{os.environ[key]} must be an integer.") + raise + return value + + +def get_logs_to_archive(today: datetime.datetime) -> list[Path]: + """Get all logs that are older than `LOG_RETENTION_DAYS`. + + Args: + today: date and time of the current day + + Returns: + A list of log file to archive. + + """ + nb_days = _get_retention_days("LOG_RETENTION_DAYS") + logs_to_archive = [] + cutoff_date = today - datetime.timedelta(days=nb_days) + + for log_file in LOGGING_PATH.iterdir(): + if is_log_file(log_file): + log_file_date = datetime.datetime.fromtimestamp( + log_file.stat().st_mtime, tz=FRENCH_TIMEZONE + ) + if log_file_date < cutoff_date: + logs_to_archive.append(log_file) + + if not logs_to_archive: + logger.warning("No log files old enough to be archived.") + + return logs_to_archive + + +def archive_logs(today: datetime.datetime, log_files: list[Path]) -> None: + """Archive old logs. + + Due to its file name, the newly created archive could overwrite the old one + if the function is called at the same seconds. + Do nothing if `log_files` is empty. + + Args: + today: date and time of the current day + log_files: log file to archive. + + """ + if not log_files: + logger.info("No file to archive. Skip.") + return + + ARCHIVE_DIR.mkdir(parents=True, exist_ok=True) + + archive_file_name = f"archive_{today.strftime(DATE_FILE_FORMAT)}.zip" + + with ZipFile(ARCHIVE_DIR / archive_file_name, "w") as archive_file: + for log_file in log_files: + logger.debug(f"Add '{log_file}' to archive.") + archive_file.write(log_file, arcname=log_file.name) + logger.debug(f"Delete '{log_file}'") + log_file.unlink() + + +def prune_logs(today: datetime.datetime) -> None: + """Detect and delete old log archives. + + Args: + today: date and time of the current day + + """ + nb_days = _get_retention_days("ARCHIVED_LOG_RETENTION_DAYS") + + cutoff_date = today - datetime.timedelta(days=nb_days) + + ARCHIVE_DIR.mkdir(parents=True, exist_ok=True) + archive_files = list(ARCHIVE_DIR.iterdir()) + + if not archive_files: + logger.warning("No archive found.") + return + + for archive_file in archive_files: + archive_date = datetime.datetime.fromtimestamp( + archive_file.stat().st_mtime, tz=FRENCH_TIMEZONE + ) + if archive_date < cutoff_date: + archive_file.unlink() + + +def prune_mails(today: datetime.datetime, account: Account) -> None: + """Get and delete all old mails from `ANALYZED_FOLDER` and `SENT_FOLDER`. + + Args: + today: date and time of the current day + account: EWS account that owns and manage log mails + + """ + analyzed_folder = account.msg_folder_root / LOG_FOLDER / ANALYZED_FOLDER + sent_folder = account.msg_folder_root / LOG_FOLDER / SENT_FOLDER + + nb_days = _get_retention_days("MAIL_RETENTION_DAYS") + cutoff_date = today - datetime.timedelta(days=nb_days) + + analyzed_mails = analyzed_folder.filter(datetime_sent__lt=cutoff_date) + sent_mails = sent_folder.filter(datetime_sent__lt=cutoff_date) + + mails = list(analyzed_mails) + list(sent_mails) + + if not mails: + logger.warning(f"No mail older than '{nb_days}' days found.") + return + + ids = [(m.id, m.changekey) for m in mails] + account.bulk_delete(ids=ids) diff --git a/src/logwatcher/cli.py b/src/logwatcher/cli.py index 5304774..d765e18 100644 --- a/src/logwatcher/cli.py +++ b/src/logwatcher/cli.py @@ -7,7 +7,20 @@ import typer from logwatcher import __version__ from logwatcher.classifier import classify_log_entries -from logwatcher.config import DATETIME_FORMAT, FRENCH_TIMEZONE, SourceType +from logwatcher.cleaner import ( + get_logs_to_archive, + archive_logs, + prune_logs, + prune_mails, +) +from logwatcher.config import ( + DATE_FILE_FORMAT, + DATETIME_FORMAT, + FRENCH_TIMEZONE, + LOGGING_PATH, + RESULT_PATH, + SourceType, +) from logwatcher.logging_config import setup_logging from logwatcher.mail_reader import ( connect_to_mailbox, @@ -19,6 +32,7 @@ from logwatcher.models import LogEntry from logwatcher.notifier import send_n2_report from logwatcher.parser import parse_file, parse_lines from logwatcher.reporter import build_mail_summary, get_period, write_log_report +from logwatcher.utils import is_log_file app = typer.Typer(help="Logwatcher - Analysis and sorting of LAME MDC logs.") @@ -26,41 +40,46 @@ app = typer.Typer(help="Logwatcher - Analysis and sorting of LAME MDC logs.") logger = logging.getLogger("logwatcher") -def _is_log_file(path: Path) -> bool: - """Return True if file is a log file (.txt, .log, or extensionless). +def _get_files_from_dir(input_dir: Path) -> list[Path]: + """Retrieve all log files from the `input_dir` directory. Args: - path: Path of a file + input_dir: directory that contains log files Returns: - True if file has .txt or .log extension or extensionless - False otherwise. + list of log files """ - if not path.is_file(): - return False - return path.suffix in {"", ".txt", ".log"} - - -def _get_files_from_dir(input_dir: Path) -> list[Path]: if not input_dir.exists(): logger.error(f"Unrecognized input directory '{input_dir}'.") raise FileNotFoundError(f"No such directory at '{input_dir}'") - return [file for file in input_dir.iterdir() if _is_log_file(file)] + return [ + current_file + for current_file in input_dir.iterdir() + if is_log_file(current_file) + ] def _files_to_log_entries(input_files: list[Path]) -> list[LogEntry]: - """Parse each input file, raising if any path is not a file.""" + """Parse each input file, raising if any path is not a file. + + Args: + input_files: list of log files + + Returns: + list of log entries + + """ log_entries = [] - for file in input_files: - if not file.is_file(): - logger.error(f"Unrecognized input file '{file}'.") - raise ValueError(f"'{file}' is not a file") - log_entries.extend(parse_file(file)) + for current_file in input_files: + if not current_file.is_file(): + logger.error(f"Unrecognized input file '{current_file}'.") + raise ValueError(f"'{current_file}' is not a file") + log_entries.extend(parse_file(current_file)) return log_entries -def version_callback(value: bool): +def version_callback(value: bool) -> None: """Print the version of package. Stop the CLI after printing version. @@ -81,16 +100,16 @@ def from_mails( typer.Option( "--output-dir", help="Output directories where reports will be stored." ), - ] = Path("output"), -): + ] = RESULT_PATH, +) -> None: """Convert mail containing logs into reports for N2 technicians. Connect to the EWS server with service account, filter most recent mail logs, generate reports based on these mails and send mails to the technicians. Args: - output_dir: Path to the generated reports folder - + output_dir: Output directories where reports will be stored. + """ logger.info("============ logwatcher mail started ============") account = connect_to_mailbox() @@ -100,7 +119,7 @@ def from_mails( nb_mails = len(messages) relevant, irrelevant = classify_log_entries(log_entries) - today = datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT) + today = datetime.now(tz=FRENCH_TIMEZONE).strftime(DATETIME_FORMAT) start_date, end_date = get_period(relevant + irrelevant) write_log_report( @@ -113,7 +132,7 @@ def from_mails( end_date, output_dir=output_dir, ) - + # send mail with n2.log as attachment mail_body = build_mail_summary( nb_mails=nb_mails, @@ -136,7 +155,7 @@ def from_files( typer.Option( "--output-dir", help="Output directories where reports will be stored." ), - ] = Path("output"), + ] = RESULT_PATH, input_files: Annotated[ list[Path] | None, typer.Option("--input-files", help="Log files to scan.") ] = None, @@ -144,7 +163,7 @@ def from_files( Path | None, typer.Option("--input-dir", help="Directories containing log files."), ] = None, -): +) -> None: """Convert log files into two reports, one of n2 related logs, one without any n2 related logs. These reports are sent by mail to n2 support technicians. @@ -153,11 +172,7 @@ def from_files( Args: input_files: Log files to scan. Either input_files or input_dir must be provided. input_dir: Directories containing log files. Either input_files or input_dir must be provided. - output_dir: Required. Output directories where reports will be stored. - version: Option that returns the version of package - - Returns: - None + output_dir: Output directories where reports will be stored. """ logger.info("============ logwatcher file started ============") @@ -176,11 +191,11 @@ def from_files( logger.error("Neither input-files nor input-dir were provided.") raise typer.BadParameter("Either input_files or input_dir must be provided") - logger.info(f"reports will be located at '{output_dir.cwd()}'") + logger.info(f"reports will be located at '{output_dir.resolve()}'") log_entries = _files_to_log_entries(log_files) relevant, irrelevant = classify_log_entries(log_entries) - today = datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT) + today = datetime.now(tz=FRENCH_TIMEZONE).strftime(DATETIME_FORMAT) start_date, end_date = get_period(relevant + irrelevant) write_log_report( source_type=SourceType.FILE, @@ -195,6 +210,47 @@ def from_files( logger.info("============ logwatcher file ended ============") +@app.command() +def prune( + mails: Annotated[ + bool, + typer.Option("-m", "--mails", help="Prune old mails."), + ] = False, + logs: Annotated[ + bool, + typer.Option("-l", "--logs", help="Prune old logs."), + ] = False, +) -> None: + """Purge mails and logs based on the value of environment variables.""" + if not mails and not logs: + raise typer.BadParameter("At least one of --mails or --logs must be provided.") + + logger.info("============ logwatcher prune started ============") + + today = datetime.now(tz=FRENCH_TIMEZONE) + if mails: + logger.debug("Mails selected to be pruned") + account = connect_to_mailbox() + logger.info("Start pruning mails.") + prune_mails(today, account) + logger.info("Pruning mails done.") + else: + logger.debug("No mails were purged.") + + if logs: + logger.debug("Logs selected to be pruned") + logger.info("Start pruning logs.") + log_files = get_logs_to_archive(today) + if log_files: + archive_logs(today, log_files) + prune_logs(today) + logger.info("Pruning logs done.") + else: + logger.debug("No logs were purged.") + + logger.info("============ logwatcher prune ended ============") + + @app.callback() def cli( version: Annotated[ @@ -222,7 +278,9 @@ def cli( version: if provided, script returns the version of package """ - setup_logging(verbose=verbose) + log_file_name = datetime.now(FRENCH_TIMEZONE).strftime(DATE_FILE_FORMAT) + ".log" + log_file = LOGGING_PATH / log_file_name # date and time of execution + setup_logging(verbose=verbose, file_path=log_file) if __name__ == "__main__": diff --git a/src/logwatcher/config.py b/src/logwatcher/config.py index b5743b3..aedd9a4 100644 --- a/src/logwatcher/config.py +++ b/src/logwatcher/config.py @@ -7,12 +7,14 @@ from pathlib import Path OUTPUT_PATH = Path("output") RESULT_PATH = OUTPUT_PATH / "results" LOGGING_PATH = OUTPUT_PATH / "logs" +ARCHIVE_DIR = LOGGING_PATH / "archives" # test paths TEST_PATH = Path("tests") FIXTURE_PATH = TEST_PATH / "fixtures" # date and time +DATE_FILE_FORMAT = "%d%m%Y_%H%M%S" DATETIME_FORMAT = "%d/%m/%Y %H:%M:%S" FRENCH_TIMEZONE = ZoneInfo("Europe/Paris") @@ -20,6 +22,6 @@ FRENCH_TIMEZONE = ZoneInfo("Europe/Paris") # file type class SourceType(Enum): """Designate which type of template to use while building reports.""" - + FILE = "fichier" MAIL = "mail" diff --git a/src/logwatcher/logging_config.py b/src/logwatcher/logging_config.py index a2c6260..deb6fda 100644 --- a/src/logwatcher/logging_config.py +++ b/src/logwatcher/logging_config.py @@ -7,13 +7,13 @@ from logwatcher.config import DATETIME_FORMAT def _setup_formatter( format: str = "[%(asctime)s] - %(levelname)s: %(message)s", datefmt: str = DATETIME_FORMAT, -) -> logging.Formatter: +) -> logging.Formatter: # TODO: Add documentation return logging.Formatter(fmt=format, datefmt=datefmt) def _setup_handler( formatter: logging.Formatter, level: int, filename: Path | None = None -) -> logging.Handler: +) -> logging.Handler: # TODO: Add documentation handler: logging.Handler if filename: handler = logging.FileHandler(filename=filename, encoding="utf-8") @@ -30,10 +30,12 @@ def setup_logging( verbose: bool = False, ) -> logging.Logger: """Create a new instance of Logger named 'logger' customized for the logwatcher package. - + The format display in order the time, the level of log and the log message - The logs are only redirect in the standard output. + The logs are only redirect in the standard output. # TODO Complete docstring """ + file_path.parent.mkdir(parents=True, exist_ok=True) + _formatter = formatter if formatter else _setup_formatter() if verbose: diff --git a/src/logwatcher/mail_reader.py b/src/logwatcher/mail_reader.py index 2700902..6c75f8c 100644 --- a/src/logwatcher/mail_reader.py +++ b/src/logwatcher/mail_reader.py @@ -14,7 +14,12 @@ from exchangelib import ( ) from exchangelib.errors import UnauthorizedError -from logwatcher.mail_utils import ANALYZED_FOLDER, LOG_IN_ATTACHMENT_PATTERN, get_or_create_folder +from logwatcher.utils import ( + ANALYZED_FOLDER, + LOG_FOLDER, + LOG_IN_ATTACHMENT_PATTERN, + get_or_create_folder, +) logger = logging.getLogger(__name__) @@ -85,7 +90,7 @@ def _has_attachment_marker(body: str) -> bool: Returns: True if the email says the report is in the attachment. False otherwise. - + """ return LOG_IN_ATTACHMENT_PATTERN in body @@ -133,9 +138,9 @@ def fetch_log_messages(account: Account) -> list[Message]: Returns: queryset of log-related mails - + """ - logs_folder = account.msg_folder_root / "Logs" + logs_folder = account.msg_folder_root / LOG_FOLDER return list(logs_folder.all()) @@ -192,15 +197,11 @@ def move_analyzed_mails(account: Account, messages: list[Message]) -> None: logger.info("No mails to move.") return - logger.info( - f"Moving {len(messages)} mails to '{ANALYZED_FOLDER}'" - ) + logger.info(f"Moving {len(messages)} mails to '{ANALYZED_FOLDER}'") analyzed_folder = get_or_create_folder(account, ANALYZED_FOLDER) ids = [(m.id, m.changekey) for m in messages] account.bulk_move(ids=ids, to_folder=analyzed_folder) - logger.info( - f"Mails were successfully moved to '{ANALYZED_FOLDER}'." - ) \ No newline at end of file + logger.info(f"Mails were successfully moved to '{ANALYZED_FOLDER}'.") diff --git a/src/logwatcher/models.py b/src/logwatcher/models.py index 10e0125..b087a05 100644 --- a/src/logwatcher/models.py +++ b/src/logwatcher/models.py @@ -24,11 +24,15 @@ class LogEntry: """Return raw line of the log entry.""" return self.raw_line - def get_start_time(self) -> str: # TODO: remove or adapt mentions of start_time in code base + def get_start_time( + self, + ) -> str: # TODO: remove or adapt mentions of start_time in code base """Return formatted start time.""" return self.start_time.strftime(DATETIME_FORMAT) - def get_error_time(self) -> str: # TODO: remove or adapt mentions of error_time in code base + def get_error_time( + self, + ) -> str: # TODO: remove or adapt mentions of error_time in code base """Return formatted error time.""" return self.error_time.strftime(DATETIME_FORMAT) diff --git a/src/logwatcher/notifier.py b/src/logwatcher/notifier.py index 6f97b46..42a98cb 100644 --- a/src/logwatcher/notifier.py +++ b/src/logwatcher/notifier.py @@ -4,7 +4,7 @@ from pathlib import Path from exchangelib import Account, FileAttachment, Message -from logwatcher.mail_utils import get_or_create_folder +from logwatcher.utils import SENT_FOLDER, get_or_create_folder logger = logging.getLogger(__name__) @@ -16,7 +16,7 @@ def send_n2_report(account: Account, summary: str, n2_log_file: Path) -> None: account: the EWS account used to send email report from summary: text body of the email report n2_log_file: path to the n2 report file - + """ try: to = os.environ["N2_REPORT_RECIPIENTS"].replace(" ", "").split(",") @@ -27,17 +27,23 @@ def send_n2_report(account: Account, summary: str, n2_log_file: Path) -> None: # build file attachment logger.info(f"Create attachment based on '{n2_log_file}' file.") - n2_log_attachment = FileAttachment(name=n2_log_file.name, content=n2_log_file.read_bytes(), is_inline=False) + n2_log_attachment = FileAttachment( + name=n2_log_file.name, content=n2_log_file.read_bytes(), is_inline=False + ) logger.info(f"Attachment '{n2_log_attachment.name}' was successfully created.") # get sent folder - sent_folder = get_or_create_folder(account, "Sent") - + sent_folder = get_or_create_folder(account, SENT_FOLDER) + # create mail - logger.info(f"Create mail with account='{account.fullname}', recipients='{to}', subject='{subject}'.") + logger.info( + f"Create mail with account='{account.fullname}', recipients='{to}', subject='{subject}'." + ) mail = Message(account=account, to_recipients=to, subject=subject, body=summary) mail.attach(n2_log_attachment) - logger.info(f"Mail '{mail.subject}' was successfully created with attachment '{n2_log_attachment.name}'.") + logger.info( + f"Mail '{mail.subject}' was successfully created with attachment '{n2_log_attachment.name}'." + ) # send and keep a copy to the Logs/Sent folder logger.info(f"Attempt to send mail '{mail.subject}'...") diff --git a/src/logwatcher/parser.py b/src/logwatcher/parser.py index 2ec9964..39b51c2 100644 --- a/src/logwatcher/parser.py +++ b/src/logwatcher/parser.py @@ -99,5 +99,5 @@ def parse_file(log_file_path: Path) -> list[LogEntry]: logger.info(f"\tparsing log file '{log_file_path.name}' started.") log_entries = parse_lines(log_file) logger.info(f"\t\t{len(log_entries)} lines were parsed.") - logger.info("\tparsing completed.") + logger.info("parsing completed.") return log_entries diff --git a/src/logwatcher/reporter.py b/src/logwatcher/reporter.py index a6a45b6..5d498e6 100644 --- a/src/logwatcher/reporter.py +++ b/src/logwatcher/reporter.py @@ -48,7 +48,7 @@ ERROR_TEMPLATE = Template(""" MAIL_TEMPLATE = Template(""" Rapport analyse de log - $generation_date -Nb de mail(s) analysé(s)\t : $nb_mail +Nb de mail(s) analysé(s)\t : $nb_mails Erreurs N2/Erreurs totales\t : $n2_errors/$total_errors Période : $start_date -> $end_date @@ -58,12 +58,12 @@ La liste des erreurs se trouve en pièce jointe `n2.log`. def get_period(entries: list[LogEntry]) -> tuple[str, str]: """Get start date and end date among dates of log entries. - - Check all log entries dates and find the timestamp of earliest and latest generated logs + + Check all log entries dates and find the timestamp of earliest and latest generated logs. Args: entries: list of log entries - + Returns: start date and end date in string format @@ -84,36 +84,36 @@ def write_log_report( end_date: str, output_dir: Path = RESULT_PATH, ) -> None: - """Write each relevant, irrelevant and general reports in their respective output file. + """Write each relevant, irrelevant and all error reports in their respective output file. Use the range date of relevant and irrelevant lists to get the period of time the logs were generated. Args: - end_date: # TODO - start_date: # TODO - today: # TODO source_type: # TODO relevant: List of n2 log entry irrelevant: List of not n2 log entry nb_files: Number of files scanned + today: # TODO + start_date: # TODO + end_date: # TODO output_dir: Location where all reports will be written """ - logger.info("\twriting reports job started.") + logger.info("writing reports job started.") reports_dict = build_reports( source_type, relevant, irrelevant, today, start_date, end_date, nb_files ) + output_dir.mkdir(parents=True, exist_ok=True) for target in reports_dict: - logger.info(f"\t\tgenerating report '{target}'...") - output_dir.mkdir(parents=True, exist_ok=True) - logger.debug(f"\t\t\t{output_dir} exists.") - target_file = output_dir / str(target + ".log") + logger.info(f"Generating report '{target}'...") + logger.debug(f"'{output_dir}' exists.") + target_file = output_dir / f"{target}.log" target_file.write_text(reports_dict[target], encoding="windows-1252") - logger.debug(f"\t\t\treport was written in {target_file.name}") - logger.info(f"\t\treport '{target}' was generated.") - logger.info("\twriting reports job completed.") + logger.debug(f"Report was written in {target_file.name}") + logger.info(f"Report '{target}' was generated.") + logger.info("Writing reports job completed.") def build_reports( @@ -143,7 +143,7 @@ def build_reports( Dictionary of reports in string format """ - logger.info("\t\tbuilding reports job started.") + logger.info("building reports job started.") relevant_report = _render_target_report( log_entries=relevant, target_template=N2_SUPPORT_TEMPLATE ) @@ -151,8 +151,7 @@ def build_reports( irrelevant_report = _render_target_report( log_entries=irrelevant, target_template=OTHER_TEMPLATE ) - logger.info("\t\t\tn2 and other reports were generated.") - logger.info("\t\tbuilding reports job completed.") + logger.info("building reports job completed.") source_value = source_type.value.capitalize() @@ -199,7 +198,7 @@ def build_mail_summary( """Build a short plain-text summary of a logwatcher run. Args: - nb_mails: Number of log files parsed. + nb_mails: Number of log mails parsed. nb_errors_total: Total number of errors across all reports. nb_errors_n2: Number of N2-relevant errors. today: # TODO @@ -213,7 +212,7 @@ def build_mail_summary( logger.info("generating mail body...") return MAIL_TEMPLATE.substitute( generation_date=today, - nb_mail=nb_mails, + nb_mails=nb_mails, n2_errors=nb_errors_n2, total_errors=nb_errors_total, start_date=start_date, @@ -231,7 +230,7 @@ def _render_target_report( Args: log_entries: List of log entries - target_template: Template to use for N2_SUPPORT_TEMPLATE + target_template: Template to use for N2_SUPPORT_TEMPLATE and OTHER_TEMPLATE Returns: content of report in string format diff --git a/src/logwatcher/mail_utils.py b/src/logwatcher/utils.py similarity index 62% rename from src/logwatcher/mail_utils.py rename to src/logwatcher/utils.py index 5ab6601..3b45edc 100644 --- a/src/logwatcher/mail_utils.py +++ b/src/logwatcher/utils.py @@ -1,11 +1,14 @@ import logging +from pathlib import Path from exchangelib import Account, Folder logger = logging.getLogger(__name__) ANALYZED_FOLDER = "Analyzed" +LOG_FOLDER = "Logs" LOG_IN_ATTACHMENT_PATTERN = "Le compte-rendu contient plus de 100 lignes." +SENT_FOLDER = "Sent" def get_or_create_folder(account: Account, folder_name: str) -> Folder: @@ -19,7 +22,7 @@ def get_or_create_folder(account: Account, folder_name: str) -> Folder: The existing or newly created 'Analyzed' folder. """ - log_folder = account.msg_folder_root / "Logs" + log_folder = account.msg_folder_root / LOG_FOLDER matches = list(log_folder.glob(folder_name)) if matches: @@ -29,4 +32,20 @@ def get_or_create_folder(account: Account, folder_name: str) -> Folder: logger.info(f"\tCreating '{folder_name}' folder.") folder = Folder(parent=log_folder, name=folder_name) folder.save() - return folder \ No newline at end of file + return folder + + +def is_log_file(path: Path) -> bool: + """Return True if file is a log file (.txt, .log, or extensionless). + + Args: + path: Path of a file + + Returns: + True if file has .txt or .log extension or extensionless + False otherwise. + + """ + if not path.is_file(): + return False + return path.suffix in {"", ".txt", ".log"} diff --git a/tests/__init__.py b/tests/__init__.py index e69de29..1df8353 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Entry point to all package tests.""" diff --git a/tests/conftest.py b/tests/conftest.py index 0204cdd..9ff605f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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 \ No newline at end of file + return _make_mock_account diff --git a/tests/test_cli.py b/tests/test_cli.py index c09bc2e..8f9a5b6 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -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"] diff --git a/tests/test_logging.py b/tests/test_logging.py index a7ff34a..0876112 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -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") diff --git a/tests/test_models.py b/tests/test_models.py index aa203b0..0e7c970 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -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 diff --git a/tests/test_notifier.py b/tests/test_notifier.py index 65ff1c8..35fa9e6 100644 --- a/tests/test_notifier.py +++ b/tests/test_notifier.py @@ -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() + ) diff --git a/tests/test_parser.py b/tests/test_parser.py index 82552b0..a05883f 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -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() diff --git a/tests/test_mail_utils.py b/tests/test_utils.py similarity index 51% rename from tests/test_mail_utils.py rename to tests/test_utils.py index 2f4382e..26a9791 100644 --- a/tests/test_mail_utils.py +++ b/tests/test_utils.py @@ -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() \ No newline at end of file + fake_folder_instance.save.assert_called_once()