x/cloudbase-init
Code Issues Proposed changes
cloudbase-init/cloudbaseinit/tests/metadata/services/test_baseopenstackservice.py
Alessandro Pilotti 1cdf43ce23 Add get_network_details_v2 in BaseOpenStackService 
Partially-Implements: blueprint json-network-config
Change-Id: Ia2a31518f87f34c234988fc512f1698617efdb10
Co-Authored-By: Adrian Vladu <avladu@cloudbasesolutions.com>
2018-09-13 17:59:16 +03:00

491 lines
20 KiB
Python
Raw Blame History

# Copyright 2014 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 functools
import posixpath
import unittest
import netaddr
try:
import unittest.mock as mock
except ImportError:
import mock
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit import exception
from cloudbaseinit.metadata.services import base
from cloudbaseinit.metadata.services import baseopenstackservice
from cloudbaseinit.models import network as network_model
from cloudbaseinit.tests.metadata import fake_json_response
from cloudbaseinit.utils import x509constants
CONF = cloudbaseinit_conf.CONF
MODPATH = "cloudbaseinit.metadata.services.baseopenstackservice"
class FinalBaseOpenStackService(baseopenstackservice.BaseOpenStackService):
def _get_data(self):
pass
class TestBaseOpenStackService(unittest.TestCase):
def setUp(self):
CONF.set_override("retry_count_interval", 0)
self._service = FinalBaseOpenStackService()
date = "2013-04-04"
fake_metadata = fake_json_response.get_fake_metadata_json(date)
self._fake_network_config = fake_metadata["network_config"]
self._fake_content = self._fake_network_config["debian_config"]
self._fake_public_keys = fake_metadata["public_keys"]
self._fake_keys = fake_metadata["keys"]
self._partial_test_get_network_details = functools.partial(
self._test_get_network_details,
network_config=self._fake_network_config,
content=self._fake_content
)
@mock.patch(MODPATH +
".BaseOpenStackService._get_cache_data")
def test_get_content(self, mock_get_cache_data):
response = self._service.get_content('fake name')
path = posixpath.join('openstack', 'content', 'fake name')
mock_get_cache_data.assert_called_once_with(path)
self.assertEqual(mock_get_cache_data.return_value, response)
@mock.patch(MODPATH +
".BaseOpenStackService._get_cache_data")
def test_get_user_data(self, mock_get_cache_data):
response = self._service.get_user_data()
path = posixpath.join('openstack', 'latest', 'user_data')
mock_get_cache_data.assert_called_once_with(path)
self.assertEqual(mock_get_cache_data.return_value, response)
@mock.patch(MODPATH +
".BaseOpenStackService._get_cache_data")
def test_get_meta_data(self, mock_get_cache_data):
mock_get_cache_data.return_value = '{"fake": "data"}'
response = self._service._get_meta_data(
version='fake version')
path = posixpath.join('openstack', 'fake version', 'meta_data.json')
mock_get_cache_data.assert_called_with(path, decode=True)
self.assertEqual({"fake": "data"}, response)
@mock.patch(MODPATH +
".BaseOpenStackService._get_meta_data")
def test_get_instance_id(self, mock_get_meta_data):
response = self._service.get_instance_id()
mock_get_meta_data.assert_called_once_with()
mock_get_meta_data().get.assert_called_once_with('uuid')
self.assertEqual(mock_get_meta_data.return_value.get.return_value,
response)
@mock.patch(MODPATH +
".BaseOpenStackService._get_meta_data")
def test_get_host_name(self, mock_get_meta_data):
response = self._service.get_host_name()
mock_get_meta_data.assert_called_once_with()
mock_get_meta_data().get.assert_called_once_with('hostname')
self.assertEqual(mock_get_meta_data.return_value.get.return_value,
response)
@mock.patch(MODPATH +
".BaseOpenStackService._get_meta_data")
def test_get_public_keys(self, mock_get_meta_data):
mock_get_meta_data.return_value.get.side_effect = \
[self._fake_public_keys, self._fake_keys]
response = self._service.get_public_keys()
mock_get_meta_data.assert_called_once_with()
mock_get_meta_data.return_value.get.assert_any_call("public_keys")
mock_get_meta_data.return_value.get.assert_any_call("keys")
public_keys = (list(self._fake_public_keys.values()) +
[key["data"] for key in self._fake_keys
if key["type"] == "ssh"])
public_keys = [key.strip() for key in public_keys]
self.assertEqual(sorted(list(set(public_keys))),
sorted(response))
@mock.patch(MODPATH +
".BaseOpenStackService._get_meta_data")
def _test_get_admin_password(self, mock_get_meta_data, meta_data):
mock_get_meta_data.return_value = meta_data
response = self._service.get_admin_password()
mock_get_meta_data.assert_called_once_with()
if meta_data and 'admin_pass' in meta_data:
self.assertEqual(meta_data['admin_pass'], response)
elif meta_data and 'admin_pass' in meta_data.get('meta'):
self.assertEqual(meta_data.get('meta')['admin_pass'], response)
else:
self.assertIsNone(response)
def test_get_admin_pass(self):
self._test_get_admin_password(meta_data={'admin_pass': 'fake pass'})
def test_get_admin_pass_in_meta(self):
self._test_get_admin_password(
meta_data={'meta': {'admin_pass': 'fake pass'}})
def test_get_admin_pass_no_pass(self):
self._test_get_admin_password(meta_data={})
@mock.patch(MODPATH +
".BaseOpenStackService._get_meta_data")
@mock.patch(MODPATH +
".BaseOpenStackService.get_user_data")
def _test_get_client_auth_certs(self, mock_get_user_data,
mock_get_meta_data, meta_data,
ret_value=None):
mock_get_meta_data.return_value = meta_data
mock_get_user_data.side_effect = [ret_value]
response = self._service.get_client_auth_certs()
mock_get_meta_data.assert_called_once_with()
if isinstance(ret_value, bytes) and ret_value.startswith(
x509constants.PEM_HEADER.encode()):
mock_get_user_data.assert_called_once_with()
self.assertEqual([ret_value.decode()], response)
elif ret_value is base.NotExistingMetadataException:
self.assertFalse(response)
else:
expected = []
expectation = {
"meta": 'fake cert',
"keys": [key["data"].strip() for key in self._fake_keys
if key["type"] == "x509"]
}
for field, value in expectation.items():
if field in meta_data:
expected.extend(value if isinstance(value, list)
else [value])
self.assertEqual(sorted(list(set(expected))), sorted(response))
def test_get_client_auth_certs(self):
self._test_get_client_auth_certs(
meta_data={'meta': {'admin_cert0': 'fake ',
'admin_cert1': 'cert'},
"keys": self._fake_keys})
def test_get_client_auth_certs_no_cert_data(self):
self._test_get_client_auth_certs(
meta_data={}, ret_value=x509constants.PEM_HEADER.encode())
def test_get_client_auth_certs_no_cert_data_exception(self):
self._test_get_client_auth_certs(
meta_data={}, ret_value=base.NotExistingMetadataException)
@mock.patch(MODPATH +
".BaseOpenStackService.get_content")
@mock.patch(MODPATH +
".BaseOpenStackService._get_meta_data")
def _test_get_network_details(self,
mock_get_meta_data,
mock_get_content,
network_config=None,
content=None,
search_fail=False,
no_path=False):
# mock obtained data
mock_get_meta_data().get.return_value = network_config
mock_get_content.return_value = content
# actual tests
if search_fail:
ret = self._service.get_network_details()
self.assertFalse(ret)
return
ret = self._service.get_network_details()
mock_get_meta_data().get.assert_called_once_with("network_config")
if network_config and not no_path:
mock_get_content.assert_called_once_with("network")
if not network_config:
self.assertIsNone(ret)
return
if no_path:
self.assertIsNone(ret)
return
# check returned NICs details
nic0 = network_model.NetworkDetails(
fake_json_response.NAME0,
fake_json_response.MAC0.upper(),
fake_json_response.ADDRESS0,
fake_json_response.ADDRESS60,
fake_json_response.NETMASK0,
fake_json_response.NETMASK60,
fake_json_response.BROADCAST0,
fake_json_response.GATEWAY0,
fake_json_response.GATEWAY60,
fake_json_response.DNSNS0.split()
)
nic1 = network_model.NetworkDetails(
fake_json_response.NAME1,
None,
fake_json_response.ADDRESS1,
fake_json_response.ADDRESS61,
fake_json_response.NETMASK1,
fake_json_response.NETMASK61,
fake_json_response.BROADCAST1,
fake_json_response.GATEWAY1,
fake_json_response.GATEWAY61,
None
)
self.assertEqual([nic0, nic1], ret)
def test_get_network_details_no_config(self):
self._partial_test_get_network_details(
network_config=None
)
def test_get_network_details_no_path(self):
self._fake_network_config.pop("content_path", None)
self._partial_test_get_network_details(
network_config=self._fake_network_config,
no_path=True
)
def test_get_network_details_search_fail(self):
self._fake_content = "invalid format"
self._partial_test_get_network_details(
content=self._fake_content,
search_fail=True
)
def test_get_network_details(self):
self._partial_test_get_network_details()
@staticmethod
def _get_network_data():
return {
"links": [{
"ethernet_mac_address": mock.sentinel.link_mac1,
"type": baseopenstackservice.NETWORK_LINK_TYPE_PHYSICAL,
"id": mock.sentinel.link_id1,
"mtu": mock.sentinel.link_mtu1,
}, {
"ethernet_mac_address": mock.sentinel.link_mac2,
"type": mock.sentinel.another_link_type,
"id": mock.sentinel.link_id2,
"mtu": mock.sentinel.link_mtu2,
}, {
"bond_miimon": mock.sentinel.bond_miimon1,
"bond_xmit_hash_policy": mock.sentinel.bond_lb_algo1,
"ethernet_mac_address": mock.sentinel.bond_mac1,
"mtu": mock.sentinel.bond_mtu1,
"bond_mode": mock.sentinel.bond_type1,
"bond_links": [
mock.sentinel.link_id1,
mock.sentinel.link_id2,
],
"type": baseopenstackservice.NETWORK_LINK_TYPE_BOND,
"id": mock.sentinel.bond_id1,
}, {
"id": mock.sentinel.vlan_link_id1,
"type": baseopenstackservice.NETWORK_LINK_TYPE_VLAN,
"vlan_link": mock.sentinel.bond_id1,
"vlan_id": mock.sentinel.vlan_id1,
"mtu": mock.sentinel.vlan_mtu1,
"ethernet_mac_address": mock.sentinel.vlan_mac1,
}],
"networks": [{
"id": mock.sentinel.network_id1,
"network_id": mock.sentinel.network_openstack_id1,
"link": mock.sentinel.bond_id1,
"type": baseopenstackservice.NETWORK_TYPE_IPV4_DHCP,
}, {
"id": mock.sentinel.network_id2,
"type": baseopenstackservice.NETWORK_TYPE_IPV4,
"link": mock.sentinel.bond_id1,
"ip_address": mock.sentinel.ip_address1,
"netmask": mock.sentinel.netmask1,
"services": [{
"type": baseopenstackservice.NETWORK_SERVICE_TYPE_DNS,
"address": mock.sentinel.dns1,
}, {
"type": baseopenstackservice.NETWORK_SERVICE_TYPE_DNS,
"address": mock.sentinel.dns2
}],
"routes": [{
"network": mock.sentinel.route_network1,
"netmask": mock.sentinel.route_netmask1,
"gateway": mock.sentinel.route_gateway1,
}, {
"network": mock.sentinel.route_network2,
"netmask": mock.sentinel.route_netmask2,
"gateway": mock.sentinel.route_gateway2,
}],
"network_id": mock.sentinel.network_openstack_id2
}, {
"id": mock.sentinel.network_id3,
"type": baseopenstackservice.NETWORK_TYPE_IPV6,
"link": mock.sentinel.bond_id1,
"ip_address": mock.sentinel.ip_address_ipv61,
"routes": [{
"network": mock.sentinel.route_network_ipv61,
"gateway": mock.sentinel.route_gateway_ipv61,
}],
"network_id": mock.sentinel.network_openstack_id3
}],
"services": [{
"type": baseopenstackservice.NETWORK_SERVICE_TYPE_DNS,
"address": mock.sentinel.dns3,
}, {
"type": baseopenstackservice.NETWORK_SERVICE_TYPE_DNS,
"address": mock.sentinel.dns4
}],
}
@mock.patch(MODPATH + ".BaseOpenStackService._get_network_data")
def _test_get_network_details_v2(self, mock_get_network_data,
invalid_bond_type=False,
invalid_bond_lb_algo=False):
mock.sentinel.ip_address1 = "10.0.0.1"
mock.sentinel.netmask1 = "255.255.255.0"
mock.sentinel.route_network1 = "172.16.0.0"
mock.sentinel.route_netmask1 = "255.255.0.0"
mock.sentinel.route_gateway1 = "172.16.1.1"
mock.sentinel.route_network2 = "0.0.0.0"
mock.sentinel.route_netmask2 = "0.0.0.0"
mock.sentinel.route_gateway2 = "10.0.0.254"
mock.sentinel.ip_address_ipv61 = "2001:cdba::3257:9652/24"
mock.sentinel.route_network_ipv61 = "::/0"
mock.sentinel.route_gateway_ipv61 = "fd00::1"
if invalid_bond_type:
mock.sentinel.bond_type1 = "invalid bond type"
else:
mock.sentinel.bond_type1 = network_model.BOND_TYPE_ACTIVE_BACKUP
if invalid_bond_lb_algo:
mock.sentinel.bond_lb_algo1 = "invalid lb algorithm"
else:
mock.sentinel.bond_lb_algo1 = network_model.BOND_LB_ALGO_L2
network_data = self._get_network_data()
mock_get_network_data.return_value = network_data
if invalid_bond_type or invalid_bond_lb_algo:
with self.assertRaises(exception.CloudbaseInitException):
self._service.get_network_details_v2()
return
network_details = self._service.get_network_details_v2()
self.assertEqual(
len(network_data["links"]), len(network_details.links))
self.assertEqual(1, len([
l for l in network_details.links if
l.type == network_model.LINK_TYPE_PHYSICAL and
l.id == mock.sentinel.link_id1 and
l.name == mock.sentinel.link_id1 and
l.mac_address == mock.sentinel.link_mac1 and
l.mtu == mock.sentinel.link_mtu1]))
self.assertEqual(1, len([
l for l in network_details.links if
l.type == network_model.LINK_TYPE_PHYSICAL and
l.id == mock.sentinel.link_id2 and
l.name == mock.sentinel.link_id2 and
l.mac_address == mock.sentinel.link_mac2 and
l.mtu == mock.sentinel.link_mtu2]))
self.assertEqual(1, len([
l for l in network_details.links if
l.type == network_model.LINK_TYPE_BOND and
l.id == mock.sentinel.bond_id1 and
l.name == mock.sentinel.bond_id1 and
l.mtu == mock.sentinel.bond_mtu1 and
l.mac_address == mock.sentinel.bond_mac1 and
l.vlan_link is None and
l.vlan_id is None and
l.bond.type == network_model.BOND_TYPE_ACTIVE_BACKUP and
l.bond.members == [
mock.sentinel.link_id1, mock.sentinel.link_id2] and
l.bond.lb_algorithm == network_model.BOND_LB_ALGO_L2 and
l.bond.lacp_rate is None]))
self.assertEqual(1, len([
l for l in network_details.links if
l.type == network_model.LINK_TYPE_VLAN and
l.id == mock.sentinel.vlan_link_id1 and
l.name == mock.sentinel.vlan_link_id1 and
l.mac_address == mock.sentinel.vlan_mac1 and
l.mtu == mock.sentinel.vlan_mtu1 and
l.vlan_link == mock.sentinel.bond_id1 and
l.vlan_id == mock.sentinel.vlan_id1]))
self.assertEqual(
len([n for n in network_data["networks"]
if n["type"] in [
baseopenstackservice.NETWORK_TYPE_IPV4,
baseopenstackservice.NETWORK_TYPE_IPV6]]),
len(network_details.networks))
def _get_cidr_address(ip_address, netmask):
prefix_len = netaddr.IPNetwork(
u"%s/%s" % (ip_address, netmask)).prefixlen
return u"%s/%s" % (ip_address, prefix_len)
address_cidr = _get_cidr_address(
mock.sentinel.ip_address1, mock.sentinel.netmask1)
network = [
n for n in network_details.networks
if n.address_cidr == address_cidr and
n.dns_nameservers == [mock.sentinel.dns1, mock.sentinel.dns2] and
n.link == mock.sentinel.bond_id1]
self.assertEqual(1, len(network))
network_cidr1 = _get_cidr_address(
mock.sentinel.route_network1, mock.sentinel.route_netmask1)
network_cidr2 = _get_cidr_address(
mock.sentinel.route_network2, mock.sentinel.route_netmask2)
self.assertEqual([
network_model.Route(
network_cidr=network_cidr1,
gateway=mock.sentinel.route_gateway1),
network_model.Route(
network_cidr=network_cidr2,
gateway=mock.sentinel.route_gateway2)],
network[0].routes)
network_ipv6 = [
n for n in network_details.networks
if n.address_cidr == mock.sentinel.ip_address_ipv61 and
n.link == mock.sentinel.bond_id1]
self.assertEqual(1, len(network_ipv6))
self.assertEqual(
[network_model.NameServerService(
addresses=[mock.sentinel.dns3, mock.sentinel.dns4],
search=None)],
network_details.services)
def test_get_network_details_v2(self):
self._test_get_network_details_v2()
def test_get_network_details_v2_invalid_bond_type(self):
self._test_get_network_details_v2(invalid_bond_type=True)
def test_get_network_details_v2_invalid_bond_lb_algo(self):
self._test_get_network_details_v2(invalid_bond_lb_algo=True)
View Git Blame Copy Permalink