diff --git a/cloudinit/sources/DataSourceOracle.py b/cloudinit/sources/DataSourceOracle.py index f113d36441b..2354648b458 100644 --- a/cloudinit/sources/DataSourceOracle.py +++ b/cloudinit/sources/DataSourceOracle.py @@ -14,7 +14,8 @@ """ import base64 -import json +from collections import namedtuple +from contextlib import suppress as noop from cloudinit import log as logging from cloudinit import net, sources, util @@ -24,7 +25,7 @@ get_interfaces_by_mac, is_netfail_master, ) -from cloudinit.url_helper import readurl +from cloudinit.url_helper import UrlError, readurl LOG = logging.getLogger(__name__) @@ -33,80 +34,13 @@ 'configure_secondary_nics': False, } CHASSIS_ASSET_TAG = "OracleCloud.com" -METADATA_ROOT = "http://169.254.169.254/opc/v1/" -METADATA_ENDPOINT = METADATA_ROOT + "instance/" -VNIC_METADATA_URL = METADATA_ROOT + "vnics/" +METADATA_ROOT = "http://169.254.169.254/opc/v{version}/" +METADATA_PATTERN = METADATA_ROOT + "{path}/" # https://docs.cloud.oracle.com/iaas/Content/Network/Troubleshoot/connectionhang.htm#Overview, # indicates that an MTU of 9000 is used within OCI MTU = 9000 - -def _add_network_config_from_opc_imds(network_config): - """ - Fetch data from Oracle's IMDS, generate secondary NIC config, merge it. - - The primary NIC configuration should not be modified based on the IMDS - values, as it should continue to be configured for DHCP. As such, this - takes an existing network_config dict which is expected to have the primary - NIC configuration already present. It will mutate the given dict to - include the secondary VNICs. - - :param network_config: - A v1 or v2 network config dict with the primary NIC already configured. - This dict will be mutated. - - :raises: - Exceptions are not handled within this function. Likely exceptions are - those raised by url_helper.readurl (if communicating with the IMDS - fails), ValueError/JSONDecodeError (if the IMDS returns invalid JSON), - and KeyError/IndexError (if the IMDS returns valid JSON with unexpected - contents). - """ - resp = readurl(VNIC_METADATA_URL) - vnics = json.loads(str(resp)) - - if 'nicIndex' in vnics[0]: - # TODO: Once configure_secondary_nics defaults to True, lower the level - # of this log message. (Currently, if we're running this code at all, - # someone has explicitly opted-in to secondary VNIC configuration, so - # we should warn them that it didn't happen. Once it's default, this - # would be emitted on every Bare Metal Machine launch, which means INFO - # or DEBUG would be more appropriate.) - LOG.warning( - 'VNIC metadata indicates this is a bare metal machine; skipping' - ' secondary VNIC configuration.' - ) - return - - interfaces_by_mac = get_interfaces_by_mac() - - for vnic_dict in vnics[1:]: - # We skip the first entry in the response because the primary interface - # is already configured by iSCSI boot; applying configuration from the - # IMDS is not required. - mac_address = vnic_dict['macAddr'].lower() - if mac_address not in interfaces_by_mac: - LOG.debug('Interface with MAC %s not found; skipping', mac_address) - continue - name = interfaces_by_mac[mac_address] - - if network_config['version'] == 1: - subnet = { - 'type': 'static', - 'address': vnic_dict['privateIp'], - } - network_config['config'].append({ - 'name': name, - 'type': 'physical', - 'mac_address': mac_address, - 'mtu': MTU, - 'subnets': [subnet], - }) - elif network_config['version'] == 2: - network_config['ethernets'][name] = { - 'addresses': [vnic_dict['privateIp']], - 'mtu': MTU, 'dhcp4': False, 'dhcp6': False, - 'match': {'macaddress': mac_address}} +OpcMetadata = namedtuple("OpcMetadata", "version instance_data vnics_data") def _ensure_netfailover_safe(network_config): @@ -175,6 +109,7 @@ class DataSourceOracle(sources.DataSource): def __init__(self, sys_cfg, *args, **kwargs): super(DataSourceOracle, self).__init__(sys_cfg, *args, **kwargs) + self._vnics_data = None self.ds_cfg = util.mergemanydict([ util.get_cfg_by_path(sys_cfg, ['datasource', self.dsname], {}), @@ -192,13 +127,23 @@ def _get_data(self): # network may be configured if iscsi root. If that is the case # then read_initramfs_config will return non-None. - if _is_iscsi_root(): - data = read_opc_metadata() - else: - with dhcp.EphemeralDHCPv4(net.find_fallback_nic()): - data = read_opc_metadata() + fetch_vnics_data = self.ds_cfg.get( + 'configure_secondary_nics', + BUILTIN_DS_CONFIG["configure_secondary_nics"] + ) + network_context = noop() + if not _is_iscsi_root(): + network_context = dhcp.EphemeralDHCPv4(net.find_fallback_nic()) + with network_context: + fetched_metadata = read_opc_metadata( + fetch_vnics_data=fetch_vnics_data + ) - self._crawled_metadata = data + data = self._crawled_metadata = fetched_metadata.instance_data + self.metadata_address = METADATA_ROOT.format( + version=fetched_metadata.version + ) + self._vnics_data = fetched_metadata.vnics_data self.metadata = { "availability-zone": data["ociAdName"], @@ -218,10 +163,6 @@ def _get_data(self): return True - def _get_subplatform(self): - """Return the subplatform metadata source details.""" - return "metadata ({})".format(METADATA_ROOT) - def check_instance_id(self, sys_cfg): """quickly check (local only) if self.instance_id is still valid @@ -246,14 +187,18 @@ def network_config(self): # this is now v2 self._network_config = self.distro.generate_fallback_config() - if self.ds_cfg.get('configure_secondary_nics'): + if self.ds_cfg.get( + 'configure_secondary_nics', + BUILTIN_DS_CONFIG["configure_secondary_nics"] + ): try: - # Mutate self._network_config to include secondary VNICs - _add_network_config_from_opc_imds(self._network_config) + # Mutate self._network_config to include secondary + # VNICs + self._add_network_config_from_opc_imds() except Exception: util.logexc( LOG, - "Failed to fetch secondary network configuration!") + "Failed to parse secondary network configuration!") # we need to verify that the nic selected is not a netfail over # device and, if it is a netfail master, then we need to avoid @@ -262,6 +207,70 @@ def network_config(self): return self._network_config + def _add_network_config_from_opc_imds(self): + """Generate secondary NIC config from IMDS and merge it. + + The primary NIC configuration should not be modified based on the IMDS + values, as it should continue to be configured for DHCP. As such, this + uses the instance's network config dict which is expected to have the + primary NIC configuration already present. + It will mutate the network config to include the secondary VNICs. + + :raises: + Exceptions are not handled within this function. Likely + exceptions are KeyError/IndexError + (if the IMDS returns valid JSON with unexpected contents). + """ + if self._vnics_data is None: + LOG.warning( + "Secondary NIC data is UNSET but should not be") + return + + if 'nicIndex' in self._vnics_data[0]: + # TODO: Once configure_secondary_nics defaults to True, lower the + # level of this log message. (Currently, if we're running this + # code at all, someone has explicitly opted-in to secondary + # VNIC configuration, so we should warn them that it didn't + # happen. Once it's default, this would be emitted on every Bare + # Metal Machine launch, which means INFO or DEBUG would be more + # appropriate.) + LOG.warning( + 'VNIC metadata indicates this is a bare metal machine; ' + 'skipping secondary VNIC configuration.' + ) + return + + interfaces_by_mac = get_interfaces_by_mac() + + for vnic_dict in self._vnics_data[1:]: + # We skip the first entry in the response because the primary + # interface is already configured by iSCSI boot; applying + # configuration from the IMDS is not required. + mac_address = vnic_dict['macAddr'].lower() + if mac_address not in interfaces_by_mac: + LOG.debug('Interface with MAC %s not found; skipping', + mac_address) + continue + name = interfaces_by_mac[mac_address] + + if self._network_config['version'] == 1: + subnet = { + 'type': 'static', + 'address': vnic_dict['privateIp'], + } + self._network_config['config'].append({ + 'name': name, + 'type': 'physical', + 'mac_address': mac_address, + 'mtu': MTU, + 'subnets': [subnet], + }) + elif self._network_config['version'] == 2: + self._network_config['ethernets'][name] = { + 'addresses': [vnic_dict['privateIp']], + 'mtu': MTU, 'dhcp4': False, 'dhcp6': False, + 'match': {'macaddress': mac_address}} + def _read_system_uuid(): sys_uuid = util.read_dmi_data('system-uuid') @@ -277,15 +286,43 @@ def _is_iscsi_root(): return bool(cmdline.read_initramfs_config()) -def read_opc_metadata(): - """ - Fetch metadata from the /opc/ routes. +def read_opc_metadata(*, fetch_vnics_data: bool = False): + """Fetch metadata from the /opc/ routes. :return: - The JSON-decoded value of the /opc/v1/instance/ endpoint on the IMDS. + A namedtuple containing: + The metadata version as an integer + The JSON-decoded value of the instance data endpoint on the IMDS + The JSON-decoded value of the vnics data endpoint if + `fetch_vnics_data` is True, else None + """ - # retries=1 as requested by Oracle to address a potential race condition - return json.loads(readurl(METADATA_ENDPOINT, retries=1)._response.text) + retries = 1 + + def _fetch(metadata_version: int, path: str) -> dict: + headers = { + "Authorization": "Bearer Oracle"} if metadata_version > 1 else None + return readurl( + url=METADATA_PATTERN.format(version=metadata_version, path=path), + headers=headers, + retries=retries, + )._response.json() + + metadata_version = 2 + try: + instance_data = _fetch(metadata_version, path="instance") + except UrlError: + metadata_version = 1 + instance_data = _fetch(metadata_version, path="instance") + + vnics_data = None + if fetch_vnics_data: + try: + vnics_data = _fetch(metadata_version, path="vnics") + except UrlError: + util.logexc(LOG, + "Failed to fetch secondary network configuration!") + return OpcMetadata(metadata_version, instance_data, vnics_data) # Used to match classes to dependencies diff --git a/cloudinit/sources/tests/test_oracle.py b/cloudinit/sources/tests/test_oracle.py index 9ee6e7faca8..2fb9a20a899 100644 --- a/cloudinit/sources/tests/test_oracle.py +++ b/cloudinit/sources/tests/test_oracle.py @@ -10,6 +10,7 @@ from cloudinit.sources import DataSourceOracle as oracle from cloudinit.sources import NetworkConfigSource +from cloudinit.sources.DataSourceOracle import OpcMetadata from cloudinit.tests import helpers as test_helpers from cloudinit.url_helper import UrlError @@ -58,7 +59,7 @@ # Fetched with `curl http://169.254.169.254/opc/v1/instance/` (and then # truncated for line length) -OPC_V1_METADATA = """\ +OPC_V2_METADATA = """\ { "availabilityDomain" : "qIZq:PHX-AD-1", "faultDomain" : "FAULT-DOMAIN-2", @@ -83,9 +84,17 @@ } }""" +# Just a small meaningless change to differentiate the two metadatas +OPC_V1_METADATA = OPC_V2_METADATA.replace("ocid1.instance", "ocid2.instance") + + +@pytest.fixture +def metadata_version(): + return 2 + @pytest.yield_fixture -def oracle_ds(request, fixture_utils, paths): +def oracle_ds(request, fixture_utils, paths, metadata_version): """ Return an instantiated DataSourceOracle. @@ -101,12 +110,13 @@ def oracle_ds(request, fixture_utils, paths): sys_cfg = fixture_utils.closest_marker_first_arg_or( request, "ds_sys_cfg", mock.MagicMock() ) + metadata = OpcMetadata(metadata_version, json.loads(OPC_V2_METADATA), None) with mock.patch(DS_PATH + "._read_system_uuid", return_value="someuuid"): with mock.patch(DS_PATH + "._is_platform_viable", return_value=True): with mock.patch(DS_PATH + "._is_iscsi_root", return_value=True): with mock.patch( DS_PATH + ".read_opc_metadata", - return_value=json.loads(OPC_V1_METADATA), + return_value=metadata, ): yield oracle.DataSourceOracle( sys_cfg=sys_cfg, distro=mock.Mock(), paths=paths, @@ -117,10 +127,20 @@ class TestDataSourceOracle: def test_platform_info(self, oracle_ds): assert "oracle" == oracle_ds.cloud_name assert "oracle" == oracle_ds.platform_type - assert ( - "metadata (http://169.254.169.254/opc/v1/)" - == oracle_ds.subplatform - ) + + def test_subplatform_before_fetch(self, oracle_ds): + assert 'unknown' == oracle_ds.subplatform + + def test_platform_info_after_fetch(self, oracle_ds): + oracle_ds._get_data() + assert 'metadata (http://169.254.169.254/opc/v2/)' == \ + oracle_ds.subplatform + + @pytest.mark.parametrize('metadata_version', [1]) + def test_v1_platform_info_after_fetch(self, oracle_ds): + oracle_ds._get_data() + assert 'metadata (http://169.254.169.254/opc/v1/)' == \ + oracle_ds.subplatform def test_secondary_nics_disabled_by_default(self, oracle_ds): assert not oracle_ds.ds_cfg["configure_secondary_nics"] @@ -153,119 +173,95 @@ def test_expected_not_viable_other(self, m_read_dmi_data): m_read_dmi_data.assert_has_calls([mock.call('chassis-asset-tag')]) -class TestNetworkConfigFromOpcImds(test_helpers.CiTestCase): - - with_logs = True - - def setUp(self): - super(TestNetworkConfigFromOpcImds, self).setUp() - self.add_patch(DS_PATH + '.readurl', 'm_readurl') - self.add_patch(DS_PATH + '.get_interfaces_by_mac', - 'm_get_interfaces_by_mac') - - def test_failure_to_readurl(self): - # readurl failures should just bubble out to the caller - self.m_readurl.side_effect = Exception('oh no') - with self.assertRaises(Exception) as excinfo: - oracle._add_network_config_from_opc_imds({}) - self.assertEqual(str(excinfo.exception), 'oh no') - - def test_empty_response(self): - # empty response error should just bubble out to the caller - self.m_readurl.return_value = '' - with self.assertRaises(Exception): - oracle._add_network_config_from_opc_imds([]) - - def test_invalid_json(self): - # invalid JSON error should just bubble out to the caller - self.m_readurl.return_value = '{' - with self.assertRaises(Exception): - oracle._add_network_config_from_opc_imds([]) - - def test_no_secondary_nics_does_not_mutate_input(self): - self.m_readurl.return_value = json.dumps([{}]) - # We test this by passing in a non-dict to ensure that no dict +class TestNetworkConfigFromOpcImds: + def test_no_secondary_nics_does_not_mutate_input(self, oracle_ds): + oracle_ds._vnics_data = [{}] + # We test this by using in a non-dict to ensure that no dict # operations are used; failure would be seen as exceptions - oracle._add_network_config_from_opc_imds(object()) + oracle_ds._network_config = object() + oracle_ds._add_network_config_from_opc_imds() - def test_bare_metal_machine_skipped(self): + def test_bare_metal_machine_skipped(self, oracle_ds, caplog): # nicIndex in the first entry indicates a bare metal machine - self.m_readurl.return_value = OPC_BM_SECONDARY_VNIC_RESPONSE - # We test this by passing in a non-dict to ensure that no dict + oracle_ds._vnics_data = json.loads(OPC_BM_SECONDARY_VNIC_RESPONSE) + # We test this by using a non-dict to ensure that no dict # operations are used - self.assertFalse(oracle._add_network_config_from_opc_imds(object())) - self.assertIn('bare metal machine', self.logs.getvalue()) + oracle_ds._network_config = object() + oracle_ds._add_network_config_from_opc_imds() + assert 'bare metal machine' in caplog.text - def test_missing_mac_skipped(self): - self.m_readurl.return_value = OPC_VM_SECONDARY_VNIC_RESPONSE - self.m_get_interfaces_by_mac.return_value = {} + def test_missing_mac_skipped(self, oracle_ds, caplog): + oracle_ds._vnics_data = json.loads(OPC_VM_SECONDARY_VNIC_RESPONSE) - network_config = {'version': 1, 'config': [{'primary': 'nic'}]} - oracle._add_network_config_from_opc_imds(network_config) + oracle_ds._network_config = { + 'version': 1, 'config': [{'primary': 'nic'}] + } + with mock.patch(DS_PATH + ".get_interfaces_by_mac", return_value={}): + oracle_ds._add_network_config_from_opc_imds() - self.assertEqual(1, len(network_config['config'])) - self.assertIn( - 'Interface with MAC 00:00:17:02:2b:b1 not found; skipping', - self.logs.getvalue()) + assert 1 == len(oracle_ds.network_config['config']) + assert 'Interface with MAC 00:00:17:02:2b:b1 not found; skipping' in \ + caplog.text - def test_missing_mac_skipped_v2(self): - self.m_readurl.return_value = OPC_VM_SECONDARY_VNIC_RESPONSE - self.m_get_interfaces_by_mac.return_value = {} + def test_missing_mac_skipped_v2(self, oracle_ds, caplog): + oracle_ds._vnics_data = json.loads(OPC_VM_SECONDARY_VNIC_RESPONSE) - network_config = {'version': 2, 'ethernets': {'primary': {'nic': {}}}} - oracle._add_network_config_from_opc_imds(network_config) + oracle_ds._network_config = { + 'version': 2, 'ethernets': {'primary': {'nic': {}}} + } + with mock.patch(DS_PATH + ".get_interfaces_by_mac", return_value={}): + oracle_ds._add_network_config_from_opc_imds() - self.assertEqual(1, len(network_config['ethernets'])) - self.assertIn( - 'Interface with MAC 00:00:17:02:2b:b1 not found; skipping', - self.logs.getvalue()) + assert 1 == len(oracle_ds.network_config['ethernets']) + assert 'Interface with MAC 00:00:17:02:2b:b1 not found; skipping' in \ + caplog.text - def test_secondary_nic(self): - self.m_readurl.return_value = OPC_VM_SECONDARY_VNIC_RESPONSE - mac_addr, nic_name = '00:00:17:02:2b:b1', 'ens3' - self.m_get_interfaces_by_mac.return_value = { - mac_addr: nic_name, + def test_secondary_nic(self, oracle_ds): + oracle_ds._vnics_data = json.loads(OPC_VM_SECONDARY_VNIC_RESPONSE) + oracle_ds._network_config = { + 'version': 1, 'config': [{'primary': 'nic'}] } - - network_config = {'version': 1, 'config': [{'primary': 'nic'}]} - oracle._add_network_config_from_opc_imds(network_config) + mac_addr, nic_name = '00:00:17:02:2b:b1', 'ens3' + with mock.patch(DS_PATH + ".get_interfaces_by_mac", + return_value={mac_addr: nic_name}): + oracle_ds._add_network_config_from_opc_imds() # The input is mutated - self.assertEqual(2, len(network_config['config'])) + assert 2 == len(oracle_ds.network_config['config']) - secondary_nic_cfg = network_config['config'][1] - self.assertEqual(nic_name, secondary_nic_cfg['name']) - self.assertEqual('physical', secondary_nic_cfg['type']) - self.assertEqual(mac_addr, secondary_nic_cfg['mac_address']) - self.assertEqual(9000, secondary_nic_cfg['mtu']) + secondary_nic_cfg = oracle_ds.network_config['config'][1] + assert nic_name == secondary_nic_cfg['name'] + assert 'physical' == secondary_nic_cfg['type'] + assert mac_addr == secondary_nic_cfg['mac_address'] + assert 9000 == secondary_nic_cfg['mtu'] - self.assertEqual(1, len(secondary_nic_cfg['subnets'])) + assert 1 == len(secondary_nic_cfg['subnets']) subnet_cfg = secondary_nic_cfg['subnets'][0] # These values are hard-coded in OPC_VM_SECONDARY_VNIC_RESPONSE - self.assertEqual('10.0.0.231', subnet_cfg['address']) + assert '10.0.0.231' == subnet_cfg['address'] - def test_secondary_nic_v2(self): - self.m_readurl.return_value = OPC_VM_SECONDARY_VNIC_RESPONSE - mac_addr, nic_name = '00:00:17:02:2b:b1', 'ens3' - self.m_get_interfaces_by_mac.return_value = { - mac_addr: nic_name, + def test_secondary_nic_v2(self, oracle_ds): + oracle_ds._vnics_data = json.loads(OPC_VM_SECONDARY_VNIC_RESPONSE) + oracle_ds._network_config = { + 'version': 2, 'ethernets': {'primary': {'nic': {}}} } - - network_config = {'version': 2, 'ethernets': {'primary': {'nic': {}}}} - oracle._add_network_config_from_opc_imds(network_config) + mac_addr, nic_name = '00:00:17:02:2b:b1', 'ens3' + with mock.patch(DS_PATH + ".get_interfaces_by_mac", + return_value={mac_addr: nic_name}): + oracle_ds._add_network_config_from_opc_imds() # The input is mutated - self.assertEqual(2, len(network_config['ethernets'])) + assert 2 == len(oracle_ds.network_config['ethernets']) - secondary_nic_cfg = network_config['ethernets']['ens3'] - self.assertFalse(secondary_nic_cfg['dhcp4']) - self.assertFalse(secondary_nic_cfg['dhcp6']) - self.assertEqual(mac_addr, secondary_nic_cfg['match']['macaddress']) - self.assertEqual(9000, secondary_nic_cfg['mtu']) + secondary_nic_cfg = oracle_ds.network_config['ethernets']['ens3'] + assert secondary_nic_cfg['dhcp4'] is False + assert secondary_nic_cfg['dhcp6'] is False + assert mac_addr == secondary_nic_cfg['match']['macaddress'] + assert 9000 == secondary_nic_cfg['mtu'] - self.assertEqual(1, len(secondary_nic_cfg['addresses'])) + assert 1 == len(secondary_nic_cfg['addresses']) # These values are hard-coded in OPC_VM_SECONDARY_VNIC_RESPONSE - self.assertEqual('10.0.0.231', secondary_nic_cfg['addresses'][0]) + assert '10.0.0.231' == secondary_nic_cfg['addresses'][0] class TestNetworkConfigFiltersNetFailover(test_helpers.CiTestCase): @@ -412,42 +408,103 @@ def _is_netfail_master(iface): self.assertEqual(expected_cfg, netcfg) +def _mock_v2_urls(httpretty): + def instance_callback(request, uri, response_headers): + print(response_headers) + assert request.headers.get("Authorization") == "Bearer Oracle" + return [200, response_headers, OPC_V2_METADATA] + + def vnics_callback(request, uri, response_headers): + assert request.headers.get("Authorization") == "Bearer Oracle" + return [200, response_headers, OPC_BM_SECONDARY_VNIC_RESPONSE] + + httpretty.register_uri( + httpretty.GET, + "http://169.254.169.254/opc/v2/instance/", + body=instance_callback + ) + httpretty.register_uri( + httpretty.GET, + "http://169.254.169.254/opc/v2/vnics/", + body=vnics_callback + ) + + +def _mock_no_v2_urls(httpretty): + httpretty.register_uri( + httpretty.GET, + "http://169.254.169.254/opc/v2/instance/", + status=404, + ) + httpretty.register_uri( + httpretty.GET, + "http://169.254.169.254/opc/v1/instance/", + body=OPC_V1_METADATA + ) + httpretty.register_uri( + httpretty.GET, + "http://169.254.169.254/opc/v1/vnics/", + body=OPC_BM_SECONDARY_VNIC_RESPONSE + ) + + class TestReadOpcMetadata: # See https://docs.pytest.org/en/stable/example # /parametrize.html#parametrizing-conditional-raising does_not_raise = ExitStack - @pytest.fixture(autouse=True) - def configure_opc_metadata_in_httpretty(self, httpretty): - """Configure HTTPretty with the various OPC metadata endpoints.""" - httpretty.register_uri( - httpretty.GET, - "http://169.254.169.254/opc/v1/instance/", - OPC_V1_METADATA, - ) + @mock.patch("cloudinit.url_helper.time.sleep", lambda _: None) + @pytest.mark.parametrize( + 'version,setup_urls,instance_data,fetch_vnics,vnics_data', [ + (2, _mock_v2_urls, json.loads(OPC_V2_METADATA), True, + json.loads(OPC_BM_SECONDARY_VNIC_RESPONSE)), + (2, _mock_v2_urls, json.loads(OPC_V2_METADATA), False, None), + (1, _mock_no_v2_urls, json.loads(OPC_V1_METADATA), True, + json.loads(OPC_BM_SECONDARY_VNIC_RESPONSE)), + (1, _mock_no_v2_urls, json.loads(OPC_V1_METADATA), False, None), + ] + ) + def test_metadata_returned( + self, version, setup_urls, instance_data, + fetch_vnics, vnics_data, httpretty + ): + setup_urls(httpretty) + metadata = oracle.read_opc_metadata(fetch_vnics_data=fetch_vnics) - def test_json_decoded_value_returned(self): - # read_opc_metadata should JSON decode the response and return it - expected = json.loads(OPC_V1_METADATA) - assert expected == oracle.read_opc_metadata() + assert version == metadata.version + assert instance_data == metadata.instance_data + assert vnics_data == metadata.vnics_data # No need to actually wait between retries in the tests @mock.patch("cloudinit.url_helper.time.sleep", lambda _: None) @pytest.mark.parametrize( - "failure_count,expectation", - [(1, does_not_raise()), (2, pytest.raises(UrlError))], + "v2_failure_count,v1_failure_count,expected_body,expectation", + [ + (1, 0, json.loads(OPC_V2_METADATA), does_not_raise()), + (2, 0, json.loads(OPC_V1_METADATA), does_not_raise()), + (2, 1, json.loads(OPC_V1_METADATA), does_not_raise()), + (2, 2, None, pytest.raises(UrlError)), + ] ) - def test_retries(self, expectation, failure_count, httpretty): - responses = [httpretty.Response("", status=404)] * failure_count - responses.append(httpretty.Response(OPC_V1_METADATA)) + def test_retries(self, v2_failure_count, v1_failure_count, + expected_body, expectation, httpretty): + v2_responses = [httpretty.Response("", status=404)] * v2_failure_count + v2_responses.append(httpretty.Response(OPC_V2_METADATA)) + v1_responses = [httpretty.Response("", status=404)] * v1_failure_count + v1_responses.append(httpretty.Response(OPC_V1_METADATA)) + httpretty.register_uri( httpretty.GET, "http://169.254.169.254/opc/v1/instance/", - responses=responses, + responses=v1_responses, + ) + httpretty.register_uri( + httpretty.GET, + "http://169.254.169.254/opc/v2/instance/", + responses=v2_responses, ) - expected = json.loads(OPC_V1_METADATA) with expectation: - assert expected == oracle.read_opc_metadata() + assert expected_body == oracle.read_opc_metadata().instance_data class TestCommon_GetDataBehaviour: @@ -516,7 +573,7 @@ def test_metadata_keys_set_correctly( @pytest.mark.parametrize( "attribute_name,expected_value", [ - ("_crawled_metadata", json.loads(OPC_V1_METADATA)), + ("_crawled_metadata", json.loads(OPC_V2_METADATA)), ( "userdata_raw", base64.b64decode(b"IyEvYmluL3NoCnRvdWNoIC90bXAvZm9v"), @@ -554,11 +611,12 @@ def test_attributes_set_correctly( def test_public_keys_handled_correctly( self, ssh_keys, expected_value, parameterized_oracle_ds ): - metadata = json.loads(OPC_V1_METADATA) + instance_data = json.loads(OPC_V1_METADATA) if ssh_keys is None: - del metadata["metadata"]["ssh_authorized_keys"] + del instance_data["metadata"]["ssh_authorized_keys"] else: - metadata["metadata"]["ssh_authorized_keys"] = ssh_keys + instance_data["metadata"]["ssh_authorized_keys"] = ssh_keys + metadata = OpcMetadata(None, instance_data, None) with mock.patch( DS_PATH + ".read_opc_metadata", mock.Mock(return_value=metadata), ): @@ -570,8 +628,9 @@ def test_public_keys_handled_correctly( def test_missing_user_data_handled_gracefully( self, parameterized_oracle_ds ): - metadata = json.loads(OPC_V1_METADATA) - del metadata["metadata"]["user_data"] + instance_data = json.loads(OPC_V1_METADATA) + del instance_data["metadata"]["user_data"] + metadata = OpcMetadata(None, instance_data, None) with mock.patch( DS_PATH + ".read_opc_metadata", mock.Mock(return_value=metadata), ): @@ -582,8 +641,9 @@ def test_missing_user_data_handled_gracefully( def test_missing_metadata_handled_gracefully( self, parameterized_oracle_ds ): - metadata = json.loads(OPC_V1_METADATA) - del metadata["metadata"] + instance_data = json.loads(OPC_V1_METADATA) + del instance_data["metadata"] + metadata = OpcMetadata(None, instance_data, None) with mock.patch( DS_PATH + ".read_opc_metadata", mock.Mock(return_value=metadata), ): @@ -617,7 +677,7 @@ def exit_context_manager(*args): exit_context_manager ) - def assert_in_context_manager(): + def assert_in_context_manager(**kwargs): assert in_context_manager return mock.MagicMock() @@ -666,10 +726,8 @@ def test_network_fallback(self, m_read_initramfs_config, oracle_ds): "configure_secondary_nics,expect_secondary_nics", [(True, True), (False, False), (None, False)], ) - @mock.patch(DS_PATH + "._add_network_config_from_opc_imds") def test_secondary_nic_addition( self, - m_add_network_config_from_opc_imds, m_read_initramfs_config, configure_secondary_nics, expect_secondary_nics, @@ -681,33 +739,38 @@ def test_secondary_nic_addition( """ m_read_initramfs_config.return_value = {"version": 1, "config": []} - def side_effect(network_config): - network_config["secondary_added"] = mock.sentinel.needle - - m_add_network_config_from_opc_imds.side_effect = side_effect - if configure_secondary_nics is not None: oracle_ds.ds_cfg[ "configure_secondary_nics" ] = configure_secondary_nics - was_secondary_added = "secondary_added" in oracle_ds.network_config + def side_effect(self): + self._network_config["secondary_added"] = mock.sentinel.needle + + oracle_ds._vnics_data = 'DummyData' + with mock.patch.object( + oracle.DataSourceOracle, "_add_network_config_from_opc_imds", + new=side_effect, + ): + was_secondary_added = "secondary_added" in oracle_ds.network_config assert expect_secondary_nics == was_secondary_added - @mock.patch(DS_PATH + "._add_network_config_from_opc_imds") def test_secondary_nic_failure_isnt_blocking( self, - m_add_network_config_from_opc_imds, m_read_initramfs_config, caplog, oracle_ds, ): - m_add_network_config_from_opc_imds.side_effect = Exception() - oracle_ds.ds_cfg["configure_secondary_nics"] = True + oracle_ds._vnics_data = "DummyData" - assert m_read_initramfs_config.return_value == oracle_ds.network_config - assert "Failed to fetch secondary network configuration" in caplog.text + with mock.patch.object( + oracle.DataSourceOracle, "_add_network_config_from_opc_imds", + side_effect=Exception() + ): + network_config = oracle_ds.network_config + assert network_config == m_read_initramfs_config.return_value + assert "Failed to parse secondary network configuration" in caplog.text def test_ds_network_cfg_preferred_over_initramfs(self, _m): """Ensure that DS net config is preferred over initramfs config"""