diff --git a/VERSION b/VERSION index fdc61d5c..961f2f93 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1!10.7.2 +1!10.7.3 diff --git a/pycloudlib/oci/utils.py b/pycloudlib/oci/utils.py index b1890fdc..4c8e5944 100644 --- a/pycloudlib/oci/utils.py +++ b/pycloudlib/oci/utils.py @@ -88,9 +88,12 @@ def get_subnet_id( raise PycloudlibError(f"Unable to determine vcn name: {vcn_name}") if len(vcns) > 1: raise PycloudlibError(f"Found multiple vcns with name: {vcn_name}") - vcn_id = vcns[0].id else: # if no vcn_name specified, use most recently created vcn - vcn_id = network_client.list_vcns(compartment_id, retry_strategy=retry_strategy).data[0].id + vcns = network_client.list_vcns(compartment_id, retry_strategy=retry_strategy).data + if len(vcns) == 0: + raise PycloudlibError("No VCNs found in compartment") + vcn_id = vcns[0].id + chosen_vcn_name = vcns[0].display_name subnets = network_client.list_subnets( compartment_id, vcn_id=vcn_id, retry_strategy=retry_strategy @@ -98,14 +101,22 @@ def get_subnet_id( subnet_id = None for subnet in subnets: if subnet.prohibit_internet_ingress: # skip subnet if it's private - log.debug("Ignoring private subnet: %s", subnet.id) + log.debug( + "Ignoring private subnet: %s [id: %s]", + subnet.display_name, + subnet.id, + ) + continue + if subnet.availability_domain and subnet.availability_domain != availability_domain: + log.debug( + "Ignoring public subnet in different availability domain: %s [id: %s]", + subnet.display_name, + subnet.id, + ) continue - log.debug("Using public subnet: %s", subnet.id) - if subnet.availability_domain == availability_domain: - subnet_id = subnet.id - break - else: - subnet_id = subnets[0].id + log.info("Using public subnet: %s [id: %s]", subnet.display_name, subnet.id) + subnet_id = subnet.id + break if not subnet_id: - raise PycloudlibError(f"Unable to determine subnet id for domain: {availability_domain}") + raise PycloudlibError(f"Unable to find suitable subnet in VCN {chosen_vcn_name}") return subnet_id diff --git a/tests/unit_tests/oci/test_instance.py b/tests/unit_tests/oci/test_instance.py index 7cc2c533..f1503262 100644 --- a/tests/unit_tests/oci/test_instance.py +++ b/tests/unit_tests/oci/test_instance.py @@ -172,6 +172,8 @@ def setup_vnic_mocks(self, oci_instance): # Mock list_subnets to return an iterable of subnets subnet_mock = mock.Mock() + subnet_mock.prohibit_internet_ingress = False + subnet_mock.availability_domain = "Uocm:PHX-AD-1" subnet_mock.id = "subnet-id" oci_instance.network_client.list_subnets.return_value = mock.Mock(data=[subnet_mock]) diff --git a/tests/unit_tests/oci/test_utils.py b/tests/unit_tests/oci/test_utils.py new file mode 100644 index 00000000..ded252fb --- /dev/null +++ b/tests/unit_tests/oci/test_utils.py @@ -0,0 +1,139 @@ +import pytest +from unittest.mock import MagicMock, patch +from pycloudlib.oci.utils import get_subnet_id +from pycloudlib.errors import PycloudlibError +from oci.retry import DEFAULT_RETRY_STRATEGY # pylint: disable=E0611,E0401 + + +@pytest.fixture +@patch("pycloudlib.oci.utils.DEFAULT_RETRY_STRATEGY", None) +def setup_environment(): + """ + Set up the test environment. + + This fixture is called before each test function execution. + + It mocks the network client and sets up the compartment ID and availability domain. + """ + network_client = MagicMock() + compartment_id = "compartment_id" + availability_domain = "availability_domain" + return network_client, compartment_id, availability_domain + + +def test_get_subnet_id_fails_with_vcn_name_not_found(setup_environment): + """Test get_subnet_id fails when VCN name is not found.""" + network_client, compartment_id, availability_domain = setup_environment + network_client.list_vcns.return_value.data = [] + with pytest.raises(PycloudlibError, match="Unable to determine vcn name"): + get_subnet_id( + network_client, + compartment_id, + availability_domain, + vcn_name="vcn_name", + ) + + +def test_get_subnet_id_fails_with_multiple_vcns_found(setup_environment): + """Test get_subnet_id fails when multiple VCNs with the same name are found.""" + network_client, compartment_id, availability_domain = setup_environment + network_client.list_vcns.return_value.data = [MagicMock(), MagicMock()] + with pytest.raises(PycloudlibError, match="Found multiple vcns with name"): + get_subnet_id( + network_client, + compartment_id, + availability_domain, + vcn_name="vcn_name", + ) + + +def test_get_subnet_id_fails_with_no_vcns_found(setup_environment): + """Test get_subnet_id fails when no VCNs are found.""" + network_client, compartment_id, availability_domain = setup_environment + network_client.list_vcns.return_value.data = [] + with pytest.raises(PycloudlibError, match="No VCNs found in compartment"): + get_subnet_id(network_client, compartment_id, availability_domain) + + +def test_get_subnet_id_fails_with_no_suitable_subnet_found(setup_environment): + """Test get_subnet_id fails when no suitable subnet is found.""" + network_client, compartment_id, availability_domain = setup_environment + vcn = MagicMock() + vcn.id = "vcn_id" + vcn.display_name = "vcn_name" + network_client.list_vcns.return_value.data = [vcn] + network_client.list_subnets.return_value.data = [] + with pytest.raises(PycloudlibError, match="Unable to find suitable subnet in VCN"): + get_subnet_id(network_client, compartment_id, availability_domain) + + +def test_get_subnet_id_fails_with_private_subnet(setup_environment): + """Test get_subnet_id fails when the subnet is private.""" + network_client, compartment_id, availability_domain = setup_environment + vcn = MagicMock() + vcn.id = "vcn_id" + vcn.display_name = "vcn_name" + network_client.list_vcns.return_value.data = [vcn] + subnet = MagicMock() + subnet.prohibit_internet_ingress = True + network_client.list_subnets.return_value.data = [subnet] + with pytest.raises(PycloudlibError, match="Unable to find suitable subnet in VCN"): + get_subnet_id(network_client, compartment_id, availability_domain) + + +def test_get_subnet_id_fails_with_different_availability_domain(setup_environment): + """Test get_subnet_id fails when the subnet is in a different availability domain.""" + network_client, compartment_id, availability_domain = setup_environment + vcn = MagicMock() + vcn.id = "vcn_id" + vcn.display_name = "vcn_name" + network_client.list_vcns.return_value.data = [vcn] + subnet = MagicMock() + subnet.prohibit_internet_ingress = False + subnet.availability_domain = "different_availability_domain" + network_client.list_subnets.return_value.data = [subnet] + with pytest.raises(PycloudlibError, match="Unable to find suitable subnet in VCN"): + get_subnet_id(network_client, compartment_id, availability_domain) + + +def test_get_subnet_id_suceeds_without_vcn_name(setup_environment): + """Test get_subnet_id suceeds without specifying a VCN name.""" + network_client, compartment_id, availability_domain = setup_environment + vcn1 = MagicMock() + vcn1.id = "vcn1_id" + vcn1.display_name = "vcn1_name" + vcn2 = MagicMock() + vcn2.id = "vcn2_id" + vcn2.display_name = "vcn2_name" + network_client.list_vcns.return_value.data = [vcn1, vcn2] + subnet = MagicMock() + subnet.prohibit_internet_ingress = False + subnet.availability_domain = None + subnet.id = "subnet_id" + network_client.list_subnets.return_value.data = [subnet] + result = get_subnet_id(network_client, compartment_id, availability_domain) + assert result == "subnet_id" + # Ensure that list_subnets is called with the first VCN's ID, not the second + network_client.list_subnets.assert_called_with( + compartment_id, vcn_id="vcn1_id", retry_strategy=DEFAULT_RETRY_STRATEGY + ) + + +def test_get_subnet_id_suceeds_with_vcn_name(setup_environment): + """Test get_subnet_id suceeds with specifying a VCN name.""" + network_client, compartment_id, availability_domain = setup_environment + vcn = MagicMock() + vcn.id = "vcn_id" + vcn.display_name = "vcn_name" + network_client.list_vcns.return_value.data = [vcn] + subnet = MagicMock() + subnet.prohibit_internet_ingress = False + subnet.availability_domain = None + subnet.id = "subnet_id" + network_client.list_subnets.return_value.data = [subnet] + result = get_subnet_id(network_client, compartment_id, availability_domain, vcn_name="vcn_name") + assert result == "subnet_id" + # Ensure that list_subnets is called with the specified VCN's ID + network_client.list_subnets.assert_called_with( + compartment_id, vcn_id="vcn_id", retry_strategy=DEFAULT_RETRY_STRATEGY + )