From 47048c31b9f03adc283ec73453436d7d3d583ba3 Mon Sep 17 00:00:00 2001 From: a-dubs Date: Thu, 13 Feb 2025 14:22:20 -0500 Subject: [PATCH] chore: fix linting issues (#470) --- examples/oracle/oracle-cluster-demo.py | 3 +- .../oracle/oracle-example-cluster-test.py | 188 +++++++++++------- pycloudlib/instance.py | 2 +- pycloudlib/oci/cloud.py | 44 +--- pycloudlib/oci/instance.py | 49 +++-- pycloudlib/oci/utils.py | 1 + .../integration_tests/oracle/test_cluster.py | 10 +- tests/unit_tests/oci/test_instance.py | 78 +++++--- 8 files changed, 221 insertions(+), 154 deletions(-) diff --git a/examples/oracle/oracle-cluster-demo.py b/examples/oracle/oracle-cluster-demo.py index 648e6c2e..8ecc06bf 100644 --- a/examples/oracle/oracle-cluster-demo.py +++ b/examples/oracle/oracle-cluster-demo.py @@ -7,13 +7,13 @@ import pycloudlib + def demo_cluster( availability_domain: str = None, compartment_id: str = None, vcn_name: str = None, ): """Show example of using the OCI library to launch a cluster instance and ping between them.""" - with pycloudlib.OCI( "pycl-oracle-cluster-demo", availability_domain=availability_domain, @@ -37,6 +37,7 @@ def demo_cluster( else: print(f"Successfully pinged {private_ip} from {instance.private_ip}") + if __name__ == "__main__": logging.basicConfig(level=logging.DEBUG) if len(sys.argv) != 3: diff --git a/examples/oracle/oracle-example-cluster-test.py b/examples/oracle/oracle-example-cluster-test.py index 8073d6a9..85857059 100644 --- a/examples/oracle/oracle-example-cluster-test.py +++ b/examples/oracle/oracle-example-cluster-test.py @@ -2,37 +2,37 @@ # This file is part of pycloudlib. See LICENSE file for license information. """Basic examples of various lifecycle with an OCI instance.""" -from datetime import datetime -import json import logging -import sys +import threading import time +from datetime import datetime +from typing import Generator import pytest import pycloudlib from pycloudlib.oci.instance import OciInstance -from typing import Generator -import logging -import threading logger = logging.getLogger(__name__) -EXISTING_INSTANCE_IDS = [ +EXISTING_INSTANCE_IDS: list[str] = [ # add the OCIDs of the instances you want to use for testing here ] + # change this to either "class" or "module" as you see fit @pytest.fixture(scope="module") def cluster() -> Generator[list[OciInstance], None, None]: """ - Launch a cluster of BM instances and yield them. - """ + Launch a cluster of BM instances. + Yields: + list[OciInstance]: The created or retrieved cluster instances. + """ with pycloudlib.OCI( "pycl-oracle-cluster-test", # use the already created "mofed-vcn" for cluster testing - vcn_name="mofed-vcn" # THIS WILL OVERRIDE THE VCN_NAME IN THE CONFIG FILE + vcn_name="mofed-vcn", # THIS WILL OVERRIDE THE VCN_NAME IN THE CONFIG FILE ) as client: if EXISTING_INSTANCE_IDS: instances = [client.get_instance(instance_id) for instance_id in EXISTING_INSTANCE_IDS] @@ -42,16 +42,21 @@ def cluster() -> Generator[list[OciInstance], None, None]: # so once this function returns, the instances are ready instances = client.create_compute_cluster( # if you create a custom image, specify its OCID here - image_id=client.released_image("noble"), + image_id=client.released_image("noble"), instance_count=2, ) yield instances class TestOracleClusterBasic: - def test_basic_ping_on_private_ips(self, cluster: list[OciInstance]): + """Test basic functionalities of Oracle Cluster.""" + + def test_basic_ping_on_private_ips(self, cluster: list[OciInstance]): # pylint: disable=W0621 """ - Verifies that the instances in the cluster can reach each other on their private IPs. + Test that cluster instances can ping each other on private IPs. + + Args: + cluster (list[OciInstance]): Instances in the cluster. """ # get the private ips of the instances private_ips = [instance.private_ip for instance in cluster] @@ -59,16 +64,27 @@ def test_basic_ping_on_private_ips(self, cluster: list[OciInstance]): for instance in cluster: for private_ip in private_ips: if private_ip != instance.private_ip: - logger.info(f"Pinging {private_ip} from {instance.private_ip}") + logger.info("Pinging %s from %s", private_ip, instance.private_ip) # ping once with a timeout of 5 seconds r = instance.execute(f"ping -c 1 -W 5 {private_ip}") assert r.ok, f"Failed to ping {private_ip} from {instance.private_ip}" - logger.info(f"Successfully pinged {private_ip} from {instance.private_ip}") + logger.info("Successfully pinged %s from %s", private_ip, instance.private_ip) def setup_mofed_iptables_rules(instance: OciInstance): + """ + Set up IPTABLES rules for RDMA usage. + + Args: + instance (OciInstance): Target instance to configure. + + Returns: + OciInstance: The same instance after configuration. + """ # Update the cloud.cfg file to set preserve_hostname to true - instance.execute("sed -i 's/preserve_hostname: false/preserve_hostname: true/' /etc/cloud/cloud.cfg") + instance.execute( + "sed -i 's/preserve_hostname: false/preserve_hostname: true/' /etc/cloud/cloud.cfg" + ) # Backup the existing iptables rules backup_file = f"/etc/iptables/rules.v4.bak.{datetime.now().strftime('%F-%T')}" instance.execute(f"cp -v /etc/iptables/rules.v4 {backup_file}") @@ -101,6 +117,12 @@ def setup_mofed_iptables_rules(instance: OciInstance): def ensure_image_is_rdma_ready(instance: OciInstance): + """ + Check if the image supports RDMA. + + Args: + instance (OciInstance): The instance to verify. + """ r = instance.execute("ibstatus") if not r.stdout or not r.ok: logger.info("Infiniband status: %s", r.stdout + "\n" + r.stderr) @@ -108,16 +130,18 @@ def ensure_image_is_rdma_ready(instance: OciInstance): class TestOracleClusterRdma: + """Test RDMA functionalities of Oracle Cluster.""" + @pytest.fixture(scope="class") - def mofed_cluster(self, cluster: list[OciInstance]) -> Generator[list[OciInstance], None, None]: + def mofed_cluster( + self, + cluster: list[OciInstance], # pylint: disable=W0621 + ) -> Generator[list[OciInstance], None, None]: """ - Custom fixture to configure the instances in the cluster for RDMA testing. - - This fixture will: - - Ensure the image being used is RDMA ready - - Create a secondary VNIC on the private subnet for each instance in the cluster - - Configure the secondary VNIC for RDMA usage - - Set up the necessary iptables rules for RDMA usage on each instance's secondary NIC + Configure cluster for RDMA testing. + + Yields: + list[OciInstance]: RDMA-ready cluster instances. """ ensure_image_is_rdma_ready(cluster[0]) for instance in cluster: @@ -130,38 +154,54 @@ def mofed_cluster(self, cluster: list[OciInstance]) -> Generator[list[OciInstanc # create a secondary VNIC on the 2nd vnic on the private subnet for RDMA usage instance.add_network_interface( nic_index=1, - subnet_name="private subnet-mofed-vcn", # use the private subnet for mofed testing + subnet_name="private subnet-mofed-vcn", # use the private subnet for mofed testing ) instance.configure_secondary_vnic() setup_mofed_iptables_rules(instance) - + yield cluster - - def test_basic_ping_on_new_rdma_ips( - self, - mofed_cluster: list[OciInstance], - ): - # get the private ips of the instances + + def test_basic_ping_on_new_rdma_ips(self, mofed_cluster: list[OciInstance]): + """ + Test ping on RDMA-enabled private IPs. + + Args: + mofed_cluster (list[OciInstance]): RDMA-enabled cluster instances. + """ + # get the private ips of the instances that are on the same RDMA-enabled subnet rdma_ips = [instance.secondary_vnic_private_ip for instance in mofed_cluster] - # try to ping each instance from each other instance at their private ip + for instance in mofed_cluster: for rdma_ip in rdma_ips: if rdma_ip != instance.secondary_vnic_private_ip: - logger.info(f"Pinging {rdma_ip} from {instance.secondary_vnic_private_ip}") - # ping once with a timeout of 5 seconds + logger.info( + "Pinging %s from %s", + rdma_ip, + instance.secondary_vnic_private_ip, + ) + # ping once with a timeout of 5 seconds so it doesn't hang r = instance.execute(f"ping -c 1 -W 5 {rdma_ip}") - assert r.ok, f"Failed to ping {rdma_ip} from {instance.secondary_vnic_private_ip}" - logger.info(f"Successfully pinged {rdma_ip} from {instance.secondary_vnic_private_ip}") - - def test_rping( - self, - mofed_cluster: list[OciInstance], - ): + assert ( + r.ok + ), f"Failed to ping {rdma_ip} from {instance.secondary_vnic_private_ip}" + logger.info( + "Successfully pinged %s from %s", + rdma_ip, + instance.secondary_vnic_private_ip, + ) + + def test_rping(self, mofed_cluster: list[OciInstance]): + """ + Test rping between two instances. + + Args: + mofed_cluster (list[OciInstance]): RDMA-enabled cluster instances + """ server_instance = mofed_cluster[0] client_instance = mofed_cluster[1] def start_server(): - # start the rping server on the first instance + """Start the rping server on the "server_instance".""" server_instance.execute(f"rping -s -a {server_instance.secondary_vnic_private_ip} -v &") server_thread = threading.Thread(target=start_server) @@ -170,74 +210,88 @@ def start_server(): # Wait for rping server to start time.sleep(5) # start the rping client on the second instance (only send 10 packets so it doesn't hang) - r = client_instance.execute(f"rping -c -a {server_instance.secondary_vnic_private_ip} -C 10 -v") + r = client_instance.execute( + f"rping -c -a {server_instance.secondary_vnic_private_ip} -C 10 -v" + ) logger.info("rping output: %s", r.stdout) assert r.ok, "Failed to run rping" - def test_ucmatose( - self, - mofed_cluster: list[OciInstance], - ): + def test_ucmatose(self, mofed_cluster: list[OciInstance]): + """ + Test ucmatose connections. + + Args: + mofed_cluster (list[OciInstance]): RDMA-enabled cluster instances + """ server_instance = mofed_cluster[0] client_instance = mofed_cluster[1] def start_server(): - # start the rping server on the first instance - server_instance.execute(f"ucmatose &") + """Start the ucmatose server on the "server_instance".""" + server_instance.execute("ucmatose &") server_thread = threading.Thread(target=start_server) server_thread.start() # Wait for server to start time.sleep(5) - # start the client on the second instance (only send 10 packets so it doesn't hang) + # start the ucmatose client r = client_instance.execute(f"ucmatose -s {server_instance.secondary_vnic_private_ip}") logger.info("ucmatose output: %s", r.stdout) assert r.ok, "Failed to run ucmatose" - def test_ucx_perftest_lat_one_node( - self, - mofed_cluster: list[OciInstance], - ): + def test_ucx_perftest_lat_one_node(self, mofed_cluster: list[OciInstance]): + """ + Run ucx_perftest latency on a single node. + + Args: + mofed_cluster (list[OciInstance]): RDMA-enabled cluster instances + """ server_instance = mofed_cluster[0] # ucx_perftest only works within a single instance on all MOFED stacks right now, so this # being 0 is intentional. (Will adjust if Oracle provides config info to resolve this) client_instance = mofed_cluster[0] def start_server(): - # start the rping server on the first instance - server_instance.execute(f"ucx_perftest -c 0 &") + """Start the ucx_perftest server on the "server_instance".""" + server_instance.execute("ucx_perftest -c 0 &") server_thread = threading.Thread(target=start_server) server_thread.start() # Wait for server to start time.sleep(5) - # start the client on the second instance (only send 10 packets so it doesn't hang) - r = client_instance.execute(f"ucx_perftest {server_instance.secondary_vnic_private_ip} -t tag_lat -c 1") + # start the ucx_perftest client + r = client_instance.execute( + f"ucx_perftest {server_instance.secondary_vnic_private_ip} -t tag_lat -c 1" + ) logger.info("ucx_perftest output: %s", r.stdout) assert r.ok, "Failed to run ucx_perftest" + def test_ucx_perftest_bw_one_node(self, mofed_cluster: list[OciInstance]): + """ + Run ucx_perftest bandwidth on a single node. - def test_ucx_perftest_bw_one_node( - self, - mofed_cluster: list[OciInstance], - ): + Args: + mofed_cluster (list[OciInstance]): RDMA-enabled cluster instances + """ server_instance = mofed_cluster[0] # ucx_perftest only works within a single instance on all MOFED stacks right now, so this # being 0 is intentional. (Will adjust if Oracle provides config info to resolve this) client_instance = mofed_cluster[0] def start_server(): - # start the rping server on the first instance - server_instance.execute(f"ucx_perftest -c 0 &") + """Start the ucx_perftest server on the "server_instance".""" + server_instance.execute("ucx_perftest -c 0 &") server_thread = threading.Thread(target=start_server) server_thread.start() # Wait for server to start time.sleep(5) - # start the client on the second instance (only send 10 packets so it doesn't hang) - r = client_instance.execute(f"ucx_perftest {server_instance.secondary_vnic_private_ip} -t tag_bw -c 1") + # start the ucx_perftest client + r = client_instance.execute( + f"ucx_perftest {server_instance.secondary_vnic_private_ip} -t tag_bw -c 1" + ) logger.info("ucx_perftest output: %s", r.stdout) assert r.ok, "Failed to run ucx_perftest" diff --git a/pycloudlib/instance.py b/pycloudlib/instance.py index 762ade10..9a2a5dbe 100644 --- a/pycloudlib/instance.py +++ b/pycloudlib/instance.py @@ -58,7 +58,7 @@ def __exit__(self, _type, _value, _traceback): def id(self) -> str: """Return instance id.""" raise NotImplementedError - + @property def private_ip(self) -> str: """Return instance private ip.""" diff --git a/pycloudlib/oci/cloud.py b/pycloudlib/oci/cloud.py index 6f3650a1..68aa5e9d 100644 --- a/pycloudlib/oci/cloud.py +++ b/pycloudlib/oci/cloud.py @@ -371,36 +371,26 @@ def _validate_tag(tag: str): if rules_failed: raise InvalidTagNameError(tag=tag, rules_failed=rules_failed) - # function that inits empty compute cluster - # then launches X number of BM instances and attaches them to the cluster at launch - # then returns an iterable of the instance objects that were created def create_compute_cluster( - self, - image_id: str, - instance_count: int = 2, - instance_type: str = "BM.Optimized3.36", - optional_launch_kwargs: Optional[dict] = None, - wait: bool = True, - ) -> list[OciInstance]: + self, + image_id: str, + instance_count: int = 2, + instance_type: str = "BM.Optimized3.36", + optional_launch_kwargs: Optional[dict] = None, + wait: bool = True, + ) -> list[OciInstance]: """ Create a compute cluster with a specified number of instances. Args: instance_count: Number of instances to create launch_kwargs: Additional arguments to pass to the launch method - + Returns: List of OciInstance objects that were created """ - - # create an empty compute cluster cluster_id = self._init_empty_cluster() - - # create a list to store the instances that are created instances: list[OciInstance] = [] - - # if image id or instance type are provided in the optional launch kwargs, pop - # them out and use them instead of the default values if optional_launch_kwargs: image_id = optional_launch_kwargs.pop("image_id", image_id) instance_type = optional_launch_kwargs.pop("instance_type", instance_type) @@ -419,26 +409,13 @@ def create_compute_cluster( instance.wait() return instances - # function that creates an empty compute cluster def _init_empty_cluster(self) -> str: """ Create an empty cluster and return the cluster id. - :return: cluster id + Returns: + str: The id of the created compute cluster. """ - # create_compute_cluster_response = core_client.create_compute_cluster( - # create_compute_cluster_details=oci.core.models.CreateComputeClusterDetails( - # availability_domain="EXAMPLE-availabilityDomain-Value", - # compartment_id="ocid1.test.oc1..EXAMPLE-compartmentId-Value", - # display_name="EXAMPLE-displayName-Value", - # defined_tags={ - # 'EXAMPLE_KEY_oDNhF': { - # 'EXAMPLE_KEY_0JbDz': 'EXAMPLE--Value'}}, - # freeform_tags={ - # 'EXAMPLE_KEY_7199B': 'EXAMPLE_VALUE_qGHycVV8GARc88jnSUIR'}), - # opc_retry_token="EXAMPLE-opcRetryToken-Value", - # opc_request_id="G968LPS24ZTZRZEWGC8Z") - create_compute_cluster_response = self.compute_client.create_compute_cluster( create_compute_cluster_details=oci.core.models.CreateComputeClusterDetails( availability_domain=self.availability_domain, @@ -447,4 +424,3 @@ def _init_empty_cluster(self) -> str: ) ) return create_compute_cluster_response.data.id - diff --git a/pycloudlib/oci/instance.py b/pycloudlib/oci/instance.py index bf29493e..c401da0b 100644 --- a/pycloudlib/oci/instance.py +++ b/pycloudlib/oci/instance.py @@ -1,11 +1,10 @@ -# pylint: disable=E1101 # This file is part of pycloudlib. See LICENSE file for license information. """OCI instance.""" import json -from time import sleep import time -from typing import Dict, List, Optional +from time import sleep +from typing import Any, Dict, List, Optional import oci @@ -70,7 +69,7 @@ def id(self) -> str: def name(self): """Return the instance name.""" return self.instance_id - + @property def ip(self): """Return IP address of instance.""" @@ -129,7 +128,6 @@ def private_ip(self): for vnic_attachment in vnic_attachment ] self._log.debug("vnics: %s", vnics) - # select vnic with is_primary = True primary_vnic = [vnic for vnic in vnics if vnic.is_primary][0] return primary_vnic.private_ip @@ -146,7 +144,6 @@ def secondary_vnic_private_ip(self) -> Optional[str]: self.network_client.get_vnic(vnic_attachment.vnic_id).data for vnic_attachment in vnic_attachments ] - # get vnic that is not primary secondary_vnic_attachment = [vnic for vnic in vnics if not vnic.is_primary][0] return secondary_vnic_attachment.private_ip @@ -263,6 +260,7 @@ def add_network_interface( nic_index: int = 0, use_private_subnet: bool = False, subnet_name: Optional[str] = None, + **kwargs: Any, ) -> str: """Add network interface to running instance. @@ -280,11 +278,16 @@ def add_network_interface( """ if subnet_name: subnet_id = get_subnet_id_by_name( - self.network_client, self.compartment_id, subnet_name, + self.network_client, + self.compartment_id, + subnet_name, ) else: subnet_id = get_subnet_id( - self.network_client, self.compartment_id, self.availability_domain, private=use_private_subnet, + self.network_client, + self.compartment_id, + self.availability_domain, + private=use_private_subnet, ) create_vnic_details = oci.core.models.CreateVnicDetails( # noqa: E501 subnet_id=subnet_id, @@ -337,10 +340,26 @@ def remove_network_interface(self, ip_address: str): raise PycloudlibError(f"Network interface with ip_address={ip_address} did not detach") def configure_secondary_vnic(self) -> str: + """ + Configure the secondary VNIC using information from the Oracle Cloud metadata service. + + This method fetches the secondary VNIC data from the Oracle Cloud metadata service, + extracts the MAC address and private IP, and assigns the IP address to the corresponding + network interface on the instance. + + Returns: + str: The private IP address assigned to the secondary VNIC. + + Raises: + ValueError: If no secondary VNIC is attached, if no interface is found for the MAC address, + or if the IP address was not successfully assigned to the interface. + PycloudlibError: If failed to fetch secondary VNIC data from the Oracle Cloud metadata service. + """ if not self.secondary_vnic_private_ip: raise ValueError("Cannot configure secondary VNIC without a secondary VNIC attached") secondary_vnic_imds_data: Optional[Dict[str, str]] = None - # it can take a bit for the + # it can take a bit for the secondary VNIC to show up in the IMDS + # so we need to retry fetching the data for roughly a minute for _ in range(60): # Fetch JSON data from the Oracle Cloud metadata service imds_req = self.execute("curl -s http://169.254.169.254/opc/v1/vnics").stdout @@ -356,8 +375,8 @@ def configure_secondary_vnic(self) -> str: raise PycloudlibError( "Failed to fetch secondary VNIC data from IMDS. Cannot configure secondary VNIC" ) - - # Extract MAC address and private IP from the second VNIC + + # Extract MAC address and private IP from the second VNIC mac_addr = secondary_vnic_imds_data["macAddr"] private_ip = secondary_vnic_imds_data["privateIp"] subnet_mask = secondary_vnic_imds_data["subnetCidrBlock"].split("/")[1] @@ -369,12 +388,12 @@ def configure_secondary_vnic(self) -> str: if not interface: raise ValueError(f"No interface found for MAC address {mac_addr}") # Add the IP address to the interface - self.execute( - f"sudo ip addr add {private_ip}/{subnet_mask} dev {interface}" - ) + self.execute(f"sudo ip addr add {private_ip}/{subnet_mask} dev {interface}") # Verify that the IP address was added r = self.execute(f"ip addr show dev {interface}") if private_ip not in r.stdout: - raise ValueError(f"IP {private_ip} was not successfully assigned to interface {interface}") + raise ValueError( + f"IP {private_ip} was not successfully assigned to interface {interface}" + ) self._log.info("Successfully assigned IP %s to interface %s", private_ip, interface) return private_ip diff --git a/pycloudlib/oci/utils.py b/pycloudlib/oci/utils.py index d18d8f96..55340d89 100644 --- a/pycloudlib/oci/utils.py +++ b/pycloudlib/oci/utils.py @@ -53,6 +53,7 @@ def wait_till_ready( ) ) + def get_subnet_id_by_name( network_client: "oci.core.VirtualNetworkClient", compartment_id: str, diff --git a/tests/integration_tests/oracle/test_cluster.py b/tests/integration_tests/oracle/test_cluster.py index 7244d910..f2134192 100644 --- a/tests/integration_tests/oracle/test_cluster.py +++ b/tests/integration_tests/oracle/test_cluster.py @@ -4,10 +4,14 @@ logger = logging.getLogger(__name__) + def test_cluster_launch(): - with OCI( - "pycl-oracle-test-cluster-integration-test" - ) as cloud: + """ + Test the launch of a cluster. + + Verifies that the instances are launched and configured correctly. + """ + with OCI("pycl-oracle-test-cluster-integration-test") as cloud: instances = cloud.create_compute_cluster( image_id=cloud.released_image("noble"), instance_count=2, diff --git a/tests/unit_tests/oci/test_instance.py b/tests/unit_tests/oci/test_instance.py index dd83e078..e9046903 100644 --- a/tests/unit_tests/oci/test_instance.py +++ b/tests/unit_tests/oci/test_instance.py @@ -490,25 +490,29 @@ def test_oci_instance_ip_parametrized( class TestConfigureSecondaryVnic: - def test_configure_secondary_vnic_success(self, oci_instance:OciInstance): + def test_configure_secondary_vnic_success(self, oci_instance: OciInstance): with mock.patch.object( OciInstance, "secondary_vnic_private_ip", new_callable=mock.PropertyMock ) as mock_secondary_ip: mock_secondary_ip.return_value = "10.0.0.5" - # Provide a side_effect matching the four execute() calls - oci_instance.execute = mock.Mock(side_effect=[ - mock.Mock(stdout='[{"macAddr": "00:16:3e:aa:bb:cc","privateIp":"10.0.0.4",' - '"subnetCidrBlock":"10.0.0.0/24"},' - '{"macAddr":"00:16:3e:ee:ff:gg","privateIp":"10.0.0.5",' - '"subnetCidrBlock":"10.0.1.0/24"}]'), - mock.Mock(stdout="eth1"), - mock.Mock(stdout=""), - mock.Mock(stdout="10.0.0.5"), - ]) + # Provide a side_effect matching the four execute() calls + oci_instance.execute = mock.Mock( + side_effect=[ + mock.Mock( + stdout='[{"macAddr": "00:16:3e:aa:bb:cc","privateIp":"10.0.0.4",' + '"subnetCidrBlock":"10.0.0.0/24"},' + '{"macAddr":"00:16:3e:ee:ff:gg","privateIp":"10.0.0.5",' + '"subnetCidrBlock":"10.0.1.0/24"}]' + ), + mock.Mock(stdout="eth1"), + mock.Mock(stdout=""), + mock.Mock(stdout="10.0.0.5"), + ] + ) ip = oci_instance.configure_secondary_vnic() assert ip == "10.0.0.5" - def test_configure_secondary_vnic_no_secondary(self, oci_instance:OciInstance): + def test_configure_secondary_vnic_no_secondary(self, oci_instance: OciInstance): with mock.patch.object( OciInstance, "secondary_vnic_private_ip", new_callable=mock.PropertyMock ) as mock_secondary_ip: @@ -516,9 +520,9 @@ def test_configure_secondary_vnic_no_secondary(self, oci_instance:OciInstance): with pytest.raises(ValueError, match="Cannot configure secondary VNIC"): oci_instance.configure_secondary_vnic() - # patch out time.sleep - @mock.patch("time.sleep", mock.MagicMock()) - def test_configure_secondary_vnic_unavailable_imds(self, oci_instance:OciInstance): + # patch out time.sleep + @mock.patch("time.sleep", mock.MagicMock()) + def test_configure_secondary_vnic_unavailable_imds(self, oci_instance: OciInstance): with mock.patch.object( OciInstance, "secondary_vnic_private_ip", new_callable=mock.PropertyMock ) as mock_secondary_ip: @@ -528,36 +532,44 @@ def test_configure_secondary_vnic_unavailable_imds(self, oci_instance:OciInstanc with pytest.raises(PycloudlibError, match="Failed to fetch secondary VNIC data"): oci_instance.configure_secondary_vnic() - def test_configure_secondary_vnic_no_interface_found(self, oci_instance:OciInstance): + def test_configure_secondary_vnic_no_interface_found(self, oci_instance: OciInstance): with mock.patch.object( OciInstance, "secondary_vnic_private_ip", new_callable=mock.PropertyMock ) as mock_secondary_ip: mock_secondary_ip.return_value = "10.0.0.5" # IMDS returns one entry - oci_instance.execute = mock.Mock(side_effect=[ - mock.Mock(stdout='[{"macAddr": "00:16:3e:aa:bb:cc","privateIp":"10.0.0.4",' - '"subnetCidrBlock":"10.0.0.0/24"},' - '{"macAddr":"00:16:3e:ee:ff:gg","privateIp":"10.0.0.5",' - '"subnetCidrBlock":"10.0.1.0/24"}]'), - mock.Mock(stdout=""), # No interface found - ]) + oci_instance.execute = mock.Mock( + side_effect=[ + mock.Mock( + stdout='[{"macAddr": "00:16:3e:aa:bb:cc","privateIp":"10.0.0.4",' + '"subnetCidrBlock":"10.0.0.0/24"},' + '{"macAddr":"00:16:3e:ee:ff:gg","privateIp":"10.0.0.5",' + '"subnetCidrBlock":"10.0.1.0/24"}]' + ), + mock.Mock(stdout=""), # No interface found + ] + ) with pytest.raises(ValueError, match="No interface found for MAC address"): oci_instance.configure_secondary_vnic() - def test_configure_secondary_vnic_ip_not_assigned(self, oci_instance:OciInstance): + def test_configure_secondary_vnic_ip_not_assigned(self, oci_instance: OciInstance): with mock.patch.object( OciInstance, "secondary_vnic_private_ip", new_callable=mock.PropertyMock ) as mock_secondary_ip: mock_secondary_ip.return_value = "10.0.0.5" # Returns the single IMDS entry, then interface, then IP add, then empty IP check - oci_instance.execute = mock.Mock(side_effect=[ - mock.Mock(stdout='[{"macAddr": "00:16:3e:aa:bb:cc","privateIp":"10.0.0.4",' - '"subnetCidrBlock":"10.0.0.0/24"},' - '{"macAddr":"00:16:3e:ee:ff:gg","privateIp":"10.0.0.5",' - '"subnetCidrBlock":"10.0.1.0/24"}]'), - mock.Mock(stdout="eth1"), - mock.Mock(stdout=""), - mock.Mock(stdout=""), # Nothing found - ]) + oci_instance.execute = mock.Mock( + side_effect=[ + mock.Mock( + stdout='[{"macAddr": "00:16:3e:aa:bb:cc","privateIp":"10.0.0.4",' + '"subnetCidrBlock":"10.0.0.0/24"},' + '{"macAddr":"00:16:3e:ee:ff:gg","privateIp":"10.0.0.5",' + '"subnetCidrBlock":"10.0.1.0/24"}]' + ), + mock.Mock(stdout="eth1"), + mock.Mock(stdout=""), + mock.Mock(stdout=""), # Nothing found + ] + ) with pytest.raises(ValueError, match="was not successfully assigned"): oci_instance.configure_secondary_vnic()