Merge pull request #25 from canonical/fix-git-ubuntu-source-clone

Refactor git-ubuntu source clone and key extraction

Author: lvoytek

pyproject.toml

@@ -136,5 +136,6 @@ disable = [
     "too-many-instance-attributes",
     "too-many-arguments",
     "too-many-positional-arguments",
-    "too-many-locals"
+    "too-many-locals",
+    "too-many-boolean-expressions",
 ]

src/charm.py

@@ -98,32 +98,43 @@ def _num_workers(self) -> int:
         return 0
 
     @property
-    def _lpuser_ssh_key(self) -> str | None:
+    def _lpuser_secret(self) -> ops.model.Secret | None:
+        secret_id: str = ""
+
         try:
             secret_id = str(self.config["lpuser_secret_id"])
-            lpuser_secret = self.model.get_secret(id=secret_id)
-            ssh_key_data = lpuser_secret.get_content().get("sshkey")
+        except KeyError:
+            logger.warning("lpuser_secret_id config not available, unable to extract keys.")
+            return None
+
+        try:
+            return self.model.get_secret(id=secret_id)
+        except (ops.SecretNotFoundError, ops.model.ModelError):
+            logger.warning("Failed to get lpuser secret with id %s", secret_id)
+
+        return None
 
-            if ssh_key_data is not None:
-                return str(ssh_key_data)
+    @property
+    def _lpuser_ssh_key(self) -> str | None:
+        secret = self._lpuser_secret
 
-        except (KeyError, ops.SecretNotFoundError, ops.model.ModelError):
-            pass
+        if secret is not None:
+            try:
+                return secret.get_content(refresh=True)["sshkey"]
+            except KeyError:
+                logger.warning("sshkey secret key not found in lpuser secret.")
 
         return None
 
     @property
     def _lpuser_lp_key(self) -> str | None:
-        try:
-            secret_id = str(self.config["lpuser_secret_id"])
-            lpuser_secret = self.model.get_secret(id=secret_id)
-            lp_key_data = lpuser_secret.get_content().get("lpkey")
+        secret = self._lpuser_secret
 
-            if lp_key_data is not None:
-                return str(lp_key_data)
-
-        except (KeyError, ops.SecretNotFoundError, ops.model.ModelError):
-            pass
+        if secret is not None:
+            try:
+                return secret.get_content(refresh=True)["lpkey"]
+            except KeyError:
+                logger.warning("lpkey secret key not found in lpuser secret.")
 
         return None
 
@@ -248,6 +259,24 @@ def _refresh_secret_keys(self) -> bool:
 
         return True
 
+    def _refresh_git_ubuntu_source(self) -> bool:
+        """Refresh the git-ubuntu source code from the configured URL.
+
+        Returns:
+            True if the source was refreshed successfully, False otherwise.
+        """
+        self.unit.status = ops.MaintenanceStatus("Refreshing git-ubuntu source.")
+
+        if not usr.refresh_git_ubuntu_source(
+            GIT_UBUNTU_SYSTEM_USER_USERNAME,
+            GIT_UBUNTU_USER_HOME_DIR,
+            self._git_ubuntu_source_url,
+        ):
+            self.unit.status = ops.BlockedStatus("Failed to refresh git-ubuntu source.")
+            return False
+
+        return True
+
     def _refresh_importer_node(self) -> None:
         """Remove old and install new git-ubuntu services."""
         self.unit.status = ops.MaintenanceStatus("Refreshing git-ubuntu services.")
@@ -383,29 +412,21 @@ def _on_install(self, _: ops.InstallEvent) -> None:
         self.unit.status = ops.ActiveStatus("Install complete.")
 
     def _on_config_changed(self, _: ops.ConfigChangedEvent) -> None:
-        """Handle updates to config items."""
-        # Update user's git and lpuser config, and git-ubuntu snap
-        if (
-            not self._update_git_user_config()
-            or not self._update_lpuser_config()
-            or not self._update_git_ubuntu_snap()
-            or not self._open_controller_port()
-            or not self._refresh_secret_keys()
-        ):
-            return
+        """Handle updates to config items.
 
-        # Refresh git-ubuntu source code
-        self.unit.status = ops.MaintenanceStatus("Refreshing git-ubuntu source.")
-        if not usr.refresh_git_ubuntu_source(
-            GIT_UBUNTU_SYSTEM_USER_USERNAME,
-            GIT_UBUNTU_USER_HOME_DIR,
-            self._git_ubuntu_source_url,
+        Update user settings, git config, the git-ubuntu snap and source, open ports, and keys.
+        If everything is successful, refresh git-ubuntu services.
+        """
+        if (
+            self._update_git_user_config()
+            and self._update_lpuser_config()
+            and self._update_git_ubuntu_snap()
+            and self._open_controller_port()
+            and self._refresh_secret_keys()
+            and self._refresh_git_ubuntu_source()
         ):
-            self.unit.status = ops.BlockedStatus("Failed to refresh git-ubuntu source.")
-            return
-
-        # Initialize or re-install git-ubuntu services as needed.
-        self._refresh_importer_node()
+            # Initialize or re-install git-ubuntu services as needed.
+            self._refresh_importer_node()
 
     def _on_leader_elected(self, _: ops.LeaderElectedEvent) -> None:
         """Refresh services and update peer data when the unit is elected as leader."""

src/user_management.py

@@ -157,7 +157,7 @@ def refresh_git_ubuntu_source(user: str, home_dir: str, source_url: str) -> bool
     clone_dir = pathops.LocalPath(directory_path, "live-allowlist-denylist-source")
 
     if clone_dir.is_dir():
-        logger.info("Updating existing git-ubuntu source in %s", clone_dir)
+        logger.info("Updating existing git-ubuntu source in %s", clone_dir.as_posix())
 
         # Update origin to the current source url
         if not _run_command_as_user(
@@ -174,8 +174,8 @@ def refresh_git_ubuntu_source(user: str, home_dir: str, source_url: str) -> bool
         return True
 
     # Clone the repository
-    logger.info("Cloning git-ubuntu source to %s", clone_dir)
-    if not _run_command_as_user(user, f"git clone {source_url} {clone_dir}"):
+    logger.info("Cloning git-ubuntu source to %s", clone_dir.as_posix())
+    if not _run_command_as_user(user, f"git clone {source_url} {clone_dir.as_posix()}"):
         logger.error("Failed to clone git-ubuntu source.")
         return False