|
Packit Service |
a04d08 |
# This file is part of cloud-init. See LICENSE file for license information.
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
from cloudinit import helpers
|
|
Packit Service |
a04d08 |
from cloudinit import util
|
|
Packit Service |
a04d08 |
from cloudinit.sources.DataSourceCloudStack import (
|
|
Packit Service |
a04d08 |
DataSourceCloudStack, get_latest_lease)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
from cloudinit.tests.helpers import CiTestCase, ExitStack, mock
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
import os
|
|
Packit Service |
a04d08 |
import time
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
MOD_PATH = 'cloudinit.sources.DataSourceCloudStack'
|
|
Packit Service |
a04d08 |
DS_PATH = MOD_PATH + '.DataSourceCloudStack'
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestCloudStackPasswordFetching(CiTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def setUp(self):
|
|
Packit Service |
a04d08 |
super(TestCloudStackPasswordFetching, self).setUp()
|
|
Packit Service |
a04d08 |
self.patches = ExitStack()
|
|
Packit Service |
a04d08 |
self.addCleanup(self.patches.close)
|
|
Packit Service |
a04d08 |
mod_name = MOD_PATH
|
|
Packit Service |
a04d08 |
self.patches.enter_context(mock.patch('{0}.ec2'.format(mod_name)))
|
|
Packit Service |
a04d08 |
self.patches.enter_context(mock.patch('{0}.uhelp'.format(mod_name)))
|
|
Packit Service |
a04d08 |
default_gw = "192.201.20.0"
|
|
Packit Service |
a04d08 |
get_latest_lease = mock.MagicMock(return_value=None)
|
|
Packit Service |
a04d08 |
self.patches.enter_context(mock.patch(
|
|
Packit Service |
a04d08 |
mod_name + '.get_latest_lease', get_latest_lease))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
get_default_gw = mock.MagicMock(return_value=default_gw)
|
|
Packit Service |
a04d08 |
self.patches.enter_context(mock.patch(
|
|
Packit Service |
a04d08 |
mod_name + '.get_default_gateway', get_default_gw))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
get_networkd_server_address = mock.MagicMock(return_value=None)
|
|
Packit Service |
a04d08 |
self.patches.enter_context(mock.patch(
|
|
Packit Service |
a04d08 |
mod_name + '.dhcp.networkd_get_option_from_leases',
|
|
Packit Service |
a04d08 |
get_networkd_server_address))
|
|
Packit Service |
a04d08 |
self.tmp = self.tmp_dir()
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def _set_password_server_response(self, response_string):
|
|
Packit Service |
a04d08 |
subp = mock.MagicMock(return_value=(response_string, ''))
|
|
Packit Service |
a04d08 |
self.patches.enter_context(
|
|
Packit Service |
751c4a |
mock.patch('cloudinit.sources.DataSourceCloudStack.subp.subp',
|
|
Packit Service |
a04d08 |
subp))
|
|
Packit Service |
a04d08 |
return subp
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_empty_password_doesnt_create_config(self):
|
|
Packit Service |
a04d08 |
self._set_password_server_response('')
|
|
Packit Service |
a04d08 |
ds = DataSourceCloudStack(
|
|
Packit Service |
a04d08 |
{}, None, helpers.Paths({'run_dir': self.tmp}))
|
|
Packit Service |
a04d08 |
ds.get_data()
|
|
Packit Service |
a04d08 |
self.assertEqual({}, ds.get_config_obj())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_saved_password_doesnt_create_config(self):
|
|
Packit Service |
a04d08 |
self._set_password_server_response('saved_password')
|
|
Packit Service |
a04d08 |
ds = DataSourceCloudStack(
|
|
Packit Service |
a04d08 |
{}, None, helpers.Paths({'run_dir': self.tmp}))
|
|
Packit Service |
a04d08 |
ds.get_data()
|
|
Packit Service |
a04d08 |
self.assertEqual({}, ds.get_config_obj())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch(DS_PATH + '.wait_for_metadata_service')
|
|
Packit Service |
a04d08 |
def test_password_sets_password(self, m_wait):
|
|
Packit Service |
a04d08 |
m_wait.return_value = True
|
|
Packit Service |
a04d08 |
password = 'SekritSquirrel'
|
|
Packit Service |
a04d08 |
self._set_password_server_response(password)
|
|
Packit Service |
a04d08 |
ds = DataSourceCloudStack(
|
|
Packit Service |
a04d08 |
{}, None, helpers.Paths({'run_dir': self.tmp}))
|
|
Packit Service |
a04d08 |
ds.get_data()
|
|
Packit Service |
a04d08 |
self.assertEqual(password, ds.get_config_obj()['password'])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch(DS_PATH + '.wait_for_metadata_service')
|
|
Packit Service |
a04d08 |
def test_bad_request_doesnt_stop_ds_from_working(self, m_wait):
|
|
Packit Service |
a04d08 |
m_wait.return_value = True
|
|
Packit Service |
a04d08 |
self._set_password_server_response('bad_request')
|
|
Packit Service |
a04d08 |
ds = DataSourceCloudStack(
|
|
Packit Service |
a04d08 |
{}, None, helpers.Paths({'run_dir': self.tmp}))
|
|
Packit Service |
a04d08 |
self.assertTrue(ds.get_data())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def assertRequestTypesSent(self, subp, expected_request_types):
|
|
Packit Service |
a04d08 |
request_types = []
|
|
Packit Service |
a04d08 |
for call in subp.call_args_list:
|
|
Packit Service |
a04d08 |
args = call[0][0]
|
|
Packit Service |
a04d08 |
for arg in args:
|
|
Packit Service |
a04d08 |
if arg.startswith('DomU_Request'):
|
|
Packit Service |
a04d08 |
request_types.append(arg.split()[1])
|
|
Packit Service |
a04d08 |
self.assertEqual(expected_request_types, request_types)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch(DS_PATH + '.wait_for_metadata_service')
|
|
Packit Service |
a04d08 |
def test_valid_response_means_password_marked_as_saved(self, m_wait):
|
|
Packit Service |
a04d08 |
m_wait.return_value = True
|
|
Packit Service |
a04d08 |
password = 'SekritSquirrel'
|
|
Packit Service |
a04d08 |
subp = self._set_password_server_response(password)
|
|
Packit Service |
a04d08 |
ds = DataSourceCloudStack(
|
|
Packit Service |
a04d08 |
{}, None, helpers.Paths({'run_dir': self.tmp}))
|
|
Packit Service |
a04d08 |
ds.get_data()
|
|
Packit Service |
a04d08 |
self.assertRequestTypesSent(subp,
|
|
Packit Service |
a04d08 |
['send_my_password', 'saved_password'])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def _check_password_not_saved_for(self, response_string):
|
|
Packit Service |
a04d08 |
subp = self._set_password_server_response(response_string)
|
|
Packit Service |
a04d08 |
ds = DataSourceCloudStack(
|
|
Packit Service |
a04d08 |
{}, None, helpers.Paths({'run_dir': self.tmp}))
|
|
Packit Service |
a04d08 |
with mock.patch(DS_PATH + '.wait_for_metadata_service') as m_wait:
|
|
Packit Service |
a04d08 |
m_wait.return_value = True
|
|
Packit Service |
a04d08 |
ds.get_data()
|
|
Packit Service |
a04d08 |
self.assertRequestTypesSent(subp, ['send_my_password'])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_password_not_saved_if_empty(self):
|
|
Packit Service |
a04d08 |
self._check_password_not_saved_for('')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_password_not_saved_if_already_saved(self):
|
|
Packit Service |
a04d08 |
self._check_password_not_saved_for('saved_password')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_password_not_saved_if_bad_request(self):
|
|
Packit Service |
a04d08 |
self._check_password_not_saved_for('bad_request')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestGetLatestLease(CiTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def _populate_dir_list(self, bdir, files):
|
|
Packit Service |
a04d08 |
"""populate_dir_list([(name, data), (name, data)])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
writes files to bdir, and updates timestamps to ensure
|
|
Packit Service |
a04d08 |
that their mtime increases with each file."""
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
start = int(time.time())
|
|
Packit Service |
a04d08 |
for num, fname in enumerate(reversed(files)):
|
|
Packit Service |
a04d08 |
fpath = os.path.sep.join((bdir, fname))
|
|
Packit Service |
a04d08 |
util.write_file(fpath, fname.encode())
|
|
Packit Service |
a04d08 |
os.utime(fpath, (start - num, start - num))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def _pop_and_test(self, files, expected):
|
|
Packit Service |
a04d08 |
lease_d = self.tmp_dir()
|
|
Packit Service |
a04d08 |
self._populate_dir_list(lease_d, files)
|
|
Packit Service |
a04d08 |
self.assertEqual(self.tmp_path(expected, lease_d),
|
|
Packit Service |
a04d08 |
get_latest_lease(lease_d))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_skips_dhcpv6_files(self):
|
|
Packit Service |
a04d08 |
"""files started with dhclient6 should be skipped."""
|
|
Packit Service |
a04d08 |
expected = "dhclient.lease"
|
|
Packit Service |
a04d08 |
self._pop_and_test([expected, "dhclient6.lease"], expected)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_selects_dhclient_dot_files(self):
|
|
Packit Service |
a04d08 |
"""files named dhclient.lease or dhclient.leases should be used.
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
Ubuntu names files dhclient.eth0.leases dhclient6.leases and
|
|
Packit Service |
a04d08 |
sometimes dhclient.leases."""
|
|
Packit Service |
a04d08 |
self._pop_and_test(["dhclient.lease"], "dhclient.lease")
|
|
Packit Service |
a04d08 |
self._pop_and_test(["dhclient.leases"], "dhclient.leases")
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_selects_dhclient_dash_files(self):
|
|
Packit Service |
a04d08 |
"""files named dhclient-lease or dhclient-leases should be used.
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
Redhat/Centos names files with dhclient--eth0.lease (centos 7) or
|
|
Packit Service |
a04d08 |
dhclient-eth0.leases (centos 6).
|
|
Packit Service |
a04d08 |
"""
|
|
Packit Service |
a04d08 |
self._pop_and_test(["dhclient-eth0.lease"], "dhclient-eth0.lease")
|
|
Packit Service |
a04d08 |
self._pop_and_test(["dhclient--eth0.lease"], "dhclient--eth0.lease")
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_ignores_by_extension(self):
|
|
Packit Service |
a04d08 |
"""only .lease or .leases file should be considered."""
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
self._pop_and_test(["dhclient.lease", "dhclient.lease.bk",
|
|
Packit Service |
a04d08 |
"dhclient.lease-old", "dhclient.leaselease"],
|
|
Packit Service |
a04d08 |
"dhclient.lease")
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_selects_newest_matching(self):
|
|
Packit Service |
a04d08 |
"""If multiple files match, the newest written should be used."""
|
|
Packit Service |
a04d08 |
lease_d = self.tmp_dir()
|
|
Packit Service |
a04d08 |
valid_1 = "dhclient.leases"
|
|
Packit Service |
a04d08 |
valid_2 = "dhclient.lease"
|
|
Packit Service |
a04d08 |
valid_1_path = self.tmp_path(valid_1, lease_d)
|
|
Packit Service |
a04d08 |
valid_2_path = self.tmp_path(valid_2, lease_d)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
self._populate_dir_list(lease_d, [valid_1, valid_2])
|
|
Packit Service |
a04d08 |
self.assertEqual(valid_2_path, get_latest_lease(lease_d))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
# now update mtime on valid_2 to be older than valid_1 and re-check.
|
|
Packit Service |
a04d08 |
mtime = int(os.path.getmtime(valid_1_path)) - 1
|
|
Packit Service |
a04d08 |
os.utime(valid_2_path, (mtime, mtime))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
self.assertEqual(valid_1_path, get_latest_lease(lease_d))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
# vi: ts=4 expandtab
|