Implement Comprehensive Log Analysis and Reporting Mechanism

.gitignore

@@ -1,22 +1,34 @@
-### AL ###
-#Template for AL projects for Dynamics 365 Business Central
-#launch.json folder
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+
+# Virtual Environments
+venv/
+env/
+.env/
+
+# Distribution / packaging
+dist/
+build/
+*.egg-info/
+
+# Logs
+*.log
+logs/
+
+# Testing
+.pytest_cache/
+.coverage
+htmlcov/
+
+# IDEs and editors
 .vscode/
-#Cache folder
-.alcache/
-#Symbols folder
-.alpackages/
-#Snapshots folder
-.snapshots/
-#Testing Output folder
-.output/
-#Extension App-file
-*.app
-#Rapid Application Development File
-rad.json
-#Translation Base-file
-*.g.xlf
-#License-file
-*.flf
-#Test results file
-TestResults.xml
+.idea/
+*.swp
+*.swo
+
+# OS generated files
+.DS_Store
+Thumbs.db

src/log_analysis/__init__.py

@@ -0,0 +1,3 @@
+from .log_parser import LogParser
+
+__all__ = ['LogParser']

src/log_analysis/log_parser.py

@@ -0,0 +1,123 @@
+from typing import List, Dict, Any
+import json
+import os
+from datetime import datetime
+
+class LogParser:
+    """
+    A class responsible for parsing and analyzing iteration logs.
+
+    This class provides methods to:
+    - Read log files
+    - Parse log entries
+    - Analyze performance metrics
+    - Generate summary reports
+    """
+
+    def __init__(self, log_directory: str = 'logs'):
+        """
+        Initialize the LogParser with a specific log directory.
+
+        Args:
+            log_directory (str): Directory containing log files. Defaults to 'logs'.
+        """
+        self.log_directory = log_directory
+
+        # Ensure log directory exists
+        os.makedirs(log_directory, exist_ok=True)
+
+    def get_log_files(self) -> List[str]:
+        """
+        Retrieve all log files in the specified directory.
+
+        Returns:
+            List[str]: List of log file paths
+        """
+        return [
+            os.path.join(self.log_directory, f) 
+            for f in os.listdir(self.log_directory) 
+            if f.endswith('.json')
+        ]
+
+    def parse_log_file(self, file_path: str) -> List[Dict[str, Any]]:
+        """
+        Parse a single log file and return its contents.
+
+        Args:
+            file_path (str): Path to the log file
+
+        Returns:
+            List[Dict[str, Any]]: Parsed log entries
+
+        Raises:
+            FileNotFoundError: If the log file doesn't exist
+            json.JSONDecodeError: If the log file is not valid JSON
+        """
+        try:
+            with open(file_path, 'r') as log_file:
+                return json.load(log_file)
+        except FileNotFoundError:
+            raise FileNotFoundError(f"Log file not found: {file_path}")
+        except json.JSONDecodeError:
+            raise ValueError(f"Invalid JSON in log file: {file_path}")
+
+    def analyze_performance(self, log_entries: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """
+        Analyze performance metrics from log entries.
+
+        Args:
+            log_entries (List[Dict[str, Any]]): List of log entries to analyze
+
+        Returns:
+            Dict[str, Any]: Performance summary metrics
+        """
+        if not log_entries:
+            return {}
+
+        # Extract performance-related metrics
+        performance_metrics = {
+            'total_iterations': len(log_entries),
+            'start_time': log_entries[0].get('timestamp'),
+            'end_time': log_entries[-1].get('timestamp'),
+            'error_rate': sum(1 for entry in log_entries if entry.get('status') == 'error') / len(log_entries),
+            'performance_scores': [entry.get('performance_score', 0) for entry in log_entries]
+        }
+
+        # Calculate additional metrics
+        if performance_metrics['performance_scores']:
+            performance_metrics.update({
+                'avg_performance': sum(performance_metrics['performance_scores']) / len(performance_metrics['performance_scores']),
+                'max_performance': max(performance_metrics['performance_scores']),
+                'min_performance': min(performance_metrics['performance_scores'])
+            })
+
+        return performance_metrics
+
+    def generate_report(self) -> Dict[str, Any]:
+        """
+        Generate a comprehensive report by analyzing all log files.
+
+        Returns:
+            Dict[str, Any]: Comprehensive log analysis report
+        """
+        log_files = self.get_log_files()
+        report = {
+            'total_log_files': len(log_files),
+            'file_analyses': []
+        }
+
+        for log_file in log_files:
+            try:
+                log_entries = self.parse_log_file(log_file)
+                file_report = {
+                    'file_name': os.path.basename(log_file),
+                    'performance': self.analyze_performance(log_entries)
+                }
+                report['file_analyses'].append(file_report)
+            except Exception as e:
+                report['file_analyses'].append({
+                    'file_name': os.path.basename(log_file),
+                    'error': str(e)
+                })
+
+        return report

tests/log_analysis/test_log_parser.py

@@ -0,0 +1,110 @@
+import os
+import json
+import pytest
+from typing import List, Dict, Any
+from src.log_analysis.log_parser import LogParser
+
+@pytest.fixture
+def sample_log_data() -> List[Dict[str, Any]]:
+    """Generate sample log data for testing."""
+    return [
+        {
+            'timestamp': '2023-01-01T00:00:00',
+            'iteration': 1,
+            'performance_score': 0.75,
+            'status': 'success'
+        },
+        {
+            'timestamp': '2023-01-01T00:01:00',
+            'iteration': 2,
+            'performance_score': 0.85,
+            'status': 'success'
+        },
+        {
+            'timestamp': '2023-01-01T00:02:00',
+            'iteration': 3,
+            'performance_score': 0.5,
+            'status': 'error'
+        }
+    ]
+
+@pytest.fixture
+def log_directory(tmp_path, sample_log_data):
+    """Create a temporary log directory with sample log files."""
+    log_dir = tmp_path / "logs"
+    log_dir.mkdir()
+
+    # Create multiple log files
+    for i in range(3):
+        log_file = log_dir / f"log_{i}.json"
+        with open(log_file, 'w') as f:
+            json.dump(sample_log_data, f)
+
+    return str(log_dir)
+
+def test_log_parser_initialization(log_directory):
+    """Test LogParser initialization."""
+    parser = LogParser(log_directory)
+    assert parser.log_directory == log_directory
+    assert os.path.exists(log_directory)
+
+def test_get_log_files(log_directory):
+    """Test retrieving log files."""
+    parser = LogParser(log_directory)
+    log_files = parser.get_log_files()
+
+    assert len(log_files) == 3
+    assert all(f.endswith('.json') for f in log_files)
+
+def test_parse_log_file(log_directory, sample_log_data):
+    """Test parsing a single log file."""
+    parser = LogParser(log_directory)
+    log_files = parser.get_log_files()
+    parsed_logs = parser.parse_log_file(log_files[0])
+
+    assert parsed_logs == sample_log_data
+    assert len(parsed_logs) == 3
+
+def test_analyze_performance(log_directory, sample_log_data):
+    """Test performance analysis of log entries."""
+    parser = LogParser(log_directory)
+    performance_metrics = parser.analyze_performance(sample_log_data)
+
+    assert performance_metrics['total_iterations'] == 3
+    assert performance_metrics['error_rate'] == pytest.approx(1/3)
+    assert performance_metrics['avg_performance'] == pytest.approx(0.7)
+    assert performance_metrics['max_performance'] == 0.85
+    assert performance_metrics['min_performance'] == 0.5
+
+def test_generate_report(log_directory):
+    """Test generating a comprehensive log report."""
+    parser = LogParser(log_directory)
+    report = parser.generate_report()
+
+    assert report['total_log_files'] == 3
+    assert len(report['file_analyses']) == 3
+
+    for file_analysis in report['file_analyses']:
+        assert 'file_name' in file_analysis
+        assert 'performance' in file_analysis
+
+def test_invalid_log_file(tmp_path):
+    """Test handling of invalid log files."""
+    invalid_log_dir = tmp_path / "invalid_logs"
+    invalid_log_dir.mkdir()
+
+    # Create an invalid JSON file
+    with open(invalid_log_dir / "invalid.json", 'w') as f:
+        f.write("Not a valid JSON")
+
+    parser = LogParser(str(invalid_log_dir))
+
+    with pytest.raises(ValueError):
+        parser.parse_log_file(str(invalid_log_dir / "invalid.json"))
+
+def test_nonexistent_log_file():
+    """Test handling of nonexistent log files."""
+    parser = LogParser('/nonexistent/path')
+
+    with pytest.raises(FileNotFoundError):
+        parser.parse_log_file('/nonexistent/path/log.json')