Provide Oracle instances with a subnet id to use and provide metadata to instance launch.

VERSION

@@ -1 +1 @@
-1!10.9.1
+1!10.10.0

pycloudlib/oci/cloud.py

@@ -6,7 +6,7 @@
 import json
 import os
 import re
-from typing import List, Optional, cast
+from typing import Dict, List, Optional, cast
 
 import oci
 
@@ -262,7 +262,9 @@ def launch(
         retry_strategy=None,
         username: Optional[str] = None,
         cluster_id: Optional[str] = None,
+        subnet_id: Optional[str] = None,
         subnet_name: Optional[str] = None,
+        metadata: Dict = {},
         **kwargs,
     ) -> OciInstance:
         """Launch an instance.
@@ -273,7 +275,12 @@ def launch(
                 https://docs.cloud.oracle.com/en-us/iaas/Content/Compute/References/computeshapes.htm
             user_data: used by Cloud-Init to run custom scripts or
                 provide custom Cloud-Init configuration
+            subnet_id: string, OCID of subnet to use for instance.
+                Takes precedence over subnet_name if both are provided.
             subnet_name: string, name of subnet to use for instance.
+                Only used if subnet_id is not provided.
+            metadata: Dict, key-value pairs provided to the launch
+                details for the instance.
             retry_strategy: a retry strategy from oci.retry module
                 to apply for this operation
             username: username to use when connecting via SSH
@@ -289,20 +296,26 @@ def launch(
         if not image_id:
             raise ValueError(f"{self._type} launch requires image_id param. Found: {image_id}")
 
-        if subnet_name:
-            subnet_id = get_subnet_id_by_name(self.network_client, self.compartment_id, subnet_name)
-        else:
-            subnet_id = get_subnet_id(
-                self.network_client,
-                self.compartment_id,
-                self.availability_domain,
-                vcn_name=self.vcn_name,
-            )
-        metadata = {
+        # provided subnet_id takes the highest precendence
+        if not subnet_id:
+            if subnet_name:
+                subnet_id = get_subnet_id_by_name(
+                    self.network_client, self.compartment_id, subnet_name
+                )
+            else:
+                subnet_id = get_subnet_id(
+                    self.network_client,
+                    self.compartment_id,
+                    self.availability_domain,
+                    vcn_name=self.vcn_name,
+                )
+        default_metadata = {
             "ssh_authorized_keys": self.key_pair.public_key_content,
         }
         if user_data:
-            metadata["user_data"] = base64.b64encode(user_data.encode("utf8")).decode("ascii")
+            default_metadata["user_data"] = base64.b64encode(user_data.encode("utf8")).decode(
+                "ascii"
+            )
 
         instance_details = oci.core.models.LaunchInstanceDetails(  # noqa: E501
             display_name=self.tag,
@@ -312,7 +325,7 @@ def launch(
             shape=instance_type,
             subnet_id=subnet_id,
             image_id=image_id,
-            metadata=metadata,
+            metadata={**default_metadata, **metadata},
             compute_cluster_id=cluster_id,
             **kwargs,
         )

tests/unit_tests/oci/test_cloud.py

@@ -290,6 +290,64 @@ def test_launch_instance(self, mock_wait_till_ready, oci_cloud, oci_mock):
         mock_network_client.list_subnets.assert_called_once()
         assert oci_cloud.get_instance.call_count == 2
 
+        # Ensure when a subnet_id is directly passed to launch
+        # no functions to obtain a subnet-id are called.
+        with mock.patch("pycloudlib.oci.cloud.get_subnet_id") as m_subnet_id, \
+            mock.patch("pycloudlib.oci.cloud.get_subnet_id_by_name") as m_subnet_name:
+            instance = oci_cloud.launch(
+                "test-image-id", instance_type="VM.Standard2.1", subnet_id="subnet-id"
+            )
+            m_subnet_name.assert_not_called()
+            m_subnet_id.assert_not_called()
+
+        # The first arg is the LaunchInstanceDetails object
+        args, _ = oci_cloud.compute_client.launch_instance.call_args
+        launch_instance_details = args[0]
+        assert launch_instance_details.subnet_id == "subnet-id"
+        assert oci_cloud.get_instance.call_count == 3
+
+
+    @mock.patch("pycloudlib.oci.cloud.wait_till_ready")
+    def test_launch_custom_metadata(self, mock_wait_till_ready, oci_cloud):
+        """Test launch method with valid inputs."""
+        # mock the key pair
+        oci_cloud.key_pair = mock.Mock(public_key_config="ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQC")
+        oci_cloud.compute_client.launch_instance.return_value = mock.Mock(
+            data=mock.Mock(id="instance-id")
+        )
+        oci_cloud.get_instance = mock.Mock(return_value=mock.Mock())
+
+        # Ensure metdata gets combined with defaults
+        metadata = {"metadata_key": "metadata_value"}
+        default_metadata = {"ssh_authorized_keys": oci_cloud.key_pair.public_key_content}
+        expected_metadata = {**default_metadata, **metadata}
+        instance = oci_cloud.launch(
+           "test-image-id", instance_type="VM.Standard2.1", subnet_id="subnet-id",
+            metadata=metadata,
+        )
+
+        # The first arg is the LaunchInstanceDetails object
+        args, _ = oci_cloud.compute_client.launch_instance.call_args
+        launch_instance_details = args[0]
+        assert launch_instance_details.metadata == expected_metadata
+        assert instance is not None
+        oci_cloud.get_instance.assert_called_once()
+
+        # Ensure default_metadata values can be overridden
+        metadata = {"ssh_authorized_keys": "override", "metadata_key": "metadata_value"}
+        expected_metadata = {**default_metadata, **metadata}
+        instance = oci_cloud.launch(
+           "test-image-id", instance_type="VM.Standard2.1", subnet_id="subnet-id",
+            metadata=metadata,
+        )
+
+        # The first arg is the LaunchInstanceDetails object
+        args, _ = oci_cloud.compute_client.launch_instance.call_args
+        launch_instance_details = args[0]
+        assert launch_instance_details.metadata == expected_metadata
+        assert oci_cloud.get_instance.call_count == 2
+
+
     def test_launch_instance_invalid_image(self, oci_cloud):
         """Test launch method raises ValueError when no image_id is provided."""
         with pytest.raises(ValueError, match="launch requires image_id param"):