x/cloudbase-init
Code Issues Proposed changes
cloudbase-init/cloudbaseinit/tests/utils/test_network.py
Zhongcheng Lao dc0367425a Support datasource specific instance metadata 
This change aligns the instance metadata with cloudinit to
add the support for datasource specific instance metadata.

The datasource specific instance metadata allows more
information to be exposed to the Jinja template for userdata
and script.

Note:
The structure is not standarized but you may refer to
https://cloudinit.readthedocs.io/en/latest/explanation/instancedata.html
to the cloudinit format.

Change-Id: I1ec7e5bdf063709c513b52a02c9251752aafe157
2024-11-27 12:29:50 +00:00

347 lines
12 KiB
Python
Raw Blame History

# Copyright 2013 Cloudbase Solutions Srl
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import collections
import unittest
import unittest.mock as mock
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit.models import network as network_model
from cloudbaseinit.tests import testutils
from cloudbaseinit.utils import network
CONF = cloudbaseinit_conf.CONF
class NetworkUtilsTest(unittest.TestCase):
link0 = network_model.Link(
id="eth0",
name="eth0",
type=network_model.LINK_TYPE_PHYSICAL,
mac_address="ab:cd:ef:ef:cd:ab",
enabled=None,
mtu=None,
bond=None,
vlan_link=None,
vlan_id=None,
)
network_private_default_route = network_model.Network(
link="eth0",
address_cidr="192.168.1.2/24",
routes=[
network_model.Route(
network_cidr=u"0.0.0.0/0",
gateway="192.168.1.1",
),
],
dns_nameservers=[]
)
network_public = network_model.Network(
link="eth0",
address_cidr="2.3.4.2/24",
routes=[
network_model.Route(
network_cidr=u"2.3.4.1/24",
gateway="2.3.4.1",
),
],
dns_nameservers=[]
)
network_public_default_route = network_model.Network(
link="eth0",
address_cidr="2.3.4.2/24",
routes=[
network_model.Route(
network_cidr=u"0.0.0.0/0",
gateway="2.3.4.1",
),
],
dns_nameservers=[]
)
network_private = network_model.Network(
link="eth0",
address_cidr="172.10.1.2/24",
routes=[
network_model.Route(
network_cidr=u"172.10.1.1/24",
gateway="172.10.1.1",
),
],
dns_nameservers=[]
)
network_local = network_model.Network(
link="eth0",
address_cidr="127.0.0.4/24",
routes=[
network_model.Route(
network_cidr=u"127.0.0.4/24",
gateway="127.0.0.2",
),
],
dns_nameservers=[]
)
ipv6_addr = '1a8f:9aaf:2904:858f:1bce:6f85:2b04:f38'
network_v6 = network_model.Network(
link="eth0",
address_cidr=ipv6_addr + '/64',
routes=[network_model.Route(
network_cidr=u"::/0",
gateway="::1",
)],
dns_nameservers=[]
)
ipv6_addr_private = 'fe80::216:3eff:fe16:db54'
network_v6_local = network_model.Network(
link="eth0",
address_cidr=ipv6_addr_private + '/64',
routes=[],
dns_nameservers=[]
)
@mock.patch('urllib.request.urlopen')
def test_check_url(self, mock_url_open):
mock_url_open.return_value = None
self.assertTrue(network.check_url("fake_url"))
@mock.patch('sys.platform', new='win32')
@mock.patch('cloudbaseinit.osutils.factory.get_os_utils')
@mock.patch('urllib.parse.urlparse')
def _test_check_metadata_ip_route(self, mock_urlparse, mock_get_os_utils,
side_effect):
mock_utils = mock.MagicMock()
mock_split = mock.MagicMock()
mock_get_os_utils.return_value = mock_utils
mock_utils.check_os_version.return_value = True
mock_urlparse().netloc.split.return_value = mock_split
mock_split[0].startswith.return_value = True
mock_utils.check_static_route_exists.return_value = False
mock_utils.get_default_gateway.return_value = (1, '0.0.0.0')
mock_utils.add_static_route.side_effect = [side_effect]
network.check_metadata_ip_route('196.254.196.254')
mock_utils.check_os_version.assert_called_once_with(6, 0)
mock_urlparse.assert_called_with('196.254.196.254')
mock_split[0].startswith.assert_called_once_with("169.254.")
mock_utils.check_static_route_exists.assert_called_once_with(
mock_split[0])
mock_utils.get_default_gateway.assert_called_once_with()
mock_utils.add_static_route.assert_called_once_with(
mock_split[0], "255.255.255.255", '0.0.0.0', 1, 10)
def test_test_check_metadata_ip_route(self):
self._test_check_metadata_ip_route(side_effect=None)
def test_test_check_metadata_ip_route_fail(self):
with testutils.LogSnatcher('cloudbaseinit.utils.network') as snatcher:
self._test_check_metadata_ip_route(side_effect=ValueError)
self.assertIn('ValueError', snatcher.output[-1])
def test_address6_to_4_truncate(self):
address_map = {
"0:0:0:0:0:ffff:c0a8:f": "192.168.0.15",
"::ffff:c0a8:e": "192.168.0.14",
"::1": "0.0.0.1",
"1:2:3:4:5::8": "0.0.0.8",
"::": "0.0.0.0",
"::7f00:1": "127.0.0.1"
}
for v6, v4 in address_map.items():
self.assertEqual(v4, network.address6_to_4_truncate(v6))
def test_netmask6_to_4_truncate(self):
netmask_map = {
"128": "255.255.255.255",
"96": "255.255.255.0",
"0": "0.0.0.0",
"100": "255.255.255.128"
}
for v6, v4 in netmask_map.items():
self.assertEqual(v4, network.netmask6_to_4_truncate(v6))
@mock.patch('socket.socket')
def test_get_local_ip(self, mock_socket):
mock_socket.return_value = mock.Mock()
mock_socket().getsockname.return_value = ["fake name"]
res = network.get_local_ip("fake address")
self.assertEqual(res, "fake name")
mock_socket().connect.assert_called_with(("fake address", 8000))
def _test_ip_netmask_to_cidr(self, expected_result, fake_ip_address,
fake_netmask):
result = network.ip_netmask_to_cidr(fake_ip_address, fake_netmask)
self.assertEqual(expected_result, result)
def test_ip_netmask_to_cidr(self):
fake_ip_address = '10.1.1.1'
expected_result = '10.1.1.1/24'
fake_netmask = '255.255.255.0'
self._test_ip_netmask_to_cidr(expected_result, fake_ip_address,
fake_netmask)
def test_ip_netmask_to_cidr_empty_netmask(self):
fake_ip_address = '10.1.1.1'
fake_netmask = None
self._test_ip_netmask_to_cidr(fake_ip_address, fake_ip_address,
fake_netmask)
def test_get_default_ip_addresses(self):
network_details = network_model.NetworkDetailsV2(
links=[
self.link0
],
networks=[
self.network_private_default_route,
],
services=[
],
)
ipv4, ipv6 = network.get_default_ip_addresses(network_details)
self.assertEqual('192.168.1.2', ipv4)
self.assertIsNone(ipv6)
def test_get_default_ip_addresses_link_local(self):
network_details = network_model.NetworkDetailsV2(
links=[
self.link0
],
networks=[
self.network_private,
],
services=[
],
)
ipv4, ipv6 = network.get_default_ip_addresses(network_details)
self.assertIsNone(ipv4)
self.assertIsNone(ipv6)
def test_get_default_ip_addresses_public_default_route(self):
network_details = network_model.NetworkDetailsV2(
links=[
self.link0
],
networks=[
self.network_public_default_route,
],
services=[
],
)
ipv4, ipv6 = network.get_default_ip_addresses(network_details)
self.assertEqual("2.3.4.2", ipv4)
self.assertIsNone(ipv6)
def test_get_default_ip_addresses_v6(self):
network_details = network_model.NetworkDetailsV2(
links=[
self.link0
],
networks=[
self.network_v6,
],
services=[
],
)
ipv4, ipv6 = network.get_default_ip_addresses(network_details)
self.assertIsNone(ipv4)
self.assertEqual(self.ipv6_addr, ipv6)
def test_get_default_ip_addresses_v6_local(self):
network_details = network_model.NetworkDetailsV2(
links=[
self.link0
],
networks=[
self.network_v6_local,
],
services=[
],
)
ipv4, ipv6 = network.get_default_ip_addresses(network_details)
self.assertIsNone(ipv4)
self.assertIsNone(ipv6)
def test_get_default_ip_addresses_dual_stack(self):
network_details = network_model.NetworkDetailsV2(
links=[
self.link0
],
networks=[
self.network_private_default_route,
self.network_public,
self.network_v6,
],
services=[
],
)
ipv4, ipv6 = network.get_default_ip_addresses(network_details)
self.assertEqual('192.168.1.2', ipv4)
self.assertEqual(self.ipv6_addr, ipv6)
def test_get_host_info(self):
network_details = network_model.NetworkDetailsV2(
links=[
self.link0
],
networks=[
self.network_private_default_route,
self.network_public, self.network_v6,
],
services=[
],
)
expect = {
'hostname': 'fake_host',
'local-hostname': 'fake_host',
'local-ipv4': '192.168.1.2',
'local_ipv4': '192.168.1.2',
'local-ipv6': '1a8f:9aaf:2904:858f:1bce:6f85:2b04:f38',
'local_ipv6': '1a8f:9aaf:2904:858f:1bce:6f85:2b04:f38',
'local_hostname': 'fake_host',
'network': {'interfaces': {
'by-ipv4': collections.OrderedDict([
('192.168.1.2',
{'broadcast': '192.168.1.255',
'mac': 'ab:cd:ef:ef:cd:ab',
'netmask': '255.255.255.0'}
),
('2.3.4.2',
{'broadcast': '2.3.4.255',
'mac': 'ab:cd:ef:ef:cd:ab',
'netmask': '255.255.255.0'}
)
]),
'by-ipv6': collections.OrderedDict([
('1a8f:9aaf:2904:858f:1bce:6f85:2b04:f38',
{'broadcast': '1a8f:9aaf:2904:858f:ffff:ffff:ffff:ffff',
'mac': 'ab:cd:ef:ef:cd:ab',
}
)
]),
'by-mac': collections.OrderedDict([
('ab:cd:ef:ef:cd:ab',
{'ipv4': {'addr': '2.3.4.2',
'broadcast': '2.3.4.255',
'netmask': '255.255.255.0'},
'ipv6': {
'addr': '1a8f:9aaf:2904:858f:1bce:6f85:2b04:f38',
'broadcast': '1a8f:9aaf:2904:858f:'
'ffff:ffff:ffff:ffff'}
}
),
])
}}
}
host_info = network.get_host_info('fake_host', network_details)
self.assertEqual(expect, host_info)
View Git Blame Copy Permalink