258 lines
9.1 KiB
Python
258 lines
9.1 KiB
Python
# 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 io
|
|
import mock
|
|
import pytest
|
|
|
|
from octane.tests import util as test_util
|
|
from octane.util import node as node_util
|
|
from octane.util import ssh
|
|
|
|
|
|
NODES = [
|
|
{'id': '1',
|
|
'fqdn': 'node-1.domain.tld',
|
|
'network_data': [{'name': 'management', 'ip': '10.20.0.2'},
|
|
{'name': 'public', 'ip': '172.167.0.2'}]},
|
|
{'id': '2',
|
|
'fqdn': 'node-2.domain.tld',
|
|
'network_data': [{'name': 'management', 'ip': '10.20.0.3'},
|
|
{'name': 'public', 'ip': '172.167.0.3'}]},
|
|
{'id': '3',
|
|
'fqdn': 'node-3.domain.tld',
|
|
'network_data': [{'name': 'management', 'ip': '10.20.0.4'},
|
|
{'name': 'public', 'ip': '172.167.0.4'}]},
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize('node_data,network_name,expected_ip', [
|
|
(NODES[0], 'management', '10.20.0.2'),
|
|
(NODES[0], 'storage', None),
|
|
({'network_data': []}, 'management', None),
|
|
])
|
|
def test_get_ip(node_data, network_name, expected_ip):
|
|
node = create_node(node_data)
|
|
ip = node_util.get_ip(network_name, node)
|
|
assert ip == expected_ip
|
|
|
|
|
|
def create_node(data):
|
|
return mock.Mock(data=data, spec_set=['data', 'env'])
|
|
|
|
|
|
@pytest.fixture
|
|
def nodes():
|
|
return map(create_node, NODES)
|
|
|
|
|
|
@pytest.mark.parametrize("network_name,expected_ips", [
|
|
('management', ['10.20.0.2', '10.20.0.3', '10.20.0.4']),
|
|
('public', ['172.167.0.2', '172.167.0.3', '172.167.0.4']),
|
|
])
|
|
def test_get_ips(nodes, network_name, expected_ips):
|
|
ips = node_util.get_ips(network_name, nodes)
|
|
assert ips == expected_ips
|
|
|
|
|
|
def test_get_hostnames(nodes):
|
|
hostnames = node_util.get_hostnames(nodes)
|
|
assert hostnames == ['node-1.domain.tld',
|
|
'node-2.domain.tld',
|
|
'node-3.domain.tld']
|
|
|
|
|
|
def test_tar_files(node, mock_ssh_popen, mock_open):
|
|
content = b'fake data\nin\nthe\narchive'
|
|
|
|
proc = mock_ssh_popen.return_value.__enter__.return_value
|
|
proc.stdout = io.BytesIO(content)
|
|
buf = io.BytesIO()
|
|
mock_open.return_value.write.side_effect = buf.write
|
|
|
|
node_util.tar_files('filename', node, 'a.file', 'b.file')
|
|
|
|
mock_ssh_popen.assert_called_once_with(
|
|
['tar', '-czvP', 'a.file', 'b.file'],
|
|
stdout=ssh.PIPE, node=node)
|
|
mock_open.assert_called_once_with('filename', 'wb')
|
|
assert buf.getvalue() == content
|
|
|
|
|
|
def test_untar_files(node, mock_ssh_popen, mock_open):
|
|
content = b'fake data\nin\nthe\narchive'
|
|
|
|
proc = mock_ssh_popen.return_value.__enter__.return_value
|
|
buf = io.BytesIO()
|
|
proc.stdin.write = buf.write
|
|
mock_open.return_value = io.BytesIO(content)
|
|
|
|
node_util.untar_files('filename', node)
|
|
|
|
mock_ssh_popen.assert_called_once_with(['tar', '-xzv', '-C', '/'],
|
|
stdin=ssh.PIPE, node=node)
|
|
mock_open.assert_called_once_with('filename', 'rb')
|
|
assert buf.getvalue() == content
|
|
|
|
|
|
NOVA_DEFAULT = b"#\u0444\n[DEFAULT]\ndebug = True\n"
|
|
NOVA_WITH_EMPTY_LEVELS = NOVA_DEFAULT + b"[upgrade_levels]\n"
|
|
NOVA_WITH_JUNO_LEVELS = NOVA_WITH_EMPTY_LEVELS + b"compute=juno\n"
|
|
NOVA_WITH_KILO_LEVELS = NOVA_WITH_EMPTY_LEVELS + b"compute=kilo\n"
|
|
NOVA_BROKEN_LEVELS = NOVA_DEFAULT + b"compute=essex\n[upgrade_levels]\n"
|
|
NOVA_BROKEN_LEVELS_WITH_KILO = NOVA_BROKEN_LEVELS + b"compute=kilo\n"
|
|
|
|
|
|
@pytest.mark.parametrize("content,expected_content,filename", [
|
|
(NOVA_DEFAULT, NOVA_WITH_KILO_LEVELS, '/etc/nova/nova.conf'),
|
|
(NOVA_WITH_EMPTY_LEVELS, NOVA_WITH_KILO_LEVELS, '/etc/nova/nova.conf'),
|
|
(NOVA_WITH_JUNO_LEVELS, NOVA_WITH_KILO_LEVELS, '/etc/nova/nova.conf'),
|
|
(NOVA_BROKEN_LEVELS, NOVA_BROKEN_LEVELS_WITH_KILO, '/etc/nova/nova.conf'),
|
|
])
|
|
def test_add_compute_upgrade_levels(mocker, node, content, expected_content,
|
|
filename):
|
|
with test_util.mock_update_file(mocker, node, content, expected_content,
|
|
filename):
|
|
node_util.add_compute_upgrade_levels(node, 'kilo')
|
|
|
|
|
|
@pytest.mark.parametrize("content,expected_content,filename", [
|
|
(NOVA_DEFAULT, NOVA_DEFAULT, '/etc/nova/nova.conf'),
|
|
(NOVA_WITH_EMPTY_LEVELS, NOVA_WITH_EMPTY_LEVELS, '/etc/nova/nova.conf'),
|
|
(NOVA_WITH_KILO_LEVELS, NOVA_WITH_EMPTY_LEVELS, '/etc/nova/nova.conf'),
|
|
(NOVA_BROKEN_LEVELS_WITH_KILO, NOVA_WITH_EMPTY_LEVELS,
|
|
'/etc/nova/nova.conf'),
|
|
])
|
|
def test_remove_compute_upgrade_levels(mocker, node, content,
|
|
expected_content, filename):
|
|
with test_util.mock_update_file(mocker, node, content, expected_content,
|
|
filename):
|
|
node_util.remove_compute_upgrade_levels(node)
|
|
|
|
|
|
NOVA_LIVE_MIGRATION_FLAG = b" live_migration_flag="
|
|
NOVA_LIVE_MIGRATION_ENABLED = NOVA_LIVE_MIGRATION_FLAG + \
|
|
b"FLAG1,VIR_MIGRATE_LIVE,FLAG2\n"
|
|
NOVA_NO_LIVE_MIGRATION_FLAG = b"no_live_migration_flag\n"
|
|
NOVA_EMPTY = b""
|
|
|
|
|
|
@pytest.mark.parametrize("content,expected_res", [
|
|
(NOVA_LIVE_MIGRATION_ENABLED, True),
|
|
(NOVA_LIVE_MIGRATION_FLAG, False),
|
|
(NOVA_NO_LIVE_MIGRATION_FLAG, False),
|
|
(NOVA_EMPTY, False),
|
|
])
|
|
def test_is_live_migration_supported(mocker, node, content, expected_res):
|
|
mock_sftp = mocker.patch("octane.util.ssh.sftp")
|
|
mock_sftp.return_value.open.return_value = io.BytesIO(content)
|
|
|
|
res = node_util.is_live_migration_supported(node)
|
|
assert res == expected_res
|
|
|
|
|
|
@pytest.mark.parametrize('node_data,fuel_version,expected_name', [
|
|
(NODES[0], '6.0', 'node-1'),
|
|
(NODES[0], '6.1', 'node-1.domain.tld'),
|
|
(NODES[0], 'invalid', None)
|
|
])
|
|
def test_get_nova_node_handle(mocker, node_data, fuel_version, expected_name):
|
|
node = create_node(node_data)
|
|
node.env.data.get.return_value = fuel_version
|
|
if expected_name:
|
|
name = node_util.get_nova_node_handle(node)
|
|
assert name == expected_name
|
|
else:
|
|
with pytest.raises(Exception):
|
|
node_util.get_nova_node_handle(node)
|
|
|
|
|
|
@pytest.mark.parametrize('stdout,nova_services_to_restart', [
|
|
(
|
|
" [ + ] nova-service-1\n"
|
|
" [ + ] nova-service-2\n"
|
|
" [ - ] nova-service-3\n"
|
|
" [ - ] not-nova-service-1\n"
|
|
" [ + ] not-nova-service-2\n"
|
|
" [ + ] nova-service-4\n",
|
|
[
|
|
'nova-service-1',
|
|
'nova-service-2',
|
|
'nova-service-4',
|
|
]
|
|
)
|
|
])
|
|
def test_restart_nova_services(mocker, node, stdout, nova_services_to_restart):
|
|
call_output_mock = mocker.patch(
|
|
"octane.util.ssh.call_output", return_value=stdout)
|
|
call_mock = mocker.patch("octane.util.ssh.call")
|
|
node_util.restart_nova_services(node)
|
|
for service in nova_services_to_restart:
|
|
call_mock.assert_any_call(["service", service, "restart"], node=node)
|
|
call_output_mock.assert_called_once_with(
|
|
["service", "--status-all"], node=node)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("parameters", "parameters_to_get", "required", "ensure", "error"), [
|
|
({
|
|
"opt1": "value1",
|
|
"opt2": "value2",
|
|
}, {
|
|
"opt1": [("section1", "option1")],
|
|
"opt2": [("section2", "option2")],
|
|
}, ("section1/option1", "section2/option2"), True, False),
|
|
|
|
({}, {
|
|
"opt1": [("section1", "option1")],
|
|
}, ("section1/option1"), True, True),
|
|
|
|
({}, {
|
|
"opt1": [("section1", "option1")],
|
|
}, (), False, False),
|
|
]
|
|
)
|
|
def test_get_parameters(mocker, parameters, parameters_to_get, required,
|
|
ensure, error):
|
|
mock_node = mock.Mock(data={"id": 1})
|
|
filename = "fake/filename.conf"
|
|
mock_sftp = mocker.patch("octane.util.ssh.sftp")
|
|
mock_get = mocker.patch("octane.util.helpers.get_parameters")
|
|
mock_get.return_value = parameters
|
|
if ensure and error:
|
|
msg = ("Could not get parameters from the file "
|
|
"node-1[fake/filename.conf]: {parameters}"
|
|
.format(parameters=", ".join(required)))
|
|
with pytest.raises(node_util.AbsentParametersError, message=msg):
|
|
node_util.get_parameters(mock_node, filename, parameters_to_get,
|
|
ensure=ensure)
|
|
else:
|
|
result = node_util.get_parameters(
|
|
mock_node, filename, parameters_to_get, ensure=ensure)
|
|
assert result == parameters
|
|
mock_get.assert_called_once_with(
|
|
mock_sftp.return_value.open.return_value.__enter__.return_value,
|
|
parameters_to_get)
|
|
|
|
|
|
@pytest.mark.parametrize(("online", "result", "error"), [
|
|
(True, True, False),
|
|
(False, None, False),
|
|
(True, False, True),
|
|
])
|
|
def test_restart_mcollective(mocker, online, result, error):
|
|
node = mock.Mock(data={"online": online, "id": 123})
|
|
mock_ssh = mocker.patch("octane.util.ssh.call")
|
|
if error:
|
|
mock_ssh.side_effect = Exception()
|
|
assert node_util.restart_mcollective(node) == result
|