This is page 2 of 3. Use http://codebase.md/djm81/log_analyzer_mcp?page={x} to view the full context.
# Directory Structure
```
├── .cursor
│ └── rules
│ ├── markdown-rules.mdc
│ ├── python-github-rules.mdc
│ └── testing-and-build-guide.mdc
├── .cursorrules
├── .env.template
├── .github
│ ├── ISSUE_TEMPLATE
│ │ └── bug_report.md
│ ├── pull_request_template.md
│ └── workflows
│ └── tests.yml
├── .gitignore
├── CHANGELOG.md
├── CODE_OF_CONDUCT.md
├── CONTRIBUTING.md
├── docs
│ ├── api_reference.md
│ ├── developer_guide.md
│ ├── getting_started.md
│ ├── LICENSE.md
│ ├── README.md
│ ├── refactoring
│ │ ├── log_analyzer_refactoring_v1.md
│ │ ├── log_analyzer_refactoring_v2.md
│ │ └── README.md
│ ├── rules
│ │ ├── markdown-rules.md
│ │ ├── python-github-rules.md
│ │ ├── README.md
│ │ └── testing-and-build-guide.md
│ └── testing
│ └── README.md
├── LICENSE.md
├── pyproject.toml
├── pyrightconfig.json
├── README.md
├── scripts
│ ├── build.sh
│ ├── cleanup.sh
│ ├── publish.sh
│ ├── release.sh
│ ├── run_log_analyzer_mcp_dev.sh
│ └── test_uvx_install.sh
├── SECURITY.md
├── setup.py
├── src
│ ├── __init__.py
│ ├── log_analyzer_client
│ │ ├── __init__.py
│ │ ├── cli.py
│ │ └── py.typed
│ └── log_analyzer_mcp
│ ├── __init__.py
│ ├── common
│ │ ├── __init__.py
│ │ ├── config_loader.py
│ │ ├── logger_setup.py
│ │ └── utils.py
│ ├── core
│ │ ├── __init__.py
│ │ └── analysis_engine.py
│ ├── log_analyzer_mcp_server.py
│ ├── py.typed
│ └── test_log_parser.py
└── tests
├── __init__.py
├── log_analyzer_client
│ ├── __init__.py
│ └── test_cli.py
└── log_analyzer_mcp
├── __init__.py
├── common
│ └── test_logger_setup.py
├── test_analysis_engine.py
├── test_log_analyzer_mcp_server.py
└── test_test_log_parser.py
```
# Files
--------------------------------------------------------------------------------
/docs/api_reference.md:
--------------------------------------------------------------------------------
```markdown
# Log Analyzer MCP API Reference
This document provides a detailed reference for the tools and endpoints exposed by the Log Analyzer MCP Server and the commands available through its CLI client.
## Table of Contents
- [MCP Server Tools](#mcp-server-tools)
- [Test Analysis and Execution](#test-analysis-and-execution)
- [Log Searching](#log-searching)
- [Server Utilities](#server-utilities)
- [CLI Client (`log-analyzer`)](#cli-client-log-analyzer)
- [Global Options](#global-options)
- [Search Commands (`log-analyzer search`)](#search-commands-log-analyzer-search)
- [Common Search Options](#common-search-options)
- [`log-analyzer search all`](#log-analyzer-search-all)
- [`log-analyzer search time`](#log-analyzer-search-time)
- [`log-analyzer search first`](#log-analyzer-search-first)
- [`log-analyzer search last`](#log-analyzer-search-last)
- [Error Handling](#error-handling)
---
## MCP Server Tools
The Log Analyzer MCP Server provides tools for test analysis, log searching, and server introspection.
### Test Analysis and Execution
Tools related to running tests, analyzing results, and managing coverage reports.
#### `analyze_tests`
Analyzes the most recent test run and provides detailed information about failures.
**Parameters:**
| Name | Type | Required | Default | Description |
|----------------|---------|----------|---------|----------------------------------------------------|
| `summary_only` | boolean | No | `False` | Whether to return only a summary of the test results |
**Returns:**
A JSON object containing the test analysis, including:
- `summary`: Overall summary (status, passed, failed, skipped).
- `error_details`: (If not `summary_only`) List of detailed error information.
- `log_file`: Path to the analyzed log file.
- `log_timestamp`: Timestamp of the log file.
- `log_age_minutes`: Age of the log file in minutes.
- `error`: (If an error occurred during analysis) Error message.
**Example Call:**
```json
{
"tool_name": "analyze_tests",
"arguments": {
"summary_only": true
}
}
```
#### `run_tests_no_verbosity`
Runs all tests with minimal output (verbosity level 0). Excludes server integration tests to prevent recursion.
**Parameters:** None
**Returns:**
A JSON object with:
- `success`: Boolean indicating if the test execution command was successful.
- `return_code`: Exit code from the test runner.
- `test_output`: Combined stdout and stderr from the test run.
- `analysis_log_path`: Path to the log file where test output was saved.
- `error`: (If an error occurred) Error message.
**Example Call:**
```json
{
"tool_name": "run_tests_no_verbosity",
"arguments": {}
}
```
#### `run_tests_verbose`
Runs all tests with verbose output (verbosity level 1). Excludes server integration tests.
**Parameters:** None
**Returns:** (Same structure as `run_tests_no_verbosity`)
**Example Call:**
```json
{
"tool_name": "run_tests_verbose",
"arguments": {}
}
```
#### `run_tests_very_verbose`
Runs all tests with very verbose output (verbosity level 2) and enables coverage. Excludes server integration tests.
**Parameters:** None
**Returns:** (Same structure as `run_tests_no_verbosity`, coverage data is generated)
**Example Call:**
```json
{
"tool_name": "run_tests_very_verbose",
"arguments": {}
}
```
#### `run_unit_test`
Runs tests for a specific agent only.
**Parameters:**
| Name | Type | Required | Default | Description |
|-------------|---------|----------|---------|----------------------------------------------------------------|
| `agent` | string | Yes | | The agent to run tests for (e.g., 'qa_agent', 'backlog_agent') |
| `verbosity` | integer | No | `1` | Verbosity level (0=minimal, 1=normal, 2=detailed) |
**Returns:** (Same structure as `run_tests_no_verbosity`)
**Example Call:**
```json
{
"tool_name": "run_unit_test",
"arguments": {
"agent": "my_agent",
"verbosity": 0
}
}
```
#### `create_coverage_report`
Runs tests with coverage and generates HTML and XML reports using `hatch`.
**Parameters:**
| Name | Type | Required | Default | Description |
|-----------------|---------|----------|---------|-----------------------------------------------------------------------|
| `force_rebuild` | boolean | No | `False` | Whether to force rebuilding the report even if it already exists |
**Returns:**
A JSON object with:
- `success`: Boolean indicating overall success of report generation steps.
- `message`: Summary message.
- `overall_coverage_percent`: Parsed overall coverage percentage.
- `coverage_xml_path`: Path to the generated XML coverage report.
- `coverage_html_dir`: Path to the directory of the HTML coverage report.
- `coverage_html_index`: Path to the main `index.html` of the HTML report.
- `text_summary_output`: Text summary from the coverage tool.
- `hatch_xml_output`: Output from the hatch XML generation command.
- `hatch_html_output`: Output from the hatch HTML generation command.
- `timestamp`: Timestamp of the report generation.
**Example Call:**
```json
{
"tool_name": "create_coverage_report",
"arguments": {
"force_rebuild": true
}
}
```
### Log Searching
Tools for searching and filtering log files managed by the `AnalysisEngine`.
#### Common Parameters for Search Tools
These parameters are available for `search_log_all_records`, `search_log_time_based`, `search_log_first_n_records`, and `search_log_last_n_records`.
| Name | Type | Required | Default | Description |
|---------------------------------|---------|----------|-----------|------------------------------------------------------------------------------------------------------------|
| `scope` | string | No | "default" | Logging scope to search within (from `.env` scopes or default). |
| `context_before` | integer | No | `2` | Number of lines before a match. |
| `context_after` | integer | No | `2` | Number of lines after a match. |
| `log_dirs_override` | string | No | `""` | Comma-separated list of log directories, files, or glob patterns (overrides `.env` for file locations). |
| `log_content_patterns_override` | string | No | `""` | Comma-separated list of REGEX patterns for log messages (overrides `.env` content filters). |
#### `search_log_all_records`
Searches for all log records, optionally filtering by scope and content patterns, with context.
**Parameters:** (Includes Common Search Parameters)
**Returns:**
A list of JSON objects, where each object represents a found log entry and includes:
- `timestamp`: Parsed timestamp of the log entry.
- `raw_line`: The original log line.
- `file_path`: Path to the log file containing the entry.
- `line_number`: Line number in the file.
- `context_before_lines`: List of lines before the matched line.
- `context_after_lines`: List of lines after the matched line.
- (Other fields from `LogEntry` model)
**Example Call:**
```json
{
"tool_name": "search_log_all_records",
"arguments": {
"scope": "my_app_scope",
"log_content_patterns_override": "ERROR.*database"
}
}
```
#### `search_log_time_based`
Searches logs within a time window, optionally filtering, with context.
**Parameters:** (Includes Common Search Parameters plus)
| Name | Type | Required | Default | Description |
|-----------|---------|----------|---------|--------------------------------------------|
| `minutes` | integer | No | `0` | Search logs from the last N minutes. |
| `hours` | integer | No | `0` | Search logs from the last N hours. |
| `days` | integer | No | `0` | Search logs from the last N days. |
**Returns:** (List of JSON objects, same structure as `search_log_all_records`)
**Example Call:**
```json
{
"tool_name": "search_log_time_based",
"arguments": {
"hours": 2,
"scope": "server_logs",
"context_after": 5
}
}
```
#### `search_log_first_n_records`
Searches for the first N (oldest) records, optionally filtering, with context.
**Parameters:** (Includes Common Search Parameters plus)
| Name | Type | Required | Default | Description |
|---------|---------|----------|---------|---------------------------------------------------------------|
| `count` | integer | Yes | | Number of first (oldest) matching records to return (must be > 0). |
**Returns:** (List of JSON objects, same structure as `search_log_all_records`)
**Example Call:**
```json
{
"tool_name": "search_log_first_n_records",
"arguments": {
"count": 10,
"log_dirs_override": "/var/log/app_archive/*.log"
}
}
```
#### `search_log_last_n_records`
Search for the last N (newest) records, optionally filtering, with context.
**Parameters:** (Includes Common Search Parameters plus)
| Name | Type | Required | Default | Description |
|---------|---------|----------|---------|--------------------------------------------------------------|
| `count` | integer | Yes | | Number of last (newest) matching records to return (must be > 0). |
**Returns:** (List of JSON objects, same structure as `search_log_all_records`)
**Example Call:**
```json
{
"tool_name": "search_log_last_n_records",
"arguments": {
"count": 50,
"scope": "realtime_feed"
}
}
```
### Server Utilities
General utility tools for the MCP server.
#### `ping`
Checks if the MCP server is alive and returns status information.
**Parameters:** None
**Returns:**
A string with status, timestamp, and a message indicating the server is running.
**Example Call:**
```json
{
"tool_name": "ping",
"arguments": {}
}
```
#### `get_server_env_details`
Returns `sys.path` and `sys.executable` and other environment details from the running MCP server.
**Parameters:** None
**Returns:**
A JSON object with:
- `sys_executable`: Path to the Python interpreter running the server.
- `sys_path`: List of paths in `sys.path`.
- `cwd`: Current working directory of the server.
- `environ_pythonpath`: Value of the `PYTHONPATH` environment variable, if set.
**Example Call:**
```json
{
"tool_name": "get_server_env_details",
"arguments": {}
}
```
#### `request_server_shutdown`
Requests the MCP server to shut down gracefully.
**Parameters:** None
**Returns:**
A string confirming that the shutdown has been initiated.
**Example Call:**
```json
{
"tool_name": "request_server_shutdown",
"arguments": {}
}
```
---
## CLI Client (`log-analyzer`)
The `log-analyzer` command-line interface provides access to log searching functionalities.
### Global Options
These options apply to the main `log-analyzer` command and are available before specifying a sub-command.
| Option | Argument Type | Description |
|---------------------|---------------|-----------------------------------------------|
| `-h`, `--help` | | Show help message and exit. |
| `--env-file` | PATH | Path to a custom `.env` file for configuration. |
### Search Commands (`log-analyzer search`)
Base command: `log-analyzer search [OPTIONS] COMMAND [ARGS]...`
#### Common Search Options
These options can be used with `all`, `time`, `first`, and `last` search commands.
| Option | Alias | Type | Default | Description |
|------------------------------------|----------|---------|-----------|------------------------------------------------------------------------------------------------------------|
| `--scope` | | STRING | "default" | Logging scope to search within (from .env or default). |
| `--before` | | INTEGER | `2` | Number of context lines before a match. |
| `--after` | | INTEGER | `2` | Number of context lines after a match. |
| `--log-dirs` | | STRING | `None` | Comma-separated list of log directories, files, or glob patterns to search (overrides .env for file locations).|
| `--log-patterns` | | STRING | `None` | Comma-separated list of REGEX patterns to filter log messages (overrides .env content filters). |
#### `log-analyzer search all`
Searches for all log records matching configured patterns.
Usage: `log-analyzer search all [COMMON_SEARCH_OPTIONS]`
**Example:**
```shell
log-analyzer search all --scope my_scope --log-patterns "CRITICAL" --before 1 --after 1
```
#### `log-analyzer search time`
Searches logs within a specified time window.
Usage: `log-analyzer search time [TIME_OPTIONS] [COMMON_SEARCH_OPTIONS]`
**Time Options:**
| Option | Type | Default | Description |
|-------------|---------|---------|--------------------------------------------|
| `--minutes` | INTEGER | `0` | Search logs from the last N minutes. |
| `--hours` | INTEGER | `0` | Search logs from the last N hours. |
| `--days` | INTEGER | `0` | Search logs from the last N days. |
**Example:**
```shell
log-analyzer search time --hours 1 --log-dirs "/var/log/app.log"
```
#### `log-analyzer search first`
Searches for the first N (oldest) matching log records.
Usage: `log-analyzer search first --count INTEGER [COMMON_SEARCH_OPTIONS]`
**Required Option:**
| Option | Type | Description |
|-----------|---------|---------------------------------------------------------------|
| `--count` | INTEGER | Number of first (oldest) matching records to return. |
**Example:**
```shell
log-analyzer search first --count 5 --scope important_logs
```
#### `log-analyzer search last`
Searches for the last N (newest) matching log records.
Usage: `log-analyzer search last --count INTEGER [COMMON_SEARCH_OPTIONS]`
**Required Option:**
| Option | Type | Description |
|-----------|---------|--------------------------------------------------------------|
| `--count` | INTEGER | Number of last (newest) matching records to return. |
**Example:**
```shell
log-analyzer search last --count 20
```
---
## Error Handling
- **MCP Server:** Errors are returned as JSON objects with `code` and `message` fields, conforming to MCP error standards.
- **CLI Client:** Errors are typically printed to stderr.
Common error types include invalid parameters, file not found, or issues with the underlying `AnalysisEngine` configuration or execution.
```
--------------------------------------------------------------------------------
/src/log_analyzer_mcp/common/logger_setup.py:
--------------------------------------------------------------------------------
```python
"""
Logging utility for standardized log setup across all agents
"""
import logging
import os
import re
import sys
from logging.handlers import RotatingFileHandler
from typing import Any, Dict, Literal, Optional
# Explicitly attempt to initialize coverage for subprocesses
# if "COVERAGE_PROCESS_START" in os.environ:
# try:
# import coverage
#
# coverage.process_startup()
# except Exception: # nosec B110 # pylint: disable=broad-exception-caught
# pass # Or handle error if coverage is mandatory
# Determine the project root directory from the location of this script
# Expected structure: /project_root/src/log_analyzer_mcp/common/logger_setup.py
# _common_dir = os.path.dirname(os.path.abspath(__file__))
# _log_analyzer_mcp_dir = os.path.dirname(_common_dir)
# _src_dir = os.path.dirname(_log_analyzer_mcp_dir)
# PROJECT_ROOT = os.path.dirname(_src_dir) # Old method
def find_project_root(start_path: str, marker_file: str = "pyproject.toml") -> str:
"""Searches upwards from start_path for a directory containing marker_file."""
current_path = os.path.abspath(start_path)
while True:
if os.path.exists(os.path.join(current_path, marker_file)):
return current_path
parent_path = os.path.dirname(current_path)
if parent_path == current_path: # Reached filesystem root
# If marker not found, CWD is the best guess for project root.
cwd = os.getcwd()
sys.stderr.write(f"Warning: '{marker_file}' not found from '{start_path}'. Falling back to CWD: {cwd}\\n")
return cwd
current_path = parent_path
PROJECT_ROOT = find_project_root(os.getcwd())
# Define the base logs directory at the project root
LOGS_BASE_DIR = os.path.join(PROJECT_ROOT, "logs")
def get_logs_dir() -> str:
"""Returns the absolute path to the base logs directory for the project."""
# Ensure the base logs directory exists
if not os.path.exists(LOGS_BASE_DIR):
try:
os.makedirs(LOGS_BASE_DIR, exist_ok=True)
except OSError as e:
# Fallback or error if cannot create logs dir, though basic logging might still work to console
sys.stderr.write(f"Warning: Could not create base logs directory {LOGS_BASE_DIR}: {e}\n")
# As a last resort, can try to use a local logs dir if in a restricted env
# For now, we assume it can be created or will be handled by calling code.
return LOGS_BASE_DIR
class MessageFlowFormatter(logging.Formatter):
"""
Custom formatter that recognizes message flow patterns and formats them accordingly
"""
# Pattern to match "sender => receiver | message" format
FLOW_PATTERN = re.compile(r"^(\w+) => (\w+) \| (.*)$")
# Pattern to match already formatted messages (both standard and flow formats)
# This includes timestamp pattern \d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}
# and agent | timestamp format
ALREADY_FORMATTED_PATTERN = re.compile(
r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}|^\w+ \| \d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3})"
)
def __init__(
self,
agent_name: str,
fmt: Optional[str] = None,
datefmt: Optional[str] = None,
style: Literal["%", "{", "$"] = "%",
session_id: Optional[str] = None,
preserve_newlines: bool = True,
):
"""
Initialize the formatter with the agent name
Args:
agent_name: Name of the agent (used when no flow information is in the message)
fmt: Format string
datefmt: Date format string
style: Style of format string
session_id: Optional unique session ID to include in log messages
preserve_newlines: Whether to preserve newlines in the original message
"""
super().__init__(fmt, datefmt, style)
self.agent_name = agent_name
self.session_id = session_id
self.preserve_newlines = preserve_newlines
def format(self, record: logging.LogRecord) -> str:
"""
Format the log record according to message flow patterns
Args:
record: The log record to format
Returns:
Formatted log string
"""
# Extract the message
original_message = record.getMessage()
# Special case for test summary format (always preserve exact format)
if "Test Summary:" in original_message or "===" in original_message:
# Special case for test analyzer compatibility - don't prepend anything
return original_message
# Guard against already formatted messages to prevent recursive formatting
# Check for timestamp pattern to identify already formatted messages
if self.ALREADY_FORMATTED_PATTERN.search(original_message):
# Log message is already formatted, return as is
return original_message
# Check if this is a message flow log
flow_match = self.FLOW_PATTERN.match(original_message)
if flow_match:
sender, receiver, message = flow_match.groups()
timestamp = self.formatTime(record, self.datefmt)
if self.session_id:
formatted_message = f"{receiver} | {timestamp} | {self.session_id} | {sender} => {receiver} | {message}"
else:
formatted_message = f"{receiver} | {timestamp} | {sender} => {receiver} | {message}"
else:
timestamp = self.formatTime(record, self.datefmt)
if self.preserve_newlines:
# Preserve newlines: if newlines are present, split and format first line, append rest
if "\\n" in original_message:
lines = original_message.split("\\n")
if self.session_id:
first_line = f"{self.agent_name} | {timestamp} | {self.session_id} | {lines[0]}"
else:
first_line = f"{self.agent_name} | {timestamp} | {lines[0]}"
formatted_message = first_line + "\\n" + "\\n".join(lines[1:])
else: # No newlines, format as single line
if self.session_id:
formatted_message = f"{self.agent_name} | {timestamp} | {self.session_id} | {original_message}"
else:
formatted_message = f"{self.agent_name} | {timestamp} | {original_message}"
else: # Not preserving newlines (preserve_newlines is False)
# Unconditionally replace newlines with spaces and format as a single line
processed_message = original_message.replace("\n", " ") # Replace actual newlines
processed_message = processed_message.replace("\\n", " ") # Also replace literal \\n, just in case
if self.session_id:
formatted_message = f"{self.agent_name} | {timestamp} | {self.session_id} | {processed_message}"
else:
formatted_message = f"{self.agent_name} | {timestamp} | {processed_message}"
record.msg = formatted_message
record.args = ()
# Return the formatted message
return formatted_message
class LoggerSetup:
"""
Utility class for standardized logging setup across all agents
"""
# Keep the old format for backward compatibility
LEGACY_LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
DEFAULT_LOG_LEVEL = "INFO"
# Store active loggers for management
_active_loggers: Dict[str, logging.Logger] = {}
@classmethod
def _clear_and_close_handlers(cls, logger: logging.Logger) -> None:
"""Helper to clear and close all handlers for a given logger."""
if logger.handlers:
for handler in list(logger.handlers): # Iterate over a copy
try:
handler.flush()
is_default_stream = False
if isinstance(handler, logging.StreamHandler):
stream = getattr(handler, "stream", None)
if stream is sys.stdout or stream is sys.stderr:
is_default_stream = True
# Check stream is not None and has fileno before comparing
if stream is not None and hasattr(stream, "fileno"):
# Also check sys.__stdout__ and sys.__stderr__ for None and fileno for completeness
if (
sys.__stdout__ is not None
and hasattr(sys.__stdout__, "fileno")
and stream is sys.stdout
):
if stream.fileno() != sys.__stdout__.fileno():
is_default_stream = False
if (
sys.__stderr__ is not None
and hasattr(sys.__stderr__, "fileno")
and stream is sys.stderr
):
if stream.fileno() != sys.__stderr__.fileno():
is_default_stream = False
if hasattr(handler, "close"):
if not (is_default_stream and not isinstance(handler, logging.FileHandler)):
try:
handler.close()
except Exception: # Broad catch for mocks or unusual states during close
pass
except ValueError:
pass # Handler already closed or removed
except Exception as e:
sys.stderr.write(f"Warning: Error during handler cleanup for {handler}: {e}\n")
logger.removeHandler(handler)
@classmethod
def get_logger(cls, name: str) -> Optional[logging.Logger]:
"""Retrieve an existing logger by name if it has been created."""
return cls._active_loggers.get(name)
@classmethod
def create_logger(
cls,
name: str,
log_file: Optional[str] = None,
agent_name: Optional[str] = None,
log_level: Optional[str] = None,
session_id: Optional[str] = None,
use_rotating_file: bool = True,
append_mode: bool = True,
preserve_test_format: bool = False,
) -> logging.Logger:
"""
Creates and configures a logger with the given name
Args:
name: Name of the logger
log_file: Optional file path for file logging. If just a filename is provided, it will be created in the centralized logs directory
agent_name: Optional agent name for message flow formatting (defaults to name)
log_level: Optional log level (defaults to environment variable or INFO)
session_id: Optional unique session ID to include in all log messages
use_rotating_file: Whether to use RotatingFileHandler (True) or simple FileHandler (False)
append_mode: Whether to append to existing log file (True) or overwrite (False)
preserve_test_format: Whether to preserve exact format of test-related messages
Returns:
Configured logger instance
"""
# Get log level from parameter, environment, or use default
log_level_str = log_level or os.getenv("LOG_LEVEL", cls.DEFAULT_LOG_LEVEL)
assert isinstance(log_level_str, str)
log_level_str = log_level_str.upper()
log_level_num = getattr(logging, log_level_str, logging.INFO)
# Use agent_name if provided, otherwise use the logger name
if agent_name:
actual_agent_name = agent_name
else:
base_name = name.lower()
if "logger" in base_name:
base_name = base_name.replace("logger", "")
if "agent" in base_name:
base_name = base_name.replace("agent", "")
base_name = base_name.strip("_") # Clean up dangling underscores
if not base_name: # if name was 'AgentLogger' or similar
actual_agent_name = "default_agent"
else:
actual_agent_name = f"{base_name}_agent"
# Create or get existing logger
logger = logging.getLogger(name)
logger.setLevel(log_level_num)
# Disable propagation to root logger to prevent duplicate logs
logger.propagate = False
# Clear existing handlers to avoid duplicates. This is crucial for overwrite mode.
# This must happen BEFORE adding new handlers, especially file handler in 'w' mode.
if name in cls._active_loggers:
# If logger exists in our tracking, it might have handlers we set up.
# We use the same logger instance, so clear its handlers.
cls._clear_and_close_handlers(logger) # logger is cls._active_loggers[name]
elif logger.hasHandlers():
# If not in _active_loggers but has handlers, it was configured elsewhere or is a pre-existing logger (e.g. root)
# Still important to clear to avoid duplication if we are taking it over.
cls._clear_and_close_handlers(logger)
# Console Handler
console_formatter: logging.Formatter
if preserve_test_format:
# For test summaries, use standard formatter on console as well
# to avoid double formatting or MessageFlowFormatter specific handling
console_formatter = logging.Formatter(cls.LEGACY_LOG_FORMAT)
else:
console_formatter = MessageFlowFormatter(
actual_agent_name,
session_id=session_id,
preserve_newlines=not preserve_test_format, # If preserving test format, don't preserve newlines for flow
)
console_handler = logging.StreamHandler(sys.stdout) # Use stdout for informational, stderr for errors
console_handler.setLevel(log_level_num)
console_handler.setFormatter(console_formatter)
logger.addHandler(console_handler)
# File Handler
if log_file:
# Determine the log file path
if os.path.isabs(log_file):
log_file_path = log_file
else:
log_file_path = os.path.join(get_logs_dir(), log_file)
# Ensure log directory exists
log_dir = os.path.dirname(log_file_path)
if not os.path.exists(log_dir):
try:
os.makedirs(log_dir, exist_ok=True)
except OSError as e:
sys.stderr.write(f"ERROR: Could not create log directory {log_dir}: {e}. File logging disabled.\\n")
log_file = None # Disable file logging
if log_file: # Check again, might have been disabled
file_mode = "w" if not append_mode else "a"
file_formatter: logging.Formatter
if preserve_test_format:
# Use a basic formatter for test log files to keep them clean
file_formatter = logging.Formatter("%(message)s")
else:
file_formatter = MessageFlowFormatter(
actual_agent_name,
session_id=session_id,
preserve_newlines=True, # Always preserve newlines in file logs unless test format
)
if use_rotating_file:
file_handler: logging.Handler = RotatingFileHandler(
log_file_path, maxBytes=10 * 1024 * 1024, backupCount=5, mode=file_mode, encoding="utf-8"
)
else:
file_handler = logging.FileHandler(log_file_path, mode=file_mode, encoding="utf-8")
file_handler.setFormatter(file_formatter)
file_handler.setLevel(log_level_num)
logger.addHandler(file_handler)
# Log file configuration message to the logger itself
logger.info(f"File logging configured for: {log_file_path}")
cls._active_loggers[name] = logger
return logger
@classmethod
def flush_all_loggers(cls) -> None:
"""Flushes all registered active loggers."""
for logger_instance in cls._active_loggers.values():
for handler in logger_instance.handlers:
handler.flush()
@classmethod
def flush_logger(cls, name: str) -> bool:
"""
Flush a specific logger by name
Args:
name: Name of the logger to flush
Returns:
True if logger was found and flushed, False otherwise
"""
if name in cls._active_loggers:
logger = cls._active_loggers[name]
for handler in logger.handlers:
handler.flush()
return True
return False
@classmethod
def write_test_summary(cls, logger: logging.Logger, summary: Dict[str, Any]) -> None:
"""
Write test summary in a format that log_analyzer.py can understand
Args:
logger: The logger to use
summary: Dictionary with test summary information
"""
# Flush any pending logs
for handler in logger.handlers:
handler.flush()
# Log summary in a format compatible with log_analyzer.py
logger.info("=" * 15 + " test session starts " + "=" * 15)
# Log test result counts
passed = summary.get("passed", 0)
failed = summary.get("failed", 0)
skipped = summary.get("skipped", 0)
duration = summary.get("duration", 0)
logger.info(f"{passed} passed, {failed} failed, {skipped} skipped in {duration:.2f}s")
logger.info(f"Test Summary: {passed} passed, {failed} failed, {skipped} skipped")
logger.info(f"Status: {'PASSED' if failed == 0 else 'FAILED'}")
logger.info(f"Duration: {duration:.2f} seconds")
# Log failed tests if any
if "failed_tests" in summary and summary["failed_tests"]:
logger.info("Failed tests by module:")
for module, tests in summary.get("failed_modules", {}).items():
logger.info(f"Module: {module} - {len(tests)} failed tests")
for test in tests:
logger.info(f"- {test}")
logger.info("=" * 50)
# Ensure everything is written
for handler in logger.handlers:
handler.flush()
@classmethod
def reset_loggers_for_testing(cls) -> None:
"""Resets all known loggers by clearing their handlers. Useful for testing."""
for logger_name in list(cls._active_loggers.keys()):
logger = cls._active_loggers.pop(logger_name)
cls._clear_and_close_handlers(logger)
# Also clear the root logger's handlers if any were added inadvertently by tests
root_logger = logging.getLogger()
cls._clear_and_close_handlers(root_logger)
def setup_logger(
agent_name: str,
log_level: str = "INFO",
session_id: Optional[str] = None,
log_file: Optional[str] = None,
use_rotating_file: bool = True,
) -> logging.Logger:
"""
Set up a logger with the given name and log level
Args:
agent_name: Name of the agent
log_level: Log level (default: INFO)
session_id: Optional unique session ID to include in all log messages
log_file: Optional file path for logging
use_rotating_file: Whether to use rotating file handler (default: True)
Returns:
Configured logger
"""
# Use the LoggerSetup class for consistent logging setup
return LoggerSetup.create_logger(
agent_name,
log_file=log_file,
agent_name=agent_name,
log_level=log_level,
session_id=session_id,
use_rotating_file=use_rotating_file,
)
```
--------------------------------------------------------------------------------
/tests/log_analyzer_mcp/common/test_logger_setup.py:
--------------------------------------------------------------------------------
```python
import logging
import os
import shutil
import sys
import tempfile
from unittest.mock import MagicMock, mock_open, patch
import pytest
from log_analyzer_mcp.common.logger_setup import (
LOGS_BASE_DIR,
PROJECT_ROOT,
LoggerSetup,
MessageFlowFormatter,
find_project_root,
get_logs_dir,
setup_logger,
)
from logging import handlers
# Helper to reset LoggerSetup._active_loggers for test isolation
@pytest.fixture(autouse=True)
def reset_active_loggers():
LoggerSetup.reset_loggers_for_testing() # Use the new robust reset method
yield
LoggerSetup.reset_loggers_for_testing() # Ensure clean state after each test
# --- Tests for find_project_root ---
def test_find_project_root_fallback(tmp_path):
"""Test find_project_root fallback when marker_file is not found."""
# Ensure no pyproject.toml is found upwards from tmp_path
# This test relies on the fallback logic calculating from __file__ of logger_setup.py
# We can't easily mock the entire filesystem structure up to root for this specific test.
# Instead, we'll check if it returns *a* path and doesn't crash.
# A more robust test would involve creating a known deep structure without the marker.
# The new fallback is os.getcwd(), so we can check against that.
expected_fallback_root = os.getcwd()
# To simulate not finding it, we pass a non-existent marker and start path far from project
# This forces it to go up to the filesystem root and trigger the fallback.
# We need to be careful as the real PROJECT_ROOT might interfere.
# Let's patch os.path.exists to simulate marker not being found
with patch("os.path.exists", return_value=False) as mock_exists:
# And patch abspath for the __file__ to be consistent if needed, though usually not.
# Call from a deep, unrelated path
unrelated_deep_path = tmp_path / "a" / "b" / "c" / "d" / "e"
unrelated_deep_path.mkdir(parents=True, exist_ok=True)
# Use a marker that definitely won't exist to force fallback
calculated_root = find_project_root(str(unrelated_deep_path), "THIS_MARKER_DOES_NOT_EXIST.txt")
# The fallback is now os.getcwd()
assert calculated_root == expected_fallback_root
# Ensure os.path.exists was called multiple times during the upward search
assert mock_exists.call_count > 1
# --- Tests for get_logs_dir ---
def test_get_logs_dir_exists(tmp_path):
"""Test get_logs_dir when the directory already exists."""
# Use a temporary logs base dir for this test
temp_logs_base = tmp_path / "test_logs"
temp_logs_base.mkdir()
with patch("log_analyzer_mcp.common.logger_setup.LOGS_BASE_DIR", str(temp_logs_base)):
assert get_logs_dir() == str(temp_logs_base)
assert temp_logs_base.exists()
def test_get_logs_dir_creates_if_not_exists(tmp_path):
"""Test get_logs_dir creates the directory if it doesn't exist."""
temp_logs_base = tmp_path / "test_logs_new"
with patch("log_analyzer_mcp.common.logger_setup.LOGS_BASE_DIR", str(temp_logs_base)):
assert not temp_logs_base.exists()
assert get_logs_dir() == str(temp_logs_base)
assert temp_logs_base.exists()
@patch("os.makedirs")
def test_get_logs_dir_os_error_on_create(mock_makedirs, tmp_path, capsys):
"""Test get_logs_dir when os.makedirs raises an OSError."""
mock_makedirs.side_effect = OSError("Test OS error")
temp_logs_base = tmp_path / "test_logs_error"
with patch("log_analyzer_mcp.common.logger_setup.LOGS_BASE_DIR", str(temp_logs_base)):
assert get_logs_dir() == str(temp_logs_base) # Should still return the path
# Check stderr for the warning
captured = capsys.readouterr()
assert f"Warning: Could not create base logs directory {str(temp_logs_base)}" in captured.err
# --- Tests for MessageFlowFormatter ---
@pytest.fixture
def mock_log_record():
record = MagicMock(spec=logging.LogRecord)
record.getMessage = MagicMock(return_value="A normal log message")
record.levelno = logging.INFO
record.levelname = "INFO"
record.created = 1678886400 # A fixed time
record.msecs = 123
record.name = "TestLogger"
record.args = () # Ensure args is an empty tuple
record.exc_info = None # Add exc_info attribute
record.exc_text = None # Add exc_text attribute
record.stack_info = None # Add stack_info attribute
return record
def test_message_flow_formatter_standard_message(mock_log_record):
formatter = MessageFlowFormatter("TestAgent")
formatted = formatter.format(mock_log_record)
assert "TestAgent |" in formatted
assert "| A normal log message" in formatted
def test_message_flow_formatter_with_session_id(mock_log_record):
formatter = MessageFlowFormatter("TestAgent", session_id="sess123")
mock_log_record.getMessage.return_value = "Another message"
formatted = formatter.format(mock_log_record)
assert "TestAgent |" in formatted
assert "| sess123 |" in formatted
assert "| Another message" in formatted
def test_message_flow_formatter_flow_pattern(mock_log_record):
formatter = MessageFlowFormatter("ReceiverAgent", session_id="s456")
mock_log_record.getMessage.return_value = "SenderAgent => ReceiverAgent | Flow details here"
formatted = formatter.format(mock_log_record)
# Receiver | Timestamp | SessionID | Sender => Receiver | Message
assert "ReceiverAgent |" in formatted # Receiver is the first part
assert "| s456 |" in formatted # Session ID
assert "| SenderAgent => ReceiverAgent | Flow details here" in formatted # The original flow part
assert "ReceiverAgent => ReceiverAgent" not in formatted # Ensure agent_name not misused
def test_message_flow_formatter_already_formatted(mock_log_record):
formatter = MessageFlowFormatter("TestAgent")
# Simulate an already formatted message (e.g., from a different handler)
already_formatted_msg = "2023-03-15 10:00:00,123 - TestAgent - INFO - Already done"
mock_log_record.getMessage.return_value = already_formatted_msg
formatted = formatter.format(mock_log_record)
assert formatted == already_formatted_msg
already_formatted_flow_msg = "OtherAgent | 2023-03-15 10:00:00,123 | SomeSender => OtherAgent | Done this way"
mock_log_record.getMessage.return_value = already_formatted_flow_msg
formatted = formatter.format(mock_log_record)
assert formatted == already_formatted_flow_msg
def test_message_flow_formatter_test_summary(mock_log_record):
formatter = MessageFlowFormatter("TestAgent")
test_summary_msg = "Test Summary: 5 passed, 0 failed"
mock_log_record.getMessage.return_value = test_summary_msg
formatted = formatter.format(mock_log_record)
assert formatted == test_summary_msg # Should be returned as-is
pytest_output_msg = "============================= test session starts =============================="
mock_log_record.getMessage.return_value = pytest_output_msg
formatted = formatter.format(mock_log_record)
assert formatted == pytest_output_msg # Should also be returned as-is
def test_message_flow_formatter_multiline(mock_log_record):
formatter = MessageFlowFormatter("TestAgent", session_id="multi789")
multiline_msg = "First line\nSecond line\nThird line"
mock_log_record.getMessage.return_value = multiline_msg
formatted = formatter.format(mock_log_record)
lines = formatted.split("\n")
assert len(lines) == 3
assert "TestAgent |" in lines[0]
assert "| multi789 |" in lines[0]
assert "| First line" in lines[0]
assert lines[1] == "Second line"
assert lines[2] == "Third line"
def test_message_flow_formatter_no_preserve_newlines(mock_log_record):
formatter = MessageFlowFormatter("TestAgent", preserve_newlines=False)
# Use an actual newline character in the message, not a literal '\\n' string
multiline_msg = "First line\nSecond line"
mock_log_record.getMessage.return_value = multiline_msg
formatted = formatter.format(mock_log_record)
# When not preserving, it should format the whole thing as one line (newlines replaced by \n in record.msg)
# The format method does `record.msg = formatted_message` then `super().format(record)` would be called.
# Our current implementation returns the formatted string directly, so it won't go to super().format.
# It handles multiline splitting itself. If preserve_newlines is false, it just formats original_message
# as a single line.
assert "\\n" not in formatted # The formatted output string should not contain raw newlines
assert "TestAgent |" in formatted
# The expected behavior now is that newlines are removed and the message is on a single line.
assert "| First line Second line" in formatted # Adjusted expectation: newlines replaced by space
# --- Tests for LoggerSetup ---
@pytest.fixture
def temp_log_file(tmp_path):
log_file = tmp_path / "test.log"
yield str(log_file)
if log_file.exists():
log_file.unlink()
def test_logger_setup_create_logger_basic(temp_log_file):
logger = LoggerSetup.create_logger("MyLogger", log_file=temp_log_file, agent_name="MyAgent")
assert logger.name == "MyLogger"
assert logger.level == logging.INFO # Default
assert len(logger.handlers) == 2 # Console and File
assert isinstance(logger.handlers[0], logging.StreamHandler) # Console
assert isinstance(logger.handlers[1], handlers.RotatingFileHandler) # File
# Check if formatter is MessageFlowFormatter
for handler in logger.handlers:
assert isinstance(handler.formatter, MessageFlowFormatter)
if isinstance(handler, logging.FileHandler): # Check path of file handler
assert handler.baseFilename == temp_log_file
# Check if it's stored
assert LoggerSetup.get_logger("MyLogger") is logger
def test_logger_setup_create_logger_levels(temp_log_file):
logger_debug = LoggerSetup.create_logger("DebugLogger", log_file=temp_log_file, log_level="DEBUG")
assert logger_debug.level == logging.DEBUG
for handler in logger_debug.handlers:
assert handler.level == logging.DEBUG
logger_warning = LoggerSetup.create_logger("WarnLogger", log_file=temp_log_file, log_level="WARNING")
assert logger_warning.level == logging.WARNING
def test_logger_setup_create_logger_no_file():
logger = LoggerSetup.create_logger("NoFileLogger")
assert len(logger.handlers) == 1 # Only console
assert isinstance(logger.handlers[0], logging.StreamHandler)
def test_logger_setup_create_logger_agent_name(temp_log_file):
logger = LoggerSetup.create_logger("AgentLoggerTest", log_file=temp_log_file, agent_name="SpecificAgent")
console_formatter = logger.handlers[0].formatter
assert isinstance(console_formatter, MessageFlowFormatter)
assert console_formatter.agent_name == "SpecificAgent"
# Test default agent_name derivation
logger_default_agent = LoggerSetup.create_logger("MyAgentLogger", log_file=temp_log_file)
default_agent_formatter = logger_default_agent.handlers[0].formatter
assert isinstance(default_agent_formatter, MessageFlowFormatter)
assert default_agent_formatter.agent_name == "my_agent" # MyAgentLogger -> my_agent
logger_simple_name = LoggerSetup.create_logger("MyLogger", log_file=temp_log_file)
simple_name_formatter = logger_simple_name.handlers[0].formatter
assert isinstance(simple_name_formatter, MessageFlowFormatter)
assert simple_name_formatter.agent_name == "my_agent" # MyLogger -> my_agent
logger_just_agent = LoggerSetup.create_logger("Agent", log_file=temp_log_file)
just_agent_formatter = logger_just_agent.handlers[0].formatter
assert isinstance(just_agent_formatter, MessageFlowFormatter)
assert just_agent_formatter.agent_name == "default_agent" # Agent -> default_agent
logger_empty_derivation = LoggerSetup.create_logger("AgentLogger", log_file=temp_log_file)
empty_deriv_formatter = logger_empty_derivation.handlers[0].formatter
assert isinstance(empty_deriv_formatter, MessageFlowFormatter)
assert empty_deriv_formatter.agent_name == "default_agent" # AgentLogger -> default_agent
def test_logger_setup_create_logger_session_id(temp_log_file):
logger = LoggerSetup.create_logger("SessionLogger", log_file=temp_log_file, session_id="sessABC")
formatter = logger.handlers[0].formatter
assert isinstance(formatter, MessageFlowFormatter)
assert formatter.session_id == "sessABC"
def test_logger_setup_create_logger_no_rotating_file(temp_log_file):
logger = LoggerSetup.create_logger("SimpleFileLogger", log_file=temp_log_file, use_rotating_file=False)
assert isinstance(logger.handlers[1], logging.FileHandler)
assert not isinstance(logger.handlers[1], handlers.RotatingFileHandler)
def test_logger_setup_create_logger_overwrite_mode(tmp_path):
log_file_overwrite = tmp_path / "overwrite.log"
log_file_overwrite.write_text("Previous content\n")
# Create logger in append mode (default)
logger_append = LoggerSetup.create_logger("AppendLogger", log_file=str(log_file_overwrite), use_rotating_file=False)
logger_append.warning("Append test")
LoggerSetup.flush_logger("AppendLogger") # Ensure written
# Create logger in overwrite mode, ensure non-rotating for this specific test of 'w' mode.
logger_overwrite = LoggerSetup.create_logger(
"OverwriteLogger",
log_file=str(log_file_overwrite),
append_mode=False,
use_rotating_file=False, # Use simple FileHandler to test 'w' mode directly
)
logger_overwrite.error("Overwrite test")
LoggerSetup.flush_logger("OverwriteLogger") # Ensure written
content = log_file_overwrite.read_text()
assert "Previous content" not in content
assert "Append test" not in content
assert "Overwrite test" in content
assert "overwrite_agent |" in content # agent name will be derived
def test_logger_setup_create_logger_preserve_test_format(temp_log_file, mock_log_record):
logger = LoggerSetup.create_logger("TestFormatLogger", log_file=temp_log_file, preserve_test_format=True)
file_handler = logger.handlers[1] # File handler
assert isinstance(file_handler.formatter, logging.Formatter) # Plain Formatter
assert not isinstance(file_handler.formatter, MessageFlowFormatter)
# Console handler should use standard Formatter when preserve_test_format is True
console_handler = logger.handlers[0]
assert isinstance(console_handler.formatter, logging.Formatter)
assert not isinstance(
console_handler.formatter, MessageFlowFormatter
) # Explicitly check it's NOT MessageFlowFormatter
# Test logging a test summary line
test_summary_msg = "Test Summary: 1 passed"
mock_log_record.getMessage.return_value = test_summary_msg
# File handler with simple formatter should just output the message
formatted_file = file_handler.formatter.format(mock_log_record)
assert formatted_file == test_summary_msg
# Console handler (standard Formatter) when preserve_test_format=True
# should output using LEGACY_LOG_FORMAT.
console_handler_formatter = console_handler.formatter
expected_console_output = console_handler_formatter.format(mock_log_record) # Format with the actual formatter
formatted_console = console_handler.formatter.format(mock_log_record)
assert formatted_console == expected_console_output
assert "Test Summary: 1 passed" in formatted_console # Check if the message is part of it
assert mock_log_record.name in formatted_console # e.g. TestLogger
assert mock_log_record.levelname in formatted_console # e.g. INFO
@patch("os.makedirs")
def test_logger_setup_create_logger_log_dir_creation_failure(mock_makedirs, tmp_path, capsys):
mock_makedirs.side_effect = OSError("Cannot create dir")
# Use a log file path that would require directory creation
log_file_in_new_dir = tmp_path / "new_log_subdir" / "error.log"
logger = LoggerSetup.create_logger("ErrorDirLogger", log_file=str(log_file_in_new_dir))
# Should have only console handler if file dir creation failed
assert len(logger.handlers) == 1
assert isinstance(logger.handlers[0], logging.StreamHandler)
captured = capsys.readouterr()
expected_dir = str(tmp_path / "new_log_subdir")
assert f"ERROR: Could not create log directory {expected_dir}" in captured.err
assert mock_makedirs.call_count == 1 # Should have attempted to create it
def test_logger_setup_clear_handlers_on_recreate(temp_log_file):
logger1 = LoggerSetup.create_logger("RecreateTest", log_file=temp_log_file)
assert len(logger1.handlers) == 2
# Get the actual underlying logger instance
underlying_logger = logging.getLogger("RecreateTest")
assert len(underlying_logger.handlers) == 2
logger2 = LoggerSetup.create_logger("RecreateTest", log_file=temp_log_file, log_level="DEBUG")
assert logger2 is logger1 # Should be the same logger object
assert len(logger2.handlers) == 2 # Handlers should be replaced, not added
assert len(underlying_logger.handlers) == 2
def test_logger_setup_flush_logger(temp_log_file):
logger = LoggerSetup.create_logger("FlushTest", log_file=temp_log_file)
mock_handler = MagicMock(spec=logging.Handler)
mock_handler.flush = MagicMock()
# Replace handlers for testing flush
original_handlers = list(logger.handlers) # Keep a copy
logger.handlers = [mock_handler]
assert LoggerSetup.flush_logger("FlushTest") is True
mock_handler.flush.assert_called_once()
logger.handlers = original_handlers # Restore original handlers
# Ensure original handlers are closed if they were file handlers, to avoid ResourceWarning
for handler in original_handlers:
if isinstance(handler, logging.FileHandler):
handler.close()
assert LoggerSetup.flush_logger("NonExistentLogger") is False
def test_logger_setup_flush_all_loggers(temp_log_file):
logger_a = LoggerSetup.create_logger("FlushAllA", log_file=temp_log_file)
logger_b = LoggerSetup.create_logger("FlushAllB", log_file=None) # Console only
# Before replacing logger_a's handlers with mocks, clear its existing (real) handlers
# to ensure its file handler is properly closed.
LoggerSetup._clear_and_close_handlers(logger_a)
mock_handler_a_file = MagicMock(spec=logging.FileHandler)
mock_handler_a_file.flush = MagicMock()
mock_handler_a_console = MagicMock(spec=logging.StreamHandler)
mock_handler_a_console.flush = MagicMock()
# Simulate stream attribute for StreamHandler mocks if _clear_and_close_handlers might access it
# However, the refined _clear_and_close_handlers uses getattr(handler, 'stream', None)
# so this might not be strictly necessary unless we want to test specific stream interactions.
# mock_handler_a_console.stream = sys.stdout
# Ensure logger_a uses these mocked handlers
logger_a.handlers = [mock_handler_a_console, mock_handler_a_file]
mock_handler_b_console = MagicMock(spec=logging.StreamHandler)
mock_handler_b_console.flush = MagicMock()
# mock_handler_b_console.stream = sys.stdout
logger_b.handlers = [mock_handler_b_console]
LoggerSetup.flush_all_loggers()
mock_handler_a_file.flush.assert_called_once()
mock_handler_a_console.flush.assert_called_once()
mock_handler_b_console.flush.assert_called_once()
# Clean up / close handlers to avoid ResourceWarning
# This is a bit tricky because flush_all_loggers doesn't return the loggers
# We rely on the autouse fixture to clear _active_loggers, which should lead to
# handlers being closed eventually if create_logger handles it well on re-creation.
# For more direct control in this specific test, we would need to access
# LoggerSetup._active_loggers, which is an internal detail.
# However, the fix in create_logger to close handlers should mitigate this.
# The new reset_loggers_for_testing in the autouse fixture should handle this.
# LoggerSetup._active_loggers.clear() # No longer needed here due to autouse fixture
def test_logger_setup_write_test_summary(temp_log_file):
logger = LoggerSetup.create_logger("TestSummaryLogger", log_file=temp_log_file, preserve_test_format=True)
# Mock the file handler to capture output
mock_file_handler_write = mock_open()
# Find the file handler and patch its write method
original_file_handler = None
for handler in logger.handlers:
if isinstance(handler, logging.FileHandler):
original_file_handler = handler
break
if original_file_handler:
# To capture output from file handler, we can check the file content
# or mock its stream's write method. Checking file content is more robust.
log_file_path = original_file_handler.baseFilename
else:
pytest.fail("File handler not found on TestSummaryLogger")
summary_data = {
"passed": 5,
"failed": 2,
"skipped": 1,
"duration": 1.234,
"failed_tests": ["test_one", "test_two"], # This structure might differ from actual use
"failed_modules": {"moduleA": ["test_one_a"], "moduleB": ["test_two_b"]},
}
LoggerSetup.write_test_summary(logger, summary_data)
LoggerSetup.flush_logger("TestSummaryLogger") # Ensure all written to file
# Read the log file content
with open(log_file_path, "r") as f:
log_content = f.read()
assert "=============== test session starts ===============" in log_content
assert "5 passed, 2 failed, 1 skipped in 1.23s" in log_content
assert "Test Summary: 5 passed, 2 failed, 1 skipped" in log_content
assert "Status: FAILED" in log_content
assert "Duration: 1.23 seconds" in log_content
assert "Failed tests by module:" in log_content
assert "Module: moduleA - 1 failed tests" in log_content
assert "- test_one_a" in log_content
assert "Module: moduleB - 1 failed tests" in log_content
assert "- test_two_b" in log_content
assert "==================================================" in log_content
# --- Tests for setup_logger (convenience function) ---
def test_setup_logger_convenience_function(temp_log_file):
with patch.object(LoggerSetup, "create_logger", wraps=LoggerSetup.create_logger) as mock_create_logger:
logger = setup_logger(
"ConvenienceAgent", log_level="DEBUG", session_id="conv123", log_file=temp_log_file, use_rotating_file=False
)
mock_create_logger.assert_called_once_with(
"ConvenienceAgent", # name
log_file=temp_log_file,
agent_name="ConvenienceAgent", # agent_name also from first arg
log_level="DEBUG",
session_id="conv123",
use_rotating_file=False,
# append_mode and preserve_test_format use defaults from create_logger
)
assert logger.name == "ConvenienceAgent"
assert logger.level == logging.DEBUG
# Test PROJECT_ROOT and LOGS_BASE_DIR for basic correctness
def test_project_root_and_logs_base_dir_paths():
# PROJECT_ROOT should be a valid directory
assert os.path.isdir(PROJECT_ROOT), f"PROJECT_ROOT is not a valid directory: {PROJECT_ROOT}"
# pyproject.toml should exist in PROJECT_ROOT
assert os.path.exists(os.path.join(PROJECT_ROOT, "pyproject.toml")), "pyproject.toml not found in PROJECT_ROOT"
# LOGS_BASE_DIR should also be valid or creatable
assert os.path.isdir(LOGS_BASE_DIR) or not os.path.exists(
LOGS_BASE_DIR
), f"LOGS_BASE_DIR is not valid or creatable: {LOGS_BASE_DIR}"
# LOGS_BASE_DIR should be under PROJECT_ROOT
assert LOGS_BASE_DIR.startswith(PROJECT_ROOT), "LOGS_BASE_DIR is not under PROJECT_ROOT"
# Test get_logs_dir() directly too
retrieved_logs_dir = get_logs_dir()
assert os.path.isdir(retrieved_logs_dir)
assert retrieved_logs_dir == LOGS_BASE_DIR
# --- Tests for find_project_root ---
def test_find_project_root_finds_marker(tmp_path):
"""Test find_project_root when pyproject.toml exists."""
marker_file = "pyproject.toml"
# Create
```
--------------------------------------------------------------------------------
/src/log_analyzer_mcp/core/analysis_engine.py:
--------------------------------------------------------------------------------
```python
# src/log_analyzer_mcp/core/analysis_engine.py
import datetime as dt # Import datetime module as dt
import glob
import os
import re # For basic parsing
from datetime import datetime as DateTimeClassForCheck # Specific import for isinstance check
from typing import Any, Dict, List, Optional # Added Any for filter_criteria flexibility
import logging # Add logging import
from ..common.config_loader import ConfigLoader
# Define a structure for a parsed log entry
# Using a simple dict for now, could be a Pydantic model later for stricter validation
ParsedLogEntry = Dict[str, Any] # Keys: 'timestamp', 'level', 'message', 'raw_line', 'file_path', 'line_number'
# Adding 'context_before_lines', 'context_after_lines' to store context directly in the entry
# And 'full_context_log' which would be the original line plus its context
class AnalysisEngine:
def __init__(
self,
logger_instance: logging.Logger,
env_file_path: Optional[str] = None,
project_root_for_config: Optional[str] = None,
):
self.logger = logger_instance
self.config_loader = ConfigLoader(env_file_path=env_file_path, project_root_for_config=project_root_for_config)
# Load configurations using the correct ConfigLoader methods
self.log_directories: List[str] = self.config_loader.get_log_directories()
self.log_content_patterns: Dict[str, List[str]] = self.config_loader.get_log_patterns()
self.default_context_lines_before: int = self.config_loader.get_context_lines_before()
self.default_context_lines_after: int = self.config_loader.get_context_lines_after()
self.logging_scopes: Dict[str, str] = self.config_loader.get_logging_scopes()
# TODO: Potentially add more sophisticated validation or processing of loaded configs
def _get_target_log_files(
self, scope: Optional[str] = None, log_dirs_override: Optional[List[str]] = None
) -> List[str]:
"""
Determines the list of log files to search.
Uses log_dirs_override if provided, otherwise falls back to scope or general config.
log_dirs_override can contain direct file paths, directory paths, or glob patterns.
If a directory path is provided, it searches for '*.log' files recursively.
"""
self.logger.info(f"[_get_target_log_files] Called with scope: {scope}, override: {log_dirs_override}")
target_paths_or_patterns: List[str] = []
project_root = self.config_loader.project_root
self.logger.info(f"[_get_target_log_files] Project root: {project_root}")
using_override_dirs = False
if log_dirs_override:
self.logger.info(f"[_get_target_log_files] Using log_dirs_override: {log_dirs_override}")
target_paths_or_patterns.extend(log_dirs_override)
using_override_dirs = True
elif scope and scope.lower() in self.logging_scopes:
path_or_pattern = self.logging_scopes[scope.lower()]
self.logger.info(f"[_get_target_log_files] Using scope '{scope}', path_or_pattern: {path_or_pattern}")
abs_scope_path = os.path.abspath(os.path.join(project_root, path_or_pattern))
if not abs_scope_path.startswith(project_root):
self.logger.warning(
f"Scope '{scope}' path '{path_or_pattern}' resolves outside project root. Skipping."
)
return []
target_paths_or_patterns.append(abs_scope_path)
elif scope: # Scope was provided but not found in self.logging_scopes
self.logger.info(
f"[AnalysisEngine] Scope '{scope}' not found in configuration. Returning no files for this scope."
)
return []
else:
self.logger.info(
f"[_get_target_log_files] Using default log_directories from config: {self.log_directories}"
)
for log_dir_pattern in self.log_directories:
abs_log_dir_pattern = os.path.abspath(os.path.join(project_root, log_dir_pattern))
if not abs_log_dir_pattern.startswith(project_root):
self.logger.warning(
f"Log directory pattern '{log_dir_pattern}' resolves outside project root. Skipping."
)
continue
target_paths_or_patterns.append(abs_log_dir_pattern)
self.logger.info(f"[_get_target_log_files] Effective target_paths_or_patterns: {target_paths_or_patterns}")
resolved_files: List[str] = []
for path_or_pattern_input in target_paths_or_patterns:
self.logger.info(f"[_get_target_log_files] Processing input: {path_or_pattern_input}")
if not os.path.isabs(path_or_pattern_input):
current_search_item = os.path.abspath(os.path.join(project_root, path_or_pattern_input))
self.logger.info(
f"[_get_target_log_files] Relative input '{path_or_pattern_input}' made absolute: {current_search_item}"
)
else:
current_search_item = os.path.abspath(path_or_pattern_input)
self.logger.info(
f"[_get_target_log_files] Absolute input '{path_or_pattern_input}' normalized to: {current_search_item}"
)
if not current_search_item.startswith(project_root):
self.logger.warning(
f"[_get_target_log_files] Item '{current_search_item}' is outside project root '{project_root}'. Skipping."
)
continue
self.logger.info(f"[_get_target_log_files] Checking item: {current_search_item}")
if os.path.isfile(current_search_item):
self.logger.info(f"[_get_target_log_files] Item '{current_search_item}' is a file.")
# If current_search_item came from a scope that resolved to a direct file,
# or from an override that was a direct file, include it.
# The `using_override_dirs` flag helps distinguish.
# If it came from a scope, `using_override_dirs` is False.
is_from_scope_direct_file = not using_override_dirs and any(
current_search_item == os.path.abspath(os.path.join(project_root, self.logging_scopes[s_key]))
for s_key in self.logging_scopes
if not glob.has_magic(self.logging_scopes[s_key])
and not os.path.isdir(os.path.join(project_root, self.logging_scopes[s_key]))
)
if using_override_dirs or is_from_scope_direct_file:
resolved_files.append(current_search_item)
elif current_search_item.endswith(".log"): # Default behavior for non-override, non-direct-scope-file
resolved_files.append(current_search_item)
elif os.path.isdir(current_search_item):
# Search for *.log files recursively in the directory
for filepath in glob.glob(
os.path.join(glob.escape(current_search_item), "**", "*.log"), recursive=True
):
if os.path.isfile(filepath) and os.path.abspath(filepath).startswith(
project_root
): # Double check resolved path
resolved_files.append(os.path.abspath(filepath))
else: # Assumed to be a glob pattern
# For glob patterns, ensure they are rooted or handled carefully.
# If an override is a glob like "specific_module/logs/*.log", it should work.
# If it's just "*.log", it will glob from CWD unless we force it relative to project_root.
# The normalization above should handle making it absolute from project_root if it was relative.
# The glob pattern itself (current_search_item) is already an absolute path or made absolute starting from project_root
is_recursive_glob = "**" in path_or_pattern_input # Check original input for "**"
for filepath in glob.glob(current_search_item, recursive=is_recursive_glob):
abs_filepath = os.path.abspath(filepath)
if (
os.path.isfile(abs_filepath)
and abs_filepath.endswith(".log")
and abs_filepath.startswith(project_root)
):
resolved_files.append(abs_filepath)
elif (
os.path.isfile(abs_filepath)
and not abs_filepath.endswith(".log")
and using_override_dirs
and not os.path.isdir(path_or_pattern_input) # Ensure original input wasn't a directory
and (
os.path.splitext(abs_filepath)[1]
in os.path.splitext(current_search_item)[1] # Check if glob was for specific ext
if not glob.has_magic(
current_search_item
) # If current_search_item was specific file (not a glob)
else True # If current_search_item itself was a glob (e.g. *.txt)
)
):
# If using override_dirs and the override was a specific file path (not a pattern or dir) that doesn't end with .log, still include it.
# This was changed above: if os.path.isfile(current_search_item) and using_override_dirs, it's added.
# This elif handles globs from override_dirs that might pick up non-.log files
# if the glob pattern itself was specific (e.g., *.txt)
# The original logic for specific file override (path_or_pattern_input == filepath) was too restrictive.
# current_search_item is the absolute version of path_or_pattern_input.
# abs_filepath is the file found by glob.
# This part needs to correctly identify if a non-.log file found by a glob from an override should be included.
# If the original glob pattern explicitly asked for non-.log (e.g. *.txt), then yes.
# If the glob was generic (e.g. dir/*) and picked up a .txt, then probably no, unless it was the only match for a specific file.
# The current logic seems to have simplified: if os.path.isfile(current_search_item) and using_override_dirs, it adds.
# This new elif is for results from glob.glob(...)
# Let's ensure that if the original path_or_pattern_input (from override) was a glob,
# and that glob resolves to a non-.log file, we include it.
# This means the user explicitly asked for it via a pattern.
if glob.has_magic(path_or_pattern_input) or glob.has_magic(current_search_item):
# If original input or its absolute form was a glob, include what it finds.
resolved_files.append(abs_filepath)
# No 'else' needed here, if it's not a .log and not from an override glob, it's skipped by the main 'if .endswith(".log")'
return sorted(list(set(resolved_files))) # Unique sorted list
def _parse_log_line(self, line: str, file_path: str, line_number: int) -> Optional[ParsedLogEntry]:
"""Parses a single log line. Attempts to match a common log format and falls back gracefully."""
# Regex for "YYYY-MM-DD HH:MM:SS[,ms] LEVEL MESSAGE"
# It captures timestamp, level, and message. Milliseconds are optional.
log_pattern = re.compile(
r"^(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(?:,\d{3})?)\s+"
r"(?P<level>[A-Z]+(?:\s+[A-Z]+)*)\s+" # Allow multi-word levels like 'INFO EXAMPLE'
r"(?P<message>.*)$"
)
match = log_pattern.match(line)
if match:
groups = match.groupdict()
timestamp_str = groups.get("timestamp")
level_str = groups.get("level", "UNKNOWN").upper()
message_str = groups.get("message", "").strip()
parsed_timestamp: Optional[dt.datetime] = None
if timestamp_str:
# self.logger.debug(f"Attempting to parse timestamp string: '{timestamp_str}' from line: '{line.strip()}'") # DEBUG
try:
# Handle optional milliseconds by splitting at comma
timestamp_to_parse = timestamp_str.split(",")[0]
parsed_timestamp = dt.datetime.strptime(timestamp_to_parse, "%Y-%m-%d %H:%M:%S")
except ValueError as e:
self.logger.debug(
f"ValueError parsing timestamp string: '{timestamp_str}' (tried '{timestamp_to_parse}'). Error: {e}. Line {line_number} in {file_path}: {line.strip()}"
)
# Fall through to return with None timestamp but other parsed fields
return {
"timestamp": parsed_timestamp,
"level": level_str,
"message": message_str,
"raw_line": line.strip(),
"file_path": file_path,
"line_number": line_number,
}
# Fallback for lines that don't match the primary pattern
# (e.g., stack traces, multi-line messages not handled by a continuation pattern)
self.logger.debug(f"Line did not match primary log pattern. Line {line_number} in {file_path}: {line.strip()}")
return {
"timestamp": None,
"level": "UNKNOWN",
"message": line.strip(),
"raw_line": line.strip(),
"file_path": file_path,
"line_number": line_number,
}
def _apply_content_filters(
self, entries: List[ParsedLogEntry], filter_criteria: Dict[str, Any]
) -> List[ParsedLogEntry]:
"""
Filters entries based on content patterns.
Uses 'log_content_patterns_override' from filter_criteria if available (as a list of general regexes).
Otherwise, uses level-specific regexes from self.log_content_patterns (config) IF a level_filter is also provided.
"""
override_patterns: Optional[List[str]] = filter_criteria.get("log_content_patterns_override")
if override_patterns is not None: # Check if the key exists, even if list is empty
# Apply general override patterns
if not override_patterns: # Empty list provided (e.g. override_patterns == [])
self.logger.info(
"[_apply_content_filters] log_content_patterns_override is empty list. Returning all entries."
)
return entries
filtered_entries: List[ParsedLogEntry] = []
for entry in entries:
message = entry.get("message", "")
# level = entry.get("level", "UNKNOWN").upper() # Not used in override path
# entry_added = False # Not strictly needed with break
for pattern_str in override_patterns:
try:
if re.search(pattern_str, message, re.IGNORECASE):
filtered_entries.append(entry)
# entry_added = True
break # Matched one pattern, include entry and move to next entry
except re.error as e:
self.logger.warning(
f"Invalid regex in override_patterns: '{pattern_str}'. Error: {e}. Skipping this pattern."
)
return filtered_entries
else:
# No override_patterns. Use configured level-specific patterns only if a level_filter is present.
level_filter_str = filter_criteria.get("level_filter", "").upper()
if not level_filter_str:
# No specific level_filter provided in criteria, and no override patterns.
# Content filtering should not apply by default from env/config in this case.
self.logger.info(
"[_apply_content_filters] No override patterns and no level_filter in criteria. Returning all entries."
)
return entries
# A specific level_filter_str IS provided. Use patterns for that level from self.log_content_patterns.
# self.log_content_patterns is Dict[str (lowercase level), List[str_patterns]]
# Ensure level_filter_str matches the key format (e.g. "error" not "ERROR")
relevant_patterns = self.log_content_patterns.get(level_filter_str.lower(), [])
self.logger.info(
f"[_apply_content_filters] Using config patterns for level_filter: '{level_filter_str}'. Relevant patterns: {relevant_patterns}"
)
# Filter by the specified level first.
# Then, if there are patterns for that level, apply them.
# If no patterns for that level, all entries of that level pass.
filtered_entries = []
for entry in entries:
entry_level = entry.get("level", "UNKNOWN").upper()
message = entry.get("message", "")
if entry_level == level_filter_str: # Entry must match the specified level
if not relevant_patterns:
# No patterns for this level, so include if level matches
filtered_entries.append(entry)
else:
# Patterns exist for this level, try to match them
for pattern_str in relevant_patterns:
try:
if re.search(pattern_str, message, re.IGNORECASE):
filtered_entries.append(entry)
break # Matched one pattern for this level, include entry
except re.error as e:
self.logger.warning(
f"Invalid regex in configured patterns for level {level_filter_str}: '{pattern_str}'. Error: {e}. Skipping pattern."
)
return filtered_entries
def _apply_time_filters(
self, entries: List[ParsedLogEntry], filter_criteria: Dict[str, Any]
) -> List[ParsedLogEntry]:
"""Filters entries based on time window from filter_criteria."""
now = dt.datetime.now() # Use dt.datetime.now()
time_window_applied = False
earliest_time: Optional[dt.datetime] = None # Use dt.datetime for type hint
if filter_criteria.get("minutes", 0) > 0:
earliest_time = now - dt.timedelta(minutes=filter_criteria["minutes"])
time_window_applied = True
elif filter_criteria.get("hours", 0) > 0:
earliest_time = now - dt.timedelta(hours=filter_criteria["hours"])
time_window_applied = True
elif filter_criteria.get("days", 0) > 0:
earliest_time = now - dt.timedelta(days=filter_criteria["days"])
time_window_applied = True
if not time_window_applied or earliest_time is None:
return entries # No time filter to apply or invalid criteria
filtered_entries: List[ParsedLogEntry] = []
for entry in entries:
entry_timestamp = entry.get("timestamp")
# Ensure entry_timestamp is a datetime.datetime object before comparison
if (
isinstance(entry_timestamp, DateTimeClassForCheck) and entry_timestamp >= earliest_time
): # Use DateTimeClassForCheck for isinstance
filtered_entries.append(entry)
return filtered_entries
def _apply_positional_filters(
self, entries: List[ParsedLogEntry], filter_criteria: Dict[str, Any]
) -> List[ParsedLogEntry]:
"""Filters entries based on positional criteria (first_n, last_n)."""
first_n = filter_criteria.get("first_n")
last_n = filter_criteria.get("last_n")
# Only filter by timestamp and sort if a positional filter is active
if (first_n is not None and isinstance(first_n, int) and first_n > 0) or (
last_n is not None and isinstance(last_n, int) and last_n > 0
):
# Filter out entries with no timestamp before sorting for positional filters
entries_with_timestamp = [e for e in entries if e.get("timestamp") is not None]
# Ensure entries are sorted by timestamp before applying positional filters
# ParsedLogEntry includes 'timestamp', which is a datetime object
# Using e["timestamp"] as we've filtered for its existence and non-None value.
sorted_entries = sorted(entries_with_timestamp, key=lambda e: e["timestamp"])
if first_n is not None and isinstance(first_n, int) and first_n > 0:
return sorted_entries[:first_n]
elif last_n is not None and isinstance(last_n, int) and last_n > 0:
return sorted_entries[-last_n:]
else:
# Should not be reached if the outer if condition is met correctly
return sorted_entries
# If no positional filter is active, return the original entries
# Order might be important, so don't sort unless a positional filter needs it.
return entries
def _extract_context_lines(
self,
entries: List[ParsedLogEntry],
all_lines_by_file: Dict[str, List[str]],
context_before: int,
context_after: int,
) -> List[ParsedLogEntry]:
"""Extracts context lines for each entry."""
if context_before == 0 and context_after == 0:
# Add empty context if no context lines are requested, to maintain structure
for entry in entries:
entry["context_before_lines"] = []
entry["context_after_lines"] = []
entry["full_context_log"] = entry["raw_line"]
return entries
entries_with_context: List[ParsedLogEntry] = []
for entry in entries:
file_path = entry["file_path"]
line_number = entry["line_number"] # 1-indexed from original file
if file_path not in all_lines_by_file:
# This shouldn't happen if all_lines_by_file is populated correctly
entry["context_before_lines"] = []
entry["context_after_lines"] = []
entry["full_context_log"] = entry["raw_line"]
entries_with_context.append(entry)
self.logger.warning(f"Warning: File {file_path} not found in all_lines_by_file for context extraction.")
continue
file_lines = all_lines_by_file[file_path]
actual_line_index = line_number - 1 # Convert to 0-indexed for list access
start_index = max(0, actual_line_index - context_before)
end_index = min(len(file_lines), actual_line_index + context_after + 1)
entry_copy = entry.copy() # Avoid modifying the original entry directly in the list
entry_copy["context_before_lines"] = [line.strip() for line in file_lines[start_index:actual_line_index]]
entry_copy["context_after_lines"] = [line.strip() for line in file_lines[actual_line_index + 1 : end_index]]
# Construct full_context_log
full_context_list = (
entry_copy["context_before_lines"] + [entry_copy["raw_line"]] + entry_copy["context_after_lines"]
)
entry_copy["full_context_log"] = "\\n".join(full_context_list)
entries_with_context.append(entry_copy)
return entries_with_context
def search_logs(self, filter_criteria: Dict[str, Any]) -> List[Dict[str, Any]]:
"""
Main method to search logs based on various criteria.
filter_criteria is a dictionary that can contain:
- log_dirs_override: List[str] (paths/globs to search instead of config)
- scope: str (e.g., "mcp", "runtime" to use predefined paths from config)
- log_content_patterns_override: List[str] (regexes for log message content)
- level_filter: str (e.g., "ERROR", "WARNING")
- time_filter_type: str ("minutes", "hours", "days") - maps to minutes, hours, days keys
- time_filter_value: int (e.g., 30 for 30 minutes) - maps to minutes, hours, days values
- positional_filter_type: str ("first_n", "last_n") - maps to first_n, last_n keys
- positional_filter_value: int (e.g., 10 for first 10 records) - maps to first_n, last_n values
- context_before: int (lines of context before match)
- context_after: int (lines of context after match)
"""
self.logger.info(f"[AnalysisEngine.search_logs] Called with filter_criteria: {filter_criteria}")
all_raw_lines_by_file: Dict[str, List[str]] = {}
parsed_entries: List[ParsedLogEntry] = []
# 1. Determine target log files
target_files = self._get_target_log_files(
scope=filter_criteria.get("scope"),
log_dirs_override=filter_criteria.get("log_dirs_override"),
)
if not target_files:
self.logger.info(
"[AnalysisEngine.search_logs] No log files found by _get_target_log_files. Returning pathway OK message."
)
# Return a specific message indicating pathway is okay but no files found
return [{"message": "No target files found, but pathway OK."}]
self.logger.info(f"[AnalysisEngine.search_logs] Target files found: {target_files}")
# 2. Parse all lines from target files
for file_path in target_files:
try:
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
lines = f.readlines()
# Store all lines for context extraction later
all_raw_lines_by_file[file_path] = [
line.rstrip("\\n") for line in lines
] # Store raw lines as they are
for i, line_content in enumerate(lines):
entry = self._parse_log_line(line_content.strip(), file_path, i + 1) # line_number is 1-indexed
if entry:
parsed_entries.append(entry)
except Exception as e: # pylint: disable=broad-exception-caught
self.logger.error(f"Error reading or parsing file {file_path}: {e}", exc_info=True)
continue # Continue with other files
self.logger.info(f"[AnalysisEngine.search_logs] Parsed {len(parsed_entries)} entries from all target files.")
if not parsed_entries:
self.logger.info("[AnalysisEngine.search_logs] No entries parsed from target files.")
return []
# 3. Apply content filters (level and regex)
filtered_entries = self._apply_content_filters(parsed_entries, filter_criteria)
if not filtered_entries:
self.logger.info("[AnalysisEngine.search_logs] No entries left after content filters.")
return []
# 4. Apply time filters
filtered_entries = self._apply_time_filters(filtered_entries, filter_criteria)
if not filtered_entries:
self.logger.info("[AnalysisEngine.search_logs] No entries left after time filters.")
return []
# 5. Apply positional filters (first_n, last_n)
# Note: _apply_positional_filters sorts by timestamp and handles entries without timestamps
filtered_entries = self._apply_positional_filters(filtered_entries, filter_criteria)
if not filtered_entries:
self.logger.info("[AnalysisEngine.search_logs] No entries left after positional filters.")
return []
# 6. Extract context lines for the final set of entries
# Use context_before and context_after from filter_criteria, or defaults from config
context_before = filter_criteria.get("context_before", self.default_context_lines_before)
context_after = filter_criteria.get("context_after", self.default_context_lines_after)
final_entries_with_context = self._extract_context_lines(
filtered_entries, all_raw_lines_by_file, context_before, context_after
)
self.logger.info(f"[AnalysisEngine.search_logs] Returning {len(final_entries_with_context)} processed entries.")
# The tool expects a list of dicts, and ParsedLogEntry is already a Dict[str, Any]
return final_entries_with_context
# TODO: Add helper functions for parsing, filtering, file handling etc. as needed.
```
--------------------------------------------------------------------------------
/src/log_analyzer_mcp/log_analyzer_mcp_server.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Test Analyzer MCP Server
Implements the Model Context Protocol (MCP) for Cursor to analyze test results.
"""
import asyncio
import anyio
import os
import re
import subprocess
import sys
import functools
from datetime import datetime
from typing import Any, Callable
from mcp.server.fastmcp import FastMCP
# MCP and Pydantic related imports
from mcp.shared.exceptions import McpError
from mcp.types import (
ErrorData,
)
from pydantic import BaseModel, Field
# Project-specific imports
from log_analyzer_mcp.common.logger_setup import LoggerSetup, get_logs_dir
from log_analyzer_mcp.common.utils import build_filter_criteria
from log_analyzer_mcp.core.analysis_engine import AnalysisEngine
from log_analyzer_mcp.test_log_parser import analyze_pytest_log_content
# Explicitly attempt to initialize coverage for subprocesses
if "COVERAGE_PROCESS_START" in os.environ:
try:
import coverage
coverage.process_startup()
# If your logger is configured very early, you could add a log here:
# print("DEBUG: coverage.process_startup() called in subprocess.", flush=True)
except ImportError:
# print("DEBUG: COVERAGE_PROCESS_START set, but coverage module not found.", flush=True)
pass # Or handle error if coverage is mandatory for the subprocess
except Exception: # pylint: disable=broad-exception-caught
# print(f"DEBUG: Error calling coverage.process_startup(): {e}", flush=True)
pass
# Define project_root and script_dir here as they are used for path definitions
script_dir = os.path.dirname(os.path.abspath(__file__))
# project_root = os.path.dirname(os.path.dirname(script_dir)) # No longer needed here if logger_setup is robust
# Set up logging using centralized configuration
logs_base_dir = get_logs_dir() # RESTORED - this should now be correct
mcp_log_dir = os.path.join(logs_base_dir, "mcp") # RESTORED
# Ensure project_root is correctly determined as the actual project root
# Forcing a known-good structure relative to where log_analyzer_mcp_server.py is.
# __file__ is src/log_analyzer_mcp/log_analyzer_mcp_server.py
# script_dir is src/log_analyzer_mcp/
# parent of script_dir is src/
# parent of parent of script_dir is PROJECT_ROOT
# actual_project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # REMOVE direct calculation here
# mcp_log_dir = os.path.join(actual_project_root, "logs", "mcp") # REMOVE direct calculation here
os.makedirs(mcp_log_dir, exist_ok=True) # This is fine, mcp_log_dir is now from get_logs_dir()
# Determine the log file path, prioritizing MCP_LOG_FILE env var
env_log_file = os.getenv("MCP_LOG_FILE")
if env_log_file:
log_file_path = os.path.abspath(env_log_file)
# Ensure the directory for the environment-specified log file exists
env_log_file_dir = os.path.dirname(log_file_path)
if not os.path.exists(env_log_file_dir):
try:
os.makedirs(env_log_file_dir, exist_ok=True)
# Temporary print to confirm this path is taken
print(
f"DEBUG_MCP_SERVER: Ensured directory exists for MCP_LOG_FILE: {env_log_file_dir}",
file=sys.stderr,
flush=True,
)
except OSError as e:
print(
f"Warning: Could not create directory for MCP_LOG_FILE {env_log_file_dir}: {e}",
file=sys.stderr,
flush=True,
)
# Fallback to default if directory creation fails for env var path
log_file_path = os.path.join(mcp_log_dir, "log_analyzer_mcp_server.log")
print(
f"DEBUG_MCP_SERVER: Using MCP_LOG_FILE from environment: {log_file_path}", file=sys.stderr, flush=True
) # ADDED
else:
log_file_path = os.path.join(mcp_log_dir, "log_analyzer_mcp_server.log")
print(f"DEBUG_MCP_SERVER: Using default log_file_path: {log_file_path}", file=sys.stderr, flush=True) # ADDED
logger = LoggerSetup.create_logger("LogAnalyzerMCP", log_file_path, agent_name="LogAnalyzerMCP")
logger.setLevel("DEBUG") # Set to debug level for MCP server
# CRITICAL DEBUG: Print to stderr immediately after logger setup
print(f"DEBUG_MCP_SERVER: Logger initialized. Attempting to log to: {log_file_path}", file=sys.stderr, flush=True)
logger.info("Log Analyzer MCP Server starting. Logging to %s", log_file_path)
# Initialize AnalysisEngine instance (can be done once)
# It will load .env settings by default upon instantiation.
analysis_engine = AnalysisEngine(logger_instance=logger)
# Update paths for scripts and logs (using project_root and script_dir)
# log_analyzer_path = os.path.join(script_dir, 'log_analyzer.py') # REMOVED
# run_tests_path = os.path.join(project_root, 'tests/run_all_tests.py') # REMOVED - using hatch test directly
# run_coverage_path = os.path.join(script_dir, 'create_coverage_report.sh') # REMOVED - using hatch run hatch-test:* directly
# analyze_runtime_errors_path = os.path.join(script_dir, 'analyze_runtime_errors.py') # REMOVED
test_log_file = os.path.join(
logs_base_dir, "run_all_tests.log" # RESTORED logs_base_dir
) # Main test log, now populated by hatch test output
# coverage_xml_path = os.path.join(logs_base_dir, 'tests', 'coverage', 'coverage.xml') # RESTORED logs_base_dir
# Initialize FastMCP server
# Add lifespan support for startup/shutdown with strong typing
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
@asynccontextmanager
async def server_lifespan(_server: FastMCP) -> AsyncIterator[None]: # Simple lifespan, no app context needed
logger.info("MCP Server Lifespan: Startup phase entered.")
try:
yield
finally:
logger.info("MCP Server Lifespan: Shutdown phase entered (finally block).")
mcp = FastMCP("log_analyzer", lifespan=server_lifespan)
# Define input models for tool validation
class AnalyzeTestsInput(BaseModel):
"""Parameters for analyzing tests."""
summary_only: bool = Field(default=False, description="Whether to return only a summary of the test results")
class RunTestsInput(BaseModel):
"""Parameters for running tests."""
verbosity: int = Field(default=1, description="Verbosity level for the test runner (0-2)", ge=0, le=2)
class CreateCoverageReportInput(BaseModel):
"""Parameters for creating coverage report."""
force_rebuild: bool = Field(
default=False, description="Whether to force rebuilding the coverage report even if it already exists"
)
class RunUnitTestInput(BaseModel):
"""Parameters for running specific unit tests."""
agent: str = Field(description="The agent to run tests for (e.g., 'qa_agent', 'backlog_agent')")
verbosity: int = Field(default=1, description="Verbosity level (0=minimal, 1=normal, 2=detailed)", ge=0, le=2)
# Define default runtime logs directory
DEFAULT_RUNTIME_LOGS_DIR = os.path.join(logs_base_dir, "runtime") # RESTORED logs_base_dir
# async def analyze_test_log(log_file_path: str, summary_only: bool = False) -> Dict[str, Any]: # REMOVED: Functionality moved to test_log_parser
# """
# Analyze a test log file and return structured results.
# ...
# """
# ...
@mcp.tool()
async def analyze_tests(summary_only: bool = False) -> dict[str, Any]:
"""Analyze the most recent test run and provide detailed information about failures.
Args:
summary_only: Whether to return only a summary of the test results
"""
logger.info("Analyzing test results (summary_only=%s)...", summary_only)
log_file = test_log_file
if not os.path.exists(log_file):
error_msg = f"Test log file not found at: {log_file}. Please run tests first."
logger.error(error_msg)
return {"error": error_msg, "summary": {"status": "ERROR", "passed": 0, "failed": 0, "skipped": 0}}
try:
with open(log_file, encoding="utf-8", errors="ignore") as f:
log_contents = f.read()
if not log_contents.strip():
error_msg = f"Test log file is empty: {log_file}"
logger.warning(error_msg)
return {"error": error_msg, "summary": {"status": "EMPTY", "passed": 0, "failed": 0, "skipped": 0}}
analysis = analyze_pytest_log_content(log_contents, summary_only=summary_only)
# Add metadata similar to the old analyze_test_log function
log_time = datetime.fromtimestamp(os.path.getmtime(log_file))
time_elapsed = (datetime.now() - log_time).total_seconds() / 60 # minutes
analysis["log_file"] = log_file
analysis["log_timestamp"] = log_time.isoformat()
analysis["log_age_minutes"] = round(time_elapsed, 1)
# The analyze_pytest_log_content already returns a structure including 'overall_summary'.
# If summary_only is true, it returns only that. Otherwise, it returns more details.
# We can directly return this analysis dictionary.
# Ensure there's always a summary structure for consistent access, even if minimal
if "overall_summary" not in analysis:
analysis["overall_summary"] = {"status": "UNKNOWN", "passed": 0, "failed": 0, "skipped": 0}
if "summary" not in analysis: # for backward compatibility or general access
analysis["summary"] = analysis["overall_summary"]
logger.info(
"Test log analysis completed using test_log_parser. Summary status: %s",
analysis.get("summary", {}).get("status"),
)
return analysis
except Exception as e: # pylint: disable=broad-exception-caught
error_msg = f"Error analyzing test log file with test_log_parser: {e}"
logger.error(error_msg, exc_info=True)
return {"error": error_msg, "summary": {"status": "ERROR", "passed": 0, "failed": 0, "skipped": 0}}
async def _run_tests(
verbosity: Any | None = None,
agent: str | None = None,
pattern: str | None = None,
run_with_coverage: bool = False,
) -> dict[str, Any]:
"""Internal helper function to run tests using hatch.
Args:
verbosity: Optional verbosity level (0=minimal, 1=normal, 2=detailed for pytest)
agent: Optional agent name to run only tests for that agent (e.g., 'qa_agent')
pattern: Optional pattern to filter test files (e.g., 'test_qa_*.py')
run_with_coverage: Whether to run tests with coverage enabled via 'hatch test --cover'.
"""
logger.info(
"Preparing to run tests via hatch (verbosity=%s, agent=%s, pattern=%s, coverage=%s)...",
verbosity,
agent,
pattern,
run_with_coverage,
)
hatch_base_cmd = ["hatch", "test"]
pytest_args = []
# ALWAYS add arguments to ignore the server integration tests to prevent recursion
# when tests are run *by this tool*.
pytest_args.extend(
[
"--ignore=tests/log_analyzer_mcp/test_log_analyzer_mcp_server.py",
"--ignore=tests/log_analyzer_mcp/test_analyze_runtime_errors.py",
]
)
logger.debug("Added ignore patterns for server integration tests (tool-invoked run).")
if run_with_coverage:
hatch_base_cmd.append("--cover")
logger.debug("Coverage enabled for hatch test run.")
# Tell pytest not to activate its own coverage plugin, as 'coverage run' is handling it.
pytest_args.append("-p")
pytest_args.append("no:cov")
logger.debug("Added '-p no:cov' to pytest arguments for coverage run.")
# Verbosity for pytest: -q (0), (1), -v (2), -vv (3+)
if verbosity is not None:
try:
v_int = int(verbosity)
if v_int == 0:
pytest_args.append("-q")
elif v_int == 2:
pytest_args.append("-v")
elif v_int >= 3:
pytest_args.append("-vv")
# Default (verbosity=1) means no specific pytest verbosity arg, relies on hatch default
except ValueError:
logger.warning("Invalid verbosity value '%s', using default.", verbosity)
# Construct pytest -k argument if agent or pattern is specified
k_expressions = []
if agent:
# Assuming agent name can be part of test names like test_agent_... or ..._agent_...
k_expressions.append(f"{agent}") # This f-string is for constructing a command argument, not direct logging.
logger.debug("Added agent '%s' to -k filter expressions.", agent)
if pattern:
k_expressions.append(pattern)
logger.debug("Added pattern '%s' to -k filter expressions.", pattern)
if k_expressions:
pytest_args.extend(["-k", " or ".join(k_expressions)]) # pytest -k "expr1 or expr2"
hatch_cmd = hatch_base_cmd
if pytest_args: # Pass pytest arguments after --
hatch_cmd.extend(["--"] + pytest_args)
logger.info("Constructed hatch command: %s", " ".join(hatch_cmd))
# Ensure the log file is cleared or managed before test run if it's always written to the same path
# For now, assuming log_analyzer.py handles this or we analyze the latest run.
test_log_output_path = os.path.join(logs_base_dir, "run_all_tests.log") # RESTORED logs_base_dir
logger.debug("Expected test output log path for analysis: %s", test_log_output_path)
try:
# Run the command using anyio.to_thread to avoid blocking asyncio event loop
# Ensure text=True for automatic decoding of stdout/stderr to string
process = await anyio.to_thread.run_sync( # type: ignore[attr-defined]
functools.partial(
subprocess.run,
hatch_cmd,
capture_output=True,
text=True, # Decode stdout/stderr as text (usually UTF-8)
check=False, # Don't raise exception for non-zero exit, handle manually
timeout=120, # Add timeout
)
)
stdout_output: str = process.stdout
stderr_output: str = process.stderr
rc = process.returncode
if rc not in [0, 1, 5]:
logger.error("Hatch test command failed with unexpected pytest return code: %s", rc)
logger.error("STDOUT:\n%s", stdout_output)
logger.error("STDERR:\n%s", stderr_output)
return {
"success": False,
"error": f"Test execution failed with code {rc}",
"test_output": stdout_output + "\n" + stderr_output,
"analysis_log_path": None,
}
logger.debug("Saving combined stdout/stderr from hatch test to %s", test_log_output_path)
with open(test_log_output_path, "w", encoding="utf-8") as f:
f.write(stdout_output)
f.write("\n")
f.write(stderr_output)
logger.debug("Content saved to %s", test_log_output_path)
# _run_tests now only runs tests and saves the log.
# Analysis is done by the analyze_tests tool or by the caller if needed.
# The old log_analyzer.main() call is removed.
# If an agent was specified, the caller of _run_tests might want to know.
# We can still populate this in the result.
if agent:
# analysis_to_return is None, so we can create a small dict or add to a base dict
# For now, let's just focus on returning the essential info
pass
return {
"success": True,
"return_code": rc,
"test_output": stdout_output + "\n" + stderr_output,
"analysis_log_path": test_log_output_path, # Provide path to the log for analysis
# "analysis" field is removed from here as _run_tests no longer parses.
}
except subprocess.TimeoutExpired as e:
stdout_output = e.stdout.decode("utf-8", errors="replace") if e.stdout else ""
stderr_output = e.stderr.decode("utf-8", errors="replace") if e.stderr else ""
stderr_output += f"\nError: Test execution timed out after 170 seconds."
rc = 1 # Indicate failure
logger.error("Test execution in _run_tests timed out: %s", e)
return {
"success": False,
"error": stderr_output,
"test_output": stdout_output + "\n" + stderr_output,
"analysis_log_path": None,
}
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("An unexpected error occurred in _run_tests: %s", e, exc_info=True)
# Capture output if process started
final_stdout = ""
final_stderr = ""
if "stdout_output" in locals() and "stderr_output" in locals(): # Check if communicate() was reached
final_stdout = stdout_output
final_stderr = stderr_output
# else: process might not have been initialized or communicate not called.
# No direct access to process.stdout/stderr here as it's out of 'with' scope.
return {
"success": False,
"error": f"Unexpected error: {e}",
"test_output": final_stdout + "\n" + final_stderr,
"analysis_log_path": None,
}
@mcp.tool()
async def run_tests_no_verbosity() -> dict[str, Any]:
"""Run all tests with minimal output (verbosity level 0)."""
return await _run_tests("0")
@mcp.tool()
async def run_tests_verbose() -> dict[str, Any]:
"""Run all tests with verbose output (verbosity level 1)."""
return await _run_tests("1")
@mcp.tool()
async def run_tests_very_verbose() -> dict[str, Any]:
"""Run all tests with very verbose output (verbosity level 2)."""
logger.info("Running tests with verbosity 2...")
return await _run_tests(verbosity=2, run_with_coverage=True)
@mcp.tool()
async def ping() -> str:
"""Check if the MCP server is alive."""
logger.debug("ping called")
return f"Status: ok\n" f"Timestamp: {datetime.now().isoformat()}\n" f"Message: Log Analyzer MCP Server is running"
async def run_coverage_script(force_rebuild: bool = False) -> dict[str, Any]:
"""
Run the coverage report script and generate HTML and XML reports.
Now uses hatch scripts for better integration.
"""
logger.info("Running coverage script...")
# Correctly reference PROJECT_ROOT from the logger_setup module
from log_analyzer_mcp.common import logger_setup as common_logger_setup # Ensure this import is here or global
current_project_root = common_logger_setup.PROJECT_ROOT
# Define different timeouts for different steps
timeout_run_cov = 300 # Longer timeout for running tests with coverage
timeout_cov_report = 120 # Shorter timeout for generating the report
# Command parts for running the coverage script via hatch
# This assumes 'run-cov' and 'cov-report' are defined in hatch envs.
# Step 1: Run tests with coverage enabled
cmd_parts_run_cov = ["hatch", "run", "hatch-test.py3.12:run-cov"] # Example: Target specific py version
# Step 2: Generate combined report (HTML and XML)
cmd_parts_report = ["hatch", "run", "hatch-test.py3.12:cov-report"] # Example
outputs = []
errors_encountered = []
steps_with_timeouts = [
("run-cov", cmd_parts_run_cov, timeout_run_cov),
("cov-report", cmd_parts_report, timeout_cov_report),
]
for step_name, cmd_parts, current_timeout_seconds in steps_with_timeouts:
logger.info(
"Executing coverage step '%s': %s (timeout: %ss)", step_name, " ".join(cmd_parts), current_timeout_seconds
)
try:
# Use functools.partial for subprocess.run
configured_subprocess_run_step = functools.partial(
subprocess.run,
cmd_parts,
cwd=current_project_root,
capture_output=True,
text=True, # Decode stdout/stderr as text
check=False, # Handle non-zero exit manually
timeout=current_timeout_seconds, # Use current step's timeout
)
process = await anyio.to_thread.run_sync(configured_subprocess_run_step) # type: ignore[attr-defined]
stdout_output: str = process.stdout
stderr_output: str = process.stderr
rc = process.returncode
outputs.append(f"--- {step_name} STDOUT ---\n{stdout_output}")
if stderr_output:
outputs.append(f"--- {step_name} STDERR ---\n{stderr_output}")
if rc != 0:
error_msg = f"Coverage step '{step_name}' failed with return code {rc}."
logger.error("%s\nSTDERR:\n%s", error_msg, stderr_output)
errors_encountered.append(error_msg)
# Optionally break if a step fails, or collect all errors
# break
except subprocess.TimeoutExpired as e:
stdout_output = e.stdout.decode("utf-8", errors="replace") if e.stdout else ""
stderr_output = e.stderr.decode("utf-8", errors="replace") if e.stderr else ""
error_msg = f"Coverage step '{step_name}' timed out after {current_timeout_seconds} seconds."
logger.error("%s: %s", error_msg, e)
errors_encountered.append(error_msg)
outputs.append(f"--- {step_name} TIMEOUT STDOUT ---\n{stdout_output}")
outputs.append(f"--- {step_name} TIMEOUT STDERR ---\n{stderr_output}")
# break
except Exception as e: # pylint: disable=broad-exception-caught
error_msg = f"Error during coverage step '{step_name}': {e}"
logger.error(error_msg, exc_info=True)
errors_encountered.append(error_msg)
# break
# Ensure a dictionary is always returned, even if errors occurred.
final_success = not errors_encountered
overall_message = (
"Coverage script steps completed." if final_success else "Errors encountered during coverage script execution."
)
# Placeholder for actual report paths, adapt as needed
coverage_xml_report_path = os.path.join(logs_base_dir, "tests", "coverage", "coverage.xml")
coverage_html_index_path = os.path.join(logs_base_dir, "tests", "coverage", "html", "index.html")
return {
"success": final_success,
"message": overall_message,
"details": "\n".join(outputs),
"errors": errors_encountered,
"coverage_xml_path": coverage_xml_report_path if final_success else None, # Example path
"coverage_html_index": coverage_html_index_path if final_success else None, # Example path
"timestamp": datetime.now().isoformat(),
}
@mcp.tool()
async def create_coverage_report(force_rebuild: bool = False) -> dict[str, Any]:
"""
Run the coverage report script and generate HTML and XML reports.
Args:
force_rebuild: Whether to force rebuilding the report even if it exists
Returns:
Dictionary containing execution results and report paths
"""
return await run_coverage_script(force_rebuild)
@mcp.tool()
async def run_unit_test(agent: str, verbosity: int = 1) -> dict[str, Any]:
"""
Run tests for a specific agent only.
This tool runs tests that match the agent's patterns including both main agent tests
and healthcheck tests, significantly reducing test execution time compared to running all tests.
Use this tool when you need to focus on testing a specific agent component.
Args:
agent: The agent to run tests for (e.g., 'qa_agent', 'backlog_agent')
verbosity: Verbosity level (0=minimal, 1=normal, 2=detailed), default is 1
Returns:
Dictionary containing test results and analysis
"""
logger.info("Running unit tests for agent: %s with verbosity %s", agent, verbosity)
# The _run_tests function now handles pattern creation from agent name.
# We call _run_tests once, and it will construct a pattern like "test_agent.py or test_healthcheck.py"
# No need for separate calls for main and healthcheck unless _run_tests logic changes.
# For verbosity, _run_tests expects 0, 1, or 2 as string or int.
# The pattern is derived by _run_tests from the agent name.
results = await _run_tests(agent=agent, verbosity=verbosity, run_with_coverage=False)
# The structure of the response from _run_tests is already good for run_unit_test.
# It includes success, return_code, test_output, and analysis (which contains agent_tested).
# No need to combine results manually here if _run_tests handles the agent pattern correctly.
return results
# --- Pydantic Models for Search Tools ---
class BaseSearchInput(BaseModel):
"""Base model for common search parameters."""
scope: str = Field(default="default", description="Logging scope to search within (from .env scopes or default).")
context_before: int = Field(default=2, description="Number of lines before a match.", ge=0)
context_after: int = Field(default=2, description="Number of lines after a match.", ge=0)
log_dirs_override: str = Field(
default="",
description="Comma-separated list of log directories, files, or glob patterns (overrides .env for file locations).",
)
log_content_patterns_override: str = Field(
default="",
description="Comma-separated list of REGEX patterns for log messages (overrides .env content filters).",
)
class SearchLogAllInput(BaseSearchInput):
"""Input for search_log_all_records."""
@mcp.tool()
async def search_log_all_records(
scope: str = "default",
context_before: int = 2,
context_after: int = 2,
log_dirs_override: str = "",
log_content_patterns_override: str = "",
) -> list[dict[str, Any]]:
"""Search for all log records, optionally filtering by scope and content patterns, with context."""
# Forcing re-initialization of analysis_engine for debugging module caching.
# Pass project_root_for_config=None to allow AnalysisEngine to determine it.
current_analysis_engine = AnalysisEngine(logger_instance=logger, project_root_for_config=None)
print(
f"DEBUG_MCP_TOOL_SEARCH_ALL: Entered search_log_all_records with log_dirs_override='{log_dirs_override}'",
file=sys.stderr,
flush=True,
)
logger.info(
"MCP search_log_all_records called with scope='%s', context=%sB/%sA, "
"log_dirs_override='%s', log_content_patterns_override='%s'",
scope,
context_before,
context_after,
log_dirs_override,
log_content_patterns_override,
)
log_dirs_list = log_dirs_override.split(",") if log_dirs_override else None
log_content_patterns_list = log_content_patterns_override.split(",") if log_content_patterns_override else None
filter_criteria = build_filter_criteria(
scope=scope,
context_before=context_before,
context_after=context_after,
log_dirs_override=log_dirs_list,
log_content_patterns_override=log_content_patterns_list,
)
try:
results = await asyncio.to_thread(current_analysis_engine.search_logs, filter_criteria)
logger.info("search_log_all_records returning %s records.", len(results))
return results
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error in search_log_all_records: %s", e, exc_info=True)
custom_message = f"Failed to search all logs: {e!s}"
raise McpError(ErrorData(code=-32603, message=custom_message)) from e
class SearchLogTimeBasedInput(BaseSearchInput):
"""Input for search_log_time_based."""
minutes: int = Field(default=0, description="Search logs from the last N minutes.", ge=0)
hours: int = Field(default=0, description="Search logs from the last N hours.", ge=0)
days: int = Field(default=0, description="Search logs from the last N days.", ge=0)
# Custom validation to ensure at least one time field is set if others are default (0)
# Pydantic v2: @model_validator(mode='after')
# Pydantic v1: @root_validator(pre=False)
# For simplicity here, relying on tool logic to handle it, or can add validator if needed.
@mcp.tool()
async def search_log_time_based(
minutes: int = 0,
hours: int = 0,
days: int = 0,
scope: str = "default",
context_before: int = 2,
context_after: int = 2,
log_dirs_override: str = "",
log_content_patterns_override: str = "",
) -> list[dict[str, Any]]:
"""Search logs within a time window, optionally filtering, with context."""
logger.info(
"MCP search_log_time_based called with time=%sd/%sh/%sm, scope='%s', "
"context=%sB/%sA, log_dirs_override='%s', "
"log_content_patterns_override='%s'",
days,
hours,
minutes,
scope,
context_before,
context_after,
log_dirs_override,
log_content_patterns_override,
)
if minutes == 0 and hours == 0 and days == 0:
logger.warning("search_log_time_based called without a time window (all minutes/hours/days are 0).")
log_dirs_list = log_dirs_override.split(",") if log_dirs_override else None
log_content_patterns_list = log_content_patterns_override.split(",") if log_content_patterns_override else None
filter_criteria = build_filter_criteria(
minutes=minutes,
hours=hours,
days=days,
scope=scope,
context_before=context_before,
context_after=context_after,
log_dirs_override=log_dirs_list,
log_content_patterns_override=log_content_patterns_list,
)
try:
results = await asyncio.to_thread(analysis_engine.search_logs, filter_criteria)
logger.info("search_log_time_based returning %s records.", len(results))
return results
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error in search_log_time_based: %s", e, exc_info=True)
custom_message = f"Failed to search time-based logs: {e!s}"
raise McpError(ErrorData(code=-32603, message=custom_message)) from e
class SearchLogFirstNInput(BaseSearchInput):
"""Input for search_log_first_n_records."""
count: int = Field(description="Number of first (oldest) matching records to return.", gt=0)
@mcp.tool()
async def search_log_first_n_records(
count: int,
scope: str = "default",
context_before: int = 2,
context_after: int = 2,
log_dirs_override: str = "",
log_content_patterns_override: str = "",
) -> list[dict[str, Any]]:
"""Search for the first N (oldest) records, optionally filtering, with context."""
logger.info(
"MCP search_log_first_n_records called with count=%s, scope='%s', "
"context=%sB/%sA, log_dirs_override='%s', "
"log_content_patterns_override='%s'",
count,
scope,
context_before,
context_after,
log_dirs_override,
log_content_patterns_override,
)
if count <= 0:
logger.error("Invalid count for search_log_first_n_records: %s. Must be > 0.", count)
raise McpError(ErrorData(code=-32602, message="Count must be a positive integer."))
log_dirs_list = log_dirs_override.split(",") if log_dirs_override else None
log_content_patterns_list = log_content_patterns_override.split(",") if log_content_patterns_override else None
filter_criteria = build_filter_criteria(
first_n=count,
scope=scope,
context_before=context_before,
context_after=context_after,
log_dirs_override=log_dirs_list,
log_content_patterns_override=log_content_patterns_list,
)
try:
results = await asyncio.to_thread(analysis_engine.search_logs, filter_criteria)
logger.info("search_log_first_n_records returning %s records.", len(results))
return results
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error in search_log_first_n_records: %s", e, exc_info=True)
custom_message = f"Failed to search first N logs: {e!s}"
raise McpError(ErrorData(code=-32603, message=custom_message)) from e
class SearchLogLastNInput(BaseSearchInput):
"""Input for search_log_last_n_records."""
count: int = Field(description="Number of last (newest) matching records to return.", gt=0)
@mcp.tool()
async def search_log_last_n_records(
count: int,
scope: str = "default",
context_before: int = 2,
context_after: int = 2,
log_dirs_override: str = "",
log_content_patterns_override: str = "",
) -> list[dict[str, Any]]:
"""Search for the last N (newest) records, optionally filtering, with context."""
logger.info(
"MCP search_log_last_n_records called with count=%s, scope='%s', "
"context=%sB/%sA, log_dirs_override='%s', "
"log_content_patterns_override='%s'",
count,
scope,
context_before,
context_after,
log_dirs_override,
log_content_patterns_override,
)
if count <= 0:
logger.error("Invalid count for search_log_last_n_records: %s. Must be > 0.", count)
raise McpError(ErrorData(code=-32602, message="Count must be a positive integer."))
log_dirs_list = log_dirs_override.split(",") if log_dirs_override else None
log_content_patterns_list = log_content_patterns_override.split(",") if log_content_patterns_override else None
filter_criteria = build_filter_criteria(
last_n=count,
scope=scope,
context_before=context_before,
context_after=context_after,
log_dirs_override=log_dirs_list,
log_content_patterns_override=log_content_patterns_list,
)
try:
results = await asyncio.to_thread(analysis_engine.search_logs, filter_criteria)
logger.info("search_log_last_n_records returning %s records.", len(results))
return results
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error in search_log_last_n_records: %s", e, exc_info=True)
custom_message = f"Failed to search last N logs: {e!s}"
raise McpError(ErrorData(code=-32603, message=custom_message)) from e
@mcp.tool()
async def get_server_env_details() -> dict[str, Any]:
"""Returns sys.path and sys.executable from the running MCP server."""
logger.info("get_server_env_details called.")
details = {
"sys_executable": sys.executable,
"sys_path": sys.path,
"cwd": os.getcwd(),
"environ_pythonpath": os.environ.get("PYTHONPATH"),
}
logger.info(f"Server env details: {details}")
return details
# Main entry point for Uvicorn or direct stdio run via script
# Ref: https://fastmcp.numaru.com/usage/server-integration/#uvicorn-integration
# Ref: https://fastmcp.numaru.com/usage/server-integration/#stdio-transport
def main() -> None:
"""Runs the MCP server, choosing transport based on arguments."""
import argparse
# Argument parsing should be done first
parser = argparse.ArgumentParser(description="Log Analyzer MCP Server")
parser.add_argument(
"--transport",
type=str,
choices=["stdio", "http"],
default=os.getenv("MCP_DEFAULT_TRANSPORT", "stdio"), # Default to stdio
help="Transport protocol to use: 'stdio' or 'http' (default: stdio or MCP_DEFAULT_TRANSPORT env var)",
)
parser.add_argument(
"--host",
type=str,
default=os.getenv("MCP_HTTP_HOST", "127.0.0.1"),
help="Host for HTTP transport (default: 127.0.0.1 or MCP_HTTP_HOST env var)",
)
parser.add_argument(
"--port",
type=int,
default=int(os.getenv("MCP_HTTP_PORT", "8000")),
help="Port for HTTP transport (default: 8000 or MCP_HTTP_PORT env var)",
)
parser.add_argument(
"--log-level",
type=str,
default=os.getenv("MCP_LOG_LEVEL", "info"),
choices=["debug", "info", "warning", "error", "critical"],
help="Logging level for Uvicorn (default: info or MCP_LOG_LEVEL env var)",
)
args = parser.parse_args()
# Uses the global mcp instance and logger already configured at module level.
# logger.info("Logger for main() using global instance.") # Optional: confirm logger usage
if args.transport == "stdio":
logger.info("Starting Log Analyzer MCP server in stdio mode via main().")
mcp.run(transport="stdio") # FastMCP handles stdio internally
elif args.transport == "http":
# Only import uvicorn and ASGIApplication if http transport is selected
try:
import uvicorn
from asgiref.typing import ASGIApplication # For type hinting
from typing import cast
except ImportError as e:
logger.error("Required packages for HTTP transport (uvicorn, asgiref) are not installed. %s", e)
sys.exit(1)
logger.info(
"Starting Log Analyzer MCP server with Uvicorn on %s:%s (log_level: %s)",
args.host,
args.port,
args.log_level,
)
uvicorn.run(cast(ASGIApplication, mcp), host=args.host, port=args.port, log_level=args.log_level)
else:
# Should not happen due to choices in argparse, but as a fallback:
logger.error("Unsupported transport type: %s. Exiting.", args.transport)
sys.exit(1)
if __name__ == "__main__":
# This block now directly calls main() to handle argument parsing and server start.
# This ensures consistency whether run as a script or via the entry point.
logger.info("Log Analyzer MCP Server script execution (__name__ == '__main__'). Calling main().")
main()
```
--------------------------------------------------------------------------------
/tests/log_analyzer_mcp/test_analysis_engine.py:
--------------------------------------------------------------------------------
```python
import os
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional
from unittest import mock # Import mock
import logging # ADDED for mock logger
import pytest
from log_analyzer_mcp.common.config_loader import ConfigLoader
# Ensure correct import path; adjust if your project structure differs
# This assumes tests/ is at the same level as src/
from log_analyzer_mcp.core.analysis_engine import AnalysisEngine, ParsedLogEntry
# --- Fixtures ---
@pytest.fixture
def temp_log_file(tmp_path):
"""Creates a temporary log file with some content for testing."""
log_content = [
"2024-05-27 10:00:00 INFO This is a normal log message.",
"2024-05-27 10:01:00 DEBUG This is a debug message with EXCEPTION details.",
"2024-05-27 10:02:00 WARNING This is a warning.",
"2024-05-27 10:03:00 ERROR This is an error log: Critical Failure.",
"2024-05-27 10:03:30 INFO Another message for context.",
"2024-05-27 10:04:00 INFO And one more after the error.",
"INVALID LOG LINE without timestamp or level",
"2024-05-27 10:05:00 ERROR Another error for positional testing.",
"2024-05-27 10:06:00 INFO Final message.",
]
log_file = tmp_path / "test_log_file.log"
with open(log_file, "w", encoding="utf-8") as f:
for line in log_content:
f.write(line + "\n")
return log_file
@pytest.fixture
def temp_another_log_file(tmp_path):
"""Creates a second temporary log file."""
log_content = [
"2024-05-27 11:00:00 INFO Log from another_module.log",
"2024-05-27 11:01:00 ERROR Specific error in another_module.",
]
log_dir = tmp_path / "another_module"
log_dir.mkdir()
log_file = log_dir / "another_module.log"
with open(log_file, "w", encoding="utf-8") as f:
for line in log_content:
f.write(line + "\n")
return log_file
@pytest.fixture
def temp_nolog_file(tmp_path):
"""Creates a temporary non-log file."""
content = ["This is not a log file.", "It has plain text."]
nolog_file = tmp_path / "notes.txt"
with open(nolog_file, "w", encoding="utf-8") as f:
for line in content:
f.write(line + "\n")
return nolog_file
@pytest.fixture
def sample_env_file(tmp_path):
"""Creates a temporary .env file for config loading tests."""
env_content = [
"LOG_DIRECTORIES=logs/,more_logs/",
"LOG_SCOPE_DEFAULT=logs/default/",
"LOG_SCOPE_MODULE_A=logs/module_a/*.log",
"LOG_SCOPE_MODULE_B=logs/module_b/specific.txt",
"LOG_PATTERNS_ERROR=Exception:.*,Traceback",
"LOG_PATTERNS_WARNING=Warning:.*",
"LOG_CONTEXT_LINES_BEFORE=1",
"LOG_CONTEXT_LINES_AFTER=1",
]
env_file = tmp_path / ".env.test"
with open(env_file, "w", encoding="utf-8") as f:
f.write("\n".join(env_content))
return env_file
@pytest.fixture
def mock_logger(): # ADDED mock_logger fixture
return mock.MagicMock(spec=logging.Logger)
@pytest.fixture
def analysis_engine_with_env(sample_env_file, mock_logger): # ADDED mock_logger
"""Provides an AnalysisEngine instance initialized with a specific .env file."""
project_root_for_env = os.path.dirname(sample_env_file) # tmp_path
os.makedirs(os.path.join(project_root_for_env, "logs", "default"), exist_ok=True)
os.makedirs(os.path.join(project_root_for_env, "logs", "module_a"), exist_ok=True)
os.makedirs(os.path.join(project_root_for_env, "logs", "module_b"), exist_ok=True)
os.makedirs(os.path.join(project_root_for_env, "more_logs"), exist_ok=True)
with open(os.path.join(project_root_for_env, "logs", "default", "default1.log"), "w") as f:
f.write("2024-01-01 00:00:00 INFO Default log 1\n")
with open(os.path.join(project_root_for_env, "logs", "module_a", "a1.log"), "w") as f:
f.write("2024-01-01 00:01:00 INFO Module A log 1\n")
with open(os.path.join(project_root_for_env, "logs", "module_b", "specific.txt"), "w") as f:
f.write("2024-01-01 00:02:00 INFO Module B specific text file\n")
with open(os.path.join(project_root_for_env, "more_logs", "another.log"), "w") as f:
f.write("2024-01-01 00:03:00 INFO More logs another log\n")
engine = AnalysisEngine(
logger_instance=mock_logger,
env_file_path=str(sample_env_file),
project_root_for_config=str(project_root_for_env),
)
# The explicit overriding of engine.config_loader.project_root and reloading attributes is no longer needed
# as it's handled by passing project_root_for_config to AnalysisEngine constructor.
return engine
@pytest.fixture
def analysis_engine_no_env(tmp_path, mock_logger): # ADDED mock_logger
"""Provides an AnalysisEngine instance without a specific .env file (uses defaults)."""
project_root_for_test = tmp_path / "test_project"
project_root_for_test.mkdir()
src_core_dir = project_root_for_test / "src" / "log_analyzer_mcp" / "core"
src_core_dir.mkdir(parents=True, exist_ok=True)
(src_core_dir / "analysis_engine.py").touch() # Still needed for AnalysisEngine to find its relative path
# Pass the test project root to the AnalysisEngine
engine = AnalysisEngine(logger_instance=mock_logger, project_root_for_config=str(project_root_for_test))
# For testing file discovery, ensure log_directories points within our test_project.
# The ConfigLoader, when no .env is found in project_root_for_test, will use its defaults.
# We need to ensure its default `get_log_directories` will be sensible for this test.
# If ConfigLoader's default is ["./"], it will become project_root_for_test relative to project_root_for_test, which is fine.
# Or, we can set it explicitly after init if the default isn't what we want for the test.
# For this fixture, let's assume we want it to search a specific subdir in our test_project.
engine.log_directories = ["logs_default_search"]
logs_default_dir = project_root_for_test / "logs_default_search"
logs_default_dir.mkdir(exist_ok=True)
with open(logs_default_dir / "default_app.log", "w") as f:
f.write("2024-01-01 10:00:00 INFO Default app log in default search path\n")
return engine
# --- Test Cases ---
class TestAnalysisEngineGetTargetLogFiles:
def test_get_target_log_files_override(
self, analysis_engine_no_env, temp_log_file, temp_another_log_file, temp_nolog_file, tmp_path, mock_logger
):
engine = analysis_engine_no_env
# engine.config_loader.project_root is now set to tmp_path / "test_project" via constructor
# For _get_target_log_files, the internal project_root is derived from AnalysisEngine.__file__,
# but config_loader.project_root is used to resolve env_file_path and default .env location.
# The actual log file paths in _get_target_log_files are resolved relative to AnalysisEngine's project_root.
# For these override tests, we are providing absolute paths from tmp_path,
# so we need to ensure the engine's _get_target_log_files method treats tmp_path as its effective root for searching.
# The most straightforward way for this test is to ensure that the AnalysisEngine used here
# has its internal project_root (used for resolving relative log_dirs_override, etc.) aligned with tmp_path.
# This is implicitly handled if AnalysisEngine is inside tmp_path (not the case here) or if paths are absolute.
# The fixture `analysis_engine_no_env` now uses `project_root_for_config` to `tmp_path / "test_project"`.
# The `_get_target_log_files` uses `os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))` for its project root.
# This will be the actual project root. The paths temp_log_file etc are in tmp_path.
# We need to ensure the test operates as if tmp_path is the root for log searching.
# This means the `log_dirs_override` paths should be absolute within tmp_path, which they are.
# The safety check `if not current_search_item.startswith(project_root):` in `_get_target_log_files`
# will compare against the *actual* project root.
# This test needs careful handling of project_root perception.
# Let's ensure the paths provided in overrides are absolute and see if the engine handles them correctly.
# The fixture `analysis_engine_no_env` project_root_for_config is `tmp_path / "test_project"`.
# The AnalysisEngine._get_target_log_files own `project_root` is the real one.
# The test below passes absolute paths from `tmp_path`, so they won't be relative to the engine's own `project_root`.
# The safety check `if not current_search_item.startswith(project_root)` will likely make these paths fail
# unless `tmp_path` is inside the real `project_root` (which it isn't usually).
# This fixture is tricky. Let's simplify: create an engine directly in the test with project_root set to tmp_path.
engine_for_test = AnalysisEngine(logger_instance=mock_logger, project_root_for_config=str(tmp_path))
# 1. Specific log file
override_paths = [str(temp_log_file)]
files = engine_for_test._get_target_log_files(log_dirs_override=override_paths)
assert len(files) == 1
assert str(temp_log_file) in files
# 2. Specific non-log file (should be included if directly specified in override)
override_paths_txt = [str(temp_nolog_file)]
files_txt = engine_for_test._get_target_log_files(log_dirs_override=override_paths_txt)
assert len(files_txt) == 1
assert str(temp_nolog_file) in files_txt
# 3. Directory containing log files
override_paths_dir = [str(temp_log_file.parent)] # tmp_path
files_dir = engine_for_test._get_target_log_files(log_dirs_override=override_paths_dir)
# Should find temp_log_file.log, temp_another_log_file.log (under another_module/)
assert len(files_dir) >= 2
assert str(temp_log_file) in files_dir
assert str(temp_another_log_file) in files_dir
assert (
str(temp_nolog_file) not in files_dir
) # .txt files not picked up from directory scan unless specified directly
# 4. Glob pattern
override_paths_glob = [str(tmp_path / "*.log")]
files_glob = engine_for_test._get_target_log_files(log_dirs_override=override_paths_glob)
assert len(files_glob) == 1
assert str(temp_log_file) in files_glob
assert str(temp_another_log_file) not in files_glob # Not at top level
# 5. Recursive Glob pattern for all .log files
override_paths_rec_glob = [str(tmp_path / "**/*.log")]
files_rec_glob = engine_for_test._get_target_log_files(log_dirs_override=override_paths_rec_glob)
# Expect temp_log_file.log, another_module/another_module.log
# And also test_project/logs_default_search/default_app.log (created by analysis_engine_no_env fixture context within tmp_path)
# if analysis_engine_no_env was used to create files in tmp_path that engine_for_test can see.
# The engine_for_test has project_root as tmp_path. The default_app.log is under tmp_path/test_project/...
assert len(files_rec_glob) == 3 # Updated from 2 to 3
assert str(temp_log_file) in files_rec_glob
assert str(temp_another_log_file) in files_rec_glob
# Find the third file: default_app.log created by analysis_engine_no_env context
# Need to construct its path carefully relative to tmp_path for the check
# analysis_engine_no_env.config_loader.project_root is tmp_path / "test_project"
# analysis_engine_no_env.log_directories is ["logs_default_search"]
# So the file is tmp_path / "test_project" / "logs_default_search" / "default_app.log"
expected_default_app_log = tmp_path / "test_project" / "logs_default_search" / "default_app.log"
assert str(expected_default_app_log) in files_rec_glob
# 6. Mixed list
override_mixed = [str(temp_log_file), str(temp_another_log_file.parent)]
files_mixed = engine_for_test._get_target_log_files(log_dirs_override=override_mixed)
assert len(files_mixed) == 2 # temp_log_file + dir scan of another_module/
assert str(temp_log_file) in files_mixed
assert str(temp_another_log_file) in files_mixed
# 7. Path outside project root (tmp_path is acting as project_root here for engine)
outside_dir = tmp_path.parent / "outside_project_logs"
outside_dir.mkdir(exist_ok=True)
outside_log = outside_dir / "external.log"
with open(outside_log, "w") as f:
f.write("external log\n")
# engine.config_loader.project_root is tmp_path
files_outside = engine_for_test._get_target_log_files(log_dirs_override=[str(outside_log)])
assert len(files_outside) == 0 # Should be skipped
def test_get_target_log_files_scope(
self, analysis_engine_with_env, sample_env_file, mock_logger
): # ADDED mock_logger
engine = analysis_engine_with_env # project_root_for_config is sample_env_file.parent (tmp_path)
# This engine from the fixture `analysis_engine_with_env` already has the mock_logger.
# No need to create a new engine here if `analysis_engine_with_env` is correctly configured
# with `project_root_for_config=str(sample_env_file.parent)`.
project_root_for_env = str(sample_env_file.parent)
# Scope "MODULE_A" -> logs/module_a/*.log (key is lowercased in ConfigLoader)
files_scope_a = engine._get_target_log_files(scope="module_a")
assert len(files_scope_a) == 1
assert os.path.join(project_root_for_env, "logs", "module_a", "a1.log") in files_scope_a
# Scope "MODULE_B" -> logs/module_b/specific.txt (key is lowercased)
files_scope_b = engine._get_target_log_files(scope="module_b")
assert len(files_scope_b) == 1
assert os.path.join(project_root_for_env, "logs", "module_b", "specific.txt") in files_scope_b
# Default scope
files_scope_default = engine._get_target_log_files(scope="default")
assert len(files_scope_default) == 1
assert os.path.join(project_root_for_env, "logs", "default", "default1.log") in files_scope_default
# Non-existent scope should return empty
files_scope_none = engine._get_target_log_files(scope="NONEXISTENT")
assert len(files_scope_none) == 0
def test_get_target_log_files_default_config(
self, analysis_engine_with_env, sample_env_file, mock_logger
): # ADDED mock_logger
engine = analysis_engine_with_env # This engine from fixture already has mock_logger
project_root_for_env = str(sample_env_file.parent)
# Default config LOG_DIRECTORIES should be logs/ and more_logs/
files_default = engine._get_target_log_files() # No scope, no override
assert len(files_default) == 3 # default1.log, a1.log, another.log (specific.txt not a .log)
def test_get_target_log_files_no_config_or_override(
self, analysis_engine_no_env, tmp_path, mock_logger
): # ADDED mock_logger
# This test uses analysis_engine_no_env. Its config_loader has project_root=tmp_path / "test_project".
# It sets engine.log_directories = ["logs_default_search"]
# And creates tmp_path / "test_project" / "logs_default_search" / "default_app.log"
engine = analysis_engine_no_env # This engine from fixture already has mock_logger
# If no .env file is loaded by ConfigLoader, and no override, it uses its internal defaults for log_directories.
# The fixture `analysis_engine_no_env` explicitly sets engine.log_directories = ["logs_default_search"].
# So, it should find the "default_app.log" created by the fixture.
files = engine._get_target_log_files()
assert len(files) == 1
expected_log = tmp_path / "test_project" / "logs_default_search" / "default_app.log"
assert str(expected_log) in files
class TestAnalysisEngineParseLogLine:
def test_parse_log_line_valid(self, analysis_engine_no_env, mock_logger): # ADDED mock_logger
engine = analysis_engine_no_env # This engine from fixture already has mock_logger
line = "2024-05-27 10:00:00 INFO This is a log message."
entry = engine._parse_log_line(line, "/test/file.log", 1)
assert entry is not None
assert entry["timestamp"] == datetime(2024, 5, 27, 10, 0, 0)
assert entry["level"] == "INFO"
assert entry["message"] == "This is a log message."
assert entry["raw_line"] == line
assert entry["file_path"] == "/test/file.log"
assert entry["line_number"] == 1
line_millis = "2024-05-27 10:00:00,123 DEBUG Another message."
parsed_millis = engine._parse_log_line(line_millis, "/test/file.log", 2)
assert parsed_millis is not None
assert parsed_millis["timestamp"] == datetime(2024, 5, 27, 10, 0, 0) # Millis are stripped for now
assert parsed_millis["level"] == "DEBUG"
assert parsed_millis["message"] == "Another message."
def test_parse_log_line_invalid(self, analysis_engine_no_env, mock_logger): # ADDED mock_logger
engine = analysis_engine_no_env # This engine from fixture already has mock_logger
line = "This is not a standard log line."
entry = engine._parse_log_line(line, "/test/file.log", 1)
assert entry is not None
assert entry["timestamp"] is None
assert entry["level"] == "UNKNOWN"
assert entry["message"] == line
assert entry["raw_line"] == line
class TestAnalysisEngineContentFilters:
@pytest.fixture
def sample_entries(self) -> List[ParsedLogEntry]:
return [
{
"level": "INFO",
"message": "Application started successfully.",
"raw_line": "...",
"file_path": "app.log",
"line_number": 1,
},
{
"level": "DEBUG",
"message": "User authentication attempt for user 'test'.",
"raw_line": "...",
"file_path": "app.log",
"line_number": 2,
},
{
"level": "WARNING",
"message": "Warning: Disk space low.",
"raw_line": "...",
"file_path": "app.log",
"line_number": 3,
},
{
"level": "ERROR",
"message": "Exception: NullPointerException occurred.",
"raw_line": "...",
"file_path": "app.log",
"line_number": 4,
},
{
"level": "ERROR",
"message": "Traceback (most recent call last):",
"raw_line": "...",
"file_path": "app.log",
"line_number": 5,
},
]
def test_apply_content_filters_override(
self, analysis_engine_no_env, sample_entries, mock_logger
): # ADDED mock_logger
engine = analysis_engine_no_env # This engine from fixture already has mock_logger
filter_criteria_exact = {"log_content_patterns_override": ["exact phrase to match"]}
results_exact = engine._apply_content_filters(sample_entries, filter_criteria_exact)
assert len(results_exact) == 0 # MODIFIED: Expect 0 results for a non-matching phrase
def test_apply_content_filters_config_based(
self, analysis_engine_with_env, sample_entries, mock_logger
): # ADDED mock_logger
engine = analysis_engine_with_env # This engine from fixture already has mock_logger
# .env.test defines LOG_PATTERNS_ERROR=Exception:.*,Traceback
# We will test that providing a level_filter correctly uses these.
filter_criteria = {"level_filter": "ERROR"} # MODIFIED: Test for ERROR level
results_config = engine._apply_content_filters(sample_entries, filter_criteria)
# Should match the two ERROR entries from sample_entries based on LOG_PATTERNS_ERROR
assert len(results_config) == 2
error_messages = {e["message"] for e in results_config}
assert "Exception: NullPointerException occurred." in error_messages
assert "Traceback (most recent call last):" in error_messages
class TestAnalysisEngineTimeFilters:
@pytest.fixture
def time_entries(self) -> List[ParsedLogEntry]:
"""Provides sample parsed log entries with varying timestamps for time filter tests."""
# Use a fixed "now" for consistent test data generation
fixed_now = datetime(2024, 5, 28, 12, 0, 0) # Example: May 28, 2024, 12:00:00 PM
def _create_entry(file_path: str, line_num: int, msg: str, ts: Optional[datetime]) -> ParsedLogEntry:
return {
"timestamp": ts,
"message": msg,
"raw_line": f"{fixed_now.strftime('%Y-%m-%d %H:%M:%S')} {msg}",
"file_path": file_path,
"line_number": line_num,
}
entries = [
_create_entry("t.log", 1, "5 mins ago", fixed_now - timedelta(minutes=5)),
_create_entry("t.log", 2, "30 mins ago", fixed_now - timedelta(minutes=30)),
_create_entry("t.log", 3, "70 mins ago", fixed_now - timedelta(hours=1, minutes=10)),
_create_entry("t.log", 4, "1 day ago", fixed_now - timedelta(days=1)),
_create_entry("t.log", 5, "2 days 1 hour ago", fixed_now - timedelta(days=2, hours=1)),
_create_entry("t.log", 6, "No timestamp", None),
]
return entries
@mock.patch("log_analyzer_mcp.core.analysis_engine.dt.datetime") # Mock dt.datetime in the SUT module
def test_apply_time_filters_minutes(
self, mock_dt_datetime, analysis_engine_no_env, time_entries, mock_logger
): # ADDED mock_logger
engine = analysis_engine_no_env # This engine from fixture already has mock_logger
# Test scenario: current time is 2024-05-28 12:00:00 (from time_entries fixture setup)
# We want to find entries from the last 10 minutes.
mock_dt_datetime.now.return_value = datetime(2024, 5, 28, 12, 0, 0) # Match fixed_now in time_entries
filter_criteria = {"minutes": 10} # Last 10 minutes
# Expected: only "5 mins ago" (2024-05-28 11:55:00) is within 10 mins of 12:00:00
results = engine._apply_time_filters(time_entries, filter_criteria)
assert len(results) == 1
assert results[0]["message"] == "5 mins ago"
@mock.patch("log_analyzer_mcp.core.analysis_engine.dt.datetime") # Mock dt.datetime
def test_apply_time_filters_hours(
self, mock_dt_datetime, analysis_engine_no_env, time_entries, mock_logger
): # ADDED mock_logger
engine = analysis_engine_no_env # This engine from fixture already has mock_logger
# Test scenario: current time is 2024-05-28 12:00:00 (from time_entries fixture setup)
# We want to find entries from the last 1 hour.
mock_dt_datetime.now.return_value = datetime(2024, 5, 28, 12, 0, 0) # Match fixed_now in time_entries
filter_criteria = {"hours": 1} # Last 1 hour (60 minutes)
# Expected: "5 mins ago" (11:55), "30 mins ago" (11:30)
# Excluded: "70 mins ago" (10:50)
results = engine._apply_time_filters(time_entries, filter_criteria)
assert len(results) == 2
assert results[0]["message"] == "5 mins ago"
assert results[1]["message"] == "30 mins ago"
@mock.patch("log_analyzer_mcp.core.analysis_engine.dt.datetime") # Mock dt.datetime
def test_apply_time_filters_days(
self, mock_dt_datetime, analysis_engine_no_env, time_entries, mock_logger
): # ADDED mock_logger
engine = analysis_engine_no_env # This engine from fixture already has mock_logger
# Test scenario: current time is 2024-05-28 12:00:00 (from time_entries fixture setup)
# We want to find entries from the last 1 day.
mock_dt_datetime.now.return_value = datetime(2024, 5, 28, 12, 0, 0) # Match fixed_now in time_entries
filter_criteria = {"days": 1} # Last 1 day
# Expected: "5 mins ago", "30 mins ago", "70 mins ago", "1 day ago"
# Excluded: "2 days 1 hour ago"
results = engine._apply_time_filters(time_entries, filter_criteria)
assert len(results) == 4
assert results[0]["message"] == "5 mins ago"
assert results[1]["message"] == "30 mins ago"
assert results[2]["message"] == "70 mins ago"
assert results[3]["message"] == "1 day ago"
@mock.patch("log_analyzer_mcp.core.analysis_engine.dt.datetime") # Mock dt.datetime
def test_apply_time_filters_no_criteria(
self, mock_dt_datetime, analysis_engine_no_env, time_entries, mock_logger
): # ADDED mock_logger
engine = analysis_engine_no_env # This engine from fixture already has mock_logger
fixed_now_for_filter = datetime(2024, 5, 28, 12, 0, 0) # Matches time_entries fixed_now for consistency
mock_dt_datetime.now.return_value = fixed_now_for_filter
filter_criteria = {} # No time filter
filtered = engine._apply_time_filters(time_entries, filter_criteria)
# If no time filter is applied, _apply_time_filters returns all original entries.
assert len(filtered) == len(time_entries) # MODIFIED: Expect all 6 entries
class TestAnalysisEnginePositionalFilters:
@pytest.fixture
def positional_entries(self, mock_logger) -> List[ParsedLogEntry]: # ADDED mock_logger
# Create a dummy engine just to use its _parse_log_line, or parse manually.
# For simplicity, manual creation or use a static method if _parse_log_line was static.
# Let's manually create them to avoid needing an engine instance here.
# engine = AnalysisEngine(logger_instance=mock_logger) # Not needed if we construct manually
base_time = datetime(2024, 1, 1, 10, 0, 0)
return [
{
"timestamp": base_time + timedelta(seconds=1), # ADDED timestamp
"level": "INFO",
"message": "Application started successfully.",
"raw_line": "2024-01-01 10:00:01 INFO Application started successfully.", # MODIFIED raw_line for consistency
"file_path": "app.log",
"line_number": 1,
},
{
"timestamp": base_time + timedelta(seconds=2), # ADDED timestamp
"level": "DEBUG",
"message": "User authentication attempt for user 'test'.",
"raw_line": "2024-01-01 10:00:02 DEBUG User authentication attempt for user 'test'.", # MODIFIED raw_line
"file_path": "app.log",
"line_number": 2,
},
{
"timestamp": base_time + timedelta(seconds=3), # ADDED timestamp
"level": "WARNING",
"message": "Warning: Disk space low.",
"raw_line": "2024-01-01 10:00:03 WARNING Warning: Disk space low.", # MODIFIED raw_line
"file_path": "app.log",
"line_number": 3,
},
{
"timestamp": base_time + timedelta(seconds=4), # ADDED timestamp
"level": "ERROR",
"message": "Exception: NullPointerException occurred.",
"raw_line": "2024-01-01 10:00:04 ERROR Exception: NullPointerException occurred.", # MODIFIED raw_line
"file_path": "app.log",
"line_number": 4,
},
{
"timestamp": base_time + timedelta(seconds=5), # ADDED timestamp
"level": "ERROR",
"message": "Traceback (most recent call last):",
"raw_line": "2024-01-01 10:00:05 ERROR Traceback (most recent call last):", # MODIFIED raw_line
"file_path": "app.log",
"line_number": 5,
},
{ # Entry with no timestamp, should be filtered out by _apply_positional_filters
"timestamp": None,
"level": "UNKNOWN",
"message": "Entry 6 No Timestamp",
"raw_line": "Entry 6 No Timestamp",
"file_path": "app.log",
"line_number": 6,
},
]
def test_apply_positional_filters_first_n(self, analysis_engine_no_env, positional_entries):
engine = analysis_engine_no_env # project_root_for_config is tmp_path / "test_project"
filter_criteria = {"first_n": 2}
filtered = engine._apply_positional_filters(positional_entries, filter_criteria)
assert len(filtered) == 2
assert filtered[0]["message"] == "Application started successfully."
assert filtered[1]["message"] == "User authentication attempt for user 'test'."
def test_apply_positional_filters_last_n(self, analysis_engine_no_env, positional_entries):
engine = analysis_engine_no_env # project_root_for_config is tmp_path / "test_project"
filter_criteria = {"last_n": 2}
# Note: the 'first' flag in _apply_positional_filters is True by default.
# The main search_logs method would set it to False for last_n.
# Here we test the direct call with first=False
filtered = engine._apply_positional_filters(positional_entries, filter_criteria)
assert len(filtered) == 2
assert filtered[0]["message"] == "Exception: NullPointerException occurred."
assert filtered[1]["message"] == "Traceback (most recent call last):"
def test_apply_positional_filters_n_larger_than_list(self, analysis_engine_no_env, positional_entries):
engine = analysis_engine_no_env # project_root_for_config is tmp_path / "test_project"
filter_criteria_first = {"first_n": 10}
filtered_first = engine._apply_positional_filters(positional_entries, filter_criteria_first)
# positional_entries has 6 items, 1 has no timestamp. _apply_positional_filters works on the 5 with timestamps.
assert len(filtered_first) == len(positional_entries) - 1 # MODIFIED
filter_criteria_last = {"last_n": 10}
filtered_last = engine._apply_positional_filters(positional_entries, filter_criteria_last)
assert len(filtered_last) == len(positional_entries) - 1 # MODIFIED
def test_apply_positional_filters_no_criteria(self, analysis_engine_no_env, positional_entries):
engine = analysis_engine_no_env # project_root_for_config is tmp_path / "test_project"
# Should return all entries because no positional filter is active.
filtered = engine._apply_positional_filters(positional_entries, {})
assert len(filtered) == len(positional_entries) # MODIFIED
# Verify that the order is preserved if no sorting was done
# or that it's sorted by original line number if timestamps are mixed.
class TestAnalysisEngineExtractContextLines:
def test_extract_context_lines(self, analysis_engine_no_env, temp_log_file, mock_logger):
# Use the fixture-provided engine, or create one specifically for the test if needed.
# engine = analysis_engine_no_env # This engine from fixture already has mock_logger
# Create an engine specifically for this test, ensuring its project_root is tmp_path
# so that it can correctly find temp_log_file if relative paths were used (though temp_log_file is absolute).
engine_for_test = AnalysisEngine(
logger_instance=mock_logger, project_root_for_config=str(temp_log_file.parent)
) # MODIFIED
all_lines_by_file = {}
with open(temp_log_file, "r") as f:
all_lines = [line.strip() for line in f.readlines()]
all_lines_by_file[str(temp_log_file)] = all_lines
# Simulate some parsed entries that matched
# Match on line "2024-05-27 10:03:00 ERROR This is an error log: Critical Failure." which is all_lines[3] (0-indexed)
parsed_entries: List[ParsedLogEntry] = [
{
"timestamp": datetime(2024, 5, 27, 10, 3, 0),
"level": "ERROR",
"message": "This is an error log: Critical Failure.",
"raw_line": all_lines[3],
"file_path": str(temp_log_file),
"line_number": 4, # 1-indexed
}
]
# Context: 1 before, 1 after
contextualized_entries = engine_for_test._extract_context_lines(parsed_entries, all_lines_by_file, 1, 1)
assert len(contextualized_entries) == 1
entry = contextualized_entries[0]
assert "context_before_lines" in entry
assert "context_after_lines" in entry
assert len(entry["context_before_lines"]) == 1
assert entry["context_before_lines"][0] == all_lines[2] # "2024-05-27 10:02:00 WARNING This is a warning."
assert len(entry["context_after_lines"]) == 1
assert (
entry["context_after_lines"][0] == all_lines[4]
) # "2024-05-27 10:03:30 INFO Another message for context."
# Context: 2 before, 2 after
contextualized_entries_2 = engine_for_test._extract_context_lines(parsed_entries, all_lines_by_file, 2, 2)
assert len(contextualized_entries_2) == 1
entry2 = contextualized_entries_2[0]
assert len(entry2["context_before_lines"]) == 2
assert entry2["context_before_lines"][0] == all_lines[1]
assert entry2["context_before_lines"][1] == all_lines[2]
assert len(entry2["context_after_lines"]) == 2
assert entry2["context_after_lines"][0] == all_lines[4]
assert entry2["context_after_lines"][1] == all_lines[5]
# Edge case: Match at the beginning of the file
parsed_entry_first: List[ParsedLogEntry] = [
{
"timestamp": datetime(2024, 5, 27, 10, 0, 0),
"level": "INFO",
"message": "This is a normal log message.",
"raw_line": all_lines[0],
"file_path": str(temp_log_file),
"line_number": 1,
}
]
contextualized_first = engine_for_test._extract_context_lines(parsed_entry_first, all_lines_by_file, 2, 2)
assert len(contextualized_first[0]["context_before_lines"]) == 0
assert len(contextualized_first[0]["context_after_lines"]) == 2
assert contextualized_first[0]["context_after_lines"][0] == all_lines[1]
assert contextualized_first[0]["context_after_lines"][1] == all_lines[2]
# Edge case: Match at the end of the file
parsed_entry_last: List[ParsedLogEntry] = [
{
"timestamp": datetime(2024, 5, 27, 10, 6, 0),
"level": "INFO",
"message": "Final message.",
"raw_line": all_lines[8], # "2024-05-27 10:06:00 INFO Final message."
"file_path": str(temp_log_file),
"line_number": 9,
}
]
contextualized_last = engine_for_test._extract_context_lines(parsed_entry_last, all_lines_by_file, 2, 2)
assert len(contextualized_last[0]["context_before_lines"]) == 2
assert contextualized_last[0]["context_before_lines"][0] == all_lines[6] # INVALID LOG LINE...
assert contextualized_last[0]["context_before_lines"][1] == all_lines[7] # 2024-05-27 10:05:00 ERROR...
assert len(contextualized_last[0]["context_after_lines"]) == 0
class TestAnalysisEngineSearchLogs:
def test_search_logs_all_records(self, analysis_engine_no_env, temp_log_file, tmp_path, mock_logger):
# For this test, we need the engine to consider tmp_path as its effective project root for searching.
# The fixture `analysis_engine_no_env` has project_root set to `tmp_path / "test_project"`.
# To simplify and ensure `temp_log_file` (which is directly under `tmp_path`) is found correctly:
engine_for_test = AnalysisEngine(logger_instance=mock_logger, project_root_for_config=str(tmp_path)) # MODIFIED
filter_criteria = {"log_dirs_override": [str(temp_log_file)]}
results = engine_for_test.search_logs(filter_criteria)
# Print mock_logger calls for debugging
print("\n---- MOCK LOGGER CALLS (test_search_logs_all_records) ----")
for call_obj in mock_logger.info.call_args_list:
print(f"INFO: {call_obj}")
for call_obj in mock_logger.debug.call_args_list:
print(f"DEBUG: {call_obj}")
print("-----------------------------------------------------------")
# temp_log_file has 9 lines, all should be parsed (some as UNKNOWN)
assert len(results) == 9
assert all("raw_line" in r for r in results)
def test_search_logs_content_filter(self, analysis_engine_no_env, temp_log_file, tmp_path, mock_logger):
# engine = analysis_engine_no_env
engine_for_test = AnalysisEngine(logger_instance=mock_logger, project_root_for_config=str(tmp_path)) # MODIFIED
filter_criteria = {
"log_dirs_override": [str(temp_log_file)],
"log_content_patterns_override": [r"\\\\bERROR\\\\b", "Critical Failure"],
}
results = engine_for_test.search_logs(filter_criteria)
# Expecting 1 line:
# "2024-05-27 10:03:00 ERROR This is an error log: Critical Failure."
# because only "Critical Failure" matches a message. r"\\bERROR\\b" does not match any message.
assert len(results) == 1
messages = sorted([r["message"] for r in results])
assert "This is an error log: Critical Failure." in messages
assert "Another error for positional testing." not in messages # This message doesn't contain "\\bERROR\\b"
def test_search_logs_time_filter(self, analysis_engine_no_env, temp_log_file, tmp_path, mock_logger):
# This test needs to mock datetime.now()
# engine = analysis_engine_no_env
engine_for_test = AnalysisEngine(logger_instance=mock_logger, project_root_for_config=str(tmp_path)) # MODIFIED
# Create a log file where entries are time-sensitive
log_content = [
"2024-05-27 10:00:00 INFO This is a normal log message.",
"2024-05-27 10:01:00 DEBUG This is a debug message with EXCEPTION details.",
"2024-05-27 10:02:00 WARNING This is a warning.",
"2024-05-27 10:03:00 ERROR This is an error log: Critical Failure.",
"2024-05-27 10:03:30 INFO Another message for context.",
"2024-05-27 10:04:00 INFO And one more after the error.",
"INVALID LOG LINE without timestamp or level",
"2024-05-27 10:05:00 ERROR Another error for positional testing.",
"2024-05-27 10:06:00 INFO Final message.",
]
log_file = tmp_path / "test_log_file.log"
with open(log_file, "w", encoding="utf-8") as f:
for line in log_content:
f.write(line + "\n")
# Placeholder for robust time test - requires mocking or more setup
pass
def test_search_logs_positional_filter(self, analysis_engine_no_env, temp_log_file, tmp_path, mock_logger):
# engine = analysis_engine_no_env
engine_for_test = AnalysisEngine(logger_instance=mock_logger, project_root_for_config=str(tmp_path)) # MODIFIED
filter_criteria_first = {
"log_dirs_override": [str(temp_log_file)],
"first_n": 2,
}
results_first = engine_for_test.search_logs(filter_criteria_first)
assert len(results_first) == 2
assert results_first[0]["raw_line"].startswith("2024-05-27 10:00:00 INFO")
assert results_first[1]["raw_line"].startswith("2024-05-27 10:01:00 DEBUG")
filter_criteria_last = {
"log_dirs_override": [str(temp_log_file)],
"last_n": 2,
}
results_last = engine_for_test.search_logs(filter_criteria_last)
assert len(results_last) == 2
# Lines are sorted by timestamp (if available), then line number within file.
# Last 2 lines from temp_log_file are:
# "2024-05-27 10:05:00 ERROR Another error for positional testing."
# "2024-05-27 10:06:00 INFO Final message."
assert results_last[0]["raw_line"].startswith("2024-05-27 10:05:00 ERROR")
assert results_last[1]["raw_line"].startswith("2024-05-27 10:06:00 INFO")
def test_search_logs_with_context(self, analysis_engine_no_env, temp_log_file, tmp_path, mock_logger):
# engine = analysis_engine_no_env
engine_for_test = AnalysisEngine(logger_instance=mock_logger, project_root_for_config=str(tmp_path)) # MODIFIED
filter_criteria = {
"log_dirs_override": [str(temp_log_file)],
"log_content_patterns_override": ["This is an error log: Critical Failure"],
"context_before": 1,
"context_after": 1,
}
results = engine_for_test.search_logs(filter_criteria)
assert len(results) == 1
assert results[0]["message"] == "This is an error log: Critical Failure."
assert "2024-05-27 10:02:00 WARNING This is a warning." in results[0]["context_before_lines"]
assert "2024-05-27 10:03:30 INFO Another message for context." in results[0]["context_after_lines"]
def test_search_logs_no_matches(self, analysis_engine_no_env, temp_log_file, tmp_path, mock_logger):
# engine = analysis_engine_no_env
engine_for_test = AnalysisEngine(logger_instance=mock_logger, project_root_for_config=str(tmp_path)) # MODIFIED
filter_criteria = {
"log_dirs_override": [str(temp_log_file)],
"log_content_patterns_override": ["NONEXISTENTPATTERNXYZ123"],
}
results = engine_for_test.search_logs(filter_criteria)
assert len(results) == 0
def test_search_logs_multiple_files_and_sorting(
self, analysis_engine_no_env, temp_log_file, temp_another_log_file, tmp_path, mock_logger
):
# engine = analysis_engine_no_env
engine_for_test = AnalysisEngine(logger_instance=mock_logger, project_root_for_config=str(tmp_path)) # MODIFIED
# Test that logs from multiple files are aggregated and sorted correctly by time
filter_criteria = {
"log_dirs_override": [str(temp_log_file), str(temp_another_log_file)],
"log_content_patterns_override": [r"\\\\bERROR\\\\b"], # Match messages containing whole word "ERROR"
}
results = engine_for_test.search_logs(filter_criteria)
# temp_log_file messages: "This is an error log: Critical Failure.", "Another error for positional testing."
# temp_another_log_file message: "Specific error in another_module."
# None of these messages contain the standalone word "ERROR".
assert len(results) == 0
```