feat: add edit-registries command

Signed-off-by: Callahan Kovacs <callahan.kovacs@canonical.com>

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/assertions.py

@@ -19,15 +19,23 @@
 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 Assertion(base.AppService):
@@ -37,13 +45,19 @@ class Assertion(base.AppService):
     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_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]:
         """Get assertions from the store.
@@ -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.
 
@@ -103,3 +140,94 @@ def list_assertions(self, *, output_format: str, name: str | None = None) -> Non
                     )
         else:
             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.",
+        )

snapcraft/services/registries.py

@@ -18,13 +18,55 @@
 
 from __future__ import annotations
 
+import textwrap
 from typing import Any
 
+from craft_application.util import dump_yaml
 from typing_extensions import override
 
 from snapcraft import models
 from snapcraft.services import Assertion
 
+_REGISTRY_SETS_TEMPLATE = textwrap.dedent(
+    """\
+    account-id: {account_id}
+    name: {set_name}
+    # summary: {summary}
+    # The revision for this registries set
+    # revision: {revision}
+    {views}
+    {body}
+    """
+)
+
+
+_REGISTRY_SETS_VIEWS_TEMPLATE = textwrap.dedent(
+    """\
+    views:
+      wifi-setup:
+        rules:
+          - request: ssids
+            storage: wifi.ssids
+            access: read
+    """
+)
+
+
+_REGISTRY_SETS_BODY_TEMPLATE = textwrap.dedent(
+    """\
+    body: |-
+      {
+        "storage": {
+          "schema": {
+            "wifi": {
+              "values": "any"
+            }
+          }
+        }
+      }
+    """
+)
+
 
 class Registries(Assertion):
     """Service for interacting with registries."""
@@ -34,6 +76,11 @@ class Registries(Assertion):
     def _assertion_name(self) -> str:
         return "registries set"
 
+    @property
+    @override
+    def _editable_assertion_class(self) -> type[models.EditableAssertion]:
+        return models.EditableRegistryAssertion
+
     @override
     def _get_assertions(self, name: str | None = None) -> list[models.Assertion]:
         return self._store_client.list_registries(name=name)
@@ -54,3 +101,27 @@ def _normalize_assertions(
         ]
 
         return headers, registries
+
+    @override
+    def _generate_yaml_from_model(self, assertion: models.Assertion) -> str:
+        return _REGISTRY_SETS_TEMPLATE.format(
+            account_id=assertion.account_id,
+            views=dump_yaml(
+                {"views": assertion.marshal().get("views")}, default_flow_style=False
+            ),
+            body=dump_yaml({"body": assertion.body}, default_flow_style=False),
+            summary=assertion.summary,
+            set_name=assertion.name,
+            revision=assertion.revision,
+        )
+
+    @override
+    def _generate_yaml_from_template(self, name: str, account_id: str) -> str:
+        return _REGISTRY_SETS_TEMPLATE.format(
+            account_id=account_id,
+            views=_REGISTRY_SETS_VIEWS_TEMPLATE,
+            body=_REGISTRY_SETS_BODY_TEMPLATE,
+            summary="A brief summary of the registries set",
+            set_name=name,
+            revision=1,
+        )

snapcraft/store/client.py

@@ -527,6 +527,8 @@ def list_registries(
         if assertions := response.json().get("assertions"):
             for assertion_data in assertions:
                 emit.debug(f"Parsing assertion: {assertion_data}")
+                # move body into model
+                assertion_data["headers"]["body"] = assertion_data["body"]
                 assertion = models.RegistryAssertion.unmarshal(
                     assertion_data["headers"]
                 )

tests/unit/commands/test_registries.py

@@ -14,6 +14,8 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+"""Unit tests for registries commands."""
+
 import sys
 
 import pytest
@@ -26,6 +28,11 @@ def mock_list_assertions(mocker):
     return mocker.patch("snapcraft.services.registries.Registries.list_assertions")
 
 
+@pytest.fixture
+def mock_edit_assertion(mocker):
+    return mocker.patch("snapcraft.services.registries.Registries.edit_assertion")
+
+
 @pytest.mark.usefixtures("memory_keyring")
 @pytest.mark.parametrize("output_format", const.OUTPUT_FORMATS)
 @pytest.mark.parametrize("name", [None, "test"])
@@ -56,3 +63,17 @@ def test_list_registries_default_format(mocker, mock_list_assertions, name):
     app.run()
 
     mock_list_assertions.assert_called_once_with(name=name, output_format="table")
+
+
+@pytest.mark.usefixtures("memory_keyring")
+def test_edit_registries(mocker, mock_edit_assertion):
+    """Test `snapcraft edit-registries`."""
+    cmd = ["snapcraft", "edit-registries", "test-account-id", "test-name"]
+    mocker.patch.object(sys, "argv", cmd)
+
+    app = application.create_app()
+    app.run()
+
+    mock_edit_assertion.assert_called_once_with(
+        name="test-name", account_id="test-account-id"
+    )