Implement click custom options/arguments/commands

griffon/commands/custom_commands.py

@@ -0,0 +1,89 @@
+"""
+Custom defined Click commands/options/etc.
+"""
+
+import click
+
+from griffon.exceptions import GriffonUsageError
+from griffon.helpers import Style
+
+
+class ListParamType(click.ParamType):
+    """Custom comma-separated list type"""
+
+    name = "list"
+
+    def convert(self, value, param, ctx):
+        if value is None:
+            return []
+        return value.split(",")
+
+
+class BaseGroupParameter:
+    """
+    Custom base parameter which handles:
+        * mutually exclusive options
+        * required group options (one of the group is required)
+    """
+
+    def handle_parse_result(self, ctx, opts, args):
+        if self.mutually_exclusive_group:
+            if opts.get(self.name) is None:
+                pass  # skip check for not supplied click.Arguments
+            elif self.name in opts and any(
+                opt in opts for opt in self.mutually_exclusive_group if opts[opt] is not None
+            ):
+                raise GriffonUsageError(
+                    (
+                        f"{Style.BOLD}{self.name} cannot be used with "
+                        f"{', '.join(self.mutually_exclusive_group)}.{Style.RESET}"
+                    ),
+                    ctx=ctx,
+                )
+
+        if self.required_group:
+            group_set = set(
+                opt for opt in opts if opt in self.required_group and opts[opt] is not None
+            )
+            if not any(group_set):
+                raise GriffonUsageError(
+                    f"{Style.BOLD}At least one of {', '.join(self.required_group)} "
+                    f"is required.{Style.RESET}",
+                    ctx=ctx,
+                )
+
+        return super().handle_parse_result(ctx, opts, args)
+
+
+class GroupOption(BaseGroupParameter, click.Option):
+    """Custom Option with BaseGroupParameter functionality"""
+
+    def __init__(self, *args, **kwargs):
+        self.mutually_exclusive_group = set(kwargs.pop("mutually_exclusive_group", []))
+        self.required_group = set(kwargs.pop("required_group", []))
+
+        if self.mutually_exclusive_group:
+            mutually_exclusive_str = ", ".join(self.mutually_exclusive_group)
+            kwargs["help"] = kwargs.get("help", "") + (
+                f", this argument is mutually exclusive "
+                f"with arguments: {Style.BOLD}[{mutually_exclusive_str}]{Style.RESET}"
+            )
+
+        if self.required_group:
+            required_str = ", ".join(self.required_group)
+            kwargs["help"] = kwargs.get("help", "") + (
+                f", at least one of these arguments: "
+                f"{Style.BOLD}[{required_str}]{Style.RESET} is required."
+            )
+
+        super().__init__(*args, **kwargs)
+
+
+class GroupArgument(BaseGroupParameter, click.Argument):
+    """Custom Argument with BaseGroupParameter functionality"""
+
+    def __init__(self, *args, **kwargs):
+        self.mutually_exclusive_group = set(kwargs.pop("mutually_exclusive_group", []))
+        self.required_group = set(kwargs.pop("required_group", []))
+
+        super().__init__(*args, **kwargs)

griffon/commands/plugins/affects.py

@@ -0,0 +1,123 @@
+"""
+Plugin for handling affects
+"""
+
+import logging
+
+import click
+import sfm2client
+from rich.console import Console
+
+from griffon import OSIDBService, console_status, get_config_option, progress_bar
+from griffon.commands.custom_commands import GroupOption, ListParamType
+from griffon.commands.queries import get_product_contain_component
+from griffon.exceptions import GriffonException
+from griffon.helpers import Color, Style
+
+console = Console()
+
+logger = logging.getLogger("griffon")
+
+
+@click.group(help="Affects plugin")
+@click.pass_context
+def plugins(ctx):
+    """Affects plugin for automating specific affects filing workflows"""
+    pass
+
+
+@plugins.command()
+@click.option(
+    "-b",
+    "--bz-id",
+    "bz_id",
+    help="Bugzilla ID",
+    cls=GroupOption,
+    mutually_exclusive_group=["flaw_id"],
+    required_group=["bz_id", "flaw_id"],
+)
+@click.option(
+    "-f",
+    "--flaw-id",
+    "flaw_id",
+    help="UUID / CVE ID",
+    cls=GroupOption,
+    mutually_exclusive_group=["bz_id"],
+    required_group=["bz_id", "flaw_id"],
+)
+@click.option(
+    "--base-compiler-components",
+    "base_compiler_components",
+    default="",
+    help="Comma-separated list of components",
+    type=ListParamType(),
+)
+@click.pass_context
+def go_stdlib(ctx, bz_id, flaw_id, base_compiler_components):
+    """Go stdlib affects handler"""
+
+    with console_status(ctx) as status:
+        status.update("Querying OSIDB for Flaw data")
+        session = OSIDBService.create_session()
+        try:
+            # TODO: acquire component/subcomponents directely instead of parsing from
+            # Bugzilla summary once OSIDB-1420 is resolved
+            if flaw_id:
+                flaw = session.flaws.retrieve(
+                    flaw_id, include_meta_attr="bz_summary,bz_id", include_fields="meta_attr"
+                )
+            elif bz_id:
+                flaw = session.flaws.retrieve_list(
+                    bz_id=bz_id, include_meta_attr="bz_summary,bz_id", include_fields="meta_attr"
+                ).results[0]
+        except Exception as e:
+            if ctx.obj["VERBOSE"]:
+                logger.error(e)
+
+            flaw_id_msg = f'with ID "{flaw_id}"' if flaw_id else f'with Bugzilla ID "{bz_id}"'
+            raise GriffonException(
+                f"Could not retrieve flaw {flaw_id_msg}",
+            )
+
+    # # base compiler affects
+    for component_name in base_compiler_components:
+        ctx.invoke(
+            get_product_contain_component,
+            component_name=component_name,
+            search_latest=True,
+            sfm2_flaw_id=flaw_id,
+            flaw_mode="replace",
+            strict_name_search=True,
+            filter_rh_naming=True,
+        )
+
+    # rpm affects
+    subcomponent = flaw.meta_attr["bz_summary"].split(":")[1].strip()
+    ctx.obj["PROFILE"] = "latest"
+    ctx.invoke(
+        get_product_contain_component,
+        component_name=subcomponent,
+        # search_latest=True,
+        sfm2_flaw_id=flaw_id,
+        flaw_mode="replace",
+        strict_name_search=True,
+        filter_rh_naming=True,
+        output_type_filter="GOLANG",
+        component_type="RPM",
+    )
+
+    # container-first, only one per product
+    ctx.invoke(
+        get_product_contain_component,
+        component_name=subcomponent,
+        # search_latest=True,
+        sfm2_flaw_id=flaw_id,
+        flaw_mode="replace",
+        strict_name_search=True,
+        filter_rh_naming=True,
+        output_type_filter="GOLANG",
+        component_type="RPM",
+    )
+
+    # special affects for all containers built with golang
+    # TODO