Decide keyword call banner and secret log suppression in run_keyword

Browser/browser.py

@@ -20,12 +20,12 @@
 import sys
 import time
 import types
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from concurrent.futures._base import Future
 from copy import copy
 from datetime import timedelta
 from pathlib import Path
-from typing import Any, Literal
+from typing import Any, Literal, get_args
 
 from assertionengine import AssertionOperator
 from overrides import overrides
@@ -86,6 +86,7 @@
     SupportedBrowsers,
     TracingGroupMode,
 )
+from .utils.types import Secret
 from .version import __version__ as VERSION
 
 KW_CALL_CONTENT_TEMPLATE = """body::before {{
@@ -107,6 +108,10 @@
     {additional_styles}
 }}"""
 
+SECRET_ARGUMENT = "secret"
+SECRET_MASK = "***"
+BANNER_MUTED_KEYWORDS = ("take_screenshot", "get_page_source")
+
 KW_CALL_BANNER_FUNCTION = """(content) => {
     const kwCallBanner = document.getElementById('kwCallBanner');
     if (kwCallBanner) {
@@ -576,6 +581,8 @@ def __init__(  # noqa: PLR0915
         self._unresolved_promises: set[Future] = set()
         self._keyword_formatters: dict = {}
         self._current_loglevel: str | None = None
+        self._logging_suppressions = 0
+        self._secret_arguments: dict[str, set[str]] = {}
         self.is_test_case_running = False
         self.auto_closing_default_run_before_unload: bool = False
         self.keyword_call_stack: list[KeywordCallStackEntry] = []
@@ -926,20 +933,13 @@ def _start_keyword(self, name, attrs):
             source,
             attrs["lineno"],
             attrs["type"],
+            kwname=attrs["kwname"],
         )
         self.keyword_call_stack.append(kw_call_stack_entry)
         if self.tracing_group_mode == TracingGroupMode.Full:
-            self._playwright_state.open_trace_group(**kw_call_stack_entry)
-        if not (
-            self.show_keyword_call_banner is False
-            or (self.show_keyword_call_banner is None and not self.presenter_mode)
-            or attrs["libname"] != "Browser"
-            or attrs["status"] == "NOT RUN"
-        ):
-            self._show_keyword_call(attrs)
-        if "secret" in attrs["kwname"].lower() and attrs["libname"] == "Browser":
-            self._set_logging(False)
-
+            self._playwright_state.open_trace_group(
+                **self._trace_group_arguments(kw_call_stack_entry)
+            )
         if attrs["type"] == "Teardown":
             timeout_pattern = "Test timeout .* exceeded."
             test = EXECUTION_CONTEXTS.current.test
@@ -957,10 +957,13 @@ def _create_keyword_call_stack_entry(
         source: str | Path | None,
         lineno: int,
         typ: str,
+        *,
+        kwname: str = "",
     ) -> KeywordCallStackEntry:
         if typ not in ["SETUP", "KEYWORD", "TEARDOWN"]:
             args = [name] if name else []
             name = typ
+            kwname = typ
         try:
             lineno = int(lineno)
         except (ValueError, TypeError):
@@ -971,15 +974,27 @@ def _create_keyword_call_stack_entry(
             ),
             "file": str(source),
             "line": lineno,
+            "kwname": kwname,
+            "args": list(args),
         }
 
+    @staticmethod
+    def _trace_group_arguments(entry: KeywordCallStackEntry) -> dict[str, Any]:
+        return {"name": entry["name"], "file": entry["file"], "line": entry["line"]}
+
     def run_keyword(self, name, args, kwargs=None):
+        is_secret_keyword = self._is_secret_keyword(name)
         try:
+            if is_secret_keyword:
+                self._set_logging(False)
+            self._show_keyword_call(name)
             if (
                 self.tracing_group_mode == TracingGroupMode.Browser
                 and self.keyword_call_stack
             ):
-                self._playwright_state.open_trace_group(**(self.keyword_call_stack[-1]))
+                self._playwright_state.open_trace_group(
+                    **self._trace_group_arguments(self.keyword_call_stack[-1])
+                )
             return DynamicCore.run_keyword(self, name, args, kwargs)
         except (AssertionError, AttributeError) as e:
             selector = self._get_selector_value_from_keyword_call(name, args, kwargs)
@@ -999,6 +1014,8 @@ def run_keyword(self, name, args, kwargs=None):
                 and self.keyword_call_stack
             ):
                 self._playwright_state.close_trace_group()
+            if is_secret_keyword:
+                self._set_logging(True)
 
     def _get_selector_value_from_keyword_call(self, name, args, kwargs):
         selector = kwargs.get("selector")
@@ -1029,8 +1046,6 @@ def _end_keyword(self, _name, attrs):
             self.keyword_call_stack.pop()
         if self.tracing_group_mode == TracingGroupMode.Full:
             self._playwright_state.close_trace_group()
-        if "secret" in attrs["kwname"].lower() and attrs["libname"] == "Browser":
-            self._set_logging(True)
 
     def _end_test(self, name, attrs):
         self._remove_from_scope_stack(attrs["id"])
@@ -1161,23 +1176,121 @@ def _set_logging(self, status: bool):
         except DataError:
             context = BuiltIn()
         if status:
-            if self._current_loglevel:
+            self._logging_suppressions = max(0, self._logging_suppressions - 1)
+            if self._logging_suppressions == 0 and self._current_loglevel:
                 context.set_log_level(self._current_loglevel)
                 self._current_loglevel = None
         else:
-            self._current_loglevel = context.set_log_level("NONE")
+            if self._logging_suppressions == 0:
+                self._current_loglevel = context.set_log_level("NONE")
+            self._logging_suppressions += 1
+
+    def _resolve_keyword_function(self, name: str) -> Callable | None:
+        """A translation replaces the registered name, the function stays the same."""
+        return self.keywords.get(name)
 
-    def _show_keyword_call(self, attrs):
+    def _keyword_argument_names(self, name: str) -> list[str]:
         try:
-            if attrs["kwname"] in ["Take Screenshot", "Get Page Source"]:
-                self.set_keyword_call_banner()
-            else:
-                args = "    ".join(attrs["args"])
-                args = BuiltIn().replace_variables(args)
-                content = f"{attrs['kwname']}{'    ' * bool(attrs['args'])}{args}"
-                self.set_keyword_call_banner(content)
+            arguments = self.get_keyword_arguments(name)
+        except Exception:
+            return []
+        names = [
+            argument[0] if isinstance(argument, tuple) else argument
+            for argument in arguments
+        ]
+        return [argument for argument in names if not argument.startswith("*")]
+
+    def _secret_argument_names(self, name: str) -> set[str]:
+        """Both rules are needed: the annotation misses a plugin that types its
+        secret as a string, the name misses `Create Credential`, whose secrets
+        are called ``privateKey`` and ``publicKey``.
+        """
+        if name not in self._secret_arguments:
+            self._secret_arguments[name] = self._find_secret_arguments(name)
+        return self._secret_arguments[name]
+
+    def _find_secret_arguments(self, name: str) -> set[str]:
+        if self._resolve_keyword_function(name) is None:
+            return set()
+        try:
+            argument_types = self.get_keyword_types(name) or {}
         except Exception:
-            pass
+            return set()
+        return {
+            argument
+            for argument, annotation in argument_types.items()
+            if argument == SECRET_ARGUMENT
+            or annotation is Secret
+            or Secret in get_args(annotation)
+        }
+
+    def _is_secret_keyword(self, name: str) -> bool:
+        return bool(self._secret_argument_names(name))
+
+    def _is_banner_muted_keyword(self, name: str) -> bool:
+        function = self._resolve_keyword_function(name)
+        return function is not None and function.__name__ in BANNER_MUTED_KEYWORDS
+
+    def _mask_secret_arguments(self, name: str, args: list[str]) -> list[str]:
+        """Counting positions is only correct because Robot Framework rejects a
+        positional argument that follows a named one, so every cell before the
+        first named argument fills its parameter in declaration order.
+        """
+        secret_arguments = self._secret_argument_names(name)
+        if not secret_arguments:
+            return list(args)
+        argument_names = self._keyword_argument_names(name)
+        masked = []
+        position = 0
+        for arg in args:
+            argument_name, separator, _ = arg.partition("=")
+            if separator and argument_name in argument_names:
+                masked.append(
+                    f"{argument_name}={SECRET_MASK}"
+                    if argument_name in secret_arguments
+                    else arg
+                )
+                continue
+            is_secret = (
+                position < len(argument_names)
+                and argument_names[position] in secret_arguments
+            )
+            masked.append(SECRET_MASK if is_secret else arg)
+            position += 1
+        return masked
+
+    def _keyword_call_banner_content(
+        self, name: str, kwname: str, args: list[str]
+    ) -> str:
+        """Resolving the variables is left to the caller, so that a masked secret
+        can never be resolved back into the banner.
+        """
+        masked = self._mask_secret_arguments(name, args)
+        return f"{kwname}{'    ' * bool(masked)}{'    '.join(masked)}"
+
+    def _keyword_call_banner_enabled(self) -> bool:
+        if self.show_keyword_call_banner is None:
+            return bool(self.presenter_mode)
+        return bool(self.show_keyword_call_banner)
+
+    def _show_keyword_call(self, name: str):
+        if not self._keyword_call_banner_enabled():
+            return
+        if not self.keyword_call_stack:
+            return
+        try:
+            if self._is_banner_muted_keyword(name):
+                self.set_keyword_call_banner()
+                return
+            entry = self.keyword_call_stack[-1]
+            content = self._keyword_call_banner_content(
+                name, entry["kwname"], entry["args"]
+            )
+            if not self._is_secret_keyword(name):
+                content = BuiltIn().replace_variables(content)
+            self.set_keyword_call_banner(content)
+        except Exception as error:
+            logger.trace(f"Keyword call banner could not be painted: {error}")
 
     def set_keyword_call_banner(self, keyword_call=None):
         if keyword_call:

Browser/keywords/playwright_state.py

@@ -937,10 +937,16 @@ def add_context_and_keyword_call_stack_to_trace(self, trace_file, ctx_id):
         self.library.tracing_contexts.append(ctx_id)
         if self.library.tracing_group_mode == TracingGroupMode.Browser:
             return self.open_trace_group(
-                **(self.library.keyword_call_stack[-1]), context_id=ctx_id
+                **self.library._trace_group_arguments(
+                    self.library.keyword_call_stack[-1]
+                ),
+                context_id=ctx_id,
             )
         for keyword_call in self.library.keyword_call_stack:
-            self.open_trace_group(**keyword_call, context_id=ctx_id)
+            self.open_trace_group(
+                **self.library._trace_group_arguments(keyword_call),
+                context_id=ctx_id,
+            )
         return None
 
     def _mask_credentials(self, data: dict):

Browser/utils/data_types.py

@@ -406,11 +406,18 @@ class AriaSnapshotReturnType(Enum):
 
 
 class KeywordCallStackEntry(TypedDict):
-    """Information about the keyword call stack."""
+    """Information about the keyword call stack.
+
+    ``kwname`` and ``args`` keep the call as it was written in the source. The
+    arguments the keyword receives are already converted and would render enums
+    and numbers instead of the text the user wrote.
+    """
 
     name: str
     file: str
     line: int
+    kwname: str
+    args: list[str]
 
 
 class SelectOptions(TypedDict):

atest/library/banner.py

@@ -0,0 +1,33 @@
+import re
+from typing import Any
+
+from assertionengine.assertion_engine import AssertionOperator, verify_assertion
+from robot.libraries.BuiltIn import BuiltIn
+
+from Browser import Browser
+
+
+def get_keyword_call_banner_text(
+    operator: AssertionOperator = None, expected: str | None = None
+):
+    """Keyword is not a Browser keyword, therefore can read without influencing the keyword call banner."""
+    browser: Browser = BuiltIn().get_library_instance("Browser")
+    text_content = browser.evaluate_javascript(
+        None,
+        "() => {const e = document.getElementById('kwCallBanner');"
+        " return e ? e.textContent : '';}",
+    )
+    content_match = re.search(r"content: '(.*?)';", text_content, re.DOTALL)
+    return verify_assertion(
+        content_match.group(1) if content_match else "", operator, expected
+    )
+
+
+def get_wrapped_page_source(
+    assertion_operator: AssertionOperator | None = None,
+    assertion_expected: Any | None = None,
+    message: str | None = None,
+) -> str | dict | tuple:
+    """Keyword is not a Browser keyword, therefore is neither shown in keyword call banner, nor is it hiding it."""
+    browser: Browser = BuiltIn().get_library_instance("Browser")
+    return browser.get_page_source(assertion_operator, assertion_expected, message)