feat: add possibility to separate log file on testCase and testRun (#557)

Author: yannpoupon

CHANGELOG.md

@@ -249,7 +249,7 @@ Commits have to follow following convention: https://www.conventionalcommits.org
 
 - Fix step report show wrong result ([#309](https://github.com/eclipse/kiso-testing/issues/309))
 - Update python version ([#314](https://github.com/eclipse/kiso-testing/issues/314))
-- Handle proxy autostart when being instanciated after auxiliaries and close channel at last
+- Handle proxy autostart when being instantiated after auxiliaries and close channel at last
 - Use shutil.move for renaming the merged trace to avoid windows errors
 - Add junitxml option to default pytest options ([#321](https://github.com/eclipse/kiso-testing/issues/321))
 
@@ -590,7 +590,7 @@ Commits have to follow following convention: https://www.conventionalcommits.org
 
 ### Documentation
 
-- Add whats new for 0.17.0
+- Add what's new for 0.17.0
 - Make usage of TestSuite elements more visible
 - Add quality goals
 
@@ -619,8 +619,8 @@ Commits have to follow following convention: https://www.conventionalcommits.org
 ### Documentation
 
 - Remove not maintained 'list of limitations' section
-- Add section whats new ([#38](https://github.com/eclipse/kiso-testing/issues/38))
-- Replace all occurences of pipenv with poetry
+- Add section what's new ([#38](https://github.com/eclipse/kiso-testing/issues/38))
+- Replace all occurrences of pipenv with poetry
 - Rework getting_started
 
 ### New Features

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pykiso"
-version = "1.4.0"
+version = "1.5.0"
 description = "Embedded integration testing framework."
 authors = ["Sebastian Fischer <sebastian.fischer@de.bosch.com>"]
 license = "Eclipse Public License - v 2.0"
@@ -58,7 +58,7 @@ rich = { version = "^13.2.0", optional = true }
 robotframework = "3.2.2"
 tabulate = ">=0.8.9,<0.10.0"
 unittest-xml-reporting = "^3.2.0"
-viztracer=  "^1.0.1"
+viztracer = "^1.0.1"
 fastapi = "^0.115.3"
 uvicorn = "^0.32.0"
 xmltodict = "^0.14.2"

src/pykiso/cli.py

@@ -183,6 +183,15 @@ def parse_args(self, ctx, args):
     multiple=True,
     help="path to log-file or folder. If not set will log to STDOUT",
 )
+@click.option(
+    "-s",
+    "--log-file-strategy",
+    required=False,
+    default=None,
+    type=click.Choice(["testRun", "testCase"], case_sensitive=True),
+    help="Strategy for the log file creation, 'testRun' for one log file per run, 'testCase' for one log file per test case. Default is one log file for all tests."
+    "This flag need to be used with --log-path option.",
+)
 @click.option(
     "--log-level",
     required=False,
@@ -237,7 +246,8 @@ def parse_args(self, ctx, args):
 def main(
     click_context: click.Context,
     test_configuration_file: Tuple[PathType],
-    log_path: Tuple[PathType] = None,
+    log_path: Tuple[PathType] | None = None,
+    log_file_strategy: str | None = None,
     log_level: str = "INFO",
     report_type: str = "text",
     step_report: Optional[PathType] = None,
@@ -258,6 +268,8 @@ def main(
     :param click_context: click context
     :param test_configuration_file: path to the YAML config file
     :param log_path: path to existing directories or files to write logs to
+    :param log_file_strategy: Strategy for the log file creation, 'testRun' for one log file per run, 'testCase'
+        for one log file per test case. Default is one log file for all tests."
     :param log_level: any of DEBUG, INFO, WARNING, ERROR
     :param report_type: if "test", the standard report, if "junit", a junit report is generated
     :param variant: allow the user to execute a subset of tests based on variants
@@ -274,6 +286,9 @@ def main(
             f"Mismatch: {len(log_path)} log files were provided for {len(test_configuration_file)} yaml configuration files"
         )
 
+    if log_file_strategy is not None and log_path is None:
+        raise click.UsageError("The --log-file-strategy option requires the --log-path option to be set.")
+
     if junit is not None:
         report_type = "junit"
 
@@ -302,14 +317,7 @@ def main(
         # Run tests
         with ConfigRegistry.provide_auxiliaries(cfg_dict):
             exit_code = test_execution.execute(
-                cfg_dict,
-                report_type,
-                yaml_name,
-                user_tags,
-                step_report,
-                pattern,
-                failfast,
-                junit,
+                cfg_dict, report_type, yaml_name, user_tags, step_report, pattern, failfast, junit, log_file_strategy
             )
 
         for handler in logging.getLogger().handlers:

src/pykiso/test_coordinator/test_execution.py

@@ -27,11 +27,11 @@
 
 import functools
 import os
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Literal, Optional, Union
 from unittest import util
 from unittest.loader import VALID_MODULE_NAME
 
-from pykiso.test_result.multi_result import MultiTestResult
+import pykiso.test_result.multi_result as multi_result
 
 if TYPE_CHECKING:
     from ..lib.connectors.cc_pcan_can import CCPCanCan
@@ -120,7 +120,7 @@ def should_skip(test_case: BasicTest) -> bool:
         for cli_tag_id, cli_tag_value in usr_tags.items():
             # skip any test case that doesn't define a CLI-provided tag name
             if cli_tag_id not in test_case.tag.keys():
-                test_case._skip_msg = f"provided tag {cli_tag_id!r} not present in test tags"
+                test_case._skip_msg = f"provided tag {cli_tag_id!r} not present in test tags"  # noqa: E713
                 return True
             # skip any test case that which tag value don't match the provided tag's value
             cli_tag_values = cli_tag_value if isinstance(cli_tag_value, list) else [cli_tag_value]
@@ -415,6 +415,7 @@ def execute(
     pattern_inject: Optional[str] = None,
     failfast: bool = False,
     junit_path: str = "reports",
+    log_file_strategy: Literal["testRun", "testCase"] | None = None,
 ) -> int:
     """Create test environment based on test configuration.
 
@@ -429,6 +430,7 @@ def execute(
         run specific tests.
     :param failfast: stop the test run on the first error or failure.
     :param junit_path: path (file or dir) to junit report
+    :param log_file_strategy: strategy for the log file creation
 
     :return: exit code corresponding to the result of the test execution
         (tests failed, unexpected exception, ...)
@@ -456,6 +458,7 @@ def execute(
             )
 
         log_file_path = get_logging_options().log_path
+        current_time = datetime.today()
         # TestRunner selection: generate or not a junit report. Start the tests and publish the results
         if report_type == "junit":
             report_path = junit_path
@@ -464,30 +467,37 @@ def execute(
                 junit_report_path = str(full_report_path)
                 full_report_path.parent.mkdir(exist_ok=True)
             else:
-                junit_report_path = str(full_report_path / Path(time.strftime(f"%Y-%m-%d_%H-%M-%S-{report_name}.xml")))
+                junit_report_path = str(
+                    full_report_path / Path(current_time.strftime(f"%Y-%m-%d_%H-%M-%S-{report_name}.xml"))
+                )
                 full_report_path.mkdir(exist_ok=True)
             with open(junit_report_path, "wb") as junit_output, ResultStream(log_file_path) as stream:
-                test_runner = xmlrunner.XMLTestRunner(
+                multi_result.test_runner_instance = xmlrunner.XMLTestRunner(
                     output=junit_output,
-                    resultclass=MultiTestResult(XmlTestResult, BannerTestResult),
+                    resultclass=multi_result.MultiTestResult(
+                        XmlTestResult, BannerTestResult, log_file_strategy=log_file_strategy
+                    ),
                     failfast=failfast,
                     verbosity=0,
                     stream=stream,
                 )
-                result = test_runner.run(all_tests_to_run)
+                result = multi_result.test_runner_instance.run(all_tests_to_run)
+                # Generate the html step report
+                if step_report is not None:
+                    generate_step_report(result, step_report)
+
         else:
             with ResultStream(log_file_path) as stream:
-                test_runner = unittest.TextTestRunner(
+                multi_result.test_runner_instance = unittest.TextTestRunner(
                     stream=stream,
-                    resultclass=MultiTestResult(BannerTestResult),
+                    resultclass=multi_result.MultiTestResult(BannerTestResult, log_file_strategy=log_file_strategy),
                     failfast=failfast,
                     verbosity=0,
                 )
-                result = test_runner.run(all_tests_to_run)
-
-        # Generate the html step report
-        if step_report is not None:
-            generate_step_report(result, step_report)
+                result = multi_result.test_runner_instance.run(all_tests_to_run)
+                # Generate the html step report
+                if step_report is not None:
+                    generate_step_report(result, step_report)
 
         exit_code = failure_and_error_handling(result)
     except NameError:

src/pykiso/test_result/multi_result.py

@@ -20,30 +20,47 @@
 """
 from __future__ import annotations
 
+import logging
+import unittest
+from datetime import datetime
 from inspect import signature
-from typing import Any, List, Optional, TextIO, Union
-from unittest import TestResult
+from pathlib import Path
+from typing import Any, List, Literal, Optional, Union
+from unittest import TestResult, util
 from unittest.case import _SubTest
 
 from pykiso.types import ExcInfoType
 
+from ..logging_initializer import get_internal_level, get_logging_options
 from ..test_coordinator.test_case import BasicTest
 from ..test_coordinator.test_suite import BaseTestSuite
 from .text_result import BannerTestResult
 from .xml_result import XmlTestResult
 
+test_runner_instance: Optional[unittest.TextTestRunner] = None
+
 
 class MultiTestResult:
     """Class that can take multiple test result classes and ran the
     test for all the classes.
     """
 
-    def __init__(self, *result_classes: TestResult):
+    def __init__(
+        self,
+        *result_classes: TestResult,
+        log_file_strategy: Literal["testRun,testCase"] | None = None,
+    ):
         """Initialize parameter
 
         :param result_classes: test result classes
+        :param log_file_strategy: strategy to use for the log file
+            - testRun: log file will be created for each test run
+            - testCase: log file will be created for each test case
         """
-        self.result_classes = result_classes
+        self.result_classes: list[TestResult] = result_classes
+        self._log_file_strategy = log_file_strategy
+        self._list_test_results: list[BasicTest] = []
+        self.current_log_file: Path | None = None
 
     def __call__(self, *args, **kwargs) -> MultiTestResult:
         """Initialize the result classes with the parameters passed in arguments."""
@@ -83,7 +100,7 @@ def __setattr__(self, name: str, value: Any) -> Any:
         """
         super().__setattr__(name, value)
         # condition to avoid infinite loop with the __getattr__
-        if name != "result_classes":
+        if name not in ["result_classes", "_log_file_strategy", "_list_test_results", "current_log_file"]:
             for result in self.result_classes:
                 setattr(result, name, value)
 
@@ -99,9 +116,43 @@ def startTest(self, test: Union[BasicTest, BaseTestSuite]) -> None:
 
         :param test: running testcase
         """
+        self.handle_log_file_strategy(test)
+
         for result in self.result_classes:
             result.startTest(test)
 
+    def handle_log_file_strategy(self, test: Union[BasicTest, BaseTestSuite]) -> None:
+        """Handle the log file strategy for the given test.
+
+        :param test: The test case or test suite being executed.
+        """
+        if self._log_file_strategy is None:
+            return
+
+        log_options = get_logging_options()
+
+        if test.__class__ not in self._list_test_results or self._log_file_strategy == "testRun":
+            log_file_name = util.strclass(test.__class__).replace(".", "_").replace("-", "_")
+            if self._log_file_strategy == "testRun":
+                log_file_name += "_" + test._testMethodName
+
+            log_file_name += f"_{datetime.today().strftime('%Y%d%m%H%M%S')}.log"
+            self.current_log_file = log_options.log_path.parent / log_file_name
+
+        # Setup the file handler for logging
+        log_format = logging.Formatter("%(asctime)s [%(levelname)s] %(module)s:%(lineno)d: %(message)s")
+        root_logger = logging.getLogger()
+        file_handler = logging.FileHandler(self.current_log_file, "a+")
+        file_handler.setFormatter(log_format)
+        file_handler.name = "strategy_log_file_handler"
+        # always include internal logs in log files
+        file_handler.setLevel(get_internal_level(log_options.log_level))
+        root_logger.addHandler(file_handler)
+
+        # Add the log file to the test runner instance to get the banner information
+        test_runner_instance.stream.stream.multifile_handler.add_file(self.current_log_file)
+        self._list_test_results.append(test.__class__)
+
     def startTestRun(self) -> None:
         """Call the startTestRun function for all result classes."""
         for result in self.result_classes:
@@ -125,6 +176,16 @@ def stopTest(self, test: Union[BasicTest, BaseTestSuite]) -> None:
         for result in self.result_classes:
             result.stopTest(test)
 
+        if self._log_file_strategy is None:
+            return
+        # Remove the log file
+        test_runner_instance.stream.stream.multifile_handler.remove_file(self.current_log_file)
+        root_logger = logging.getLogger()
+        for handler in root_logger.handlers:
+            if handler.name == "strategy_log_file_handler":
+                root_logger.removeHandler(handler)
+                handler.close()
+
     def addSuccess(self, test: Union[BasicTest, BaseTestSuite]) -> None:
         """Call the addSuccess function for all result classes.