fix: stop overriding team-operator image when not explicitly configured

python-pulumi/src/ptd/__init__.py

@@ -409,7 +409,7 @@ class Toleration:
 
 @dataclasses.dataclass(frozen=True)
 class WorkloadClusterConfig:
-    team_operator_image: str = "latest"
+    team_operator_image: str | None = None
     # Overrides team_operator_image when set. Can be a tag (e.g., "test", "dev")
     # or a full image reference. For adhoc images from posit-dev/team-operator PRs:
     #   ghcr.io/posit-dev/team-operator:adhoc-{branch}-{version}

python-pulumi/src/ptd/aws_workload.py

@@ -408,8 +408,10 @@ def _load_workload_cluster_config_dict(
         for key in list(cluster_spec.keys()):
             cluster_spec[key.replace("-", "_")] = cluster_spec.pop(key)
 
-        team_operator_image = cluster_spec.pop("team_operator_image", "latest").strip().lower()
-        cluster_spec["team_operator_image"] = {"": "latest"}.get(team_operator_image, team_operator_image)
+        team_operator_image = cluster_spec.pop("team_operator_image", None)
+        if team_operator_image is not None:
+            team_operator_image = team_operator_image.strip().lower() or None
+        cluster_spec["team_operator_image"] = team_operator_image
 
         ptd_controller_image = cluster_spec.pop("ptd_controller_image", "latest").strip().lower()
 

python-pulumi/src/ptd/azure_workload.py

@@ -183,8 +183,10 @@ def _load_workload_cluster_config_dict(
         for key in list(cluster_spec.keys()):
             cluster_spec[key.replace("-", "_")] = cluster_spec.pop(key)
 
-        team_operator_image = cluster_spec.pop("team_operator_image", "latest").strip().lower()
-        cluster_spec["team_operator_image"] = {"": "latest"}.get(team_operator_image, team_operator_image)
+        team_operator_image = cluster_spec.pop("team_operator_image", None)
+        if team_operator_image is not None:
+            team_operator_image = team_operator_image.strip().lower() or None
+        cluster_spec["team_operator_image"] = team_operator_image
 
         # Handle user_node_pools if present
         if cluster_spec.get("user_node_pools"):

python-pulumi/src/ptd/pulumi_resources/team_operator.py

@@ -71,9 +71,13 @@ def __init__(
         self._define_helm_release()
 
     def _define_image(self):
-        # Use adhoc_team_operator_image if set, otherwise use team_operator_image
-        # adhoc images can be tags like "test", "dev", or full image references
+        # Use adhoc_team_operator_image if set, otherwise use team_operator_image if explicitly set.
+        # If neither is set (team_operator_image is None), self.image stays None so the Helm chart
+        # defaults to its appVersion.
         image_config = self.cluster_cfg.adhoc_team_operator_image or self.cluster_cfg.team_operator_image
+        if image_config is None:
+            self.image = None
+            return
         self.image = ptd.define_component_image(
             image_config=image_config,
             component_image=ptd.ComponentImages.TEAM_OPERATOR,
@@ -244,17 +248,19 @@ def _build_migration_script(self, namespace: str) -> str:
     def _define_helm_release(self):
         # Parse self.image (from _define_image) into repository and tag
         # Format is either "repo@sha256:digest" or "repo:tag"
-        if "@" in self.image:
-            # Image with digest: "hostname/repo@sha256:abc123"
-            image_repository, image_tag = self.image.rsplit("@", 1)
-            image_tag = f"@{image_tag}"  # Helm needs the @ prefix for digests
-        elif ":" in self.image.split("/")[-1]:
-            # Image with tag: "hostname/repo:tag"
-            image_repository, image_tag = self.image.rsplit(":", 1)
-        else:
-            # No tag specified, use latest
-            image_repository = self.image
-            image_tag = "latest"
+        # If self.image is None, we skip image configuration to let the Helm chart use its default appVersion
+        if self.image is not None:
+            if "@" in self.image:
+                # Image with digest: "hostname/repo@sha256:abc123"
+                image_repository, image_tag = self.image.rsplit("@", 1)
+                image_tag = f"@{image_tag}"  # Helm needs the @ prefix for digests
+            elif ":" in self.image.split("/")[-1]:
+                # Image with tag: "hostname/repo:tag"
+                image_repository, image_tag = self.image.rsplit(":", 1)
+            else:
+                # No tag specified, use latest
+                image_repository = self.image
+                image_tag = "latest"
 
         # Build environment variables
         env_vars = {
@@ -265,17 +271,19 @@ def _define_helm_release(self):
         if self.workload.cfg.region:
             env_vars["AWS_REGION"] = self.workload.cfg.region
 
+        # Build container config - only include image if explicitly set
+        container_config = {"env": env_vars}
+        if self.image is not None:
+            container_config["image"] = {
+                "repository": image_repository,
+                "tag": image_tag,
+            }
+
         # Helm values for the team-operator chart
         helm_values = {
             "controllerManager": {
                 "replicas": 1,
-                "container": {
-                    "image": {
-                        "repository": image_repository,
-                        "tag": image_tag,
-                    },
-                    "env": env_vars,
-                },
+                "container": container_config,
                 # Use default serviceAccountName from chart (team-operator-controller-manager)
                 # to match existing kustomize resources for seamless migration
                 "serviceAccount": {

python-pulumi/tests/test_workload_cluster_config.py

@@ -10,7 +10,7 @@ def test_workload_cluster_config_default_initialization():
     config = ptd.WorkloadClusterConfig()
 
     # Test default values
-    assert config.team_operator_image == "latest"
+    assert config.team_operator_image is None
     assert config.ptd_controller_image == "latest"
     assert config.eks_access_entries.enabled is True
     assert config.eks_access_entries.additional_entries == []
@@ -175,7 +175,7 @@ def test_workload_cluster_config_dataclass_fields():
 
     # team_operator_image field
     team_op_field = field_dict["team_operator_image"]
-    assert team_op_field.default == "latest"
+    assert team_op_field.default is None
 
     # ptd_controller_image field
     ptd_ctrl_field = field_dict["ptd_controller_image"]