fix: trust config plumbing for instance

azext_edge/edge/providers/orchestration/targets.py

@@ -164,7 +164,28 @@ def get_ops_enablement_template(
         # TODO - @digimaun - expand trustSource for self managed & trustBundleSettings
         return template.content, parameters
 
-    def get_ops_instance_template(self, cl_extension_ids: List[str]) -> Tuple[dict, dict]:
+    def get_ops_instance_template(
+        self, cl_extension_ids: List[str], ops_extension_config: Dict[str, str]
+    ) -> Tuple[dict, dict]:
+        # Set the schema registry resource Id from the extension config
+        self.schema_registry_resource_id = ops_extension_config.get("schemaRegistry.values.resourceId")
+        trust_source = ops_extension_config.get("trustSource")
+
+        # TODO - This conditional should be temporary until the AIO extension and instance are deployed
+        # in the same flow.
+        if trust_source == "CustomerManaged":
+            trust_issuer_name = ops_extension_config.get("trustBundleSettings.issuer.name")
+            trust_issuer_kind = ops_extension_config.get("trustBundleSettings.issuer.kind")
+            trust_configmap_name = ops_extension_config.get("trustBundleSettings.configMap.name")
+            trust_configmap_key = ops_extension_config.get("trustBundleSettings.configMap.key")
+            self.trust_settings = {
+                "issuerName": trust_issuer_name,
+                "issuerKind": trust_issuer_kind,
+                "configMapName": trust_configmap_name,
+                "configMapKey": trust_configmap_key,
+            }
+        self.trust_config = self.get_trust_settings_target_map()
+
         template, parameters = self._handle_apply_targets(
             param_to_target={
                 "clusterName": self.cluster_name,
@@ -176,6 +197,7 @@ def get_ops_instance_template(self, cl_extension_ids: List[str]) -> Tuple[dict,
                 "schemaRegistryId": self.schema_registry_resource_id,
                 "defaultDataflowinstanceCount": self.dataflow_profile_instances,
                 "brokerConfig": self.broker_config,
+                "trustConfig": self.trust_config,
             },
             template_blueprint=M2_INSTANCE_TEMPLATE,
         )

azext_edge/edge/providers/orchestration/work.py

@@ -373,15 +373,14 @@ def _do_work(self):  # noqa: C901
                             "Foundational service installation not detected. "
                             "Instance deployment will not continue. Please run init."
                         )
-                    # Set the schema registry resource Id from the extension config
-                    self._targets.schema_registry_resource_id = self._extension_map[IOT_OPS_EXTENSION_TYPE][
-                        "properties"
-                    ]["configurationSettings"]["schemaRegistry.values.resourceId"]
 
                 instance_work_name = self._work_format_str.format(op="instance")
                 self.render_display(category=WorkCategoryKey.DEPLOY_IOT_OPS, active_step=WorkStepKey.WHAT_IF_INSTANCE)
                 instance_content, instance_parameters = self._targets.get_ops_instance_template(
-                    cl_extension_ids=[self._extension_map[ext]["id"] for ext in self._extension_map]
+                    cl_extension_ids=[self._extension_map[ext]["id"] for ext in self._extension_map],
+                    ops_extension_config=self._extension_map[IOT_OPS_EXTENSION_TYPE]["properties"][
+                        "configurationSettings"
+                    ],
                 )
                 self._deploy_template(
                     content=instance_content,

azext_edge/tests/edge/orchestration/test_targets_unit.py

@@ -54,6 +54,7 @@ def get_trust_settings():
     "schemaRegistryId": "schema_registry_resource_id",
     "defaultDataflowinstanceCount": "dataflow_profile_instances",
     "brokerConfig": "broker_config",
+    "trustConfig": "trust_config",
 }
 
 
@@ -126,16 +127,7 @@ def test_init_targets(target_scenario: dict):
     if target_scenario.get("enable_fault_tolerance"):
         assert targets.advanced_config == {"edgeStorageAccelerator": {"faultToleranceEnabled": True}}
 
-    if target_scenario.get("trust_settings"):
-        assert targets.trust_config == {
-            "source": "CustomerManaged",
-            "settings": {
-                "issuerKind": target_scenario["trust_settings"]["issuerKind"],
-                "configMapKey": target_scenario["trust_settings"]["configMapKey"],
-                "issuerName": target_scenario["trust_settings"]["issuerName"],
-                "configMapName": target_scenario["trust_settings"]["configMapName"],
-            },
-        }
+    verify_user_trust_settings(targets, target_scenario)
 
     enablement_template, enablement_parameters = targets.get_ops_enablement_template()
     for parameter in enablement_parameters:
@@ -156,7 +148,17 @@ def test_init_targets(target_scenario: dict):
         assert enablement_template["variables"]["VERSIONS"]["aio"] == targets.ops_version
 
     extension_ids = [generate_random_string(), generate_random_string()]
-    instance_template, instance_parameters = targets.get_ops_instance_template(extension_ids)
+    extension_config = {"schemaRegistry.values.resourceId": target_scenario.get("schema_registry_resource_id")}
+    target_scenario_has_user_trust = target_scenario.get("trust_settings")
+    if target_scenario_has_user_trust:
+        extension_config["trustSource"] = "CustomerManaged"
+        extension_config["trustBundleSettings.issuer.name"] = target_scenario["trust_settings"]["issuerName"]
+        extension_config["trustBundleSettings.issuer.kind"] = target_scenario["trust_settings"]["issuerKind"]
+        extension_config["trustBundleSettings.configMap.name"] = target_scenario["trust_settings"]["configMapName"]
+        extension_config["trustBundleSettings.configMap.key"] = target_scenario["trust_settings"]["configMapKey"]
+        targets.trust_config = None
+
+    instance_template, instance_parameters = targets.get_ops_instance_template(extension_ids, extension_config)
     for parameter in instance_parameters:
         if parameter == "clExtentionIds":
             assert instance_parameters[parameter]["value"] == extension_ids
@@ -188,3 +190,22 @@ def test_init_targets(target_scenario: dict):
         assert instance_template["resources"]["broker_listener_insecure"] == get_insecure_listener(
             targets.instance_name, "default"
         )
+
+    verify_user_trust_settings(targets, target_scenario)
+
+
+def verify_user_trust_settings(targets: InitTargets, target_scenario: dict):
+    target_scenario_has_user_trust = target_scenario.get("trust_settings")
+    if not target_scenario_has_user_trust:
+        assert targets.trust_config == {"source": "SelfSigned"}
+        return
+
+    assert targets.trust_config == {
+        "source": "CustomerManaged",
+        "settings": {
+            "issuerKind": target_scenario["trust_settings"]["issuerKind"],
+            "configMapKey": target_scenario["trust_settings"]["configMapKey"],
+            "issuerName": target_scenario["trust_settings"]["issuerName"],
+            "configMapName": target_scenario["trust_settings"]["configMapName"],
+        },
+    }