Vulnerability

pulp-glue-console/pulp_glue/console/__init__.py

@@ -1 +1 @@
-__version__ = "0.1.0.dev"
+__version__ = "0.1.1.dev"

pulp-glue-console/pulp_glue/console/context.py

@@ -1,87 +1,33 @@
+import typing as t
 from gettext import gettext as _
 
-from pulp_glue.common.context import (
-    EntityDefinition,
-    PluginRequirement,
-    PulpContentContext,
-    PulpEntityContext,
-    PulpRemoteContext,
-    PulpRepositoryContext,
-    PulpRepositoryVersionContext,
-)
+from pulp_glue.common.context import PulpEntityContext
 
 
-class PulpConsoleContentContext(PulpContentContext):
-    """Context for Console Content."""
+class PulpVulnerabilityReportContext(PulpEntityContext):
+    """Context for working with vulnerability reports."""
 
-    PLUGIN = "console"
-    RESOURCE_TYPE = "content"
-    ENTITY = _("console content")
-    ENTITIES = _("console content")
-    HREF = "console_console_content_href"
-    ID_PREFIX = "content_console_content"
-    NEEDS_PLUGINS = [PluginRequirement("console", specifier=">=1.0.0")]
+    ENTITY = _("vulnerability report")
+    ENTITIES = _("vulnerability reports")
+    ID_PREFIX = "vuln_report"
+    HREF = "service_vulnerability_report_href"
 
+    def upload(self, file: t.IO[bytes], chunk_size: int = 1000000) -> t.Dict[str, t.Any]:
+        """Upload a vulnerability report from a JSON file.
 
-class PulpConsoleDistributionContext(PulpEntityContext):
-    """Context for Console Distribution."""
+        Args:
+            file: The file object to upload
+            chunk_size: The chunk size for the upload
 
-    PLUGIN = "console"
-    RESOURCE_TYPE = "console"
-    ENTITY = _("console distribution")
-    ENTITIES = _("console distributions")
-    HREF = "console_console_distribution_href"
-    ID_PREFIX = "distributions_console_console"
-    NEEDS_PLUGINS = [PluginRequirement("console", specifier=">=1.0.0")]
-
-    def preprocess_entity(self, body: EntityDefinition, partial: bool = False) -> EntityDefinition:
-        body = super().preprocess_entity(body, partial)
-        version = body.pop("version", None)
-        if version is not None:
-            repository_href = body.pop("repository")
-            body["repository_version"] = f"{repository_href}versions/{version}/"
-        return body
-
-
-class PulpConsoleRemoteContext(PulpRemoteContext):
-    """Context for Console Remote."""
-
-    PLUGIN = "console"
-    RESOURCE_TYPE = "console"
-    ENTITY = _("console remote")
-    ENTITIES = _("console remotes")
-    HREF = "console_console_remote_href"
-    ID_PREFIX = "remotes_console_console"
-    NEEDS_PLUGINS = [PluginRequirement("console", specifier=">=1.0.0")]
-
-
-class PulpConsoleRepositoryVersionContext(PulpRepositoryVersionContext):
-    """Context for Console Repository Version."""
-
-    PLUGIN = "console"
-    RESOURCE_TYPE = "console"
-    HREF = "console_console_repository_version_href"
-    ID_PREFIX = "repositories_console_console_versions"
-    NEEDS_PLUGINS = [PluginRequirement("console", specifier=">=1.0.0")]
-
-
-class PulpConsoleRepositoryContext(PulpRepositoryContext):
-    """Context for Console Repository."""
-
-    PLUGIN = "console"
-    RESOURCE_TYPE = "console"
-    HREF = "console_console_repository_href"
-    ID_PREFIX = "repositories_console_console"
-    VERSION_CONTEXT = PulpConsoleRepositoryVersionContext
-    NEEDS_PLUGINS = [PluginRequirement("console", specifier=">=1.0.0")]
-    CAPABILITIES = {
-        "sync": [PluginRequirement("console", specifier=">=1.0.0")],
-    }
-
-    # Add custom methods for your repository operations here
-    # For example:
-    # def import_content(self, href: str, artifact: str, ...) -> Any:
-    #     body = {...}
-    #     return self.pulp_ctx.call("your_import_method_id",
-    #                               parameters={self.HREF: href},
-    #                               body=body)
+        Returns:
+            The created vulnerability report entity
+        """
+        # Read the raw file content
+        file_content = file.read()
+        # Submit the file content to the Pulp API
+        response = self.pulp_ctx.call(
+            operation_id="vuln_report_create",
+            body={"package_json": file_content},
+            validate_body=False,
+        )
+        return t.cast(t.Dict[str, t.Any], response)

pulpcore/cli/console/__init__.py

@@ -1,28 +1,40 @@
-from typing import Any
+import json
+import typing as t
 
 import click
-from pulp_glue.common.i18n import get_translation
-from pulpcore.cli.common.generic import pulp_group
-
-from pulpcore.cli.console.distribution import distribution
-from pulpcore.cli.console.remote import remote
-from pulpcore.cli.console.repository import repository
-
-translation = get_translation(__package__)
-_ = translation.gettext
-
-__version__ = "0.1.0.dev"
-
-
-@pulp_group("console")
-def console_group() -> None:
-    """Manage Console plugin."""
-    pass
-
-
-def mount(main: click.Group, **kwargs: Any) -> None:
-    """Mount the console commands to the CLI."""
-    console_group.add_command(distribution)
-    console_group.add_command(remote)
-    console_group.add_command(repository)
-    main.add_command(console_group)
+from pulp_glue.common.openapi import OpenAPI
+
+
+def mount(main: click.Group, **kwargs: t.Any) -> None:
+    if hasattr(OpenAPI, "_parse_response"):
+        original_parse_response = OpenAPI._parse_response
+        parse_response_attr = "_parse_response"
+    else:
+        original_parse_response = OpenAPI.parse_response  # type: ignore[attr-defined]
+        parse_response_attr = "parse_response"
+
+    # Define our custom implementation that handles 202 responses (Original one throws an error)
+    def custom_parse_response(
+        self: OpenAPI, method_spec: t.Dict[str, t.Any], response: t.Any
+    ) -> t.Any:
+        # Handle 202 responses directly
+        if response.status_code == 202:
+            content_type = response.headers.get("content-type")
+            if content_type is not None and content_type.startswith("application/json"):
+                return json.loads(response.body)
+            return {"status": "accepted"}
+
+        # For all other responses, use the original implementation
+        return original_parse_response(self, method_spec, response)
+
+    setattr(OpenAPI, parse_response_attr, custom_parse_response)
+
+    # Continue with normal mounting
+    from pulpcore.cli.console.vulnerability import attach_vulnerability_commands
+
+    @main.group()
+    def console() -> None:
+        """Pulp Console commands."""
+        pass
+
+    attach_vulnerability_commands(console)  # type: ignore

pulpcore/cli/console/py.typed

@@ -0,0 +1 @@
+# This file exists to mark this package as having type annotations