Align new logging approach with current implementation in MCM 1

Author: KarolinaPomian

tests/validation/Engine/README.md

@@ -0,0 +1,105 @@
+# CSV and HTML Reporting for Media Communications Mesh Test Framework
+
+This directory contains utilities for generating CSV and HTML reports from test results.
+
+## Overview
+
+The reporting framework consists of the following components:
+
+1. `csv_report.py` - Core functionality for recording test results in CSV format
+2. `logging_utils.py` - Helper functions and decorators for logging
+3. `output_validator.py` - Enhanced output validation with reporting integration
+4. `html_report.py` - HTML report generation from CSV data
+5. `pytest_hooks.py` - Pytest hooks for automatic report generation
+
+## Usage
+
+### Basic Usage
+
+The reporting framework is automatically enabled when running tests. Test results will be recorded in CSV format, and an HTML report will be generated at the end of the test run.
+
+```bash
+pytest -v tests/validation
+```
+
+### Logging Test Results
+
+To manually log test results from test functions, use the decorators from `logging_utils.py`:
+
+```python
+from Engine.logging_utils import log_ffmpeg_test, log_media_proxy_test, log_mesh_agent_test, log_rxtx_app_test
+
+@log_ffmpeg_test(test_name="test_ffmpeg_encoding", test_description="Test FFmpeg encoding performance")
+def test_ffmpeg_encoding():
+    # Your test code here
+    pass
+```
+
+### Output Validation
+
+For validating component output and logging results:
+
+```python
+from Engine.output_validator import output_validator
+
+def test_media_proxy():
+    # Run media proxy
+    # ...
+    
+    # Validate output
+    result = output_validator.validate_media_proxy(
+        output=output_text,
+        test_name="test_media_proxy",
+        duration=execution_time
+    )
+    
+    assert result["is_pass"], f"Media proxy test failed: {result['errors']}"
+```
+
+### Command Line Options
+
+Additional command line options are available:
+
+```bash
+pytest -v tests/validation --timestamp  # Add timestamp to log directory
+```
+
+## Report Files
+
+The following report files are generated:
+
+- `test_report.csv` - CSV file containing test results
+- `compliance_report.csv` - CSV file containing compliance results
+- `test_report.html` - HTML report with interactive filtering and visualization
+
+## API Reference
+
+### csv_report.py
+
+- `initialize_csv_report(log_dir)` - Initialize CSV report files
+- `csv_add_test(test_name, test_description, component, result, duration, error_message)` - Add a test result
+- `update_compliance_result(requirement_id, is_compliant)` - Update compliance result
+- `csv_write_report(log_dir)` - Write test results and compliance results to CSV files
+- `get_test_summary()` - Get summary of test results
+- `get_component_summary()` - Get summary of results by component
+- `get_compliance_summary()` - Get summary of compliance results
+
+### logging_utils.py
+
+- `log_test_result(component, test_name, test_description, error_message)` - Decorator for logging test results
+- `log_ffmpeg_test(test_name, test_description, error_message)` - FFmpeg test decorator
+- `log_media_proxy_test(test_name, test_description, error_message)` - Media Proxy test decorator
+- `log_mesh_agent_test(test_name, test_description, error_message)` - Mesh Agent test decorator
+- `log_rxtx_app_test(test_name, test_description, error_message)` - RxTx App test decorator
+- `parse_output_for_errors(component, output, custom_error_keywords)` - Parse output for errors
+- `streamline_component_output_validation(component, output, test_name, duration, custom_error_keywords)` - Validate component output
+
+### output_validator.py
+
+- `OutputValidator` - Class for validating component output and logging results
+- `output_validator` - Singleton instance of OutputValidator
+- `set_output_validator_log_path(log_path)` - Set log path for the validator
+
+### html_report.py
+
+- `generate_html_report(log_dir)` - Generate HTML report from CSV data

tests/validation/Engine/csv_report.py

@@ -0,0 +1,216 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright 2024-2025 Intel Corporation
+# Media Communications Mesh
+
+import csv
+import logging
+import os
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+# Constants
+CSV_REPORT_FILE = "test_report.csv"
+COMPLIANCE_REPORT_FILE = "compliance_report.csv"
+TEST_RESULT_COLUMN = "Test Result"
+TEST_NAME_COLUMN = "Test Name"
+TEST_DESC_COLUMN = "Test Description"
+TEST_TIME_COLUMN = "Test Time"
+TEST_DURATION_COLUMN = "Test Duration"
+TEST_ERROR_COLUMN = "Error Message"
+COMPONENT_COLUMN = "Component"
+
+# Components
+COMPONENT_FFMPEG = "FFmpeg"
+COMPONENT_MEDIA_PROXY = "Media Proxy"
+COMPONENT_MESH_AGENT = "Mesh Agent"
+COMPONENT_RXTX_APP = "RxTx App"
+
+# Standard column headers
+CSV_HEADERS = [
+    TEST_NAME_COLUMN,
+    TEST_DESC_COLUMN,
+    COMPONENT_COLUMN,
+    TEST_RESULT_COLUMN,
+    TEST_TIME_COLUMN,
+    TEST_DURATION_COLUMN,
+    TEST_ERROR_COLUMN,
+]
+
+# Global storage for test results
+_test_results = []
+_compliance_results = {}
+
+def initialize_csv_report(log_dir: Optional[str] = None) -> None:
+    """
+    Initialize CSV report files.
+    
+    Args:
+        log_dir: Directory where report files will be stored
+    """
+    global _test_results, _compliance_results
+    _test_results = []
+    _compliance_results = {}
+    
+    # Create log directory if it doesn't exist
+    if log_dir:
+        Path(log_dir).mkdir(parents=True, exist_ok=True)
+
+
+def csv_add_test(
+    test_name: str,
+    test_description: str,
+    component: str,
+    result: bool,
+    duration: float,
+    error_message: str = "",
+) -> None:
+    """
+    Add a test result to the CSV report.
+    
+    Args:
+        test_name: Name of the test
+        test_description: Description of the test
+        component: Component being tested (FFmpeg, Media Proxy, Mesh Agent, RxTx App)
+        result: True if test passed, False if test failed
+        duration: Test duration in seconds
+        error_message: Error message if test failed
+    """
+    global _test_results
+    
+    test_result = "PASS" if result else "FAIL"
+    test_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    
+    _test_results.append({
+        TEST_NAME_COLUMN: test_name,
+        TEST_DESC_COLUMN: test_description,
+        COMPONENT_COLUMN: component,
+        TEST_RESULT_COLUMN: test_result,
+        TEST_TIME_COLUMN: test_time,
+        TEST_DURATION_COLUMN: f"{duration:.2f}s",
+        TEST_ERROR_COLUMN: error_message
+    })
+    
+    logger.info(f"Added test result: {test_name} - {test_result}")
+
+
+def update_compliance_result(requirement_id: str, is_compliant: bool) -> None:
+    """
+    Update compliance result for a specific requirement.
+    
+    Args:
+        requirement_id: ID of the requirement
+        is_compliant: True if compliant, False if not
+    """
+    global _compliance_results
+    
+    _compliance_results[requirement_id] = is_compliant
+    logger.info(f"Updated compliance for {requirement_id}: {'COMPLIANT' if is_compliant else 'NON-COMPLIANT'}")
+
+
+def csv_write_report(log_dir: str = ".") -> None:
+    """
+    Write test results and compliance results to CSV files.
+    
+    Args:
+        log_dir: Directory where report files will be stored
+    """
+    report_path = Path(log_dir) / CSV_REPORT_FILE
+    compliance_path = Path(log_dir) / COMPLIANCE_REPORT_FILE
+    
+    # Write test results
+    if _test_results:
+        try:
+            with open(report_path, 'w', newline='') as csvfile:
+                writer = csv.DictWriter(csvfile, fieldnames=CSV_HEADERS)
+                writer.writeheader()
+                for result in _test_results:
+                    writer.writerow(result)
+            logger.info(f"Test report written to {report_path}")
+        except Exception as e:
+            logger.error(f"Failed to write test report: {e}")
+    
+    # Write compliance results
+    if _compliance_results:
+        try:
+            with open(compliance_path, 'w', newline='') as csvfile:
+                writer = csv.DictWriter(csvfile, fieldnames=['Requirement ID', 'Status'])
+                writer.writeheader()
+                for req_id, is_compliant in _compliance_results.items():
+                    writer.writerow({
+                        'Requirement ID': req_id,
+                        'Status': 'COMPLIANT' if is_compliant else 'NON-COMPLIANT'
+                    })
+            logger.info(f"Compliance report written to {compliance_path}")
+        except Exception as e:
+            logger.error(f"Failed to write compliance report: {e}")
+
+
+def get_test_summary() -> Dict[str, int | str]:
+    """
+    Get a summary of test results.
+    
+    Returns:
+        Dictionary with total, passed, and failed test counts and pass rate
+    """
+    pass_count = sum(1 for r in _test_results if r[TEST_RESULT_COLUMN] == "PASS")
+    fail_count = sum(1 for r in _test_results if r[TEST_RESULT_COLUMN] == "FAIL")
+    total_count = len(_test_results)
+    
+    return {
+        "total": total_count,
+        "passed": pass_count,
+        "failed": fail_count,
+        "pass_rate": f"{(pass_count / total_count * 100):.1f}%" if total_count > 0 else "N/A"
+    }
+
+
+def get_component_summary() -> Dict[str, Dict[str, int | str]]:
+    """
+    Get a summary of test results by component.
+    
+    Returns:
+        Dictionary with component-wise test summaries
+    """
+    components = {}
+    
+    for result in _test_results:
+        component = result[COMPONENT_COLUMN]
+        if component not in components:
+            components[component] = {"total": 0, "passed": 0, "failed": 0}
+        
+        components[component]["total"] += 1
+        if result[TEST_RESULT_COLUMN] == "PASS":
+            components[component]["passed"] += 1
+        else:
+            components[component]["failed"] += 1
+    
+    # Add pass rate
+    for component in components:
+        total = components[component]["total"]
+        passed = components[component]["passed"]
+        components[component]["pass_rate"] = f"{(passed / total * 100):.1f}%" if total > 0 else "N/A"
+    
+    return components
+
+
+def get_compliance_summary() -> Dict[str, int | str]:
+    """
+    Get a summary of compliance results.
+    
+    Returns:
+        Dictionary with total, compliant, and non-compliant counts and compliance rate
+    """
+    compliant_count = sum(1 for v in _compliance_results.values() if v)
+    non_compliant_count = sum(1 for v in _compliance_results.values() if not v)
+    total_count = len(_compliance_results)
+    
+    return {
+        "total": total_count,
+        "compliant": compliant_count,
+        "non_compliant": non_compliant_count,
+        "compliance_rate": f"{(compliant_count / total_count * 100):.1f}%" if total_count > 0 else "N/A"
+    }