248 lines
8.6 KiB
Python
248 lines
8.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# 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 os
|
|
|
|
import etcd
|
|
import mock
|
|
|
|
from fuel_ccp_entrypoint import start_script
|
|
from fuel_ccp_entrypoint.tests import base
|
|
|
|
|
|
class TestGetIpAddress(base.TestCase):
|
|
|
|
def setUp(self):
|
|
super(TestGetIpAddress, self).setUp()
|
|
self.private_iface = 'eth0'
|
|
self.public_iface = 'eth1'
|
|
|
|
@mock.patch('netifaces.interfaces')
|
|
@mock.patch('netifaces.ifaddresses')
|
|
def test_get_ip_address_iface_wrong(self, m_ifaddresses, m_interfaces):
|
|
m_interfaces.return_value = ['eth10', 'eth99']
|
|
r_value = start_script.get_ip_address(self.private_iface)
|
|
self.assertEqual('127.0.0.1', r_value)
|
|
self.assertEqual(1, len(m_interfaces.mock_calls))
|
|
self.assertEqual(0, len(m_ifaddresses.mock_calls))
|
|
|
|
@mock.patch('netifaces.interfaces')
|
|
@mock.patch('netifaces.ifaddresses')
|
|
def test_get_ip_address_address_family_wrong(self, m_ifaddresses,
|
|
m_interfaces):
|
|
m_interfaces.return_value = ['eth0', 'eth99']
|
|
m_ifaddresses.return_value = {3: [{"addr": "8.8.8.8"}]}
|
|
r_value = start_script.get_ip_address(self.private_iface)
|
|
self.assertEqual('127.0.0.1', r_value)
|
|
self.assertEqual(1, len(m_interfaces.mock_calls))
|
|
self.assertEqual(1, len(m_ifaddresses.mock_calls))
|
|
|
|
@mock.patch('netifaces.interfaces')
|
|
@mock.patch('netifaces.ifaddresses')
|
|
def test_get_ip_address_address_wrong(self, m_ifaddresses, m_interfaces):
|
|
m_interfaces.return_value = ['eth0', 'eth99']
|
|
m_ifaddresses.return_value = {2: [{"notaddr": "8.8.8.8"}]}
|
|
r_value = start_script.get_ip_address(self.private_iface)
|
|
self.assertEqual('127.0.0.1', r_value)
|
|
self.assertEqual(1, len(m_interfaces.mock_calls))
|
|
self.assertEqual(2, len(m_ifaddresses.mock_calls))
|
|
|
|
@mock.patch('netifaces.interfaces')
|
|
@mock.patch('netifaces.ifaddresses')
|
|
def test_get_ip_address_address_good(self, m_ifaddresses, m_interfaces):
|
|
m_interfaces.return_value = ['eth0', 'eth99']
|
|
m_ifaddresses.return_value = {2: [{"addr": "8.8.8.8"}]}
|
|
r_value = start_script.get_ip_address(self.private_iface)
|
|
self.assertEqual('8.8.8.8', r_value)
|
|
self.assertEqual(1, len(m_interfaces.mock_calls))
|
|
self.assertEqual(2, len(m_ifaddresses.mock_calls))
|
|
|
|
|
|
class TestGetVariables(base.TestCase):
|
|
|
|
def setUp(self):
|
|
super(TestGetVariables, self).setUp()
|
|
os.environ['CCP_NODE_NAME'] = 'node1'
|
|
os.environ['CCP_POD_NAME'] = 'pod1'
|
|
|
|
def tearDown(self):
|
|
super(TestGetVariables, self).tearDown()
|
|
del os.environ['CCP_NODE_NAME']
|
|
del os.environ['CCP_POD_NAME']
|
|
|
|
@mock.patch('six.moves.builtins.open', mock.mock_open())
|
|
@mock.patch('json.load')
|
|
@mock.patch('fuel_ccp_entrypoint.start_script.create_network_topology')
|
|
def test_get_variables(self, m_create_network_topology, m_json_load):
|
|
m_json_load.side_effect = [{'glob': 'glob_val'}, {}]
|
|
m_create_network_topology.return_value = 'network_topology'
|
|
r_value = start_script.get_variables('role')
|
|
e_value = {
|
|
'glob': 'glob_val',
|
|
'role_name': 'role',
|
|
'network_topology': 'network_topology',
|
|
'node_name': 'node1',
|
|
'pod_name': 'pod1'
|
|
}
|
|
self.assertEqual(r_value, e_value)
|
|
|
|
@mock.patch('six.moves.builtins.open', mock.mock_open())
|
|
@mock.patch('json.load')
|
|
@mock.patch('fuel_ccp_entrypoint.start_script.create_network_topology')
|
|
def test_get_variables_with_node_config(self, m_create_network_topology,
|
|
m_json_load):
|
|
m_json_load.side_effect = [
|
|
# globals
|
|
{
|
|
'a': {
|
|
'b': {
|
|
'c': ['d', 'e', 'f'],
|
|
'g': 'h',
|
|
},
|
|
'i': ['j', 'k'],
|
|
'l': 'm'
|
|
},
|
|
'n': ['o', 'p', 'q'],
|
|
'r': 's'
|
|
},
|
|
# nodes configs
|
|
{
|
|
'node[1-3]': {
|
|
'a': {
|
|
'b': {
|
|
'c': ['e', 'f', 't'],
|
|
'u': 'v'
|
|
},
|
|
'w': {
|
|
'x': 'y'
|
|
}
|
|
},
|
|
'n': ['o', 'p'],
|
|
'z': 'NaN'
|
|
},
|
|
'node[1-2]': {
|
|
'aa': {'ab': 'ac'},
|
|
'r': {'ad': ['ae', 'af', 'ag']}
|
|
|
|
}
|
|
}
|
|
]
|
|
m_create_network_topology.return_value = 'network_topology'
|
|
actual = start_script.get_variables('fake_role')
|
|
expected = {
|
|
'role_name': 'fake_role',
|
|
'network_topology': 'network_topology',
|
|
'node_name': 'node1',
|
|
'pod_name': 'pod1',
|
|
'a': {
|
|
'b': {
|
|
'c': ['e', 'f', 't'],
|
|
'g': 'h',
|
|
'u': 'v'
|
|
},
|
|
'w': {'x': 'y'},
|
|
'i': ['j', 'k'],
|
|
'l': 'm'
|
|
},
|
|
'n': ['o', 'p'],
|
|
'r': {'ad': ['ae', 'af', 'ag']},
|
|
'z': 'NaN',
|
|
'aa': {'ab': 'ac'},
|
|
}
|
|
self.assertEqual(expected, actual)
|
|
|
|
|
|
class TestRetry(base.TestCase):
|
|
def setUp(self):
|
|
super(TestRetry, self).setUp()
|
|
start_script.VARIABLES = {'etcd': {
|
|
'connection_attempts': 3,
|
|
'connection_delay': 0
|
|
}}
|
|
|
|
@start_script.retry
|
|
def func_test(self):
|
|
return self.func_ret()
|
|
|
|
def test_retry_succeeded(self):
|
|
self.func_ret = mock.Mock(side_effect=[
|
|
etcd.EtcdException('test_error'), 'test_result'])
|
|
self.assertEqual('test_result', self.func_test())
|
|
self.assertEqual(2, self.func_ret.call_count)
|
|
|
|
def test_retry_failed(self):
|
|
self.func_ret = mock.Mock(side_effect=[
|
|
etcd.EtcdException('test_error') for _ in range(3)])
|
|
|
|
self.assertRaisesRegexp(
|
|
etcd.EtcdException, 'test_error', self.func_test)
|
|
self.assertEqual(3, self.func_ret.call_count)
|
|
|
|
|
|
class TestGetETCDClient(base.TestCase):
|
|
def test_get_etcd_local_client(self):
|
|
start_script.VARIABLES = {
|
|
"role_name": "etcd",
|
|
"etcd": {
|
|
"client_port": {
|
|
"cont": 10042
|
|
},
|
|
"connection_attempts": 3,
|
|
"connection_delay": 0,
|
|
},
|
|
"network_topology": {
|
|
"private": {
|
|
"address": "192.0.2.1"
|
|
}
|
|
}
|
|
}
|
|
with mock.patch("etcd.Client") as m_etcd:
|
|
expected_value = object()
|
|
m_etcd.return_value = expected_value
|
|
etcd_client = start_script.get_etcd_client()
|
|
self.assertIs(expected_value, etcd_client)
|
|
m_etcd.assert_called_once_with(
|
|
host=(("192.0.2.1", 10042),),
|
|
allow_reconnect=True,
|
|
read_timeout=2)
|
|
|
|
def test_get_etcd_client(self):
|
|
start_script.VARIABLES = {
|
|
"role_name": "banana",
|
|
"namespace": "ccp",
|
|
"cluster_domain": 'cluster.local',
|
|
"etcd": {
|
|
"client_port": {
|
|
"cont": 1234
|
|
},
|
|
"connection_attempts": 3,
|
|
"connection_delay": 0,
|
|
},
|
|
}
|
|
with mock.patch("etcd.Client") as m_etcd:
|
|
expected_value = object()
|
|
m_etcd.return_value = expected_value
|
|
etcd_client = start_script.get_etcd_client()
|
|
self.assertIs(expected_value, etcd_client)
|
|
m_etcd.assert_called_once_with(
|
|
host=(('etcd.ccp.svc.cluster.local', 1234),),
|
|
allow_reconnect=True,
|
|
read_timeout=2)
|
|
|
|
def test_get_etcd_client_wrong(self):
|
|
start_script.VARIABLES = {
|
|
"role_nmae": "banana"
|
|
}
|
|
self.assertRaises(KeyError, start_script.get_etcd_client)
|