summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--cloudinit/dmi.py10
-rw-r--r--cloudinit/net/__init__.py2
-rw-r--r--cloudinit/sources/DataSourceAzure.py2
-rw-r--r--cloudinit/sources/DataSourceOracle.py172
-rw-r--r--cloudinit/sources/__init__.py8
-rw-r--r--integration-requirements.txt2
-rw-r--r--tests/integration_tests/datasources/test_lxd_discovery.py4
-rw-r--r--tests/integration_tests/datasources/test_network_dependency.py4
-rw-r--r--tests/integration_tests/datasources/test_oci_networking.py118
-rw-r--r--tests/integration_tests/integration_settings.py5
-rw-r--r--tests/unittests/distros/test_networking.py2
-rw-r--r--tests/unittests/helpers.py43
-rw-r--r--tests/unittests/sources/test_oracle.py756
-rw-r--r--tox.ini1
14 files changed, 760 insertions, 369 deletions
diff --git a/cloudinit/dmi.py b/cloudinit/dmi.py
index 23ca047e..dff9ab0f 100644
--- a/cloudinit/dmi.py
+++ b/cloudinit/dmi.py
@@ -1,6 +1,7 @@
# This file is part of cloud-init. See LICENSE file for license information.
import os
from collections import namedtuple
+from typing import Optional
from cloudinit import log as logging
from cloudinit import subp
@@ -58,7 +59,7 @@ DMIDECODE_TO_KERNEL = {
}
-def _read_dmi_syspath(key):
+def _read_dmi_syspath(key: str) -> Optional[str]:
"""
Reads dmi data from /sys/class/dmi/id
"""
@@ -96,7 +97,7 @@ def _read_dmi_syspath(key):
return None
-def _read_kenv(key):
+def _read_kenv(key: str) -> Optional[str]:
"""
Reads dmi data from FreeBSD's kenv(1)
"""
@@ -114,12 +115,11 @@ def _read_kenv(key):
return result
except subp.ProcessExecutionError as e:
LOG.debug("failed kenv cmd: %s\n%s", cmd, e)
- return None
return None
-def _call_dmidecode(key, dmidecode_path):
+def _call_dmidecode(key: str, dmidecode_path: str) -> Optional[str]:
"""
Calls out to dmidecode to get the data out. This is mostly for supporting
OS's without /sys/class/dmi/id support.
@@ -137,7 +137,7 @@ def _call_dmidecode(key, dmidecode_path):
return None
-def read_dmi_data(key):
+def read_dmi_data(key: str) -> Optional[str]:
"""
Wrapper for reading DMI data.
diff --git a/cloudinit/net/__init__.py b/cloudinit/net/__init__.py
index d9fcaf10..7f534e9c 100644
--- a/cloudinit/net/__init__.py
+++ b/cloudinit/net/__init__.py
@@ -250,7 +250,7 @@ def has_netfail_standby_feature(devname):
return features[62] == "1"
-def is_netfail_master(devname, driver=None):
+def is_netfail_master(devname, driver=None) -> bool:
"""A device is a "netfail master" device if:
- The device does NOT have the 'master' sysfs attribute
diff --git a/cloudinit/sources/DataSourceAzure.py b/cloudinit/sources/DataSourceAzure.py
index 4cf857a6..e63e223d 100644
--- a/cloudinit/sources/DataSourceAzure.py
+++ b/cloudinit/sources/DataSourceAzure.py
@@ -100,7 +100,7 @@ class PPSType(Enum):
PLATFORM_ENTROPY_SOURCE: Optional[str] = "/sys/firmware/acpi/tables/OEM0"
# List of static scripts and network config artifacts created by
-# stock ubuntu suported images.
+# stock ubuntu supported images.
UBUNTU_EXTENDED_NETWORK_SCRIPTS = [
"/etc/netplan/90-hotplug-azure.yaml",
"/usr/local/sbin/ephemeral_eth.sh",
diff --git a/cloudinit/sources/DataSourceOracle.py b/cloudinit/sources/DataSourceOracle.py
index 3fd8d753..d4a3f133 100644
--- a/cloudinit/sources/DataSourceOracle.py
+++ b/cloudinit/sources/DataSourceOracle.py
@@ -15,12 +15,12 @@ Notes:
import base64
from collections import namedtuple
-from contextlib import suppress as noop
-from typing import Tuple
+from typing import Optional, Tuple
from cloudinit import dmi
from cloudinit import log as logging
from cloudinit import net, sources, util
+from cloudinit.distros.networking import NetworkConfig
from cloudinit.net import (
cmdline,
dhcp,
@@ -46,7 +46,19 @@ V2_HEADERS = {"Authorization": "Bearer Oracle"}
OpcMetadata = namedtuple("OpcMetadata", "version instance_data vnics_data")
-def _ensure_netfailover_safe(network_config):
+class KlibcOracleNetworkConfigSource(cmdline.KlibcNetworkConfigSource):
+ """Override super class to lower the applicability conditions.
+
+ If any `/run/net-*.cfg` files exist, then it is applicable. Even if
+ `/run/initramfs/open-iscsi.interface` does not exist.
+ """
+
+ def is_applicable(self) -> bool:
+ """Override is_applicable"""
+ return bool(self._files)
+
+
+def _ensure_netfailover_safe(network_config: NetworkConfig) -> None:
"""
Search network config physical interfaces to see if any of them are
a netfailover master. If found, we prevent matching by MAC as the other
@@ -110,7 +122,7 @@ class DataSourceOracle(sources.DataSource):
sources.NetworkConfigSource.SYSTEM_CFG,
)
- _network_config = sources.UNSET
+ _network_config: dict = {"config": [], "version": 1}
def __init__(self, sys_cfg, *args, **kwargs):
super(DataSourceOracle, self).__init__(sys_cfg, *args, **kwargs)
@@ -122,8 +134,12 @@ class DataSourceOracle(sources.DataSource):
BUILTIN_DS_CONFIG,
]
)
+ self._network_config_source = KlibcOracleNetworkConfigSource()
+
+ def _has_network_config(self) -> bool:
+ return bool(self._network_config.get("config", []))
- def _is_platform_viable(self):
+ def _is_platform_viable(self) -> bool:
"""Check platform environment to report if this datasource may run."""
return _is_platform_viable()
@@ -133,24 +149,21 @@ class DataSourceOracle(sources.DataSource):
self.system_uuid = _read_system_uuid()
- # network may be configured if iscsi root. If that is the case
- # then read_initramfs_config will return non-None.
- fetch_vnics_data = self.ds_cfg.get(
+ network_context = dhcp.EphemeralDHCPv4(
+ iface=net.find_fallback_nic(),
+ connectivity_url_data={
+ "url": METADATA_PATTERN.format(version=2, path="instance"),
+ "headers": V2_HEADERS,
+ },
+ )
+ fetch_primary_nic = not self._is_iscsi_root()
+ fetch_secondary_nics = 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(
- iface=net.find_fallback_nic(),
- connectivity_url_data={
- "url": METADATA_PATTERN.format(version=2, path="instance"),
- "headers": V2_HEADERS,
- },
- )
with network_context:
fetched_metadata = read_opc_metadata(
- fetch_vnics_data=fetch_vnics_data
+ fetch_vnics_data=fetch_primary_nic or fetch_secondary_nics
)
data = self._crawled_metadata = fetched_metadata.instance_data
@@ -177,7 +190,7 @@ class DataSourceOracle(sources.DataSource):
return True
- def check_instance_id(self, sys_cfg):
+ def check_instance_id(self, sys_cfg) -> bool:
"""quickly check (local only) if self.instance_id is still valid
On Oracle, the dmi-provided system uuid differs from the instance-id
@@ -187,59 +200,75 @@ class DataSourceOracle(sources.DataSource):
def get_public_ssh_keys(self):
return sources.normalize_pubkey_data(self.metadata.get("public_keys"))
+ def _is_iscsi_root(self) -> bool:
+ """Return whether we are on a iscsi machine."""
+ return self._network_config_source.is_applicable()
+
+ def _get_iscsi_config(self) -> dict:
+ return self._network_config_source.render_config()
+
@property
def network_config(self):
"""Network config is read from initramfs provided files
+ Priority for primary network_config selection:
+ - iscsi
+ - imds
+
If none is present, then we fall back to fallback configuration.
"""
- if self._network_config == sources.UNSET:
- # this is v1
- self._network_config = cmdline.read_initramfs_config()
-
- if not self._network_config:
- # this is now v2
- self._network_config = self.distro.generate_fallback_config()
-
- if self.ds_cfg.get(
- "configure_secondary_nics",
- BUILTIN_DS_CONFIG["configure_secondary_nics"],
- ):
- try:
- # Mutate self._network_config to include secondary
- # VNICs
- self._add_network_config_from_opc_imds()
- except Exception:
- util.logexc(
- LOG, "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
- # emitting any match by mac
- _ensure_netfailover_safe(self._network_config)
+ if self._has_network_config():
+ return self._network_config
+
+ set_primary = False
+ # this is v1
+ if self._is_iscsi_root():
+ self._network_config = self._get_iscsi_config()
+ if not self._has_network_config():
+ LOG.warning(
+ "Could not obtain network configuration from initramfs. "
+ "Falling back to IMDS."
+ )
+ set_primary = True
+
+ set_secondary = self.ds_cfg.get(
+ "configure_secondary_nics",
+ BUILTIN_DS_CONFIG["configure_secondary_nics"],
+ )
+ if set_primary or set_secondary:
+ try:
+ # Mutate self._network_config to include primary and/or
+ # secondary VNICs
+ self._add_network_config_from_opc_imds(set_primary)
+ except Exception:
+ util.logexc(
+ LOG,
+ "Failed to parse IMDS 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
+ # emitting any match by mac
+ _ensure_netfailover_safe(self._network_config)
return self._network_config
- def _add_network_config_from_opc_imds(self):
- """Generate secondary NIC config from IMDS and merge it.
+ def _add_network_config_from_opc_imds(self, set_primary: bool = False):
+ """Generate primary and/or 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.
+ :param set_primary: If True set primary interface.
: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")
+ LOG.warning("NIC data is UNSET but should not be")
return
- if "nicIndex" in self._vnics_data[0]:
+ if not set_primary and ("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
@@ -255,14 +284,14 @@ class DataSourceOracle(sources.DataSource):
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.
+ vnics_data = self._vnics_data if set_primary else self._vnics_data[1:]
+
+ for vnic_dict in vnics_data:
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
+ LOG.warning(
+ "Interface with MAC %s not found; skipping",
+ mac_address,
)
continue
name = interfaces_by_mac[mac_address]
@@ -291,21 +320,25 @@ class DataSourceOracle(sources.DataSource):
}
-def _read_system_uuid():
+def _read_system_uuid() -> Optional[str]:
sys_uuid = dmi.read_dmi_data("system-uuid")
return None if sys_uuid is None else sys_uuid.lower()
-def _is_platform_viable():
+def _is_platform_viable() -> bool:
asset_tag = dmi.read_dmi_data("chassis-asset-tag")
return asset_tag == CHASSIS_ASSET_TAG
-def _is_iscsi_root():
- return bool(cmdline.read_initramfs_config())
+def _fetch(metadata_version: int, path: str, retries: int = 2) -> dict:
+ return readurl(
+ url=METADATA_PATTERN.format(version=metadata_version, path=path),
+ headers=V2_HEADERS if metadata_version > 1 else None,
+ retries=retries,
+ )._response.json()
-def read_opc_metadata(*, fetch_vnics_data: bool = False):
+def read_opc_metadata(*, fetch_vnics_data: bool = False) -> OpcMetadata:
"""Fetch metadata from the /opc/ routes.
:return:
@@ -319,15 +352,6 @@ def read_opc_metadata(*, fetch_vnics_data: bool = False):
# Per Oracle, there are short windows (measured in milliseconds) throughout
# an instance's lifetime where the IMDS is being updated and may 404 as a
# result. To work around these windows, we retry a couple of times.
- retries = 2
-
- def _fetch(metadata_version: int, path: str) -> dict:
- return readurl(
- url=METADATA_PATTERN.format(version=metadata_version, path=path),
- headers=V2_HEADERS if metadata_version > 1 else None,
- retries=retries,
- )._response.json()
-
metadata_version = 2
try:
instance_data = _fetch(metadata_version, path="instance")
@@ -340,9 +364,7 @@ def read_opc_metadata(*, fetch_vnics_data: bool = False):
try:
vnics_data = _fetch(metadata_version, path="vnics")
except UrlError:
- util.logexc(
- LOG, "Failed to fetch secondary network configuration!"
- )
+ util.logexc(LOG, "Failed to fetch IMDS network configuration!")
return OpcMetadata(metadata_version, instance_data, vnics_data)
diff --git a/cloudinit/sources/__init__.py b/cloudinit/sources/__init__.py
index effff379..b621fb6e 100644
--- a/cloudinit/sources/__init__.py
+++ b/cloudinit/sources/__init__.py
@@ -361,7 +361,7 @@ class DataSource(CloudInitPickleMixin, metaclass=abc.ABCMeta):
if not attr_defaults:
self._dirty_cache = False
- def get_data(self):
+ def get_data(self) -> bool:
"""Datasources implement _get_data to setup metadata and userdata_raw.
Minimally, the datasource should return a boolean True on success.
@@ -442,7 +442,7 @@ class DataSource(CloudInitPickleMixin, metaclass=abc.ABCMeta):
write_json(json_file, redact_sensitive_keys(processed_data))
return True
- def _get_data(self):
+ def _get_data(self) -> bool:
"""Walk metadata sources, process crawled data and save attributes."""
raise NotImplementedError(
"Subclasses of DataSource must implement _get_data which"
@@ -986,7 +986,9 @@ def list_sources(cfg_list, depends, pkg_list):
return src_list
-def instance_id_matches_system_uuid(instance_id, field="system-uuid"):
+def instance_id_matches_system_uuid(
+ instance_id, field: str = "system-uuid"
+) -> bool:
# quickly (local check only) if self.instance_id is still valid
# we check kernel command line or files.
if not instance_id:
diff --git a/integration-requirements.txt b/integration-requirements.txt
index ebfdf8dc..7b64554d 100644
--- a/integration-requirements.txt
+++ b/integration-requirements.txt
@@ -1,5 +1,5 @@
# PyPI requirements for cloud-init integration testing
# https://cloudinit.readthedocs.io/en/latest/topics/integration_tests.html
#
-pycloudlib @ git+https://github.com/canonical/pycloudlib.git@2eba2592d598562425016867a119f7675a85f42c
+pycloudlib @ git+https://github.com/canonical/pycloudlib.git@c42341990cb35460946ee04e2623d0f9fffe2b3c
pytest
diff --git a/tests/integration_tests/datasources/test_lxd_discovery.py b/tests/integration_tests/datasources/test_lxd_discovery.py
index f72b1b4b..feae52a9 100644
--- a/tests/integration_tests/datasources/test_lxd_discovery.py
+++ b/tests/integration_tests/datasources/test_lxd_discovery.py
@@ -8,7 +8,7 @@ from tests.integration_tests.instances import IntegrationInstance
from tests.integration_tests.util import verify_clean_log
-def _customize_envionment(client: IntegrationInstance):
+def _customize_environment(client: IntegrationInstance):
# Assert our platform can detect LXD during systemd generator timeframe.
ds_id_log = client.execute("cat /run/cloud-init/ds-identify.log").stdout
assert "check for 'LXD' returned found" in ds_id_log
@@ -54,7 +54,7 @@ def _customize_envionment(client: IntegrationInstance):
def test_lxd_datasource_discovery(client: IntegrationInstance):
"""Test that DataSourceLXD is detected instead of NoCloud."""
- _customize_envionment(client)
+ _customize_environment(client)
result = client.execute("cloud-init status --wait --long")
if not result.ok:
raise AssertionError("cloud-init failed:\n%s", result.stderr)
diff --git a/tests/integration_tests/datasources/test_network_dependency.py b/tests/integration_tests/datasources/test_network_dependency.py
index 32ac7053..bd7fe658 100644
--- a/tests/integration_tests/datasources/test_network_dependency.py
+++ b/tests/integration_tests/datasources/test_network_dependency.py
@@ -3,7 +3,7 @@ import pytest
from tests.integration_tests.instances import IntegrationInstance
-def _customize_envionment(client: IntegrationInstance):
+def _customize_environment(client: IntegrationInstance):
# Insert our "disable_network_activation" file here
client.write_to_file(
"/etc/cloud/cloud.cfg.d/99-disable-network-activation.cfg",
@@ -19,7 +19,7 @@ def _customize_envionment(client: IntegrationInstance):
@pytest.mark.ubuntu # Because netplan
def test_network_activation_disabled(client: IntegrationInstance):
"""Test that the network is not activated during init mode."""
- _customize_envionment(client)
+ _customize_environment(client)
result = client.execute("systemctl status google-guest-agent.service")
if not result.ok:
raise AssertionError(
diff --git a/tests/integration_tests/datasources/test_oci_networking.py b/tests/integration_tests/datasources/test_oci_networking.py
new file mode 100644
index 00000000..f569650e
--- /dev/null
+++ b/tests/integration_tests/datasources/test_oci_networking.py
@@ -0,0 +1,118 @@
+import re
+from typing import Iterator, Set
+
+import pytest
+import yaml
+
+from tests.integration_tests.clouds import IntegrationCloud
+from tests.integration_tests.instances import IntegrationInstance
+from tests.integration_tests.util import verify_clean_log
+
+DS_CFG = """\
+datasource:
+ Oracle:
+ configure_secondary_nics: {configure_secondary_nics}
+"""
+
+
+def customize_environment(
+ client: IntegrationInstance,
+ tmpdir,
+ configure_secondary_nics: bool = False,
+):
+ cfg = tmpdir.join("01_oracle_datasource.cfg")
+ with open(cfg, "w") as f:
+ f.write(
+ DS_CFG.format(configure_secondary_nics=configure_secondary_nics)
+ )
+ client.push_file(cfg, "/etc/cloud/cloud.cfg.d/01_oracle_datasource.cfg")
+
+ client.execute("cloud-init clean --logs")
+ client.restart()
+
+
+def extract_interface_names(network_config: dict) -> Set[str]:
+ if network_config["version"] == 1:
+ interfaces = map(lambda conf: conf["name"], network_config["config"])
+ elif network_config["version"] == 2:
+ interfaces = network_config["ethernets"].keys()
+ else:
+ raise NotImplementedError(
+ f'Implement me for version={network_config["version"]}'
+ )
+ return set(interfaces)
+
+
+@pytest.mark.oci
+def test_oci_networking_iscsi_instance(client: IntegrationInstance, tmpdir):
+ customize_environment(client, tmpdir, configure_secondary_nics=False)
+ result_net_files = client.execute("ls /run/net-*.conf")
+ assert result_net_files.ok, "No net files found under /run"
+
+ log = client.read_from_file("/var/log/cloud-init.log")
+ verify_clean_log(log)
+
+ assert (
+ "opc/v2/vnics/" not in log
+ ), "vnic data was fetched and it should not have been"
+
+ netplan_yaml = client.read_from_file("/etc/netplan/50-cloud-init.yaml")
+ netplan_cfg = yaml.safe_load(netplan_yaml)
+ configured_interfaces = extract_interface_names(netplan_cfg["network"])
+ assert 1 <= len(
+ configured_interfaces
+ ), "Expected at least 1 primary network configuration."
+
+ expected_interfaces = set(
+ re.findall(r"/run/net-(.+)\.conf", result_net_files.stdout)
+ )
+ for expected_interface in expected_interfaces:
+ assert (
+ f"Reading from /run/net-{expected_interface}.conf" in log
+ ), "Expected {expected_interface} not found in: {log}"
+
+ not_found_interfaces = expected_interfaces.difference(
+ configured_interfaces
+ )
+ assert not not_found_interfaces, (
+ f"Interfaces, {not_found_interfaces}, expected to be configured in"
+ f" {netplan_cfg['network']}"
+ )
+ assert client.execute("ping -c 2 canonical.com").ok
+
+
+@pytest.fixture(scope="function")
+def client_with_secondary_vnic(
+ session_cloud: IntegrationCloud,
+) -> Iterator[IntegrationInstance]:
+ """Create an instance client and attach a temporary vnic"""
+ with session_cloud.launch() as client:
+ ip_address = client.instance.add_network_interface()
+ yield client
+ client.instance.remove_network_interface(ip_address)
+
+
+@pytest.mark.oci
+def test_oci_networking_iscsi_instance_secondary_vnics(
+ client_with_secondary_vnic: IntegrationInstance, tmpdir
+):
+ client = client_with_secondary_vnic
+ customize_environment(client, tmpdir, configure_secondary_nics=True)
+
+ log = client.read_from_file("/var/log/cloud-init.log")
+ verify_clean_log(log)
+
+ assert "opc/v2/vnics/" in log, f"vnics data not fetched in {log}"
+ netplan_yaml = client.read_from_file("/etc/netplan/50-cloud-init.yaml")
+ netplan_cfg = yaml.safe_load(netplan_yaml)
+ configured_interfaces = extract_interface_names(netplan_cfg["network"])
+ assert 2 <= len(
+ configured_interfaces
+ ), "Expected at least 1 primary and 1 secondary network configurations"
+
+ result_net_files = client.execute("ls /run/net-*.conf")
+ expected_interfaces = set(
+ re.findall(r"/run/net-(.+)\.conf", result_net_files.stdout)
+ )
+ assert len(expected_interfaces) + 1 == len(configured_interfaces)
+ assert client.execute("ping -c 2 canonical.com").ok
diff --git a/tests/integration_tests/integration_settings.py b/tests/integration_tests/integration_settings.py
index f27e4f12..abc70fe4 100644
--- a/tests/integration_tests/integration_settings.py
+++ b/tests/integration_tests/integration_settings.py
@@ -1,5 +1,6 @@
# This file is part of cloud-init. See LICENSE file for license information.
import os
+from typing import Optional
from cloudinit.util import is_false, is_true
@@ -26,7 +27,7 @@ PLATFORM = "lxd_container"
# The cloud-specific instance type to run. E.g., a1.medium on AWS
# If the pycloudlib instance provides a default, this can be left None
-INSTANCE_TYPE = None
+INSTANCE_TYPE: Optional[str] = None
# Determines the base image to use or generate new images from.
#
@@ -38,7 +39,7 @@ OS_IMAGE = "focal"
# Populate if you want to use a pre-launched instance instead of
# creating a new one. The exact contents will be platform dependent
-EXISTING_INSTANCE_ID = None
+EXISTING_INSTANCE_ID: Optional[str] = None
##################################################################
# IMAGE GENERATION SETTINGS
diff --git a/tests/unittests/distros/test_networking.py b/tests/unittests/distros/test_networking.py
index f56b34ad..6f7465c9 100644
--- a/tests/unittests/distros/test_networking.py
+++ b/tests/unittests/distros/test_networking.py
@@ -2,7 +2,6 @@
# /parametrize.html#parametrizing-conditional-raising
import textwrap
-from contextlib import ExitStack as does_not_raise
from unittest import mock
import pytest
@@ -14,6 +13,7 @@ from cloudinit.distros.networking import (
LinuxNetworking,
Networking,
)
+from tests.unittests.helpers import does_not_raise
@pytest.fixture
diff --git a/tests/unittests/helpers.py b/tests/unittests/helpers.py
index 064d89db..9d5a7ed2 100644
--- a/tests/unittests/helpers.py
+++ b/tests/unittests/helpers.py
@@ -18,6 +18,7 @@ from unittest import mock
from unittest.util import strclass
import httpretty
+import pytest
import cloudinit
from cloudinit import cloud, distros
@@ -72,6 +73,13 @@ def retarget_many_wrapper(new_base, am, old_func):
return wrapper
+def random_string(length=8):
+ """return a random lowercase string with default length of 8"""
+ return "".join(
+ random.choice(string.ascii_lowercase) for _ in range(length)
+ )
+
+
class TestCase(unittest.TestCase):
def reset_global_state(self):
"""Reset any global state to its original settings.
@@ -86,9 +94,7 @@ class TestCase(unittest.TestCase):
In the future this should really be done with some registry that
can then be cleaned in a more obvious way.
"""
- util.PROC_CMDLINE = None
util._DNS_REDIRECT_IP = None
- util._LSB_RELEASE = {}
def setUp(self):
super(TestCase, self).setUp()
@@ -227,10 +233,7 @@ class CiTestCase(TestCase):
@classmethod
def random_string(cls, length=8):
- """return a random lowercase string with default length of 8"""
- return "".join(
- random.choice(string.ascii_lowercase) for _ in range(length)
- )
+ return random_string(length)
class ResourceUsingTestCase(CiTestCase):
@@ -552,4 +555,32 @@ def cloud_init_project_dir(sub_path: str) -> str:
return str(get_top_level_dir() / sub_path)
+@contextmanager
+def does_not_raise():
+ """Context manager to parametrize tests raising and not raising exceptions
+
+ Note: In python-3.7+, this can be substituted by contextlib.nullcontext
+ More info:
+ https://docs.pytest.org/en/6.2.x/example/parametrize.html?highlight=does_not_raise#parametrizing-conditional-raising
+
+ Example:
+ --------
+ >>> @pytest.mark.parametrize(
+ >>> "example_input,expectation",
+ >>> [
+ >>> (1, does_not_raise()),
+ >>> (0, pytest.raises(ZeroDivisionError)),
+ >>> ],
+ >>> )
+ >>> def test_division(example_input, expectation):
+ >>> with expectation:
+ >>> assert (0 / example_input) is not None
+
+ """
+ try:
+ yield
+ except Exception as ex:
+ raise pytest.fail("DID RAISE {0}".format(ex))
+
+
# vi: ts=4 expandtab
diff --git a/tests/unittests/sources/test_oracle.py b/tests/unittests/sources/test_oracle.py
index b7b16952..b5d47178 100644
--- a/tests/unittests/sources/test_oracle.py
+++ b/tests/unittests/sources/test_oracle.py
@@ -3,7 +3,7 @@
import base64
import copy
import json
-from contextlib import ExitStack
+import logging
from unittest import mock
import pytest
@@ -13,6 +13,7 @@ from cloudinit.sources import NetworkConfigSource
from cloudinit.sources.DataSourceOracle import OpcMetadata
from cloudinit.url_helper import UrlError
from tests.unittests import helpers as test_helpers
+from tests.unittests.helpers import does_not_raise
DS_PATH = "cloudinit.sources.DataSourceOracle"
@@ -87,6 +88,25 @@ OPC_V2_METADATA = """\
# Just a small meaningless change to differentiate the two metadatas
OPC_V1_METADATA = OPC_V2_METADATA.replace("ocid1.instance", "ocid2.instance")
+MAC_ADDR = "00:00:17:02:2b:b1"
+
+DHCP = {
+ "name": "eth0",
+ "type": "physical",
+ "subnets": [
+ {
+ "broadcast": "192.168.122.255",
+ "control": "manual",
+ "gateway": "192.168.122.1",
+ "dns_search": ["foo.com"],
+ "type": "dhcp",
+ "netmask": "255.255.255.0",
+ "dns_nameservers": ["192.168.122.1"],
+ }
+ ],
+}
+KLIBC_NET_CFG = {"version": 1, "config": [DHCP]}
+
@pytest.fixture
def metadata_version():
@@ -94,15 +114,20 @@ def metadata_version():
@pytest.fixture
-def oracle_ds(request, fixture_utils, paths, metadata_version):
+def oracle_ds(request, fixture_utils, paths, metadata_version, mocker):
"""
Return an instantiated DataSourceOracle.
- This also performs the mocking required for the default test case:
+ This also performs the mocking required:
* ``_read_system_uuid`` returns something,
* ``_is_platform_viable`` returns True,
- * ``_is_iscsi_root`` returns True (the simpler code path),
- * ``read_opc_metadata`` returns ``OPC_V1_METADATA``
+ * ``DataSourceOracle._is_iscsi_root`` returns True by default or what
+ pytest.mark.is_iscsi gives as first param,
+ * ``DataSourceOracle._get_iscsi_config`` returns a network cfg if
+ is_iscsi else an empty network config,
+ * ``read_opc_metadata`` returns ``OPC_V1_METADATA``,
+ * ``dhcp.EphemeralDHCPv4`` and ``net.find_fallback_nic`` mocked to
+ avoid subp calls
(This uses the paths fixture for the required helpers.Paths object, and the
fixture_utils fixture for fetching markers.)
@@ -110,19 +135,29 @@ def oracle_ds(request, fixture_utils, paths, metadata_version):
sys_cfg = fixture_utils.closest_marker_first_arg_or(
request, "ds_sys_cfg", mock.MagicMock()
)
+ is_iscsi = fixture_utils.closest_marker_first_arg_or(
+ request, "is_iscsi", True
+ )
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=metadata,
- ):
- yield oracle.DataSourceOracle(
- sys_cfg=sys_cfg,
- distro=mock.Mock(),
- paths=paths,
- )
+
+ mocker.patch(DS_PATH + ".net.find_fallback_nic")
+ mocker.patch(DS_PATH + ".dhcp.EphemeralDHCPv4")
+ mocker.patch(DS_PATH + "._read_system_uuid", return_value="someuuid")
+ mocker.patch(DS_PATH + "._is_platform_viable", return_value=True)
+ mocker.patch(DS_PATH + ".read_opc_metadata", return_value=metadata)
+ mocker.patch(DS_PATH + ".KlibcOracleNetworkConfigSource")
+ ds = oracle.DataSourceOracle(
+ sys_cfg=sys_cfg,
+ distro=mock.Mock(),
+ paths=paths,
+ )
+ mocker.patch.object(ds, "_is_iscsi_root", return_value=is_iscsi)
+ if is_iscsi:
+ iscsi_config = copy.deepcopy(KLIBC_NET_CFG)
+ else:
+ iscsi_config = {"version": 1, "config": []}
+ mocker.patch.object(ds, "_get_iscsi_config", return_value=iscsi_config)
+ yield ds
class TestDataSourceOracle:
@@ -158,28 +193,27 @@ class TestDataSourceOracle:
assert oracle_ds.ds_cfg["configure_secondary_nics"]
-class TestIsPlatformViable(test_helpers.CiTestCase):
- @mock.patch(
- DS_PATH + ".dmi.read_dmi_data", return_value=oracle.CHASSIS_ASSET_TAG
+class TestIsPlatformViable:
+ @pytest.mark.parametrize(
+ "dmi_data, platform_viable",
+ [
+ # System with known chassis tag is viable.
+ (oracle.CHASSIS_ASSET_TAG, True),
+ # System without known chassis tag is not viable.
+ (None, False),
+ # System with unknown chassis tag is not viable.
+ ("LetsGoCubs", False),
+ ],
)
- def test_expected_viable(self, m_read_dmi_data):
- """System with known chassis tag is viable."""
- self.assertTrue(oracle._is_platform_viable())
- m_read_dmi_data.assert_has_calls([mock.call("chassis-asset-tag")])
-
- @mock.patch(DS_PATH + ".dmi.read_dmi_data", return_value=None)
- def test_expected_not_viable_dmi_data_none(self, m_read_dmi_data):
- """System without known chassis tag is not viable."""
- self.assertFalse(oracle._is_platform_viable())
- m_read_dmi_data.assert_has_calls([mock.call("chassis-asset-tag")])
-
- @mock.patch(DS_PATH + ".dmi.read_dmi_data", return_value="LetsGoCubs")
- def test_expected_not_viable_other(self, m_read_dmi_data):
- """System with unnown chassis tag is not viable."""
- self.assertFalse(oracle._is_platform_viable())
+ def test_is_platform_viable(self, dmi_data, platform_viable):
+ with mock.patch(
+ DS_PATH + ".dmi.read_dmi_data", return_value=dmi_data
+ ) as m_read_dmi_data:
+ assert platform_viable == oracle._is_platform_viable()
m_read_dmi_data.assert_has_calls([mock.call("chassis-asset-tag")])
+@pytest.mark.is_iscsi(False)
@mock.patch(
"cloudinit.net.is_openvswitch_internal_interface",
mock.Mock(return_value=False),
@@ -190,7 +224,7 @@ class TestNetworkConfigFromOpcImds:
# We test this by using in a non-dict to ensure that no dict
# operations are used; failure would be seen as exceptions
oracle_ds._network_config = object()
- oracle_ds._add_network_config_from_opc_imds()
+ oracle_ds._add_network_config_from_opc_imds(set_primary=False)
def test_bare_metal_machine_skipped(self, oracle_ds, caplog):
# nicIndex in the first entry indicates a bare metal machine
@@ -198,40 +232,47 @@ class TestNetworkConfigFromOpcImds:
# We test this by using a non-dict to ensure that no dict
# operations are used
oracle_ds._network_config = object()
- oracle_ds._add_network_config_from_opc_imds()
+ oracle_ds._add_network_config_from_opc_imds(set_primary=False)
assert "bare metal machine" in caplog.text
- def test_missing_mac_skipped(self, oracle_ds, caplog):
- oracle_ds._vnics_data = json.loads(OPC_VM_SECONDARY_VNIC_RESPONSE)
-
- 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()
-
- 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, oracle_ds, caplog):
+ @pytest.mark.parametrize(
+ "network_config, network_config_key",
+ [
+ pytest.param(
+ {
+ "version": 1,
+ "config": [{"primary": "nic"}],
+ },
+ "config",
+ id="v1",
+ ),
+ pytest.param(
+ {
+ "version": 2,
+ "ethernets": {"primary": {"nic": {}}},
+ },
+ "ethernets",
+ id="v2",
+ ),
+ ],
+ )
+ def test_missing_mac_skipped(
+ self,
+ oracle_ds,
+ network_config,
+ network_config_key,
+ caplog,
+ ):
oracle_ds._vnics_data = json.loads(OPC_VM_SECONDARY_VNIC_RESPONSE)
-
- oracle_ds._network_config = {
- "version": 2,
- "ethernets": {"primary": {"nic": {}}},
- }
+ oracle_ds._network_config = network_config
with mock.patch(DS_PATH + ".get_interfaces_by_mac", return_value={}):
- oracle_ds._add_network_config_from_opc_imds()
+ oracle_ds._add_network_config_from_opc_imds(set_primary=False)
- assert 1 == len(oracle_ds.network_config["ethernets"])
+ assert 1 == len(oracle_ds._network_config[network_config_key])
assert (
- "Interface with MAC 00:00:17:02:2b:b1 not found; skipping"
- in caplog.text
+ f"Interface with MAC {MAC_ADDR} not found; skipping" in caplog.text
)
+ assert 1 == caplog.text.count(" not found; skipping")
def test_secondary_nic(self, oracle_ds):
oracle_ds._vnics_data = json.loads(OPC_VM_SECONDARY_VNIC_RESPONSE)
@@ -239,12 +280,12 @@ class TestNetworkConfigFromOpcImds:
"version": 1,
"config": [{"primary": "nic"}],
}
- mac_addr, nic_name = "00:00:17:02:2b:b1", "ens3"
+ mac_addr, nic_name = MAC_ADDR, "ens3"
with mock.patch(
DS_PATH + ".get_interfaces_by_mac",
return_value={mac_addr: nic_name},
):
- oracle_ds._add_network_config_from_opc_imds()
+ oracle_ds._add_network_config_from_opc_imds(set_primary=False)
# The input is mutated
assert 2 == len(oracle_ds.network_config["config"])
@@ -266,12 +307,12 @@ class TestNetworkConfigFromOpcImds:
"version": 2,
"ethernets": {"primary": {"nic": {}}},
}
- mac_addr, nic_name = "00:00:17:02:2b:b1", "ens3"
+ mac_addr, nic_name = MAC_ADDR, "ens3"
with mock.patch(
DS_PATH + ".get_interfaces_by_mac",
return_value={mac_addr: nic_name},
):
- oracle_ds._add_network_config_from_opc_imds()
+ oracle_ds._add_network_config_from_opc_imds(set_primary=False)
# The input is mutated
assert 2 == len(oracle_ds.network_config["ethernets"])
@@ -286,77 +327,180 @@ class TestNetworkConfigFromOpcImds:
# These values are hard-coded in OPC_VM_SECONDARY_VNIC_RESPONSE
assert "10.0.0.231" == secondary_nic_cfg["addresses"][0]
+ @pytest.mark.parametrize("error_add_network", [None, Exception])
+ @pytest.mark.parametrize(
+ "configure_secondary_nics",
+ [False, True],
+ )
+ @mock.patch(DS_PATH + "._ensure_netfailover_safe")
+ def test_network_config_log_errors(
+ self,
+ m_ensure_netfailover_safe,
+ configure_secondary_nics,
+ error_add_network,
+ oracle_ds,
+ caplog,
+ capsys,
+ ):
+ assert not oracle_ds._has_network_config()
+ oracle_ds.ds_cfg["configure_secondary_nics"] = configure_secondary_nics
+ with mock.patch.object(
+ oracle.DataSourceOracle,
+ "_add_network_config_from_opc_imds",
+ ) as m_add_network_config_from_opc_imds:
+ if error_add_network:
+ m_add_network_config_from_opc_imds.side_effect = (
+ error_add_network
+ )
+ oracle_ds.network_config # pylint: disable=pointless-statement # noqa: E501
+ assert [
+ mock.call(True, False)
+ == m_add_network_config_from_opc_imds.call_args_list
+ ]
+ assert 1 == oracle_ds._is_iscsi_root.call_count
+ assert 1 == m_ensure_netfailover_safe.call_count
+
+ assert ("", "") == capsys.readouterr()
+ if not error_add_network:
+ log_initramfs_index = -1
+ else:
+ log_initramfs_index = -3
+ # Primary
+ assert (
+ logging.WARNING,
+ "Failed to parse IMDS network configuration!",
+ ) == caplog.record_tuples[-2][1:]
+ # Secondary
+ assert (
+ logging.DEBUG,
+ "Failed to parse IMDS network configuration!",
+ ) == caplog.record_tuples[-1][1:]
-class TestNetworkConfigFiltersNetFailover(test_helpers.CiTestCase):
- def setUp(self):
- super(TestNetworkConfigFiltersNetFailover, self).setUp()
- self.add_patch(
- DS_PATH + ".get_interfaces_by_mac", "m_get_interfaces_by_mac"
- )
- self.add_patch(DS_PATH + ".is_netfail_master", "m_netfail_master")
+ assert (
+ logging.WARNING,
+ "Could not obtain network configuration from initramfs."
+ " Falling back to IMDS.",
+ ) == caplog.record_tuples[log_initramfs_index][1:]
- def test_ignore_bogus_network_config(self):
- netcfg = {"something": "here"}
- passed_netcfg = copy.copy(netcfg)
- oracle._ensure_netfailover_safe(passed_netcfg)
- self.assertEqual(netcfg, passed_netcfg)
- def test_ignore_network_config_unknown_versions(self):
- netcfg = {"something": "here", "version": 3}
+@mock.patch(DS_PATH + ".get_interfaces_by_mac")
+@mock.patch(DS_PATH + ".is_netfail_master")
+class TestNetworkConfigFiltersNetFailover:
+ @pytest.mark.parametrize(
+ "netcfg",
+ [
+ pytest.param({"something": "here"}, id="bogus"),
+ pytest.param(
+ {"something": "here", "version": 3}, id="unknown_version"
+ ),
+ ],
+ )
+ def test_ignore_network_config(
+ self, m_netfail_master, m_get_interfaces_by_mac, netcfg
+ ):
passed_netcfg = copy.copy(netcfg)
oracle._ensure_netfailover_safe(passed_netcfg)
- self.assertEqual(netcfg, passed_netcfg)
+ assert netcfg == passed_netcfg
- def test_checks_v1_type_physical_interfaces(self):
- mac_addr, nic_name = "00:00:17:02:2b:b1", "ens3"
- self.m_get_interfaces_by_mac.return_value = {
- mac_addr: nic_name,
- }
- netcfg = {
- "version": 1,
- "config": [
+ @pytest.mark.parametrize(
+ "nic_name, netcfg, netfail_master_return, call_args_list",
+ [
+ pytest.param(
+ "ens3",
{
- "type": "physical",
- "name": nic_name,
- "mac_address": mac_addr,
- "subnets": [{"type": "dhcp4"}],
- }
- ],
- }
- passed_netcfg = copy.copy(netcfg)
- self.m_netfail_master.return_value = False
- oracle._ensure_netfailover_safe(passed_netcfg)
- self.assertEqual(netcfg, passed_netcfg)
- self.assertEqual(
- [mock.call(nic_name)], self.m_netfail_master.call_args_list
- )
-
- def test_checks_v1_skips_non_phys_interfaces(self):
- mac_addr, nic_name = "00:00:17:02:2b:b1", "bond0"
- self.m_get_interfaces_by_mac.return_value = {
- mac_addr: nic_name,
- }
- netcfg = {
- "version": 1,
- "config": [
+ "version": 1,
+ "config": [
+ {
+ "type": "physical",
+ "name": "ens3",
+ "mac_address": MAC_ADDR,
+ "subnets": [{"type": "dhcp4"}],
+ }
+ ],
+ },
+ False,
+ [mock.call("ens3")],
+ id="checks_v1_type_physical_interfaces",
+ ),
+ pytest.param(
+ "bond0",
{
- "type": "bond",
- "name": nic_name,
- "mac_address": mac_addr,
- "subnets": [{"type": "dhcp4"}],
- }
- ],
+ "version": 1,
+ "config": [
+ {
+ "type": "bond",
+ "name": "bond0",
+ "mac_address": MAC_ADDR,
+ "subnets": [{"type": "dhcp4"}],
+ }
+ ],
+ },
+ None,
+ [],
+ id="skips_v1_non_phys_interfaces",
+ ),
+ pytest.param(
+ "ens3",
+ {
+ "version": 2,
+ "ethernets": {
+ "ens3": {
+ "dhcp4": True,
+ "critical": True,
+ "set-name": "ens3",
+ "match": {"macaddress": MAC_ADDR},
+ }
+ },
+ },
+ False,
+ [mock.call("ens3")],
+ id="checks_v2_type_ethernet_interfaces",
+ ),
+ pytest.param(
+ "wlps0",
+ {
+ "version": 2,
+ "ethernets": {
+ "wlps0": {
+ "dhcp4": True,
+ "critical": True,
+ "set-name": "wlps0",
+ "match": {"macaddress": MAC_ADDR},
+ }
+ },
+ },
+ None,
+ [mock.call("wlps0")],
+ id="skips_v2_non_ethernet_interfaces",
+ ),
+ ],
+ )
+ def test__ensure_netfailover_safe(
+ self,
+ m_netfail_master,
+ m_get_interfaces_by_mac,
+ nic_name,
+ netcfg,
+ netfail_master_return,
+ call_args_list,
+ ):
+ m_get_interfaces_by_mac.return_value = {
+ MAC_ADDR: nic_name,
}
passed_netcfg = copy.copy(netcfg)
+ if netfail_master_return is not None:
+ m_netfail_master.return_value = netfail_master_return
oracle._ensure_netfailover_safe(passed_netcfg)
- self.assertEqual(netcfg, passed_netcfg)
- self.assertEqual(0, self.m_netfail_master.call_count)
-
- def test_removes_master_mac_property_v1(self):
- nic_master, mac_master = "ens3", self.random_string()
- nic_other, mac_other = "ens7", self.random_string()
- nic_extra, mac_extra = "enp0s1f2", self.random_string()
- self.m_get_interfaces_by_mac.return_value = {
+ assert netcfg == passed_netcfg
+ assert call_args_list == m_netfail_master.call_args_list
+
+ def test_removes_master_mac_property_v1(
+ self, m_netfail_master, m_get_interfaces_by_mac
+ ):
+ nic_master, mac_master = "ens3", test_helpers.random_string()
+ nic_other, mac_other = "ens7", test_helpers.random_string()
+ nic_extra, mac_extra = "enp0s1f2", test_helpers.random_string()
+ m_get_interfaces_by_mac.return_value = {
mac_master: nic_master,
mac_other: nic_other,
mac_extra: nic_extra,
@@ -387,7 +531,7 @@ class TestNetworkConfigFiltersNetFailover(test_helpers.CiTestCase):
return True
return False
- self.m_netfail_master.side_effect = _is_netfail_master
+ m_netfail_master.side_effect = _is_netfail_master
expected_cfg = {
"version": 1,
"config": [
@@ -405,58 +549,15 @@ class TestNetworkConfigFiltersNetFailover(test_helpers.CiTestCase):
],
}
oracle._ensure_netfailover_safe(netcfg)
- self.assertEqual(expected_cfg, netcfg)
-
- def test_checks_v2_type_ethernet_interfaces(self):
- mac_addr, nic_name = "00:00:17:02:2b:b1", "ens3"
- self.m_get_interfaces_by_mac.return_value = {
- mac_addr: nic_name,
- }
- netcfg = {
- "version": 2,
- "ethernets": {
- nic_name: {
- "dhcp4": True,
- "critical": True,
- "set-name": nic_name,
- "match": {"macaddress": mac_addr},
- }
- },
- }
- passed_netcfg = copy.copy(netcfg)
- self.m_netfail_master.return_value = False
- oracle._ensure_netfailover_safe(passed_netcfg)
- self.assertEqual(netcfg, passed_netcfg)
- self.assertEqual(
- [mock.call(nic_name)], self.m_netfail_master.call_args_list
- )
+ assert expected_cfg == netcfg
- def test_skips_v2_non_ethernet_interfaces(self):
- mac_addr, nic_name = "00:00:17:02:2b:b1", "wlps0"
- self.m_get_interfaces_by_mac.return_value = {
- mac_addr: nic_name,
- }
- netcfg = {
- "version": 2,
- "wifis": {
- nic_name: {
- "dhcp4": True,
- "critical": True,
- "set-name": nic_name,
- "match": {"macaddress": mac_addr},
- }
- },
- }
- passed_netcfg = copy.copy(netcfg)
- oracle._ensure_netfailover_safe(passed_netcfg)
- self.assertEqual(netcfg, passed_netcfg)
- self.assertEqual(0, self.m_netfail_master.call_count)
-
- def test_removes_master_mac_property_v2(self):
- nic_master, mac_master = "ens3", self.random_string()
- nic_other, mac_other = "ens7", self.random_string()
- nic_extra, mac_extra = "enp0s1f2", self.random_string()
- self.m_get_interfaces_by_mac.return_value = {
+ def test_removes_master_mac_property_v2(
+ self, m_netfail_master, m_get_interfaces_by_mac
+ ):
+ nic_master, mac_master = "ens3", test_helpers.random_string()
+ nic_other, mac_other = "ens7", test_helpers.random_string()
+ nic_extra, mac_extra = "enp0s1f2", test_helpers.random_string()
+ m_get_interfaces_by_mac.return_value = {
mac_master: nic_master,
mac_other: nic_other,
mac_extra: nic_extra,
@@ -487,7 +588,7 @@ class TestNetworkConfigFiltersNetFailover(test_helpers.CiTestCase):
return True
return False
- self.m_netfail_master.side_effect = _is_netfail_master
+ m_netfail_master.side_effect = _is_netfail_master
expected_cfg = {
"version": 2,
@@ -511,7 +612,7 @@ class TestNetworkConfigFiltersNetFailover(test_helpers.CiTestCase):
pprint.pprint(netcfg)
print("---- ^^ modified ^^ ---- vv original vv ----")
pprint.pprint(expected_cfg)
- self.assertEqual(expected_cfg, netcfg)
+ assert expected_cfg == netcfg
def _mock_v2_urls(httpretty):
@@ -557,7 +658,6 @@ def _mock_no_v2_urls(httpretty):
class TestReadOpcMetadata:
# See https://docs.pytest.org/en/stable/example
# /parametrize.html#parametrizing-conditional-raising
- does_not_raise = ExitStack
@mock.patch("cloudinit.url_helper.time.sleep", lambda _: None)
@pytest.mark.parametrize(
@@ -636,7 +736,29 @@ class TestReadOpcMetadata:
with expectation:
assert expected_body == oracle.read_opc_metadata().instance_data
+ # No need to actually wait between retries in the tests
+ @mock.patch("cloudinit.url_helper.time.sleep", lambda _: None)
+ def test_fetch_vnics_error(self, caplog):
+ def mocked_fetch(*args, path="instance", **kwargs):
+ if path == "vnics":
+ raise UrlError("cause")
+
+ with mock.patch(DS_PATH + "._fetch", side_effect=mocked_fetch):
+ opc_metadata = oracle.read_opc_metadata(fetch_vnics_data=True)
+ assert None is opc_metadata.vnics_data
+ assert (
+ logging.WARNING,
+ "Failed to fetch IMDS network configuration!",
+ ) == caplog.record_tuples[-2][1:]
+
+@pytest.mark.parametrize(
+ "",
+ [
+ pytest.param(marks=pytest.mark.is_iscsi(True), id="iscsi"),
+ pytest.param(marks=pytest.mark.is_iscsi(False), id="non-iscsi"),
+ ],
+)
class TestCommon_GetDataBehaviour:
"""This test class tests behaviour common to iSCSI and non-iSCSI root.
@@ -649,33 +771,14 @@ class TestCommon_GetDataBehaviour:
separate class for that case.)
"""
- @pytest.fixture(params=[True, False])
- def parameterized_oracle_ds(self, request, oracle_ds):
- """oracle_ds parameterized for iSCSI and non-iSCSI root respectively"""
- is_iscsi_root = request.param
- with ExitStack() as stack:
- stack.enter_context(
- mock.patch(
- DS_PATH + "._is_iscsi_root", return_value=is_iscsi_root
- )
- )
- if not is_iscsi_root:
- stack.enter_context(
- mock.patch(DS_PATH + ".net.find_fallback_nic")
- )
- stack.enter_context(
- mock.patch(DS_PATH + ".dhcp.EphemeralDHCPv4")
- )
- yield oracle_ds
-
@mock.patch(
DS_PATH + "._is_platform_viable", mock.Mock(return_value=False)
)
def test_false_if_platform_not_viable(
self,
- parameterized_oracle_ds,
+ oracle_ds,
):
- assert not parameterized_oracle_ds._get_data()
+ assert not oracle_ds._get_data()
@pytest.mark.parametrize(
"keyname,expected_value",
@@ -699,10 +802,10 @@ class TestCommon_GetDataBehaviour:
self,
keyname,
expected_value,
- parameterized_oracle_ds,
+ oracle_ds,
):
- assert parameterized_oracle_ds._get_data()
- assert expected_value == parameterized_oracle_ds.metadata[keyname]
+ assert oracle_ds._get_data()
+ assert expected_value == oracle_ds.metadata[keyname]
@pytest.mark.parametrize(
"attribute_name,expected_value",
@@ -722,12 +825,10 @@ class TestCommon_GetDataBehaviour:
self,
attribute_name,
expected_value,
- parameterized_oracle_ds,
+ oracle_ds,
):
- assert parameterized_oracle_ds._get_data()
- assert expected_value == getattr(
- parameterized_oracle_ds, attribute_name
- )
+ assert oracle_ds._get_data()
+ assert expected_value == getattr(oracle_ds, attribute_name)
@pytest.mark.parametrize(
"ssh_keys,expected_value",
@@ -746,7 +847,7 @@ class TestCommon_GetDataBehaviour:
],
)
def test_public_keys_handled_correctly(
- self, ssh_keys, expected_value, parameterized_oracle_ds
+ self, ssh_keys, expected_value, oracle_ds
):
instance_data = json.loads(OPC_V1_METADATA)
if ssh_keys is None:
@@ -758,14 +859,10 @@ class TestCommon_GetDataBehaviour:
DS_PATH + ".read_opc_metadata",
mock.Mock(return_value=metadata),
):
- assert parameterized_oracle_ds._get_data()
- assert (
- expected_value == parameterized_oracle_ds.get_public_ssh_keys()
- )
+ assert oracle_ds._get_data()
+ assert expected_value == oracle_ds.get_public_ssh_keys()
- def test_missing_user_data_handled_gracefully(
- self, parameterized_oracle_ds
- ):
+ def test_missing_user_data_handled_gracefully(self, oracle_ds):
instance_data = json.loads(OPC_V1_METADATA)
del instance_data["metadata"]["user_data"]
metadata = OpcMetadata(None, instance_data, None)
@@ -773,13 +870,11 @@ class TestCommon_GetDataBehaviour:
DS_PATH + ".read_opc_metadata",
mock.Mock(return_value=metadata),
):
- assert parameterized_oracle_ds._get_data()
+ assert oracle_ds._get_data()
- assert parameterized_oracle_ds.userdata_raw is None
+ assert oracle_ds.userdata_raw is None
- def test_missing_metadata_handled_gracefully(
- self, parameterized_oracle_ds
- ):
+ def test_missing_metadata_handled_gracefully(self, oracle_ds):
instance_data = json.loads(OPC_V1_METADATA)
del instance_data["metadata"]
metadata = OpcMetadata(None, instance_data, None)
@@ -787,17 +882,17 @@ class TestCommon_GetDataBehaviour:
DS_PATH + ".read_opc_metadata",
mock.Mock(return_value=metadata),
):
- assert parameterized_oracle_ds._get_data()
+ assert oracle_ds._get_data()
- assert parameterized_oracle_ds.userdata_raw is None
- assert [] == parameterized_oracle_ds.get_public_ssh_keys()
+ assert oracle_ds.userdata_raw is None
+ assert [] == oracle_ds.get_public_ssh_keys()
-@mock.patch(DS_PATH + "._is_iscsi_root", lambda: False)
+@pytest.mark.is_iscsi(False)
class TestNonIscsiRoot_GetDataBehaviour:
@mock.patch(DS_PATH + ".dhcp.EphemeralDHCPv4")
@mock.patch(DS_PATH + ".net.find_fallback_nic")
- def test_read_opc_metadata_called_with_ephemeral_dhcp(
+ def test_run_net_files(
self, m_find_fallback_nic, m_EphemeralDHCPv4, oracle_ds
):
in_context_manager = False
@@ -837,74 +932,122 @@ class TestNonIscsiRoot_GetDataBehaviour:
)
] == m_EphemeralDHCPv4.call_args_list
+ @mock.patch(DS_PATH + ".dhcp.EphemeralDHCPv4")
+ @mock.patch(DS_PATH + ".net.find_fallback_nic")
+ def test_read_opc_metadata_called_with_ephemeral_dhcp(
+ self, m_find_fallback_nic, m_EphemeralDHCPv4, oracle_ds
+ ):
+ in_context_manager = False
-@mock.patch(DS_PATH + ".get_interfaces_by_mac", lambda: {})
-@mock.patch(DS_PATH + ".cmdline.read_initramfs_config")
-class TestNetworkConfig:
- def test_network_config_cached(self, m_read_initramfs_config, oracle_ds):
- """.network_config should be cached"""
- assert 0 == m_read_initramfs_config.call_count
- oracle_ds.network_config # pylint: disable=pointless-statement
- assert 1 == m_read_initramfs_config.call_count
- oracle_ds.network_config # pylint: disable=pointless-statement
- assert 1 == m_read_initramfs_config.call_count
+ def enter_context_manager():
+ nonlocal in_context_manager
+ in_context_manager = True
- def test_network_cmdline(self, m_read_initramfs_config, oracle_ds):
- """network_config should prefer initramfs config over fallback"""
- ncfg = {"version": 1, "config": [{"a": "b"}]}
- m_read_initramfs_config.return_value = copy.deepcopy(ncfg)
+ def exit_context_manager(*args):
+ nonlocal in_context_manager
+ in_context_manager = False
- assert ncfg == oracle_ds.network_config
- assert 0 == oracle_ds.distro.generate_fallback_config.call_count
+ m_EphemeralDHCPv4.return_value.__enter__.side_effect = (
+ enter_context_manager
+ )
+ m_EphemeralDHCPv4.return_value.__exit__.side_effect = (
+ exit_context_manager
+ )
- def test_network_fallback(self, m_read_initramfs_config, oracle_ds):
- """network_config should prefer initramfs config over fallback"""
- ncfg = {"version": 1, "config": [{"a": "b"}]}
+ def assert_in_context_manager(**kwargs):
+ assert in_context_manager
+ return mock.MagicMock()
- m_read_initramfs_config.return_value = None
- oracle_ds.distro.generate_fallback_config.return_value = copy.deepcopy(
- ncfg
- )
+ with mock.patch(
+ DS_PATH + ".read_opc_metadata",
+ mock.Mock(side_effect=assert_in_context_manager),
+ ):
+ assert oracle_ds._get_data()
+
+ assert [
+ mock.call(
+ iface=m_find_fallback_nic.return_value,
+ connectivity_url_data={
+ "headers": {"Authorization": "Bearer Oracle"},
+ "url": "http://169.254.169.254/opc/v2/instance/",
+ },
+ )
+ ] == m_EphemeralDHCPv4.call_args_list
- assert ncfg == oracle_ds.network_config
+
+@mock.patch(DS_PATH + ".get_interfaces_by_mac", return_value={})
+class TestNetworkConfig:
+ def test_network_config_cached(self, m_get_interfaces_by_mac, oracle_ds):
+ """.network_config should be cached"""
+ assert 0 == oracle_ds._get_iscsi_config.call_count
+ oracle_ds.network_config # pylint: disable=pointless-statement
+ assert 1 == oracle_ds._get_iscsi_config.call_count
+ oracle_ds.network_config # pylint: disable=pointless-statement
+ assert 1 == oracle_ds._get_iscsi_config.call_count
@pytest.mark.parametrize(
- "configure_secondary_nics,expect_secondary_nics",
- [(True, True), (False, False), (None, False)],
+ "configure_secondary_nics,is_iscsi,expected_set_primary",
+ [
+ pytest.param(
+ True,
+ True,
+ [mock.call(False)],
+ marks=pytest.mark.is_iscsi(True),
+ ),
+ pytest.param(
+ True,
+ False,
+ [mock.call(True)],
+ marks=pytest.mark.is_iscsi(False),
+ ),
+ pytest.param(False, True, [], marks=pytest.mark.is_iscsi(True)),
+ pytest.param(
+ False,
+ False,
+ [mock.call(True)],
+ marks=pytest.mark.is_iscsi(False),
+ ),
+ pytest.param(None, True, [], marks=pytest.mark.is_iscsi(True)),
+ pytest.param(
+ None,
+ False,
+ [mock.call(True)],
+ marks=pytest.mark.is_iscsi(False),
+ ),
+ ],
)
def test_secondary_nic_addition(
self,
- m_read_initramfs_config,
+ m_get_interfaces_by_mac,
configure_secondary_nics,
- expect_secondary_nics,
+ is_iscsi,
+ expected_set_primary,
oracle_ds,
):
"""Test that _add_network_config_from_opc_imds is called as expected
(configure_secondary_nics=None is used to test the default behaviour.)
"""
- m_read_initramfs_config.return_value = {"version": 1, "config": []}
if configure_secondary_nics is not None:
oracle_ds.ds_cfg[
"configure_secondary_nics"
] = configure_secondary_nics
- def side_effect(self):
- self._network_config["secondary_added"] = mock.sentinel.needle
-
oracle_ds._vnics_data = "DummyData"
with mock.patch.object(
- oracle.DataSourceOracle,
+ oracle_ds,
"_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
+ ) as m_add_network_config_from_opc_imds:
+ oracle_ds.network_config # pylint: disable=pointless-statement
+ assert (
+ expected_set_primary
+ == m_add_network_config_from_opc_imds.call_args_list
+ )
def test_secondary_nic_failure_isnt_blocking(
self,
- m_read_initramfs_config,
+ m_get_interfaces_by_mac,
caplog,
oracle_ds,
):
@@ -917,15 +1060,88 @@ class TestNetworkConfig:
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
+ assert network_config == oracle_ds._get_iscsi_config.return_value
+ assert 2 == caplog.text.count(
+ "Failed to parse IMDS network configuration"
+ )
- def test_ds_network_cfg_preferred_over_initramfs(self, _m):
+ def test_ds_network_cfg_preferred_over_initramfs(
+ self, m_get_interfaces_by_mac
+ ):
"""Ensure that DS net config is preferred over initramfs config"""
config_sources = oracle.DataSourceOracle.network_config_sources
ds_idx = config_sources.index(NetworkConfigSource.DS)
initramfs_idx = config_sources.index(NetworkConfigSource.INITRAMFS)
assert ds_idx < initramfs_idx
+ @pytest.mark.parametrize("set_primary", [True, False])
+ def test__add_network_config_from_opc_imds_no_vnics_data(
+ self,
+ m_get_interfaces_by_mac,
+ set_primary,
+ oracle_ds,
+ caplog,
+ ):
+ assert not oracle_ds._has_network_config()
+ with mock.patch.object(oracle_ds, "_vnics_data", None):
+ oracle_ds._add_network_config_from_opc_imds(set_primary)
+ assert not oracle_ds._has_network_config()
+ assert (
+ logging.WARNING,
+ "NIC data is UNSET but should not be",
+ ) == caplog.record_tuples[-1][1:]
+
+ def test_missing_mac_skipped(
+ self,
+ m_get_interfaces_by_mac,
+ oracle_ds,
+ caplog,
+ ):
+ """If no intefaces by mac found, then _network_config not setted and
+ correct logs.
+ """
+ vnics_data = json.loads(OPC_VM_SECONDARY_VNIC_RESPONSE)
+ assert not oracle_ds._has_network_config()
+ with mock.patch.object(oracle_ds, "_vnics_data", vnics_data):
+ oracle_ds._add_network_config_from_opc_imds(set_primary=True)
+ assert not oracle_ds._has_network_config()
+ assert (
+ logging.WARNING,
+ "Interface with MAC 02:00:17:05:d1:db not found; skipping",
+ ) == caplog.record_tuples[-2][1:]
+ assert (
+ logging.WARNING,
+ f"Interface with MAC {MAC_ADDR} not found; skipping",
+ ) == caplog.record_tuples[-1][1:]
+
+ @pytest.mark.parametrize("set_primary", [True, False])
+ def test_nics(
+ self,
+ m_get_interfaces_by_mac,
+ set_primary,
+ oracle_ds,
+ caplog,
+ mocker,
+ ):
+ """Correct number of configs added"""
+ vnics_data = json.loads(OPC_VM_SECONDARY_VNIC_RESPONSE)
+ if set_primary:
+ assert not oracle_ds._has_network_config()
+ else:
+ # Simulate primary config was taken from iscsi
+ oracle_ds._network_config = copy.deepcopy(KLIBC_NET_CFG)
+
+ mocker.patch(
+ DS_PATH + ".get_interfaces_by_mac",
+ return_value={"02:00:17:05:d1:db": "eth_0", MAC_ADDR: "name_1"},
+ )
+ mocker.patch.object(oracle_ds, "_vnics_data", vnics_data)
+
+ oracle_ds._add_network_config_from_opc_imds(set_primary)
+ assert 2 == len(
+ oracle_ds._network_config["config"]
+ ), "Config not added"
+ assert "" == caplog.text
+
# vi: ts=4 expandtab
diff --git a/tox.ini b/tox.ini
index 9173db43..6e335d96 100644
--- a/tox.ini
+++ b/tox.ini
@@ -242,3 +242,4 @@ markers =
ubuntu: this test should run on Ubuntu
unstable: skip this test because it is flakey
adhoc: only run on adhoc basis, not in any CI environment (travis or jenkins)
+ is_iscsi: whether is an instance has iscsi net cfg or not
View Lorry Controller Status