[VM] No Depot switch

For those with Depot access defined in their OS environment there is no
way to get a true vault pin of a VM.  This is an unusual use case so add
a negative flag to prevent installation of depot and setting the
parameters.

Author: PlaidCat

kt/commands/vm/command.py

@@ -71,8 +71,9 @@
 @click.option("--test", is_flag=True, help="Build the kernel and run kselftests")
 @click.option("--vcpus", type=int, default=12, help="Number of virtual CPUs (default: 12)")
 @click.option("--memory", type=int, default=32768, help="Memory in MiB (default: 32768)")
+@click.option("--no-depot", is_flag=True, help="Skip depot client installation and channel setup")
 @click.argument("kernel_workspace", required=False, shell_complete=ShellCompletion.show_kernel_workspaces)
-def vm(kernel_workspace, console, destroy, override, override_base, list_all, test, vcpus, memory):
+def vm(kernel_workspace, console, destroy, override, override_base, list_all, test, vcpus, memory, no_depot):
     if not list_all and not kernel_workspace:
         raise click.UsageError("kernel_workspace is required unless --list-all is specified")
 
@@ -86,4 +87,5 @@ def vm(kernel_workspace, console, destroy, override, override_base, list_all, te
         test=test,
         vcpus=vcpus,
         memory=memory,
+        no_depot=no_depot,
     )

kt/commands/vm/impl.py

@@ -16,6 +16,7 @@ def main(
     test: bool = False,
     vcpus: int = 12,
     memory: int = 32768,
+    no_depot: bool = False,
 ):
     if list_all:
         VmCommand.list_all()
@@ -31,7 +32,12 @@ def main(
         return
 
     vm_instance = Vm.setup_and_spinup(
-        kernel_workspace_name=name, override=override, override_base=override_base, vcpus=vcpus, memory=memory
+        kernel_workspace_name=name,
+        override=override,
+        override_base=override_base,
+        vcpus=vcpus,
+        memory=memory,
+        no_depot=no_depot,
     )
     config = Config.load()
 

kt/ktlib/vm.py

@@ -142,6 +142,7 @@ def setup_and_spinup(
         override_base: bool = False,
         vcpus: int = 12,
         memory: int = 32768,
+        no_depot: bool = False,
     ):
         """
         Setup and spin up a VM from a kernel workspace name.
@@ -152,6 +153,7 @@ def setup_and_spinup(
             override_base: If True, destroy and recreate the VM but override the base image as well
             vcpus: Number of virtual CPUs
             memory: Memory in MiB
+            no_depot: If True, skip depot client installation and channel setup
 
         Returns:
             VmInstance: The running VM instance
@@ -163,7 +165,7 @@ def setup_and_spinup(
             vm.destroy()
 
         vm.setup(override_base=override_base)
-        vm_instance = vm.spin_up(config=config, vcpus=vcpus, memory=memory)
+        vm_instance = vm.spin_up(config=config, vcpus=vcpus, memory=memory, no_depot=no_depot)
 
         return vm_instance
 
@@ -186,7 +188,7 @@ def _download_source_image(self, override_base: bool = False):
         logging.info(f"Downloading image from {self._get_vm_url()}")
         wget.download(self._get_vm_url(), out=str(self.qcow2_source_path))
 
-    def _setup_cloud_init(self, config: Config):
+    def _setup_cloud_init(self, config: Config, no_depot: bool = False):
         yaml = YAML()
         yaml.preserve_quotes = True
         yaml.width = 4096
@@ -231,26 +233,29 @@ def _setup_cloud_init(self, config: Config):
             )
             data["runcmd"].append("dnf clean all")
 
-        # Depot: always install the client, but only login+enable if credentials are present
-        data["runcmd"].append("dnf install -y https://depot.ciq.com/public/files/depot-client/depot/depot.x86_64.rpm")
-        depot_user = os.environ.get("DEPOT_USER")
-        depot_token = os.environ.get("DEPOT_TOKEN")
-        if depot_user and depot_token and self.depot_channels:
-            data["runcmd"].append(f"depot login -u {depot_user} -t {depot_token}")
-            for channel in self.depot_channels:
-                data["runcmd"].append(f"depot enable {channel} -y")
+        # Depot: install the client and login+enable if credentials are present
+        if not no_depot:
+            data["runcmd"].append(
+                "dnf install -y https://depot.ciq.com/public/files/depot-client/depot/depot.x86_64.rpm"
+            )
+            depot_user = os.environ.get("DEPOT_USER")
+            depot_token = os.environ.get("DEPOT_TOKEN")
+            if depot_user and depot_token and self.depot_channels:
+                data["runcmd"].append(f"depot login -u {depot_user} -t {depot_token}")
+                for channel in self.depot_channels:
+                    data["runcmd"].append(f"depot enable {channel} -y")
 
         # Install packages needed later
         data["runcmd"].append([str(config.base_path / Path("kernel-src-tree-tools") / Path("kernel_install_dep.sh"))])
         # Write this to image cloud_init
         with open(self.cloud_init_path, "w") as f:
             yaml.dump(data, f)
 
-    def _create_image(self, config: Config, vcpus: int = 12, memory: int = 32768):
+    def _create_image(self, config: Config, vcpus: int = 12, memory: int = 32768, no_depot: bool = False):
         # Make sure the dir exists
         self.qcow2_path.parent.mkdir(parents=True, exist_ok=True)
 
-        self._setup_cloud_init(config=config)
+        self._setup_cloud_init(config=config, no_depot=no_depot)
         # Copy qcow2 image to work dir
         self.qcow2_source_path.copy(self.qcow2_path)
 
@@ -282,11 +287,11 @@ def _resize_disk(self):
     def setup(self, override_base: bool = False):
         self._download_source_image(override_base=override_base)
 
-    def spin_up(self, config: Config, vcpus: int = 12, memory: int = 32768) -> VmInstance:
+    def spin_up(self, config: Config, vcpus: int = 12, memory: int = 32768, no_depot: bool = False) -> VmInstance:
         if not VirtHelper.exists(vm_name=self.name):
             logging.info(f"VM {self.name} does not exist, creating from scratch...")
 
-            self._create_image(config=config, vcpus=vcpus, memory=memory)
+            self._create_image(config=config, vcpus=vcpus, memory=memory, no_depot=no_depot)
             return VmInstance(name=self.name, kernel_workspace=self.kernel_workspace)
 
         logging.info(f"Vm {self.name} already exists")

tests/kt/ktlib/test_vm.py

@@ -90,7 +90,7 @@ def _make_config():
     return config
 
 
-def _build_vm_and_get_runcmd(vm_image_url=None, depot_channels=None, depot_env=None):
+def _build_vm_and_get_runcmd(vm_image_url=None, depot_channels=None, depot_env=None, no_depot=False):
     """Build a Vm, call _setup_cloud_init, and capture the generated runcmd."""
     config = _make_config()
 
@@ -139,7 +139,7 @@ def __exit__(self_inner, *args):
         return _real_open(path, mode)
 
     with patch("builtins.open", side_effect=fake_open), patch.dict(os.environ, env):
-        vm._setup_cloud_init(config=config)
+        vm._setup_cloud_init(config=config, no_depot=no_depot)
 
     content = captured_yaml.get("content", "")
     if content.startswith("#cloud-config\n"):
@@ -166,7 +166,7 @@ def test_cloud_init_no_vault_pin_when_no_image_url():
     assert not any("vault/rocky" in s for s in runcmd_strs)
 
 
-def test_cloud_init_depot_install_always_present():
+def test_cloud_init_depot_install_present_by_default():
     runcmd = _build_vm_and_get_runcmd(vm_image_url=None)
     runcmd_strs = [str(c) for c in runcmd]
     assert any("depot.x86_64.rpm" in s for s in runcmd_strs)
@@ -205,3 +205,23 @@ def test_cloud_init_ordering():
     dep_script_idx = next(i for i, s in enumerate(runcmd_strs) if "kernel_install_dep.sh" in s)
 
     assert vault_idx < depot_install_idx < depot_login_idx < dep_script_idx
+
+
+def test_cloud_init_no_depot_skips_all_depot_commands():
+    """no_depot=True skips depot even when creds and channels are present."""
+    runcmd = _build_vm_and_get_runcmd(
+        depot_channels=["lts-8.6"],
+        depot_env=("myuser", "mytoken"),
+        no_depot=True,
+    )
+    runcmd_strs = [str(c) for c in runcmd]
+    assert not any("depot" in s for s in runcmd_strs)
+    assert any("kernel_install_dep.sh" in s for s in runcmd_strs)
+
+
+def test_cloud_init_no_depot_skips_depot_install():
+    """no_depot=True omits the depot RPM install (no creds variant)."""
+    runcmd = _build_vm_and_get_runcmd(no_depot=True)
+    runcmd_strs = [str(c) for c in runcmd]
+    assert not any("depot.x86_64.rpm" in s for s in runcmd_strs)
+    assert any("kernel_install_dep.sh" in s for s in runcmd_strs)