Added --code flag for collecting code context.

doc/en/introduction.rst

@@ -455,6 +455,7 @@ Command line
 
       --label LABEL         Label the test report with the given name, useful to categorize or classify similar reports (aka "run-id").
       --driver-info         Display drivers startup and teardown information, and visualise driver connections in the report.
+      --code                Collects file path, line number and code context of the assertions.
 
 
 Highlighted features

doc/newsfragments/2981_new.code_context.rst

@@ -0,0 +1,2 @@
+Added ``--code`` flag to collect code context for the assertions. Code context one-liner will be displayed on the web UI if enabled.
+Note that file path information is no longer collected by default. To collect file path information, enable code context.

testplan/base.py

@@ -146,6 +146,8 @@ class Testplan(entity.RunnableManager):
         categorize or classify similar reports .
     :param driver_info: Display driver setup / teardown time and driver
         interconnection information in UI report.
+    :param collect_code_context: Collects the file path, line number and code
+        context of the assertions.
     """
 
     CONFIG = TestplanConfig
@@ -194,6 +196,7 @@ def __init__(
         extra_deps: Optional[List[Union[str, ModuleType]]] = None,
         label: Optional[str] = None,
         driver_info: bool = False,
+        collect_code_context: bool = False,
         auto_part_runtime_limit: int = defaults.AUTO_PART_RUNTIME_LIMIT,
         plan_runtime_target: int = defaults.PLAN_RUNTIME_TARGET,
         **options,
@@ -256,6 +259,7 @@ def __init__(
             extra_deps=extra_deps,
             label=label,
             driver_info=driver_info,
+            collect_code_context=collect_code_context,
             auto_part_runtime_limit=auto_part_runtime_limit,
             plan_runtime_target=plan_runtime_target,
             **options,
@@ -401,6 +405,7 @@ def main_wrapper(
         extra_deps=None,
         label=None,
         driver_info=False,
+        collect_code_context=False,
         auto_part_runtime_limit=defaults.AUTO_PART_RUNTIME_LIMIT,
         plan_runtime_target=defaults.PLAN_RUNTIME_TARGET,
         **options,
@@ -462,6 +467,7 @@ def test_plan_inner_inner():
                     extra_deps=extra_deps,
                     label=label,
                     driver_info=driver_info,
+                    collect_code_context=collect_code_context,
                     auto_part_runtime_limit=auto_part_runtime_limit,
                     plan_runtime_target=plan_runtime_target,
                     **options,

testplan/parser.py

@@ -499,6 +499,14 @@ def generate_parser(self) -> HelpParser:
             help="Display drivers setup / teardown timing and interconnection information in UI report.",
         )
 
+        report_group.add_argument(
+            "--code",
+            dest="collect_code_context",
+            action="store_true",
+            default=self._default_options["collect_code_context"],
+            help="Collects file path, line number and code context of the assertions.",
+        )
+
         self.add_arguments(parser)
         return parser
 

testplan/runnable/base.py

@@ -267,6 +267,7 @@ def get_options(cls):
                 "skip_strategy", default=common.SkipStrategy.noop()
             ): Use(common.SkipStrategy.from_option_or_none),
             ConfigOption("driver_info", default=False): bool,
+            ConfigOption("collect_code_context", default=False): bool,
         }
 
 

testplan/testing/base.py

@@ -901,6 +901,14 @@ def driver_info(self) -> bool:
             return False
         return self.cfg.driver_info
 
+    @property
+    def collect_code_context(self) -> bool:
+        """
+        Collecting the file path, line number and code context of the assertions
+        if enabled.
+        """
+        return getattr(self.cfg, "collect_code_context", False)
+
 
 class ProcessRunnerTestConfig(TestConfig):
     """

testplan/testing/multitest/base.py

@@ -30,7 +30,7 @@
 )
 from testplan.testing.multitest import suite as mtest_suite
 from testplan.testing.multitest.entries import base as entries_base
-from testplan.testing.result import report_target
+from testplan.testing.result import report_target, collect_code_context
 from testplan.testing.multitest.suite import (
     get_suite_metadata,
     get_testcase_metadata,
@@ -1158,6 +1158,9 @@ def _run_testcase(
                 ),
             )
 
+        if getattr(self.cfg, "collect_code_context", False):
+            testcase = collect_code_context(func=testcase)
+
         # specially handle skipped testcases
         if hasattr(testcase, "__should_skip__"):
             with compose_contexts(

testplan/testing/multitest/entries/base.py

@@ -50,6 +50,7 @@ def __init__(self, description, category=None, flag=None):
         # Will be set explicitly via containers
         self.line_no = None
         self.file_path = None
+        self.code_context = None
 
     def __str__(self):
         return repr(self)

testplan/testing/multitest/entries/schemas/base.py

@@ -33,6 +33,7 @@ class BaseSchema(Schema):
     category = fields.String()
     flag = fields.String()
     file_path = fields.String()
+    code_context = fields.String()
     custom_style = fields.Dict(keys=fields.String(), values=fields.String())
 
     def load(self, *args, **kwargs):

testplan/testing/result.py

@@ -93,12 +93,14 @@ def __exit__(self, exc_type, exc_value, tb):
             description=self.description,
         )
 
-        with MOD_LOCK:
-            # TODO: see https://github.com/python/cpython/commit/85cf1d514b84dc9a4bcb40e20a12e1d82ff19f20
-            caller_frame = inspect.stack()[1]
+        if getattr(assertion_state, "collect_code_context", False):
+            with MOD_LOCK:
+                # TODO: see https://github.com/python/cpython/commit/85cf1d514b84dc9a4bcb40e20a12e1d82ff19f20
+                caller_frame = inspect.stack()[1]
 
-        exc_assertion.file_path = os.path.abspath(caller_frame[1])
-        exc_assertion.line_no = caller_frame[2]
+            exc_assertion.file_path = os.path.abspath(caller_frame[1])
+            exc_assertion.line_no = caller_frame[2]
+            exc_assertion.code_context = caller_frame.code_context[0].strip()
 
         # We cannot use `bind_entry` here as this block will
         # be run when an exception is raised
@@ -113,6 +115,19 @@ def __exit__(self, exc_type, exc_value, tb):
 assertion_state = threading.local()
 
 
+def collect_code_context(func: Callable) -> Callable:
+    """
+    Sets the decorated function to collect code context
+    """
+
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        assertion_state.collect_code_context = True
+        func(*args, **kwargs)
+
+    return wrapper
+
+
 def report_target(func: Callable, ref_func: Callable = None) -> Callable:
     """
     Sets the decorated function's filepath and line-range in assertion state.
@@ -165,7 +180,10 @@ def wrapper(result, *args, **kwargs):
             custom_style = kwargs.pop("custom_style", None)
             dryrun = kwargs.pop("dryrun", False)
             entry = func(result, *args, **kwargs)
-            if top_assertion:
+            if not top_assertion:
+                return entry
+
+            if getattr(assertion_state, "collect_code_context", False):
                 with MOD_LOCK:
                     call_stack = inspect.stack()
                     try:
@@ -183,34 +201,35 @@ def wrapper(result, *args, **kwargs):
                                 frame = call_stack[1]
                         entry.file_path = os.path.abspath(frame.filename)
                         entry.line_no = frame.lineno
+                        entry.code_context = frame.code_context[0].strip()
                     finally:
                         # https://docs.python.org/3/library/inspect.html
                         del frame
                         del call_stack
 
-                if custom_style is not None:
-                    if not isinstance(custom_style, dict):
-                        raise TypeError(
-                            "Use `dict[str, str]` to specify custom CSS style"
-                        )
-                    entry.custom_style = custom_style
+            if custom_style is not None:
+                if not isinstance(custom_style, dict):
+                    raise TypeError(
+                        "Use `dict[str, str]` to specify custom CSS style"
+                    )
+                entry.custom_style = custom_style
 
-                assert isinstance(result, AssertionNamespace) or isinstance(
-                    result, Result
-                ), "Incorrect usage of assertion decorator"
+            assert isinstance(result, AssertionNamespace) or isinstance(
+                result, Result
+            ), "Incorrect usage of assertion decorator"
 
-                if isinstance(result, AssertionNamespace):
-                    result = result.result
+            if isinstance(result, AssertionNamespace):
+                result = result.result
 
-                if not dryrun:
-                    result.entries.append(entry)
+            if not dryrun:
+                result.entries.append(entry)
 
-                stdout_registry.log_entry(
-                    entry=entry, stdout_style=result.stdout_style
-                )
+            stdout_registry.log_entry(
+                entry=entry, stdout_style=result.stdout_style
+            )
 
-                if not entry and not result.continue_on_failure:
-                    raise AssertionError(entry)
+            if not entry and not result.continue_on_failure:
+                raise AssertionError(entry)
 
             return entry
         finally:
@@ -1371,10 +1390,13 @@ def __exit__(self, exc_type, exc_value, traceback):
             return False
         super().__exit__(exc_type, exc_value, traceback)
 
-        with MOD_LOCK:
-            # TODO: see https://github.com/python/cpython/commit/85cf1d514b84dc9a4bcb40e20a12e1d82ff19f20
-            # XXX: do we have concrete ideas about thread-safety here?
-            caller_frame = inspect.stack()[1]
+        if getattr(assertion_state, "collect_code_context", False):
+            with MOD_LOCK:
+                # TODO: see https://github.com/python/cpython/commit/85cf1d514b84dc9a4bcb40e20a12e1d82ff19f20
+                # XXX: do we have concrete ideas about thread-safety here?
+                caller_frame = inspect.stack()[1]
+        else:
+            caller_frame = None
 
         assertion = assertions.LogfileMatch(
             self.timeout,
@@ -1383,8 +1405,11 @@ def __exit__(self, exc_type, exc_value, traceback):
             self.description,
             self.category,
         )
-        assertion.file_path = os.path.abspath(caller_frame[1])
-        assertion.line_no = caller_frame[2]
+
+        if caller_frame:
+            assertion.file_path = os.path.abspath(caller_frame[1])
+            assertion.line_no = caller_frame[2]
+            # assertion.code_context = caller_frame.code_context[0].strip()
 
         stdout_registry.log_entry(
             entry=assertion, stdout_style=self.result.stdout_style