Repository: aristanetworks/anta (python)

Issue:
Add test to search syslog for specific entries
Need a test case to search the syslog for a specific entry.  Test case should accept a regular expression to match within the log and the number of previous log entries to search.

Example of input would be 
match: ".*ACCOUNTING-5-EXEC: cvpadmin ssh.*"
previous entries: 30 
Arista EOS CLI command: `show logging 30`

This should pass based on the following line in the syslog:
```
Mar  9 19:07:23 DC1-SPINE1 Aaa: %ACCOUNTING-5-EXEC: cvpadmin ssh 192.168.0.6 start task_id=2 start_time=1741547242 timezone=UTC service=shell
```


Pull request:
feat(anta.tests): Added test to search syslog for specific entries
# Description
Added test to search syslog for specific entries

Fixes #1077 

# Checklist:
- [x] My code follows the style guidelines of this project
- [x] I have performed a self-review of my code
- [x] I have run pre-commit for code linting and typing (`pre-commit run`)
- [ ] I have made corresponding changes to the documentation
- [x] I have added tests that prove my fix is effective or that my feature works
- [x] New and existing unit tests pass locally with my changes (`tox -e testenv`)


Patch:
diff --git a/anta/input_models/logging.py b/anta/input_models/logging.py
new file mode 100644
index 00000000..977f1ab0
--- /dev/null
+++ b/anta/input_models/logging.py
@@ -0,0 +1,21 @@
+# Copyright (c) 2023-2025 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Module containing input models for logging tests."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from anta.custom_types import LogSeverityLevel, RegexString
+
+
+class LoggingQuery(BaseModel):
+    """Logging query model representing the logging details."""
+
+    regex_match: RegexString
+    """Log regex pattern to be searched in last log entries."""
+    last_number_messages: int = Field(ge=1, le=9999)
+    """Last number of messages to check in the logging buffers."""
+    severity_level: LogSeverityLevel = "informational"
+    """Log severity level."""
diff --git a/pyproject.toml b/pyproject.toml
index 176ca0ff..12914cdc 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -434,6 +434,9 @@ runtime-evaluated-base-classes = ["pydantic.BaseModel", "anta.models.AntaTest.In
 "anta/tests/logging.py" = [
   "A005",  # TODO: Module `logging` shadows a Python standard-library module
 ]
+"anta/input_models/logging.py" = [
+  "A005",  # TODO: Module `logging` shadows a Python standard-library module
+]
 "anta/decorators.py" = [
   "ANN401",  # Ok to use Any type hint in our decorators
 ]


Test patch:
diff --git a/anta/tests/logging.py b/anta/tests/logging.py
index 51b83aa6..f13860e2 100644
--- a/anta/tests/logging.py
+++ b/anta/tests/logging.py
@@ -15,6 +15,7 @@ from ipaddress import IPv4Address
 from typing import TYPE_CHECKING, ClassVar
 
 from anta.custom_types import LogSeverityLevel
+from anta.input_models.logging import LoggingQuery
 from anta.models import AntaCommand, AntaTemplate, AntaTest
 
 if TYPE_CHECKING:
@@ -433,3 +434,53 @@ class VerifyLoggingErrors(AntaTest):
             self.result.is_success()
         else:
             self.result.is_failure("Device has reported syslog messages with a severity of ERRORS or higher")
+
+
+class VerifyLoggingEntries(AntaTest):
+    """Verifies that the expected log string is present in the last specified log messages.
+
+    Expected Results
+    ----------------
+    * Success: The test will pass if the expected log string for the mentioned severity level is present in the last specified log messages.
+    * Failure: The test will fail if the specified log string is not present in the last specified log messages.
+
+    Examples
+    --------
+    ```yaml
+    anta.tests.logging:
+      - VerifyLoggingEntries:
+          logging_entries:
+            - regex_match: ".ACCOUNTING-5-EXEC: cvpadmin ssh."
+              last_number_messages: 30
+              severity_level: alerts
+            - regex_match: ".SPANTREE-6-INTERFACE_ADD:."
+              last_number_messages: 10
+              severity_level: critical
+    ```
+    """
+
+    categories: ClassVar[list[str]] = ["logging"]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [
+        AntaTemplate(template="show logging {last_number_messages} {severity_level}", ofmt="text", use_cache=False)
+    ]
+
+    class Input(AntaTest.Input):
+        """Input model for the VerifyLoggingEntries test."""
+
+        logging_entries: list[LoggingQuery]
+        """List of logging entries and regex match."""
+
+    def render(self, template: AntaTemplate) -> list[AntaCommand]:
+        """Render the template for last number messages and log severity level in the input."""
+        return [template.render(last_number_messages=entry.last_number_messages, severity_level=entry.severity_level) for entry in self.inputs.logging_entries]
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Main test function for VerifyLoggingEntries."""
+        self.result.is_success()
+        for command_output, logging_entry in zip(self.instance_commands, self.inputs.logging_entries):
+            output = command_output.text_output
+            if not re.search(logging_entry.regex_match, output):
+                self.result.is_failure(
+                    f"Pattern: {logging_entry.regex_match} - Not found in last {logging_entry.last_number_messages} {logging_entry.severity_level} log entries"
+                )
diff --git a/docs/api/tests/logging.md b/docs/api/tests/logging.md
index c59d3a94..b305c42f 100644
--- a/docs/api/tests/logging.md
+++ b/docs/api/tests/logging.md
@@ -8,6 +8,8 @@ anta_title: ANTA catalog for logging tests
   ~ that can be found in the LICENSE file.
   -->
 
+# Tests
+
 ::: anta.tests.logging
 
     options:
@@ -22,3 +24,20 @@ anta_title: ANTA catalog for logging tests
       show_root_toc_entry: false
       show_symbol_type_heading: false
       show_symbol_type_toc: false
+
+# Input models
+
+::: anta.input_models.logging
+
+    options:
+      anta_hide_test_module_description: true
+      filters:
+        - "!test"
+        - "!render"
+      merge_init_into_class: false
+      show_bases: false
+      show_labels: true
+      show_root_heading: false
+      show_root_toc_entry: false
+      show_symbol_type_heading: false
+      show_symbol_type_toc: false
diff --git a/examples/tests.yaml b/examples/tests.yaml
index 673702b5..ea36e8c3 100644
--- a/examples/tests.yaml
+++ b/examples/tests.yaml
@@ -294,6 +294,15 @@ anta.tests.lanz:
 anta.tests.logging:
   - VerifyLoggingAccounting:
       # Verifies if AAA accounting logs are generated.
+  - VerifyLoggingEntries:
+      # Verifies that the expected log string is present in the last specified log messages.
+      logging_entries:
+        - regex_match: ".ACCOUNTING-5-EXEC: cvpadmin ssh."
+          last_number_messages: 30
+          severity_level: alerts
+        - regex_match: ".SPANTREE-6-INTERFACE_ADD:."
+          last_number_messages: 10
+          severity_level: critical
   - VerifyLoggingErrors:
       # Verifies there are no syslog messages with a severity of ERRORS or higher.
   - VerifyLoggingHostname:
diff --git a/tests/units/anta_tests/test_logging.py b/tests/units/anta_tests/test_logging.py
index 9ad03630..5f795895 100644
--- a/tests/units/anta_tests/test_logging.py
+++ b/tests/units/anta_tests/test_logging.py
@@ -9,6 +9,7 @@ from typing import Any
 
 from anta.tests.logging import (
     VerifyLoggingAccounting,
+    VerifyLoggingEntries,
     VerifyLoggingErrors,
     VerifyLoggingHostname,
     VerifyLoggingHosts,
@@ -319,4 +320,44 @@ DATA: list[dict[str, Any]] = [
         "inputs": None,
         "expected": {"result": "failure", "messages": ["Syslog logging is disabled"]},
     },
+    {
+        "name": "success",
+        "test": VerifyLoggingEntries,
+        "eos_data": [
+            """Mar 13 04:10:45 s1-leaf1 ProcMgr: %PROCMGR-6-TERMINATE_RUNNING_PROCESS: Terminating deconfigured/reconfigured process 'SystemInitMonitor' (PID=859)
+ Mar 13 04:10:45 s1-leaf1 ProcMgr: %PROCMGR-6-PROCESS_TERMINATED: 'SystemInitMonitor' (PID=859, status=9) has terminated.""",
+            """Mar 13 04:10:45 s1-leaf1 ProcMgr: %PROCMGR-7-WORKER_WARMSTART_DONE: ProcMgr worker warm start done. (PID=547)""",
+        ],
+        "inputs": {
+            "logging_entries": [
+                {"regex_match": ".*PROCMGR-6-PROCESS_TERMINATED:.*", "last_number_messages": 3},
+                {"regex_match": ".*ProcMgr worker warm start.*", "last_number_messages": 2, "severity_level": "debugging"},
+            ]
+        },
+        "expected": {"result": "success"},
+    },
+    {
+        "name": "failure-log-str-not-found",
+        "test": VerifyLoggingEntries,
+        "eos_data": [
+            """Mar 12 04:34:01 s1-leaf1 ProcMgr: %PROCMGR-7-WORKER_WARMSTART_DONE: ProcMgr worker warm start done. (PID=559)
+Mar 12 04:34:01 s1-leaf1 ProcMgr: %PROCMGR-6-PROCESS_TERMINATED: 'SystemInitMonitor' (PID=867, status=9) has terminated.""",
+            """Mar 13 03:58:12 s1-leaf1 ConfigAgent: %SYS-5-CONFIG_SESSION_ABORTED: User cvpsystem aborted
+             configuration session capiVerify-612-612b34a2ffbf11ef96ba3a348d538ba0 on TerminAttr (localhost)
+ Mar 13 04:10:45 s1-leaf1 SystemInitMonitor: %SYS-5-SYSTEM_INITIALIZED: System is initialized""",
+        ],
+        "inputs": {
+            "logging_entries": [
+                {"regex_match": ".ACCOUNTING-5-EXEC: cvpadmin ssh.", "last_number_messages": 3},
+                {"regex_match": ".*ProcMgr worker warm start.*", "last_number_messages": 10, "severity_level": "debugging"},
+            ]
+        },
+        "expected": {
+            "result": "failure",
+            "messages": [
+                "Pattern: .ACCOUNTING-5-EXEC: cvpadmin ssh. - Not found in last 3 informational log entries",
+                "Pattern: .*ProcMgr worker warm start.* - Not found in last 10 debugging log entries",
+            ],
+        },
+    },
 ]