diff options
| -rw-r--r-- | cloudinit/net/dhcp.py | 42 | ||||
| -rw-r--r-- | cloudinit/net/tests/test_dhcp.py | 113 | ||||
| -rw-r--r-- | cloudinit/sources/DataSourceCloudStack.py | 17 | ||||
| -rw-r--r-- | cloudinit/sources/helpers/azure.py | 20 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_azure_helper.py | 143 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_cloudstack.py | 11 |
6 files changed, 282 insertions, 64 deletions
diff --git a/cloudinit/net/dhcp.py b/cloudinit/net/dhcp.py index 05350639..0cba7032 100644 --- a/cloudinit/net/dhcp.py +++ b/cloudinit/net/dhcp.py @@ -4,6 +4,7 @@ # # This file is part of cloud-init. See LICENSE file for license information. +import configobj import logging import os import re @@ -11,9 +12,12 @@ import re from cloudinit.net import find_fallback_nic, get_devicelist from cloudinit import temp_utils from cloudinit import util +from six import StringIO LOG = logging.getLogger(__name__) +NETWORKD_LEASES_DIR = '/run/systemd/netif/leases' + class InvalidDHCPLeaseFileError(Exception): """Raised when parsing an empty or invalid dhcp.leases file. @@ -118,4 +122,42 @@ def dhcp_discovery(dhclient_cmd_path, interface, cleandir): return parse_dhcp_lease_file(lease_file) +def networkd_parse_lease(content): + """Parse a systemd lease file content as in /run/systemd/netif/leases/ + + Parse this (almost) ini style file even though it says: + # This is private data. Do not parse. + + Simply return a dictionary of key/values.""" + + return dict(configobj.ConfigObj(StringIO(content), list_values=False)) + + +def networkd_load_leases(leases_d=None): + """Return a dictionary of dictionaries representing each lease + found in lease_d.i + + The top level key will be the filename, which is typically the ifindex.""" + + if leases_d is None: + leases_d = NETWORKD_LEASES_DIR + + ret = {} + if not os.path.isdir(leases_d): + return ret + for lfile in os.listdir(leases_d): + ret[lfile] = networkd_parse_lease( + util.load_file(os.path.join(leases_d, lfile))) + return ret + + +def networkd_get_option_from_leases(keyname, leases_d=None): + if leases_d is None: + leases_d = NETWORKD_LEASES_DIR + leases = networkd_load_leases(leases_d=leases_d) + for ifindex, data in sorted(leases.items()): + if data.get(keyname): + return data[keyname] + return None + # vi: ts=4 expandtab diff --git a/cloudinit/net/tests/test_dhcp.py b/cloudinit/net/tests/test_dhcp.py index a38edaec..1c1f504a 100644 --- a/cloudinit/net/tests/test_dhcp.py +++ b/cloudinit/net/tests/test_dhcp.py @@ -6,9 +6,9 @@ from textwrap import dedent from cloudinit.net.dhcp import ( InvalidDHCPLeaseFileError, maybe_perform_dhcp_discovery, - parse_dhcp_lease_file, dhcp_discovery) + parse_dhcp_lease_file, dhcp_discovery, networkd_load_leases) from cloudinit.util import ensure_file, write_file -from cloudinit.tests.helpers import CiTestCase, wrap_and_call +from cloudinit.tests.helpers import CiTestCase, wrap_and_call, populate_dir class TestParseDHCPLeasesFile(CiTestCase): @@ -149,3 +149,112 @@ class TestDHCPDiscoveryClean(CiTestCase): [os.path.join(tmpdir, 'dhclient'), '-1', '-v', '-lf', lease_file, '-pf', os.path.join(tmpdir, 'dhclient.pid'), 'eth9', '-sf', '/bin/true'], capture=True)]) + + +class TestSystemdParseLeases(CiTestCase): + + lxd_lease = dedent("""\ + # This is private data. Do not parse. + ADDRESS=10.75.205.242 + NETMASK=255.255.255.0 + ROUTER=10.75.205.1 + SERVER_ADDRESS=10.75.205.1 + NEXT_SERVER=10.75.205.1 + BROADCAST=10.75.205.255 + T1=1580 + T2=2930 + LIFETIME=3600 + DNS=10.75.205.1 + DOMAINNAME=lxd + HOSTNAME=a1 + CLIENTID=ffe617693400020000ab110c65a6a0866931c2 + """) + + lxd_parsed = { + 'ADDRESS': '10.75.205.242', + 'NETMASK': '255.255.255.0', + 'ROUTER': '10.75.205.1', + 'SERVER_ADDRESS': '10.75.205.1', + 'NEXT_SERVER': '10.75.205.1', + 'BROADCAST': '10.75.205.255', + 'T1': '1580', + 'T2': '2930', + 'LIFETIME': '3600', + 'DNS': '10.75.205.1', + 'DOMAINNAME': 'lxd', + 'HOSTNAME': 'a1', + 'CLIENTID': 'ffe617693400020000ab110c65a6a0866931c2', + } + + azure_lease = dedent("""\ + # This is private data. Do not parse. + ADDRESS=10.132.0.5 + NETMASK=255.255.255.255 + ROUTER=10.132.0.1 + SERVER_ADDRESS=169.254.169.254 + NEXT_SERVER=10.132.0.1 + MTU=1460 + T1=43200 + T2=75600 + LIFETIME=86400 + DNS=169.254.169.254 + NTP=169.254.169.254 + DOMAINNAME=c.ubuntu-foundations.internal + DOMAIN_SEARCH_LIST=c.ubuntu-foundations.internal google.internal + HOSTNAME=tribaal-test-171002-1349.c.ubuntu-foundations.internal + ROUTES=10.132.0.1/32,0.0.0.0 0.0.0.0/0,10.132.0.1 + CLIENTID=ff405663a200020000ab11332859494d7a8b4c + OPTION_245=624c3620 + """) + + azure_parsed = { + 'ADDRESS': '10.132.0.5', + 'NETMASK': '255.255.255.255', + 'ROUTER': '10.132.0.1', + 'SERVER_ADDRESS': '169.254.169.254', + 'NEXT_SERVER': '10.132.0.1', + 'MTU': '1460', + 'T1': '43200', + 'T2': '75600', + 'LIFETIME': '86400', + 'DNS': '169.254.169.254', + 'NTP': '169.254.169.254', + 'DOMAINNAME': 'c.ubuntu-foundations.internal', + 'DOMAIN_SEARCH_LIST': 'c.ubuntu-foundations.internal google.internal', + 'HOSTNAME': 'tribaal-test-171002-1349.c.ubuntu-foundations.internal', + 'ROUTES': '10.132.0.1/32,0.0.0.0 0.0.0.0/0,10.132.0.1', + 'CLIENTID': 'ff405663a200020000ab11332859494d7a8b4c', + 'OPTION_245': '624c3620'} + + def setUp(self): + super(TestSystemdParseLeases, self).setUp() + self.lease_d = self.tmp_dir() + + def test_no_leases_returns_empty_dict(self): + """A leases dir with no lease files should return empty dictionary.""" + self.assertEqual({}, networkd_load_leases(self.lease_d)) + + def test_no_leases_dir_returns_empty_dict(self): + """A non-existing leases dir should return empty dict.""" + enodir = os.path.join(self.lease_d, 'does-not-exist') + self.assertEqual({}, networkd_load_leases(enodir)) + + def test_single_leases_file(self): + """A leases dir with one leases file.""" + populate_dir(self.lease_d, {'2': self.lxd_lease}) + self.assertEqual( + {'2': self.lxd_parsed}, networkd_load_leases(self.lease_d)) + + def test_single_azure_leases_file(self): + """On Azure, option 245 should be present, verify it specifically.""" + populate_dir(self.lease_d, {'1': self.azure_lease}) + self.assertEqual( + {'1': self.azure_parsed}, networkd_load_leases(self.lease_d)) + + def test_multiple_files(self): + """Multiple leases files on azure with one found return that value.""" + self.maxDiff = None + populate_dir(self.lease_d, {'1': self.azure_lease, + '9': self.lxd_lease}) + self.assertEqual({'1': self.azure_parsed, '9': self.lxd_parsed}, + networkd_load_leases(self.lease_d)) diff --git a/cloudinit/sources/DataSourceCloudStack.py b/cloudinit/sources/DataSourceCloudStack.py index 7e0f9bb8..9dc473fc 100644 --- a/cloudinit/sources/DataSourceCloudStack.py +++ b/cloudinit/sources/DataSourceCloudStack.py @@ -19,6 +19,7 @@ import time from cloudinit import ec2_utils as ec2 from cloudinit import log as logging +from cloudinit.net import dhcp from cloudinit import sources from cloudinit import url_helper as uhelp from cloudinit import util @@ -224,20 +225,28 @@ def get_vr_address(): # Get the address of the virtual router via dhcp leases # If no virtual router is detected, fallback on default gateway. # See http://docs.cloudstack.apache.org/projects/cloudstack-administration/en/4.8/virtual_machines/user-data.html # noqa + + # Try networkd first... + latest_address = dhcp.networkd_get_option_from_leases('SERVER_ADDRESS') + if latest_address: + LOG.debug("Found SERVER_ADDRESS '%s' via networkd_leases", + latest_address) + return latest_address + + # Try dhcp lease files next... lease_file = get_latest_lease() if not lease_file: LOG.debug("No lease file found, using default gateway") return get_default_gateway() - latest_address = None with open(lease_file, "r") as fd: for line in fd: if "dhcp-server-identifier" in line: words = line.strip(" ;\r\n").split(" ") if len(words) > 2: - dhcp = words[2] - LOG.debug("Found DHCP identifier %s", dhcp) - latest_address = dhcp + dhcptok = words[2] + LOG.debug("Found DHCP identifier %s", dhcptok) + latest_address = dhcptok if not latest_address: # No virtual router found, fallback on default gateway LOG.debug("No DHCP found, using default gateway") diff --git a/cloudinit/sources/helpers/azure.py b/cloudinit/sources/helpers/azure.py index 28ed0ae2..959b1bda 100644 --- a/cloudinit/sources/helpers/azure.py +++ b/cloudinit/sources/helpers/azure.py @@ -8,6 +8,7 @@ import socket import struct import time +from cloudinit.net import dhcp from cloudinit import stages from cloudinit import temp_utils from contextlib import contextmanager @@ -15,7 +16,6 @@ from xml.etree import ElementTree from cloudinit import util - LOG = logging.getLogger(__name__) @@ -239,6 +239,11 @@ class WALinuxAgentShim(object): return socket.inet_ntoa(packed_bytes) @staticmethod + def _networkd_get_value_from_leases(leases_d=None): + return dhcp.networkd_get_option_from_leases( + 'OPTION_245', leases_d=leases_d) + + @staticmethod def _get_value_from_leases_file(fallback_lease_file): leases = [] content = util.load_file(fallback_lease_file) @@ -287,12 +292,15 @@ class WALinuxAgentShim(object): @staticmethod def find_endpoint(fallback_lease_file=None): - LOG.debug('Finding Azure endpoint...') value = None - # Option-245 stored in /run/cloud-init/dhclient.hooks/<ifc>.json - # a dhclient exit hook that calls cloud-init-dhclient-hook - dhcp_options = WALinuxAgentShim._load_dhclient_json() - value = WALinuxAgentShim._get_value_from_dhcpoptions(dhcp_options) + LOG.debug('Finding Azure endpoint from networkd...') + value = WALinuxAgentShim._networkd_get_value_from_leases() + if value is None: + # Option-245 stored in /run/cloud-init/dhclient.hooks/<ifc>.json + # a dhclient exit hook that calls cloud-init-dhclient-hook + LOG.debug('Finding Azure endpoint from hook json...') + dhcp_options = WALinuxAgentShim._load_dhclient_json() + value = WALinuxAgentShim._get_value_from_dhcpoptions(dhcp_options) if value is None: # Fallback and check the leases file if unsuccessful LOG.debug("Unable to find endpoint in dhclient logs. " diff --git a/tests/unittests/test_datasource/test_azure_helper.py b/tests/unittests/test_datasource/test_azure_helper.py index 44b99eca..b42b073f 100644 --- a/tests/unittests/test_datasource/test_azure_helper.py +++ b/tests/unittests/test_datasource/test_azure_helper.py @@ -1,10 +1,12 @@ # This file is part of cloud-init. See LICENSE file for license information. import os +from textwrap import dedent from cloudinit.sources.helpers import azure as azure_helper -from cloudinit.tests.helpers import ExitStack, mock, TestCase +from cloudinit.tests.helpers import CiTestCase, ExitStack, mock, populate_dir +from cloudinit.sources.helpers.azure import WALinuxAgentShim as wa_shim GOAL_STATE_TEMPLATE = """\ <?xml version="1.0" encoding="utf-8"?> @@ -45,7 +47,7 @@ GOAL_STATE_TEMPLATE = """\ """ -class TestFindEndpoint(TestCase): +class TestFindEndpoint(CiTestCase): def setUp(self): super(TestFindEndpoint, self).setUp() @@ -56,18 +58,19 @@ class TestFindEndpoint(TestCase): mock.patch.object(azure_helper.util, 'load_file')) self.dhcp_options = patches.enter_context( - mock.patch.object(azure_helper.WALinuxAgentShim, - '_load_dhclient_json')) + mock.patch.object(wa_shim, '_load_dhclient_json')) + + self.networkd_leases = patches.enter_context( + mock.patch.object(wa_shim, '_networkd_get_value_from_leases')) + self.networkd_leases.return_value = None def test_missing_file(self): - self.assertRaises(ValueError, - azure_helper.WALinuxAgentShim.find_endpoint) + self.assertRaises(ValueError, wa_shim.find_endpoint) def test_missing_special_azure_line(self): self.load_file.return_value = '' self.dhcp_options.return_value = {'eth0': {'key': 'value'}} - self.assertRaises(ValueError, - azure_helper.WALinuxAgentShim.find_endpoint) + self.assertRaises(ValueError, wa_shim.find_endpoint) @staticmethod def _build_lease_content(encoded_address): @@ -80,8 +83,7 @@ class TestFindEndpoint(TestCase): def test_from_dhcp_client(self): self.dhcp_options.return_value = {"eth0": {"unknown_245": "5:4:3:2"}} - self.assertEqual('5.4.3.2', - azure_helper.WALinuxAgentShim.find_endpoint(None)) + self.assertEqual('5.4.3.2', wa_shim.find_endpoint(None)) def test_latest_lease_used(self): encoded_addresses = ['5:4:3:2', '4:3:2:1'] @@ -89,53 +91,38 @@ class TestFindEndpoint(TestCase): for encoded_address in encoded_addresses]) self.load_file.return_value = file_content self.assertEqual(encoded_addresses[-1].replace(':', '.'), - azure_helper.WALinuxAgentShim.find_endpoint("foobar")) + wa_shim.find_endpoint("foobar")) -class TestExtractIpAddressFromLeaseValue(TestCase): +class TestExtractIpAddressFromLeaseValue(CiTestCase): def test_hex_string(self): ip_address, encoded_address = '98.76.54.32', '62:4c:36:20' self.assertEqual( - ip_address, - azure_helper.WALinuxAgentShim.get_ip_from_lease_value( - encoded_address - )) + ip_address, wa_shim.get_ip_from_lease_value(encoded_address)) def test_hex_string_with_single_character_part(self): ip_address, encoded_address = '4.3.2.1', '4:3:2:1' self.assertEqual( - ip_address, - azure_helper.WALinuxAgentShim.get_ip_from_lease_value( - encoded_address - )) + ip_address, wa_shim.get_ip_from_lease_value(encoded_address)) def test_packed_string(self): ip_address, encoded_address = '98.76.54.32', 'bL6 ' self.assertEqual( - ip_address, - azure_helper.WALinuxAgentShim.get_ip_from_lease_value( - encoded_address - )) + ip_address, wa_shim.get_ip_from_lease_value(encoded_address)) def test_packed_string_with_escaped_quote(self): ip_address, encoded_address = '100.72.34.108', 'dH\\"l' self.assertEqual( - ip_address, - azure_helper.WALinuxAgentShim.get_ip_from_lease_value( - encoded_address - )) + ip_address, wa_shim.get_ip_from_lease_value(encoded_address)) def test_packed_string_containing_a_colon(self): ip_address, encoded_address = '100.72.58.108', 'dH:l' self.assertEqual( - ip_address, - azure_helper.WALinuxAgentShim.get_ip_from_lease_value( - encoded_address - )) + ip_address, wa_shim.get_ip_from_lease_value(encoded_address)) -class TestGoalStateParsing(TestCase): +class TestGoalStateParsing(CiTestCase): default_parameters = { 'incarnation': 1, @@ -195,7 +182,7 @@ class TestGoalStateParsing(TestCase): self.assertIsNone(certificates_xml) -class TestAzureEndpointHttpClient(TestCase): +class TestAzureEndpointHttpClient(CiTestCase): regular_headers = { 'x-ms-agent-name': 'WALinuxAgent', @@ -258,7 +245,7 @@ class TestAzureEndpointHttpClient(TestCase): self.read_file_or_url.call_args) -class TestOpenSSLManager(TestCase): +class TestOpenSSLManager(CiTestCase): def setUp(self): super(TestOpenSSLManager, self).setUp() @@ -300,7 +287,7 @@ class TestOpenSSLManager(TestCase): self.assertEqual([mock.call(manager.tmpdir)], del_dir.call_args_list) -class TestWALinuxAgentShim(TestCase): +class TestWALinuxAgentShim(CiTestCase): def setUp(self): super(TestWALinuxAgentShim, self).setUp() @@ -310,8 +297,7 @@ class TestWALinuxAgentShim(TestCase): self.AzureEndpointHttpClient = patches.enter_context( mock.patch.object(azure_helper, 'AzureEndpointHttpClient')) self.find_endpoint = patches.enter_context( - mock.patch.object( - azure_helper.WALinuxAgentShim, 'find_endpoint')) + mock.patch.object(wa_shim, 'find_endpoint')) self.GoalState = patches.enter_context( mock.patch.object(azure_helper, 'GoalState')) self.OpenSSLManager = patches.enter_context( @@ -320,7 +306,7 @@ class TestWALinuxAgentShim(TestCase): mock.patch.object(azure_helper.time, 'sleep', mock.MagicMock())) def test_http_client_uses_certificate(self): - shim = azure_helper.WALinuxAgentShim() + shim = wa_shim() shim.register_with_azure_and_fetch_data() self.assertEqual( [mock.call(self.OpenSSLManager.return_value.certificate)], @@ -328,7 +314,7 @@ class TestWALinuxAgentShim(TestCase): def test_correct_url_used_for_goalstate(self): self.find_endpoint.return_value = 'test_endpoint' - shim = azure_helper.WALinuxAgentShim() + shim = wa_shim() shim.register_with_azure_and_fetch_data() get = self.AzureEndpointHttpClient.return_value.get self.assertEqual( @@ -340,7 +326,7 @@ class TestWALinuxAgentShim(TestCase): self.GoalState.call_args_list) def test_certificates_used_to_determine_public_keys(self): - shim = azure_helper.WALinuxAgentShim() + shim = wa_shim() data = shim.register_with_azure_and_fetch_data() self.assertEqual( [mock.call(self.GoalState.return_value.certificates_xml)], @@ -351,13 +337,13 @@ class TestWALinuxAgentShim(TestCase): def test_absent_certificates_produces_empty_public_keys(self): self.GoalState.return_value.certificates_xml = None - shim = azure_helper.WALinuxAgentShim() + shim = wa_shim() data = shim.register_with_azure_and_fetch_data() self.assertEqual([], data['public-keys']) def test_correct_url_used_for_report_ready(self): self.find_endpoint.return_value = 'test_endpoint' - shim = azure_helper.WALinuxAgentShim() + shim = wa_shim() shim.register_with_azure_and_fetch_data() expected_url = 'http://test_endpoint/machine?comp=health' self.assertEqual( @@ -368,7 +354,7 @@ class TestWALinuxAgentShim(TestCase): self.GoalState.return_value.incarnation = 'TestIncarnation' self.GoalState.return_value.container_id = 'TestContainerId' self.GoalState.return_value.instance_id = 'TestInstanceId' - shim = azure_helper.WALinuxAgentShim() + shim = wa_shim() shim.register_with_azure_and_fetch_data() posted_document = ( self.AzureEndpointHttpClient.return_value.post.call_args[1]['data'] @@ -378,11 +364,11 @@ class TestWALinuxAgentShim(TestCase): self.assertIn('TestInstanceId', posted_document) def test_clean_up_can_be_called_at_any_time(self): - shim = azure_helper.WALinuxAgentShim() + shim = wa_shim() shim.clean_up() def test_clean_up_will_clean_up_openssl_manager_if_instantiated(self): - shim = azure_helper.WALinuxAgentShim() + shim = wa_shim() shim.register_with_azure_and_fetch_data() shim.clean_up() self.assertEqual( @@ -393,12 +379,12 @@ class TestWALinuxAgentShim(TestCase): pass self.AzureEndpointHttpClient.return_value.get.side_effect = ( SentinelException) - shim = azure_helper.WALinuxAgentShim() + shim = wa_shim() self.assertRaises(SentinelException, shim.register_with_azure_and_fetch_data) -class TestGetMetadataFromFabric(TestCase): +class TestGetMetadataFromFabric(CiTestCase): @mock.patch.object(azure_helper, 'WALinuxAgentShim') def test_data_from_shim_returned(self, shim): @@ -422,4 +408,65 @@ class TestGetMetadataFromFabric(TestCase): azure_helper.get_metadata_from_fabric) self.assertEqual(1, shim.return_value.clean_up.call_count) + +class TestExtractIpAddressFromNetworkd(CiTestCase): + + azure_lease = dedent("""\ + # This is private data. Do not parse. + ADDRESS=10.132.0.5 + NETMASK=255.255.255.255 + ROUTER=10.132.0.1 + SERVER_ADDRESS=169.254.169.254 + NEXT_SERVER=10.132.0.1 + MTU=1460 + T1=43200 + T2=75600 + LIFETIME=86400 + DNS=169.254.169.254 + NTP=169.254.169.254 + DOMAINNAME=c.ubuntu-foundations.internal + DOMAIN_SEARCH_LIST=c.ubuntu-foundations.internal google.internal + HOSTNAME=tribaal-test-171002-1349.c.ubuntu-foundations.internal + ROUTES=10.132.0.1/32,0.0.0.0 0.0.0.0/0,10.132.0.1 + CLIENTID=ff405663a200020000ab11332859494d7a8b4c + OPTION_245=624c3620 + """) + + def setUp(self): + super(TestExtractIpAddressFromNetworkd, self).setUp() + self.lease_d = self.tmp_dir() + + def test_no_valid_leases_is_none(self): + """No valid leases should return None.""" + self.assertIsNone( + wa_shim._networkd_get_value_from_leases(self.lease_d)) + + def test_option_245_is_found_in_single(self): + """A single valid lease with 245 option should return it.""" + populate_dir(self.lease_d, {'9': self.azure_lease}) + self.assertEqual( + '624c3620', wa_shim._networkd_get_value_from_leases(self.lease_d)) + + def test_option_245_not_found_returns_None(self): + """A valid lease, but no option 245 should return None.""" + populate_dir( + self.lease_d, + {'9': self.azure_lease.replace("OPTION_245", "OPTION_999")}) + self.assertIsNone( + wa_shim._networkd_get_value_from_leases(self.lease_d)) + + def test_multiple_returns_first(self): + """Somewhat arbitrarily return the first address when multiple. + + Most important at the moment is that this is consistent behavior + rather than changing randomly as in order of a dictionary.""" + myval = "624c3601" + populate_dir( + self.lease_d, + {'9': self.azure_lease, + '2': self.azure_lease.replace("624c3620", myval)}) + self.assertEqual( + myval, wa_shim._networkd_get_value_from_leases(self.lease_d)) + + # vi: ts=4 expandtab diff --git a/tests/unittests/test_datasource/test_cloudstack.py b/tests/unittests/test_datasource/test_cloudstack.py index 8e98e1bb..96144b64 100644 --- a/tests/unittests/test_datasource/test_cloudstack.py +++ b/tests/unittests/test_datasource/test_cloudstack.py @@ -23,13 +23,16 @@ class TestCloudStackPasswordFetching(CiTestCase): default_gw = "192.201.20.0" get_latest_lease = mock.MagicMock(return_value=None) self.patches.enter_context(mock.patch( - 'cloudinit.sources.DataSourceCloudStack.get_latest_lease', - get_latest_lease)) + mod_name + '.get_latest_lease', get_latest_lease)) get_default_gw = mock.MagicMock(return_value=default_gw) self.patches.enter_context(mock.patch( - 'cloudinit.sources.DataSourceCloudStack.get_default_gateway', - get_default_gw)) + mod_name + '.get_default_gateway', get_default_gw)) + + get_networkd_server_address = mock.MagicMock(return_value=None) + self.patches.enter_context(mock.patch( + mod_name + '.dhcp.networkd_get_option_from_leases', + get_networkd_server_address)) def _set_password_server_response(self, response_string): subp = mock.MagicMock(return_value=(response_string, '')) |