Make color and other logging method params keyword-only

examples/plugins/example/provision.py

@@ -104,7 +104,7 @@ def go(self):
             val = self.get(opt)
             # You can hide some not important information about provisioning.
             if opt != 'switch':
-                self.info(opt, val, 'green')
+                self.info(opt, val, color='green')
             data[opt] = val
 
         self._guest = GuestExample(data, name=self.name, parent=self.step)
@@ -212,7 +212,7 @@ def start(self):
         if self.opt('dry'):
             return
 
-        self.verbose('what', self.what, 'green')
+        self.verbose('what', self.what, color='green')
 
         if self._some_your_internal_stuff():
             return
@@ -265,6 +265,6 @@ def remove(self):
         """
 
         if self.what:
-            self.info('guest', 'removed', 'green')
+            self.info('guest', 'removed', color='green')
             self.delete()
             self.what = None

tmt/base/plan.py

@@ -1259,7 +1259,7 @@ def header(self) -> None:
         self.info('')
         self.info(self.name, color='red')
         if self.summary:
-            self.verbose('summary', self.summary, 'green')
+            self.verbose('summary', self.summary, color='green')
 
     def go(self) -> None:
         """
@@ -1270,8 +1270,8 @@ def go(self) -> None:
         # Additional debug info like plan environment
         self.debug('info', color='cyan', shift=0, level=3)
         # TODO: something better than str()?
-        self.debug('environment', self.environment, 'magenta', level=3)
-        self.debug('context', self.fmf_context, 'magenta', level=3)
+        self.debug('environment', self.environment, color='magenta', level=3)
+        self.debug('context', self.fmf_context, color='magenta', level=3)
 
         # Wake up all steps
         self.wake()

tmt/guest/__init__.py

@@ -2101,7 +2101,7 @@ def _ansible_summary(self, output: Optional[str]) -> None:
             matched = re.search(rf'^.*\s:\s.*{key}=(\d+).*$', output, re.MULTILINE)
             if matched and int(matched.group(1)) > 0:
                 tasks = fmf.utils.listed(matched.group(1), 'task')
-                self.verbose(key, tasks, 'green')
+                self.verbose(key, tasks, color='green')
 
     def _sanitize_ansible_playbook_path(
         self, playbook: AnsibleApplicable, playbook_root: Optional[Path]

tmt/libraries/beakerlib.py

@@ -417,7 +417,7 @@ def _do_fetch(self, directory: Path) -> None:
         self.parent.verbose(
             'commit-hash',
             tmt.utils.git.git_hash(directory=clone_dir, logger=self._logger),
-            'green',
+            color='green',
         )
 
         # Copy only the required library
@@ -432,7 +432,7 @@ def _do_fetch(self, directory: Path) -> None:
         self.parent.verbose(
             'using remote git library',
             cast(dict[str, str], self.identifier.to_minimal_dict()),
-            'green',
+            color='green',
             level=3,
         )
 

tmt/log.py

@@ -579,6 +579,7 @@ def __call__(
         self,
         key: str,
         value: Optional[str] = None,
+        *,
         color: 'tmt.utils.themes.Style' = None,
         shift: int = 0,
         level: VerbosityLevel = 1,
@@ -592,6 +593,7 @@ class Print(Protocol):
     def __call__(
         self,
         text: Optional[str] = None,
+        *,
         color: 'tmt.utils.themes.Style' = None,
         file: Optional[TextIO] = None,
         nl: bool = True,
@@ -964,6 +966,7 @@ def _log(
     def print_format(
         self,
         text: str,
+        *,
         color: 'tmt.utils.themes.Style' = None,
     ) -> str:
         """
@@ -984,6 +987,7 @@ def print_format(
     def print(
         self,
         text: Optional[str] = None,
+        *,
         color: 'tmt.utils.themes.Style' = None,
         file: Optional[TextIO] = None,
         nl: bool = True,
@@ -998,6 +1002,7 @@ def info(
         self,
         key: str,
         value: Optional[LoggableValue] = None,
+        *,
         color: 'tmt.utils.themes.Style' = None,
         shift: int = 0,
         topic: Optional[Topic] = None,
@@ -1013,6 +1018,7 @@ def verbose(
         self,
         key: str,
         value: Optional[LoggableValue] = None,
+        *,
         color: 'tmt.utils.themes.Style' = None,
         shift: int = 0,
         level: VerbosityLevel = 1,
@@ -1036,6 +1042,7 @@ def debug(
         self,
         key: str,
         value: Optional[LoggableValue] = None,
+        *,
         color: 'tmt.utils.themes.Style' = None,
         shift: int = 0,
         level: DebugLevel = 1,
@@ -1058,6 +1065,7 @@ def debug(
     def warning(
         self,
         message: str,
+        *,
         shift: int = 0,
         stacklevel: int = 1,
         source: Optional[str] = None,
@@ -1080,14 +1088,16 @@ def warning(
     def warn(
         self,
         message: str,
+        *,
         shift: int,
         stacklevel: int = 1,
     ) -> None:
-        return self.warning(message, shift, stacklevel=stacklevel + 1)
+        return self.warning(message, shift=shift, stacklevel=stacklevel + 1)
 
     def fail(
         self,
         message: str,
+        *,
         shift: int = 0,
         stacklevel: int = 1,
     ) -> None:

tmt/package_managers/bootc.py

@@ -274,7 +274,7 @@ def build_container(self) -> Optional[CommandOutput]:
 
                 self.debug(f"containerfile content: {containerfile}")
                 # Build the container image
-                self.info("package", "building container image with dependencies", "green")
+                self.info("package", "building container image with dependencies", color="green")
 
                 assert self.guest.parent is not None
 
@@ -287,14 +287,14 @@ def build_container(self) -> Optional[CommandOutput]:
                 )
 
                 # Switch to the new image for next boot
-                self.info("package", f"switching to new image {image_tag}", "green")
+                self.info("package", f"switching to new image {image_tag}", color="green")
 
                 bootc_command, _ = self.engine.prepare_command()
                 bootc_command += Command('switch', '--transport', 'containers-storage', image_tag)
                 self.guest.execute(bootc_command)
 
                 # Reboot into the new image
-                self.info("package", "rebooting to apply new image", "green")
+                self.info("package", "rebooting to apply new image", color="green")
                 self.guest.reboot()
 
                 return build_output

tmt/package_managers/dnf.py

@@ -349,7 +349,7 @@ def enable_copr(self, *repositories: str) -> None:
         self.debug('Make sure the copr plugin is available.')
         self.install(Package(self.copr_plugin))
         for repository in repositories:
-            self.info('copr', repository, 'green')
+            self.info('copr', repository, color='green')
             self.guest.execute(
                 ShellScript(
                     f"{self.engine.command.to_script()} copr "

tmt/package_managers/rpm_ostree.py

@@ -239,7 +239,7 @@ def install(
         required, recommended = self.sort_packages(*installables, options=options)
 
         for package in recommended:
-            self.info('package', str(package), 'green')
+            self.info('package', str(package), color='green')
             try:
                 super().install(package, options=options)
             except RunError as error:

tmt/steps/__init__.py

@@ -1549,7 +1549,7 @@ def go(self, force: bool = False) -> None:
         self.info(self.name, color='blue')
         # Show workdir in verbose mode
         if self.workdir:
-            self.debug('workdir', self.workdir, 'magenta')
+            self.debug('workdir', self.workdir, color='magenta')
 
     def prune(self, logger: tmt.log.Logger) -> None:
         """
@@ -2400,15 +2400,15 @@ def go_prolog(self, logger: tmt.log.Logger) -> None:
         logger = logger or self._logger
 
         # Show the method
-        logger.info('how', self.get('how'), 'magenta')
+        logger.info('how', self.get('how'), color='magenta')
         # Give summary if provided
         if self.get('summary'):
-            logger.info('summary', self.get('summary'), 'magenta')
+            logger.info('summary', self.get('summary'), color='magenta')
         # Show name only if it's not the default one
         if not self.name.startswith(tmt.utils.DEFAULT_NAME):
-            logger.info('name', self.name, 'magenta')
+            logger.info('name', self.name, color='magenta')
         # Include order in verbose mode
-        logger.verbose('order', self.order, 'magenta', level=3)
+        logger.verbose('order', self.order, color='magenta', level=3)
 
     def essential_requires(self) -> list['tmt.base.core.Dependency']:
         """

tmt/steps/cleanup/__init__.py

@@ -115,7 +115,7 @@ def summary(self) -> None:
 
         # TODO Provide a number of stopped guests
         tasks = fmf.utils.listed(self.phases(), 'task')
-        self.info('summary', f'{tasks} completed', 'green', shift=1)
+        self.info('summary', f'{tasks} completed', color='green', shift=1)
 
     def go(self, force: bool = False) -> None:
         """
@@ -126,7 +126,7 @@ def go(self, force: bool = False) -> None:
 
         # Nothing more to do if already done
         if self.status() == 'done':
-            self.info('status', 'done', 'green', shift=1)
+            self.info('status', 'done', color='green', shift=1)
             self.summary()
             self.actions()
             return

tmt/steps/discover/__init__.py

@@ -267,7 +267,7 @@ def log_import_plan_details(self) -> None:
                 # Note the missing Optional for values - to_minimal_dict() would
                 # not include unset keys, therefore all values should be valid.
                 for key, value in cast(dict[str, str], remote_plan_id.to_minimal_spec()).items():
-                    self.verbose(f'import {key}', value, 'green')
+                    self.verbose(f'import {key}', value, color='green')
 
     def post_dist_git(self, created_content: list[Path]) -> None:
         """
@@ -281,7 +281,7 @@ def _fetch_remote_source(self, url: str) -> Optional[Path]:
         :param url: URL of the remote source.
         :returns: Potential path to the metadata tree root within the fetched source.
         """
-        self.info('url', url, 'green')
+        self.info('url', url, color='green')
         if self.data.url_content_type == "git":
             self.debug(f"Clone '{url}' to '{self.test_dir}'.")
             tmt.utils.git.git_clone(
@@ -323,7 +323,7 @@ def fetch_source(self) -> Path:
 
         if path is None or path.resolve() == Path.cwd().resolve():
             return Path('')
-        self.info('path', path, 'green')
+        self.info('path', path, color='green')
         return path
 
     def checkout_ref(self) -> None:
@@ -351,7 +351,7 @@ def checkout_ref(self) -> None:
             raise tmt.utils.DiscoverError("Could not resolve dynamic reference") from error
 
         if ref:
-            self.info('ref', ref, 'green')
+            self.info('ref', ref, color='green')
             self.debug(f"Checkout ref '{ref}'.")
             self.run(Command('git', 'checkout', '-f', ref), cwd=self.test_dir)
 
@@ -361,7 +361,7 @@ def checkout_ref(self) -> None:
                 self.verbose(
                     'commit-hash',
                     tmt.utils.git.git_hash(directory=self.test_dir, logger=self._logger),
-                    'green',
+                    color='green',
                 )
 
     def prune_tree(
@@ -881,7 +881,7 @@ def summary(self) -> None:
 
         # Summary of selected tests
         text = listed(len(self.tests(enabled=True)), 'test') + ' selected'
-        self.info('summary', text, 'green', shift=1)
+        self.info('summary', text, color='green', shift=1)
         # Test list in verbose mode
         for test_origin in self.tests(enabled=True):
             self.verbose(test_origin.test.name, color='red', shift=2)
@@ -897,7 +897,7 @@ def go(self, force: bool = False) -> None:
 
         # Nothing more to do if already done
         if self.status() == 'done':
-            self.info('status', 'done', 'green', shift=1)
+            self.info('status', 'done', color='green', shift=1)
             self.summary()
             self.actions()
             return

tmt/steps/discover/fmf.py

@@ -635,7 +635,7 @@ def _fetch_local_repository(self) -> Optional[Path]:
         else:
             assert fmf_root is not None  # narrow type
             directory = fmf_root
-        self.info('directory', directory, 'green')
+        self.info('directory', directory, color='green')
         if not self.data.dist_git_source or self.data.dist_git_merge:
             self.debug(f"Copy '{directory}' to '{self.test_dir}'.")
             if not self.is_dry_run:
@@ -736,10 +736,12 @@ def do_the_discovery(self, path: Optional[Path] = None) -> list['tmt.base.core.T
         # Check the 'test --filter' option first, then from discover
         filters = list(tmt.base.core.Test._opt('filters') or self.get('filter', []))
         for filter_ in filters:
-            self.info('filter', filter_, 'green')
+            self.info('filter', filter_, color='green')
         # Names of tests selected by --test option
         if self.data.test:
-            self.info('tests', fmf.utils.listed([test.name for test in self.data.test]), 'green')
+            self.info(
+                'tests', fmf.utils.listed([test.name for test in self.data.test]), color='green'
+            )
 
         # Check the 'test --link' option first, then from discover
         # FIXME: cast() - typeless "dispatcher" method
@@ -749,7 +751,7 @@ def do_the_discovery(self, path: Optional[Path] = None) -> list['tmt.base.core.T
         ]
 
         for link_needle in link_needles:
-            self.info('link', str(link_needle), 'green')
+            self.info('link', str(link_needle), color='green')
 
         excludes = list(tmt.base.core.Test._opt('exclude') or self.data.exclude)
         includes = list(tmt.base.core.Test._opt('include') or self.data.include)
@@ -760,7 +762,7 @@ def do_the_discovery(self, path: Optional[Path] = None) -> list['tmt.base.core.T
         if modified_url:
             previous = modified_url
             modified_url = tmt.utils.git.clonable_git_url(modified_url)
-            self.info('modified-url', modified_url, 'green')
+            self.info('modified-url', modified_url, color='green')
             if previous != modified_url:
                 self.debug(f"Original url was '{previous}'.")
             self.debug(f"Fetch also '{modified_url}' as 'reference'.")
@@ -777,13 +779,13 @@ def do_the_discovery(self, path: Optional[Path] = None) -> list['tmt.base.core.T
                 'modified-ref',
                 tmt.utils.git.default_branch(repository=self.test_dir, logger=self._logger),
             )
-            self.info('modified-ref', modified_ref, 'green')
+            self.info('modified-ref', modified_ref, color='green')
             ref_commit = self.run(
                 Command('git', 'rev-parse', '--short', str(modified_ref)),
                 cwd=self.test_dir,
             )
             assert ref_commit.stdout is not None
-            self.verbose('modified-ref hash', ref_commit.stdout.strip(), 'green')
+            self.verbose('modified-ref hash', ref_commit.stdout.strip(), color='green')
             output = self.run(
                 Command(
                     'git', 'log', '--format=', '--stat', '--name-only', f"{modified_ref}..HEAD"
@@ -950,7 +952,7 @@ def post_dist_git(self, created_content: list[Path]) -> None:
         if path is None or path.resolve() == Path.cwd().resolve():
             path = Path('')
         else:
-            self.info('path', path, 'green')
+            self.info('path', path, color='green')
 
         # Discover tests
         self._tests = self.do_the_discovery(path)