summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/unittests/test_data.py5
-rw-r--r--tests/unittests/test_datasource/test_altcloud.py23
-rw-r--r--tests/unittests/test_datasource/test_azure.py15
-rw-r--r--tests/unittests/test_datasource/test_azure_helper.py78
-rw-r--r--tests/unittests/test_datasource/test_configdrive.py32
-rw-r--r--tests/unittests/test_datasource/test_maas.py16
-rw-r--r--tests/unittests/test_datasource/test_opennebula.py2
-rw-r--r--tests/unittests/test_datasource/test_smartos.py84
-rw-r--r--tests/unittests/test_distros/test_user_data_normalize.py29
-rw-r--r--tests/unittests/test_handler/test_handler_lxd.py75
-rw-r--r--tests/unittests/test_handler/test_handler_power_state.py5
-rw-r--r--tests/unittests/test_handler/test_handler_seed_random.py13
-rw-r--r--tests/unittests/test_handler/test_handler_snappy.py3
-rw-r--r--tests/unittests/test_sshutil.py3
-rw-r--r--tests/unittests/test_templating.py3
-rw-r--r--tests/unittests/test_util.py10
16 files changed, 258 insertions, 138 deletions
diff --git a/tests/unittests/test_data.py b/tests/unittests/test_data.py
index c603bfdb..9c1ec1d4 100644
--- a/tests/unittests/test_data.py
+++ b/tests/unittests/test_data.py
@@ -27,11 +27,12 @@ from cloudinit import stages
from cloudinit import user_data as ud
from cloudinit import util
-INSTANCE_ID = "i-testing"
-
from . import helpers
+INSTANCE_ID = "i-testing"
+
+
class FakeDataSource(sources.DataSource):
def __init__(self, userdata=None, vendordata=None):
diff --git a/tests/unittests/test_datasource/test_altcloud.py b/tests/unittests/test_datasource/test_altcloud.py
index e9cd2fa5..85759c68 100644
--- a/tests/unittests/test_datasource/test_altcloud.py
+++ b/tests/unittests/test_datasource/test_altcloud.py
@@ -134,8 +134,7 @@ class TestGetCloudType(TestCase):
'''
util.read_dmi_data = _dmi_data('RHEV')
dsrc = DataSourceAltCloud({}, None, self.paths)
- self.assertEquals('RHEV', \
- dsrc.get_cloud_type())
+ self.assertEquals('RHEV', dsrc.get_cloud_type())
def test_vsphere(self):
'''
@@ -144,8 +143,7 @@ class TestGetCloudType(TestCase):
'''
util.read_dmi_data = _dmi_data('VMware Virtual Platform')
dsrc = DataSourceAltCloud({}, None, self.paths)
- self.assertEquals('VSPHERE', \
- dsrc.get_cloud_type())
+ self.assertEquals('VSPHERE', dsrc.get_cloud_type())
def test_unknown(self):
'''
@@ -154,8 +152,7 @@ class TestGetCloudType(TestCase):
'''
util.read_dmi_data = _dmi_data('Unrecognized Platform')
dsrc = DataSourceAltCloud({}, None, self.paths)
- self.assertEquals('UNKNOWN', \
- dsrc.get_cloud_type())
+ self.assertEquals('UNKNOWN', dsrc.get_cloud_type())
class TestGetDataCloudInfoFile(TestCase):
@@ -412,27 +409,27 @@ class TestReadUserDataCallback(TestCase):
'''Test read_user_data_callback() with both files.'''
self.assertEquals('test user data',
- read_user_data_callback(self.mount_dir))
+ read_user_data_callback(self.mount_dir))
def test_callback_dc(self):
'''Test read_user_data_callback() with only DC file.'''
_remove_user_data_files(self.mount_dir,
- dc_file=False,
- non_dc_file=True)
+ dc_file=False,
+ non_dc_file=True)
self.assertEquals('test user data',
- read_user_data_callback(self.mount_dir))
+ read_user_data_callback(self.mount_dir))
def test_callback_non_dc(self):
'''Test read_user_data_callback() with only non-DC file.'''
_remove_user_data_files(self.mount_dir,
- dc_file=True,
- non_dc_file=False)
+ dc_file=True,
+ non_dc_file=False)
self.assertEquals('test user data',
- read_user_data_callback(self.mount_dir))
+ read_user_data_callback(self.mount_dir))
def test_callback_none(self):
'''Test read_user_data_callback() no files are found.'''
diff --git a/tests/unittests/test_datasource/test_azure.py b/tests/unittests/test_datasource/test_azure.py
index 3933794f..444e2799 100644
--- a/tests/unittests/test_datasource/test_azure.py
+++ b/tests/unittests/test_datasource/test_azure.py
@@ -207,7 +207,7 @@ class TestAzureDataSource(TestCase):
yaml_cfg = "{agent_command: my_command}\n"
cfg = yaml.safe_load(yaml_cfg)
odata = {'HostName': "myhost", 'UserName': "myuser",
- 'dscfg': {'text': yaml_cfg, 'encoding': 'plain'}}
+ 'dscfg': {'text': yaml_cfg, 'encoding': 'plain'}}
data = {'ovfcontent': construct_valid_ovf_env(data=odata)}
dsrc = self._get_ds(data)
@@ -219,8 +219,8 @@ class TestAzureDataSource(TestCase):
# set dscfg in via base64 encoded yaml
cfg = {'agent_command': "my_command"}
odata = {'HostName': "myhost", 'UserName': "myuser",
- 'dscfg': {'text': b64e(yaml.dump(cfg)),
- 'encoding': 'base64'}}
+ 'dscfg': {'text': b64e(yaml.dump(cfg)),
+ 'encoding': 'base64'}}
data = {'ovfcontent': construct_valid_ovf_env(data=odata)}
dsrc = self._get_ds(data)
@@ -267,7 +267,8 @@ class TestAzureDataSource(TestCase):
# should equal that after the '$'
pos = defuser['passwd'].rfind("$") + 1
self.assertEqual(defuser['passwd'],
- crypt.crypt(odata['UserPassword'], defuser['passwd'][0:pos]))
+ crypt.crypt(odata['UserPassword'],
+ defuser['passwd'][0:pos]))
def test_userdata_plain(self):
mydata = "FOOBAR"
@@ -364,8 +365,8 @@ class TestAzureDataSource(TestCase):
# Make sure that user can affect disk aliases
dscfg = {'disk_aliases': {'ephemeral0': '/dev/sdc'}}
odata = {'HostName': "myhost", 'UserName': "myuser",
- 'dscfg': {'text': b64e(yaml.dump(dscfg)),
- 'encoding': 'base64'}}
+ 'dscfg': {'text': b64e(yaml.dump(dscfg)),
+ 'encoding': 'base64'}}
usercfg = {'disk_setup': {'/dev/sdc': {'something': '...'},
'ephemeral0': False}}
userdata = '#cloud-config' + yaml.dump(usercfg) + "\n"
@@ -634,7 +635,7 @@ class TestReadAzureOvf(TestCase):
def test_invalid_xml_raises_non_azure_ds(self):
invalid_xml = "<foo>" + construct_valid_ovf_env(data={})
self.assertRaises(DataSourceAzure.BrokenAzureDataSource,
- DataSourceAzure.read_azure_ovf, invalid_xml)
+ DataSourceAzure.read_azure_ovf, invalid_xml)
def test_load_with_pubkeys(self):
mypklist = [{'fingerprint': 'fp1', 'path': 'path1', 'value': ''}]
diff --git a/tests/unittests/test_datasource/test_azure_helper.py b/tests/unittests/test_datasource/test_azure_helper.py
index 0638c974..1134199b 100644
--- a/tests/unittests/test_datasource/test_azure_helper.py
+++ b/tests/unittests/test_datasource/test_azure_helper.py
@@ -1,6 +1,4 @@
import os
-import struct
-import unittest
from cloudinit.sources.helpers import azure as azure_helper
from ..helpers import TestCase
@@ -75,48 +73,64 @@ class TestFindEndpoint(TestCase):
self.assertRaises(Exception,
azure_helper.WALinuxAgentShim.find_endpoint)
- def _build_lease_content(self, ip_address, use_hex=True):
- ip_address_repr = ':'.join(
- [hex(int(part)).replace('0x', '')
- for part in ip_address.split('.')])
- if not use_hex:
- ip_address_repr = struct.pack(
- '>L', int(ip_address_repr.replace(':', ''), 16))
- ip_address_repr = '"{0}"'.format(ip_address_repr.decode('utf-8'))
+ @staticmethod
+ def _build_lease_content(encoded_address):
return '\n'.join([
'lease {',
' interface "eth0";',
- ' option unknown-245 {0};'.format(ip_address_repr),
+ ' option unknown-245 {0};'.format(encoded_address),
'}'])
- def test_hex_string(self):
- ip_address = '98.76.54.32'
- file_content = self._build_lease_content(ip_address)
+ def test_latest_lease_used(self):
+ encoded_addresses = ['5:4:3:2', '4:3:2:1']
+ file_content = '\n'.join([self._build_lease_content(encoded_address)
+ for encoded_address in encoded_addresses])
self.load_file.return_value = file_content
- self.assertEqual(ip_address,
+ self.assertEqual(encoded_addresses[-1].replace(':', '.'),
azure_helper.WALinuxAgentShim.find_endpoint())
+
+class TestExtractIpAddressFromLeaseValue(TestCase):
+
+ def test_hex_string(self):
+ ip_address, encoded_address = '98.76.54.32', '62:4c:36:20'
+ self.assertEqual(
+ ip_address,
+ azure_helper.WALinuxAgentShim.get_ip_from_lease_value(
+ encoded_address
+ ))
+
def test_hex_string_with_single_character_part(self):
- ip_address = '4.3.2.1'
- file_content = self._build_lease_content(ip_address)
- self.load_file.return_value = file_content
- self.assertEqual(ip_address,
- azure_helper.WALinuxAgentShim.find_endpoint())
+ ip_address, encoded_address = '4.3.2.1', '4:3:2:1'
+ self.assertEqual(
+ ip_address,
+ azure_helper.WALinuxAgentShim.get_ip_from_lease_value(
+ encoded_address
+ ))
def test_packed_string(self):
- ip_address = '98.76.54.32'
- file_content = self._build_lease_content(ip_address, use_hex=False)
- self.load_file.return_value = file_content
- self.assertEqual(ip_address,
- azure_helper.WALinuxAgentShim.find_endpoint())
+ ip_address, encoded_address = '98.76.54.32', 'bL6 '
+ self.assertEqual(
+ ip_address,
+ azure_helper.WALinuxAgentShim.get_ip_from_lease_value(
+ encoded_address
+ ))
- def test_latest_lease_used(self):
- ip_addresses = ['4.3.2.1', '98.76.54.32']
- file_content = '\n'.join([self._build_lease_content(ip_address)
- for ip_address in ip_addresses])
- self.load_file.return_value = file_content
- self.assertEqual(ip_addresses[-1],
- azure_helper.WALinuxAgentShim.find_endpoint())
+ def test_packed_string_with_escaped_quote(self):
+ ip_address, encoded_address = '100.72.34.108', 'dH\\"l'
+ self.assertEqual(
+ ip_address,
+ azure_helper.WALinuxAgentShim.get_ip_from_lease_value(
+ encoded_address
+ ))
+
+ def test_packed_string_containing_a_colon(self):
+ ip_address, encoded_address = '100.72.58.108', 'dH:l'
+ self.assertEqual(
+ ip_address,
+ azure_helper.WALinuxAgentShim.get_ip_from_lease_value(
+ encoded_address
+ ))
class TestGoalStateParsing(TestCase):
diff --git a/tests/unittests/test_datasource/test_configdrive.py b/tests/unittests/test_datasource/test_configdrive.py
index 83aca505..bfd787d1 100644
--- a/tests/unittests/test_datasource/test_configdrive.py
+++ b/tests/unittests/test_datasource/test_configdrive.py
@@ -61,16 +61,16 @@ CONTENT_0 = b'This is contents of /etc/foo.cfg\n'
CONTENT_1 = b'# this is /etc/bar/bar.cfg\n'
CFG_DRIVE_FILES_V2 = {
- 'ec2/2009-04-04/meta-data.json': json.dumps(EC2_META),
- 'ec2/2009-04-04/user-data': USER_DATA,
- 'ec2/latest/meta-data.json': json.dumps(EC2_META),
- 'ec2/latest/user-data': USER_DATA,
- 'openstack/2012-08-10/meta_data.json': json.dumps(OSTACK_META),
- 'openstack/2012-08-10/user_data': USER_DATA,
- 'openstack/content/0000': CONTENT_0,
- 'openstack/content/0001': CONTENT_1,
- 'openstack/latest/meta_data.json': json.dumps(OSTACK_META),
- 'openstack/latest/user_data': USER_DATA}
+ 'ec2/2009-04-04/meta-data.json': json.dumps(EC2_META),
+ 'ec2/2009-04-04/user-data': USER_DATA,
+ 'ec2/latest/meta-data.json': json.dumps(EC2_META),
+ 'ec2/latest/user-data': USER_DATA,
+ 'openstack/2012-08-10/meta_data.json': json.dumps(OSTACK_META),
+ 'openstack/2012-08-10/user_data': USER_DATA,
+ 'openstack/content/0000': CONTENT_0,
+ 'openstack/content/0001': CONTENT_1,
+ 'openstack/latest/meta_data.json': json.dumps(OSTACK_META),
+ 'openstack/latest/user_data': USER_DATA}
class TestConfigDriveDataSource(TestCase):
@@ -293,9 +293,8 @@ class TestConfigDriveDataSource(TestCase):
util.is_partition = my_is_partition
devs_with_answers = {"TYPE=vfat": [],
- "TYPE=iso9660": ["/dev/vdb"],
- "LABEL=config-2": ["/dev/vdb"],
- }
+ "TYPE=iso9660": ["/dev/vdb"],
+ "LABEL=config-2": ["/dev/vdb"]}
self.assertEqual(["/dev/vdb"], ds.find_candidate_devs())
# add a vfat item
@@ -306,9 +305,10 @@ class TestConfigDriveDataSource(TestCase):
# verify that partitions are considered, that have correct label.
devs_with_answers = {"TYPE=vfat": ["/dev/sda1"],
- "TYPE=iso9660": [], "LABEL=config-2": ["/dev/vdb3"]}
+ "TYPE=iso9660": [],
+ "LABEL=config-2": ["/dev/vdb3"]}
self.assertEqual(["/dev/vdb3"],
- ds.find_candidate_devs())
+ ds.find_candidate_devs())
finally:
util.find_devs_with = orig_find_devs_with
@@ -319,7 +319,7 @@ class TestConfigDriveDataSource(TestCase):
populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
myds = cfg_ds_from_dir(self.tmp)
self.assertEqual(myds.get_public_ssh_keys(),
- [OSTACK_META['public_keys']['mykey']])
+ [OSTACK_META['public_keys']['mykey']])
def cfg_ds_from_dir(seed_d):
diff --git a/tests/unittests/test_datasource/test_maas.py b/tests/unittests/test_datasource/test_maas.py
index eb97b692..77d15cac 100644
--- a/tests/unittests/test_datasource/test_maas.py
+++ b/tests/unittests/test_datasource/test_maas.py
@@ -25,9 +25,9 @@ class TestMAASDataSource(TestCase):
"""Verify a valid seeddir is read as such."""
data = {'instance-id': 'i-valid01',
- 'local-hostname': 'valid01-hostname',
- 'user-data': b'valid01-userdata',
- 'public-keys': 'ssh-rsa AAAAB3Nz...aC1yc2E= keyname'}
+ 'local-hostname': 'valid01-hostname',
+ 'user-data': b'valid01-userdata',
+ 'public-keys': 'ssh-rsa AAAAB3Nz...aC1yc2E= keyname'}
my_d = os.path.join(self.tmp, "valid")
populate_dir(my_d, data)
@@ -45,8 +45,8 @@ class TestMAASDataSource(TestCase):
"""Verify extra files do not affect seed_dir validity."""
data = {'instance-id': 'i-valid-extra',
- 'local-hostname': 'valid-extra-hostname',
- 'user-data': b'valid-extra-userdata', 'foo': 'bar'}
+ 'local-hostname': 'valid-extra-hostname',
+ 'user-data': b'valid-extra-userdata', 'foo': 'bar'}
my_d = os.path.join(self.tmp, "valid_extra")
populate_dir(my_d, data)
@@ -64,7 +64,7 @@ class TestMAASDataSource(TestCase):
"""Verify that invalid seed_dir raises MAASSeedDirMalformed."""
valid = {'instance-id': 'i-instanceid',
- 'local-hostname': 'test-hostname', 'user-data': ''}
+ 'local-hostname': 'test-hostname', 'user-data': ''}
my_based = os.path.join(self.tmp, "valid_extra")
@@ -94,8 +94,8 @@ class TestMAASDataSource(TestCase):
def test_seed_dir_missing(self):
"""Verify that missing seed_dir raises MAASSeedDirNone."""
self.assertRaises(DataSourceMAAS.MAASSeedDirNone,
- DataSourceMAAS.read_maas_seed_dir,
- os.path.join(self.tmp, "nonexistantdirectory"))
+ DataSourceMAAS.read_maas_seed_dir,
+ os.path.join(self.tmp, "nonexistantdirectory"))
def test_seed_url_valid(self):
"""Verify that valid seed_url is read as such."""
diff --git a/tests/unittests/test_datasource/test_opennebula.py b/tests/unittests/test_datasource/test_opennebula.py
index 27adf21b..d796f030 100644
--- a/tests/unittests/test_datasource/test_opennebula.py
+++ b/tests/unittests/test_datasource/test_opennebula.py
@@ -20,7 +20,7 @@ TEST_VARS = {
'VAR7': 'single\\t',
'VAR8': 'double\\tword',
'VAR9': 'multi\\t\nline\n',
- 'VAR10': '\\', # expect \
+ 'VAR10': '\\', # expect '\'
'VAR11': '\'', # expect '
'VAR12': '$', # expect $
}
diff --git a/tests/unittests/test_datasource/test_smartos.py b/tests/unittests/test_datasource/test_smartos.py
index adee9019..5c49966a 100644
--- a/tests/unittests/test_datasource/test_smartos.py
+++ b/tests/unittests/test_datasource/test_smartos.py
@@ -56,12 +56,13 @@ MOCK_RETURNS = {
'cloud-init:user-data': '\n'.join(['#!/bin/sh', '/bin/true', '']),
'sdc:datacenter_name': 'somewhere2',
'sdc:operator-script': '\n'.join(['bin/true', '']),
+ 'sdc:uuid': str(uuid.uuid4()),
'sdc:vendor-data': '\n'.join(['VENDOR_DATA', '']),
'user-data': '\n'.join(['something', '']),
'user-script': '\n'.join(['/bin/true', '']),
}
-DMI_DATA_RETURN = (str(uuid.uuid4()), 'smartdc')
+DMI_DATA_RETURN = 'smartdc'
def get_mock_client(mockdata):
@@ -111,7 +112,8 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase):
ret = apply_patches(patches)
self.unapply += ret
- def _get_ds(self, sys_cfg=None, ds_cfg=None, mockdata=None, dmi_data=None):
+ def _get_ds(self, sys_cfg=None, ds_cfg=None, mockdata=None, dmi_data=None,
+ is_lxbrand=False):
mod = DataSourceSmartOS
if mockdata is None:
@@ -124,9 +126,13 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase):
return dmi_data
def _os_uname():
- # LP: #1243287. tests assume this runs, but running test on
- # arm would cause them all to fail.
- return ('LINUX', 'NODENAME', 'RELEASE', 'VERSION', 'x86_64')
+ if not is_lxbrand:
+ # LP: #1243287. tests assume this runs, but running test on
+ # arm would cause them all to fail.
+ return ('LINUX', 'NODENAME', 'RELEASE', 'VERSION', 'x86_64')
+ else:
+ return ('LINUX', 'NODENAME', 'RELEASE', 'BRANDZ VIRTUAL LINUX',
+ 'X86_64')
if sys_cfg is None:
sys_cfg = {}
@@ -136,7 +142,6 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase):
sys_cfg['datasource']['SmartOS'] = ds_cfg
self.apply_patches([(mod, 'LEGACY_USER_D', self.legacy_user_d)])
- self.apply_patches([(mod, 'get_serial', mock.MagicMock())])
self.apply_patches([
(mod, 'JoyentMetadataClient', get_mock_client(mockdata))])
self.apply_patches([(mod, 'dmi_data', _dmi_data)])
@@ -144,6 +149,7 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase):
self.apply_patches([(mod, 'device_exists', lambda d: True)])
dsrc = mod.DataSourceSmartOS(sys_cfg, distro=None,
paths=self.paths)
+ self.apply_patches([(dsrc, '_get_seed_file_object', mock.MagicMock())])
return dsrc
def test_seed(self):
@@ -151,14 +157,29 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase):
dsrc = self._get_ds()
ret = dsrc.get_data()
self.assertTrue(ret)
+ self.assertEquals('kvm', dsrc.smartos_type)
self.assertEquals('/dev/ttyS1', dsrc.seed)
+ def test_seed_lxbrand(self):
+ # default seed should be /dev/ttyS1
+ dsrc = self._get_ds(is_lxbrand=True)
+ ret = dsrc.get_data()
+ self.assertTrue(ret)
+ self.assertEquals('lx-brand', dsrc.smartos_type)
+ self.assertEquals('/native/.zonecontrol/metadata.sock', dsrc.seed)
+
def test_issmartdc(self):
dsrc = self._get_ds()
ret = dsrc.get_data()
self.assertTrue(ret)
self.assertTrue(dsrc.is_smartdc)
+ def test_issmartdc_lxbrand(self):
+ dsrc = self._get_ds(is_lxbrand=True)
+ ret = dsrc.get_data()
+ self.assertTrue(ret)
+ self.assertTrue(dsrc.is_smartdc)
+
def test_no_base64(self):
ds_cfg = {'no_base64_decode': ['test_var1'], 'all_base': True}
dsrc = self._get_ds(ds_cfg=ds_cfg)
@@ -169,7 +190,8 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase):
dsrc = self._get_ds(mockdata=MOCK_RETURNS)
ret = dsrc.get_data()
self.assertTrue(ret)
- self.assertEquals(DMI_DATA_RETURN[0], dsrc.metadata['instance-id'])
+ self.assertEquals(MOCK_RETURNS['sdc:uuid'],
+ dsrc.metadata['instance-id'])
def test_root_keys(self):
dsrc = self._get_ds(mockdata=MOCK_RETURNS)
@@ -407,18 +429,6 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase):
self.assertEqual(dsrc.device_name_to_device('FOO'),
mydscfg['disk_aliases']['FOO'])
- @mock.patch('cloudinit.sources.DataSourceSmartOS.JoyentMetadataClient')
- @mock.patch('cloudinit.sources.DataSourceSmartOS.get_serial')
- def test_serial_console_closed_on_error(self, get_serial, metadata_client):
- class OurException(Exception):
- pass
- metadata_client.side_effect = OurException
- try:
- DataSourceSmartOS.query_data('noun', 'device', 0)
- except OurException:
- pass
- self.assertEqual(1, get_serial.return_value.close.call_count)
-
def apply_patches(patches):
ret = []
@@ -447,14 +457,25 @@ class TestJoyentMetadataClient(helpers.FilesystemMockingTestCase):
}
def make_response():
- payload = ''
- if self.response_parts['payload']:
- payload = ' {0}'.format(self.response_parts['payload'])
- del self.response_parts['payload']
- return (
- 'V2 {length} {crc} {request_id} {command}{payload}\n'.format(
- payload=payload, **self.response_parts).encode('ascii'))
- self.serial.readline.side_effect = make_response
+ payloadstr = ''
+ if 'payload' in self.response_parts:
+ payloadstr = ' {0}'.format(self.response_parts['payload'])
+ return ('V2 {length} {crc} {request_id} '
+ '{command}{payloadstr}\n'.format(
+ payloadstr=payloadstr,
+ **self.response_parts).encode('ascii'))
+
+ self.metasource_data = None
+
+ def read_response(length):
+ if not self.metasource_data:
+ self.metasource_data = make_response()
+ self.metasource_data_len = len(self.metasource_data)
+ resp = self.metasource_data[:length]
+ self.metasource_data = self.metasource_data[length:]
+ return resp
+
+ self.serial.read.side_effect = read_response
self.patched_funcs.enter_context(
mock.patch('cloudinit.sources.DataSourceSmartOS.random.randint',
mock.Mock(return_value=self.request_id)))
@@ -477,7 +498,9 @@ class TestJoyentMetadataClient(helpers.FilesystemMockingTestCase):
client.get_metadata('some_key')
self.assertEqual(1, self.serial.write.call_count)
written_line = self.serial.write.call_args[0][0]
- self.assertEndsWith(written_line, b'\n')
+ print(type(written_line))
+ self.assertEndsWith(written_line.decode('ascii'),
+ b'\n'.decode('ascii'))
self.assertEqual(1, written_line.count(b'\n'))
def _get_written_line(self, key='some_key'):
@@ -489,7 +512,8 @@ class TestJoyentMetadataClient(helpers.FilesystemMockingTestCase):
self.assertIsInstance(self._get_written_line(), six.binary_type)
def test_get_metadata_line_starts_with_v2(self):
- self.assertStartsWith(self._get_written_line(), b'V2')
+ foo = self._get_written_line()
+ self.assertStartsWith(foo.decode('ascii'), b'V2'.decode('ascii'))
def test_get_metadata_uses_get_command(self):
parts = self._get_written_line().decode('ascii').strip().split(' ')
@@ -526,7 +550,7 @@ class TestJoyentMetadataClient(helpers.FilesystemMockingTestCase):
def test_get_metadata_reads_a_line(self):
client = self._get_client()
client.get_metadata('some_key')
- self.assertEqual(1, self.serial.readline.call_count)
+ self.assertEqual(self.metasource_data_len, self.serial.read.call_count)
def test_get_metadata_returns_valid_value(self):
client = self._get_client()
diff --git a/tests/unittests/test_distros/test_user_data_normalize.py b/tests/unittests/test_distros/test_user_data_normalize.py
index e4488e2a..4525f487 100644
--- a/tests/unittests/test_distros/test_user_data_normalize.py
+++ b/tests/unittests/test_distros/test_user_data_normalize.py
@@ -6,13 +6,13 @@ from ..helpers import TestCase
bcfg = {
- 'name': 'bob',
- 'plain_text_passwd': 'ubuntu',
- 'home': "/home/ubuntu",
- 'shell': "/bin/bash",
- 'lock_passwd': True,
- 'gecos': "Ubuntu",
- 'groups': ["foo"]
+ 'name': 'bob',
+ 'plain_text_passwd': 'ubuntu',
+ 'home': "/home/ubuntu",
+ 'shell': "/bin/bash",
+ 'lock_passwd': True,
+ 'gecos': "Ubuntu",
+ 'groups': ["foo"]
}
@@ -34,16 +34,11 @@ class TestUGNormalize(TestCase):
def test_group_dict(self):
distro = self._make_distro('ubuntu')
g = {'groups': [
- {
- 'ubuntu': ['foo', 'bar'],
- 'bob': 'users',
- },
- 'cloud-users',
- {
- 'bob': 'users2',
- },
- ]
- }
+ {'ubuntu': ['foo', 'bar'],
+ 'bob': 'users'},
+ 'cloud-users',
+ {'bob': 'users2'}
+ ]}
(_users, groups) = self._norm(g, distro)
self.assertIn('ubuntu', groups)
ub_members = groups['ubuntu']
diff --git a/tests/unittests/test_handler/test_handler_lxd.py b/tests/unittests/test_handler/test_handler_lxd.py
new file mode 100644
index 00000000..7ffa2a53
--- /dev/null
+++ b/tests/unittests/test_handler/test_handler_lxd.py
@@ -0,0 +1,75 @@
+from cloudinit.config import cc_lxd
+from cloudinit import (distros, helpers, cloud)
+from cloudinit.sources import DataSourceNoCloud
+from .. import helpers as t_help
+
+import logging
+
+try:
+ from unittest import mock
+except ImportError:
+ import mock
+
+LOG = logging.getLogger(__name__)
+
+
+class TestLxd(t_help.TestCase):
+ lxd_cfg = {
+ 'lxd': {
+ 'init': {
+ 'network_address': '0.0.0.0',
+ 'storage_backend': 'zfs',
+ 'storage_pool': 'poolname',
+ }
+ }
+ }
+
+ def setUp(self):
+ super(TestLxd, self).setUp()
+
+ def _get_cloud(self, distro):
+ cls = distros.fetch(distro)
+ paths = helpers.Paths({})
+ d = cls(distro, {}, paths)
+ ds = DataSourceNoCloud.DataSourceNoCloud({}, d, paths)
+ cc = cloud.Cloud(ds, paths, {}, d, None)
+ return cc
+
+ @mock.patch("cloudinit.config.cc_lxd.util")
+ def test_lxd_init(self, mock_util):
+ cc = self._get_cloud('ubuntu')
+ mock_util.which.return_value = True
+ cc_lxd.handle('cc_lxd', self.lxd_cfg, cc, LOG, [])
+ self.assertTrue(mock_util.which.called)
+ init_call = mock_util.subp.call_args_list[0][0][0]
+ self.assertEquals(init_call,
+ ['lxd', 'init', '--auto',
+ '--network-address=0.0.0.0',
+ '--storage-backend=zfs',
+ '--storage-pool=poolname'])
+
+ @mock.patch("cloudinit.config.cc_lxd.util")
+ def test_lxd_install(self, mock_util):
+ cc = self._get_cloud('ubuntu')
+ cc.distro = mock.MagicMock()
+ mock_util.which.return_value = None
+ cc_lxd.handle('cc_lxd', self.lxd_cfg, cc, LOG, [])
+ self.assertTrue(cc.distro.install_packages.called)
+ install_pkg = cc.distro.install_packages.call_args_list[0][0][0]
+ self.assertEquals(sorted(install_pkg), ['lxd', 'zfs'])
+
+ @mock.patch("cloudinit.config.cc_lxd.util")
+ def test_no_init_does_nothing(self, mock_util):
+ cc = self._get_cloud('ubuntu')
+ cc.distro = mock.MagicMock()
+ cc_lxd.handle('cc_lxd', {'lxd': {}}, cc, LOG, [])
+ self.assertFalse(cc.distro.install_packages.called)
+ self.assertFalse(mock_util.subp.called)
+
+ @mock.patch("cloudinit.config.cc_lxd.util")
+ def test_no_lxd_does_nothing(self, mock_util):
+ cc = self._get_cloud('ubuntu')
+ cc.distro = mock.MagicMock()
+ cc_lxd.handle('cc_lxd', {'package_update': True}, cc, LOG, [])
+ self.assertFalse(cc.distro.install_packages.called)
+ self.assertFalse(mock_util.subp.called)
diff --git a/tests/unittests/test_handler/test_handler_power_state.py b/tests/unittests/test_handler/test_handler_power_state.py
index 5687b10d..04ce5687 100644
--- a/tests/unittests/test_handler/test_handler_power_state.py
+++ b/tests/unittests/test_handler/test_handler_power_state.py
@@ -74,7 +74,7 @@ class TestLoadPowerState(t_help.TestCase):
class TestCheckCondition(t_help.TestCase):
def cmd_with_exit(self, rc):
return([sys.executable, '-c', 'import sys; sys.exit(%s)' % rc])
-
+
def test_true_is_true(self):
self.assertEqual(psc.check_condition(True), True)
@@ -94,7 +94,6 @@ class TestCheckCondition(t_help.TestCase):
self.assertEqual(mocklog.warn.call_count, 1)
-
def check_lps_ret(psc_return, mode=None):
if len(psc_return) != 3:
raise TypeError("length returned = %d" % len(psc_return))
@@ -107,7 +106,7 @@ def check_lps_ret(psc_return, mode=None):
if 'shutdown' not in psc_return[0][0]:
errs.append("string 'shutdown' not in cmd")
- if 'condition' is None:
+ if condition is None:
errs.append("condition was not returned")
if mode is not None:
diff --git a/tests/unittests/test_handler/test_handler_seed_random.py b/tests/unittests/test_handler/test_handler_seed_random.py
index 0bcdcb31..98bc9b81 100644
--- a/tests/unittests/test_handler/test_handler_seed_random.py
+++ b/tests/unittests/test_handler/test_handler_seed_random.py
@@ -170,27 +170,30 @@ class TestRandomSeed(t_help.TestCase):
contents = util.load_file(self._seed_file)
self.assertEquals('tiny-tim-was-here-so-was-josh', contents)
- def test_seed_command_not_provided_pollinate_available(self):
+ def test_seed_command_provided_and_available(self):
c = self._get_cloud('ubuntu', {})
self.whichdata = {'pollinate': '/usr/bin/pollinate'}
- cc_seed_random.handle('test', {}, c, LOG, [])
+ cfg = {'random_seed': {'command': ['pollinate', '-q']}}
+ cc_seed_random.handle('test', cfg, c, LOG, [])
subp_args = [f['args'] for f in self.subp_called]
self.assertIn(['pollinate', '-q'], subp_args)
- def test_seed_command_not_provided_pollinate_not_available(self):
+ def test_seed_command_not_provided(self):
c = self._get_cloud('ubuntu', {})
self.whichdata = {}
cc_seed_random.handle('test', {}, c, LOG, [])
# subp should not have been called as which would say not available
- self.assertEquals(self.subp_called, list())
+ self.assertFalse(self.subp_called)
def test_unavailable_seed_command_and_required_raises_error(self):
c = self._get_cloud('ubuntu', {})
self.whichdata = {}
+ cfg = {'random_seed': {'command': ['THIS_NO_COMMAND'],
+ 'command_required': True}}
self.assertRaises(ValueError, cc_seed_random.handle,
- 'test', {'random_seed': {'command_required': True}}, c, LOG, [])
+ 'test', cfg, c, LOG, [])
def test_seed_command_and_required(self):
c = self._get_cloud('ubuntu', {})
diff --git a/tests/unittests/test_handler/test_handler_snappy.py b/tests/unittests/test_handler/test_handler_snappy.py
index eceb14d9..8aeff53c 100644
--- a/tests/unittests/test_handler/test_handler_snappy.py
+++ b/tests/unittests/test_handler/test_handler_snappy.py
@@ -125,8 +125,7 @@ class TestInstallPackages(t_help.TestCase):
"pkg1.smoser.config": "pkg1.smoser.config-data",
"pkg1.config": "pkg1.config-data",
"pkg2.smoser_0.0_amd64.snap": "pkg2-snapdata",
- "pkg2.smoser_0.0_amd64.config": "pkg2.config",
- })
+ "pkg2.smoser_0.0_amd64.config": "pkg2.config"})
ret = get_package_ops(
packages=[], configs={}, installed=[], fspath=self.tmp)
diff --git a/tests/unittests/test_sshutil.py b/tests/unittests/test_sshutil.py
index 3b317121..9aeb1cde 100644
--- a/tests/unittests/test_sshutil.py
+++ b/tests/unittests/test_sshutil.py
@@ -32,7 +32,8 @@ VALID_CONTENT = {
),
}
-TEST_OPTIONS = ("no-port-forwarding,no-agent-forwarding,no-X11-forwarding,"
+TEST_OPTIONS = (
+ "no-port-forwarding,no-agent-forwarding,no-X11-forwarding,"
'command="echo \'Please login as the user \"ubuntu\" rather than the'
'user \"root\".\';echo;sleep 10"')
diff --git a/tests/unittests/test_templating.py b/tests/unittests/test_templating.py
index 0c19a2c2..b9863650 100644
--- a/tests/unittests/test_templating.py
+++ b/tests/unittests/test_templating.py
@@ -114,5 +114,6 @@ $a,$b'''
codename)
out_data = templater.basic_render(in_data,
- {'mirror': mirror, 'codename': codename})
+ {'mirror': mirror,
+ 'codename': codename})
self.assertEqual(ex_data, out_data)
diff --git a/tests/unittests/test_util.py b/tests/unittests/test_util.py
index 95990165..0a986fec 100644
--- a/tests/unittests/test_util.py
+++ b/tests/unittests/test_util.py
@@ -385,6 +385,16 @@ class TestReadDMIData(helpers.FilesystemMockingTestCase):
self.patch_mapping({})
self.assertEqual(None, util.read_dmi_data('expect-fail'))
+ def test_dots_returned_instead_of_foxfox(self):
+ # uninitialized dmi values show as \xff, return those as .
+ my_len = 32
+ dmi_value = b'\xff' * my_len + b'\n'
+ expected = '.' * my_len
+ dmi_key = 'system-product-name'
+ sysfs_key = 'product_name'
+ self._create_sysfs_file(sysfs_key, dmi_value)
+ self.assertEqual(expected, util.read_dmi_data(dmi_key))
+
class TestMultiLog(helpers.FilesystemMockingTestCase):
generated by cgit v1.2.3 (git 2.39.1) at 2025-10-31 07:00:50 +0000