feat: add edit-registries command

snapcraft/cli.py

@@ -143,6 +143,7 @@
     craft_cli.CommandGroup(
         "Store Registries",
         [
+            commands.StoreEditRegistriesCommand,
             commands.StoreListRegistriesCommand,
         ],
     ),

snapcraft/commands/__init__.py

@@ -52,7 +52,7 @@
     StoreRegisterCommand,
 )
 from .plugins import ListPluginsCommand, PluginsCommand
-from .registries import StoreListRegistriesCommand
+from .registries import StoreEditRegistriesCommand, StoreListRegistriesCommand
 from .remote import RemoteBuildCommand
 from .status import (
     StoreListRevisionsCommand,
@@ -77,6 +77,7 @@
     "SnapCommand",
     "StoreCloseCommand",
     "StoreEditValidationSetsCommand",
+    "StoreEditRegistriesCommand",
     "StoreExportLoginCommand",
     "StoreLegacyCreateKeyCommand",
     "StoreLegacyGatedCommand",

snapcraft/commands/registries.py

@@ -29,7 +29,7 @@ class StoreListRegistriesCommand(craft_application.commands.AppCommand):
     """List registries."""
 
     name = "list-registries"
-    help_msg = "List registries"
+    help_msg = "List registries sets"
     overview = textwrap.dedent(
         """
         List all registries for the authenticated account.
@@ -69,3 +69,41 @@ def run(self, parsed_args: "argparse.Namespace"):
             name=parsed_args.name,
             output_format=parsed_args.format,
         )
+
+
+class StoreEditRegistriesCommand(craft_application.commands.AppCommand):
+    """Edit a registries set."""
+
+    name = "edit-registries"
+    help_msg = "Edit or create a registries set"
+    overview = textwrap.dedent(
+        """
+        Edit a registries set.
+
+        If the registries set does not exist, then a new registries set will be created.
+
+        The account ID of the authenticated account can be determined with the
+        ``snapcraft whoami`` command.
+
+        Use the ``list-registries`` command to view existing registries.
+        """
+    )
+    _services: services.SnapcraftServiceFactory  # type: ignore[reportIncompatibleVariableOverride]
+
+    @override
+    def fill_parser(self, parser: "argparse.ArgumentParser") -> None:
+        parser.add_argument(
+            "account_id",
+            metavar="account-id",
+            help="The account ID of the registries set to edit",
+        )
+        parser.add_argument(
+            "name", metavar="name", help="Name of the registries set to edit"
+        )
+
+    @override
+    def run(self, parsed_args: "argparse.Namespace"):
+        self._services.registries.edit_assertion(
+            name=parsed_args.name,
+            account_id=parsed_args.account_id,
+        )

snapcraft/services/__init__.py

@@ -16,20 +16,20 @@
 
 """Snapcraft services."""
 
-from snapcraft.services.assertions import AssertionService
+from snapcraft.services.assertions import Assertion
 from snapcraft.services.lifecycle import Lifecycle
 from snapcraft.services.package import Package
 from snapcraft.services.provider import Provider
-from snapcraft.services.registries import RegistriesService
+from snapcraft.services.registries import Registries
 from snapcraft.services.remotebuild import RemoteBuild
 from snapcraft.services.service_factory import SnapcraftServiceFactory
 
 __all__ = [
-    "AssertionService",
+    "Assertion",
     "Lifecycle",
     "Package",
     "Provider",
-    "RegistriesService",
+    "Registries",
     "RemoteBuild",
     "SnapcraftServiceFactory",
 ]

snapcraft/services/assertions.py

@@ -19,30 +19,44 @@
 from __future__ import annotations
 
 import abc
+import io
 import json
+import os
+import pathlib
+import subprocess
+import tempfile
 from typing import Any
 
 import craft_cli
 import tabulate
+import yaml
+from craft_application.errors import CraftValidationError
 from craft_application.services import base
+from craft_application.util import safe_yaml_load
 from typing_extensions import override
 
-from snapcraft import const, errors, models, store
+from snapcraft import const, errors, models, store, utils
 
 
-class AssertionService(base.AppService):
+class Assertion(base.AppService):
     """Abstract service for interacting with assertions."""
 
     @override
     def setup(self) -> None:
         """Application-specific service setup."""
         self._store_client = store.StoreClientCLI()
+        self._editor_cmd = os.getenv("EDITOR", "vi")
         super().setup()
 
     @property
     @abc.abstractmethod
-    def _assertion_type(self) -> str:
-        """The pluralized name of the assertion type."""
+    def _assertion_name(self) -> str:
+        """The lowercase name of the assertion type."""
+
+    @property
+    @abc.abstractmethod
+    def _editable_assertion_class(self) -> type[models.EditableAssertion]:
+        """The type of the editable assertion."""
 
     @abc.abstractmethod
     def _get_assertions(self, name: str | None = None) -> list[models.Assertion]:
@@ -65,6 +79,29 @@ def _normalize_assertions(
         :returns: A tuple containing the headers and normalized assertions.
         """
 
+    @abc.abstractmethod
+    def _generate_yaml_from_model(self, assertion: models.Assertion) -> str:
+        """Generate a multi-line yaml string from an existing assertion.
+
+        This string should contain only user-editable data.
+
+        :param assertion: The assertion to generate a yaml string from.
+
+        :returns: A multi-line yaml string.
+        """
+
+    @abc.abstractmethod
+    def _generate_yaml_from_template(self, name: str, account_id: str) -> str:
+        """Generate a multi-line yaml string of a default assertion.
+
+        This string should contain only user-editable data.
+
+        :param name: The name of the assertion.
+        :param account_id: The account ID of the authenticated user.
+
+        :returns: A multi-line yaml string.
+        """
+
     def list_assertions(self, *, output_format: str, name: str | None = None) -> None:
         """List assertions from the store.
 
@@ -81,7 +118,7 @@ def list_assertions(self, *, output_format: str, name: str | None = None) -> Non
             match output_format:
                 case const.OutputFormat.json:
                     json_assertions = {
-                        self._assertion_type.lower(): [
+                        f"{self._assertion_name}s": [
                             {
                                 header.lower(): value
                                 for header, value in zip(headers, assertion)
@@ -102,4 +139,95 @@ def list_assertions(self, *, output_format: str, name: str | None = None) -> Non
                         msg=f"'--format {output_format}'",
                     )
         else:
-            craft_cli.emit.message(f"No {self._assertion_type} found.")
+            craft_cli.emit.message(f"No {self._assertion_name}s found.")
+
+    def _edit_yaml_file(self, filepath: pathlib.Path) -> models.EditableAssertion:
+        """Edit a yaml file and unmarshal it to an editable assertion.
+
+        If the file is not valid, the user is prompted to amend it.
+
+        :param filepath: The path to the yaml file to edit.
+
+        :returns: The edited assertion.
+        """
+        while True:
+            craft_cli.emit.debug(f"Using {self._editor_cmd} to edit file.")
+            with craft_cli.emit.pause():
+                subprocess.run([self._editor_cmd, filepath], check=True)
+            try:
+                with filepath.open() as file:
+                    data = safe_yaml_load(file)
+                edited_assertion = self._editable_assertion_class.from_yaml_data(
+                    data=data,
+                    # filepath is only shown for pydantic errors and snapcraft should
+                    # not expose the temp file name
+                    filepath=pathlib.Path(self._assertion_name.replace(" ", "-")),
+                )
+                return edited_assertion
+            except (yaml.YAMLError, CraftValidationError) as err:
+                craft_cli.emit.message(f"{err!s}")
+                if not utils.confirm_with_user(
+                    f"Do you wish to amend the {self._assertion_name}?"
+                ):
+                    raise errors.SnapcraftError("operation aborted") from err
+
+    def _get_yaml_data(self, name: str, account_id: str) -> str:
+        craft_cli.emit.progress(
+            f"Requesting {self._assertion_name} '{name}' from the store."
+        )
+
+        if assertions := self._get_assertions(name=name):
+            yaml_data = self._generate_yaml_from_model(assertions[0])
+        else:
+            craft_cli.emit.progress(
+                f"Creating a new {self._assertion_name} because no existing "
+                f"{self._assertion_name} named '{name}' was found for the "
+                "authenticated account.",
+                permanent=True,
+            )
+            yaml_data = self._generate_yaml_from_template(
+                name=name, account_id=account_id
+            )
+
+        return yaml_data
+
+    @staticmethod
+    def _write_to_file(yaml_data: str) -> pathlib.Path:
+        with tempfile.NamedTemporaryFile() as temp_file:
+            filepath = pathlib.Path(temp_file.name)
+        craft_cli.emit.trace(f"Writing yaml data to temporary file '{filepath}'.")
+        filepath.write_text(yaml_data, encoding="utf-8")
+        return filepath
+
+    @staticmethod
+    def _remove_temp_file(filepath: pathlib.Path) -> None:
+        craft_cli.emit.trace(f"Removing temporary file '{filepath}'.")
+        filepath.unlink()
+
+    def edit_assertion(self, *, name: str, account_id: str) -> None:
+        """Edit, sign and upload an assertion.
+
+         If the assertion does not exist, a new assertion is created from a template.
+
+        :param name: The name of the assertion to edit.
+        :param account_id: The account ID associated with the registries set.
+        """
+        yaml_data = self._get_yaml_data(name=name, account_id=account_id)
+        yaml_file = self._write_to_file(yaml_data)
+        original_assertion = self._editable_assertion_class.unmarshal(
+            safe_yaml_load(io.StringIO(yaml_data))
+        )
+        edited_assertion = self._edit_yaml_file(yaml_file)
+
+        if edited_assertion == original_assertion:
+            craft_cli.emit.message("No changes made.")
+            self._remove_temp_file(yaml_file)
+            return
+
+        # TODO: build, sign, and push assertion (#5018)
+
+        self._remove_temp_file(yaml_file)
+        craft_cli.emit.message(f"Successfully edited {self._assertion_name} {name!r}.")
+        raise errors.FeatureNotImplemented(
+            f"Building, signing and uploading {self._assertion_name} is not implemented.",
+        )