summaryrefslogtreecommitdiff
path: root/tests/unittests
diff options
context:
space:
mode:
authorJoe VLcek <JoeV@RedHat.com>2012-08-10 15:29:42 -0400
committerJoe VLcek <JoeV@RedHat.com>2012-08-10 15:29:42 -0400
commit6205d39ad5fed1bba31e4a97656beb213db251bc (patch)
treedda214d4f2ba842a76fb12d1e8ad2c17f1acc1e3 /tests/unittests
parent6f2dc0d943d3f3527983b38f20085a6cf071c43e (diff)
downloadvyos-cloud-init-6205d39ad5fed1bba31e4a97656beb213db251bc.tar.gz
vyos-cloud-init-6205d39ad5fed1bba31e4a97656beb213db251bc.zip
Address review feedback for:
https://code.launchpad.net/~joev-n/cloud-init/altcloud-changes/+merge/116542/comments/255564 https://code.launchpad.net/~joev-n/cloud-init/altcloud-changes/+merge/116542/comments/255565
Diffstat (limited to 'tests/unittests')
-rw-r--r--tests/unittests/test_datasource/test_altcloud.py255
1 files changed, 131 insertions, 124 deletions
diff --git a/tests/unittests/test_datasource/test_altcloud.py b/tests/unittests/test_datasource/test_altcloud.py
index 27912652..b828711c 100644
--- a/tests/unittests/test_datasource/test_altcloud.py
+++ b/tests/unittests/test_datasource/test_altcloud.py
@@ -1,6 +1,29 @@
-#! /usr/bin/env python
+# vi: ts=4 expandtab
+#
+# Copyright (C) 2009-2010 Canonical Ltd.
+# Copyright (C) 2012 Hewlett-Packard Development Company, L.P.
+# Copyright (C) 2012 Yahoo! Inc.
+#
+# Author: Joe VLcek <JVLcek@RedHat.com>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+'''
+This test file exercises the code in sources DataSourceAltCloud.py
+'''
import os
+import shutil
+import tempfile
from unittest import TestCase
from cloudinit import helpers
@@ -16,9 +39,9 @@ def _write_cloud_info_file(value):
with a cloud backend identifier ImageFactory when building
an image with ImageFactory.
'''
- f = open(cloudinit.sources.DataSourceAltCloud.CLOUD_INFO_FILE, 'w')
- f.write(value)
- f.close()
+ cifile = open(cloudinit.sources.DataSourceAltCloud.CLOUD_INFO_FILE, 'w')
+ cifile.write(value)
+ cifile.close()
os.chmod(cloudinit.sources.DataSourceAltCloud.CLOUD_INFO_FILE, 0664)
def _remove_cloud_info_file():
@@ -35,14 +58,14 @@ def _write_user_data_files(mount_dir, value):
deltacloud_user_data_file = mount_dir + '/deltacloud-user-data.txt'
user_data_file = mount_dir + '/user-data.txt'
- f = open(deltacloud_user_data_file, 'w')
- f.write(value)
- f.close()
+ udfile = open(deltacloud_user_data_file, 'w')
+ udfile.write(value)
+ udfile.close()
os.chmod(deltacloud_user_data_file, 0664)
- f = open(user_data_file, 'w')
- f.write(value)
- f.close()
+ udfile = open(user_data_file, 'w')
+ udfile.write(value)
+ udfile.close()
os.chmod(user_data_file, 0664)
def _remove_user_data_files(mount_dir,
@@ -68,8 +91,7 @@ def _remove_user_data_files(mount_dir,
except OSError:
pass
-
-class TestDataSouceAltCloud_get_cloud_type(TestCase):
+class TestGetCloudType(TestCase):
'''
Test to exercise method: DataSourceAltCloud.get_cloud_type()
'''
@@ -77,70 +99,66 @@ class TestDataSouceAltCloud_get_cloud_type(TestCase):
def setUp(self):
''' Set up '''
self.paths = helpers.Paths({ 'cloud_dir': '/tmp' })
- cloudinit.sources.DataSourceAltCloud.RETRY_TIMES = 1
- cloudinit.sources.DataSourceAltCloud.SLEEP_SECS = 1
def tearDown(self):
# Reset
cloudinit.sources.DataSourceAltCloud.CMD_DMI_SYSTEM = \
['dmidecode', '--string', 'system-product-name']
- cloudinit.sources.DataSourceAltCloud.RETRY_TIMES = 3
- cloudinit.sources.DataSourceAltCloud.SLEEP_SECS = 3
- def test_get_cloud_type_RHEV(self):
+ def test_rhev(self):
'''
Test method get_cloud_type() for RHEVm systems.
Forcing dmidecode return to match a RHEVm system: RHEV Hypervisor
'''
cloudinit.sources.DataSourceAltCloud.CMD_DMI_SYSTEM = \
['echo', 'RHEV Hypervisor']
- ds = DataSourceAltCloud({}, None, self.paths)
+ dsrc = DataSourceAltCloud({}, None, self.paths)
self.assertEquals('RHEV', \
- ds.get_cloud_type())
+ dsrc.get_cloud_type())
- def test_get_cloud_type_VSPHERE(self):
+ def test_vsphere(self):
'''
Test method get_cloud_type() for vSphere systems.
Forcing dmidecode return to match a vSphere system: RHEV Hypervisor
'''
cloudinit.sources.DataSourceAltCloud.CMD_DMI_SYSTEM = \
['echo', 'VMware Virtual Platform']
- ds = DataSourceAltCloud({}, None, self.paths)
+ dsrc = DataSourceAltCloud({}, None, self.paths)
self.assertEquals('VSPHERE', \
- ds.get_cloud_type())
+ dsrc.get_cloud_type())
- def test_get_cloud_type_UNKNOWN(self):
+ def test_unknown(self):
'''
Test method get_cloud_type() for unknown systems.
Forcing dmidecode return to match an unrecognized return.
'''
cloudinit.sources.DataSourceAltCloud.CMD_DMI_SYSTEM = \
['echo', 'Unrecognized Platform']
- ds = DataSourceAltCloud({}, None, self.paths)
+ dsrc = DataSourceAltCloud({}, None, self.paths)
self.assertEquals('UNKNOWN', \
- ds.get_cloud_type())
+ dsrc.get_cloud_type())
- def test_get_cloud_type_exception1(self):
+ def test_exception1(self):
'''
Test method get_cloud_type() where command dmidecode fails.
'''
cloudinit.sources.DataSourceAltCloud.CMD_DMI_SYSTEM = \
['ls', 'bad command']
- ds = DataSourceAltCloud({}, None, self.paths)
+ dsrc = DataSourceAltCloud({}, None, self.paths)
self.assertEquals('UNKNOWN', \
- ds.get_cloud_type())
+ dsrc.get_cloud_type())
- def test_get_cloud_type_exception(self):
+ def test_exception2(self):
'''
Test method get_cloud_type() where command dmidecode is not available.
'''
cloudinit.sources.DataSourceAltCloud.CMD_DMI_SYSTEM = \
['bad command']
- ds = DataSourceAltCloud({}, None, self.paths)
+ dsrc = DataSourceAltCloud({}, None, self.paths)
self.assertEquals('UNKNOWN', \
- ds.get_cloud_type())
+ dsrc.get_cloud_type())
-class TestDataSouceAltCloud_get_data_cloud_info_file(TestCase):
+class TestGetDataCloudInfoFile(TestCase):
'''
Test to exercise method: DataSourceAltCloud.get_data()
With a contrived CLOUD_INFO_FILE
@@ -148,58 +166,62 @@ class TestDataSouceAltCloud_get_data_cloud_info_file(TestCase):
def setUp(self):
''' Set up '''
self.paths = helpers.Paths({ 'cloud_dir': '/tmp' })
+ self.cloud_info_file = tempfile.mkstemp()[1]
cloudinit.sources.DataSourceAltCloud.CLOUD_INFO_FILE = \
- '/tmp/cloudinit_test_etc_sysconfig_cloud-info'
- cloudinit.sources.DataSourceAltCloud.RETRY_TIMES = 1
- cloudinit.sources.DataSourceAltCloud.SLEEP_SECS = 1
+ self.cloud_info_file
def tearDown(self):
# Reset
+
+ # Attempt to remove the temp file ignoring errors
+ try:
+ os.remove(self.cloud_info_file)
+ except OSError:
+ pass
+
cloudinit.sources.DataSourceAltCloud.CLOUD_INFO_FILE = \
'/etc/sysconfig/cloud-info'
- cloudinit.sources.DataSourceAltCloud.RETRY_TIMES = 3
- cloudinit.sources.DataSourceAltCloud.SLEEP_SECS = 3
- def test_get_data_RHEV_cloud_file(self):
+ def test_rhev(self):
'''Success Test module get_data() forcing RHEV '''
_write_cloud_info_file('RHEV')
- ds = DataSourceAltCloud({}, None, self.paths)
- ds.user_data_rhevm = lambda : True
- self.assertEquals(True, ds.get_data())
+ dsrc = DataSourceAltCloud({}, None, self.paths)
+ dsrc.user_data_rhevm = lambda : True
+ self.assertEquals(True, dsrc.get_data())
- def test_get_data_VSPHERE_cloud_file(self):
+ def test_vsphere(self):
'''Success Test module get_data() forcing VSPHERE '''
_write_cloud_info_file('VSPHERE')
- ds = DataSourceAltCloud({}, None, self.paths)
- ds.user_data_vsphere = lambda : True
- self.assertEquals(True, ds.get_data())
+ dsrc = DataSourceAltCloud({}, None, self.paths)
+ dsrc.user_data_vsphere = lambda : True
+ self.assertEquals(True, dsrc.get_data())
- def test_failure_get_data_RHEV_cloud_file(self):
+ def test_fail_rhev(self):
'''Failure Test module get_data() forcing RHEV '''
_write_cloud_info_file('RHEV')
- ds = DataSourceAltCloud({}, None, self.paths)
- ds.user_data_rhevm = lambda : False
- self.assertEquals(False, ds.get_data())
+ dsrc = DataSourceAltCloud({}, None, self.paths)
+ dsrc.user_data_rhevm = lambda : False
+ self.assertEquals(False, dsrc.get_data())
- def test_failure_get_data_VSPHERE_cloud_file(self):
+ def test_fail_vsphere(self):
'''Failure Test module get_data() forcing VSPHERE '''
_write_cloud_info_file('VSPHERE')
- ds = DataSourceAltCloud({}, None, self.paths)
- ds.user_data_vsphere = lambda : False
- self.assertEquals(False, ds.get_data())
+ dsrc = DataSourceAltCloud({}, None, self.paths)
+ dsrc.user_data_vsphere = lambda : False
+ self.assertEquals(False, dsrc.get_data())
- def test_failure_get_data_unrecognized_cloud_file(self):
+ def test_unrecognized(self):
'''Failure Test module get_data() forcing unrecognized '''
_write_cloud_info_file('unrecognized')
- ds = DataSourceAltCloud({}, None, self.paths)
- self.assertEquals(False, ds.get_data())
+ dsrc = DataSourceAltCloud({}, None, self.paths)
+ self.assertEquals(False, dsrc.get_data())
-class TestDataSouceAltCloud_get_data_no_cloud_info_file(TestCase):
+class TestGetDataNoCloudInfoFile(TestCase):
'''
Test to exercise method: DataSourceAltCloud.get_data()
Without a CLOUD_INFO_FILE
@@ -209,8 +231,6 @@ class TestDataSouceAltCloud_get_data_no_cloud_info_file(TestCase):
self.paths = helpers.Paths({ 'cloud_dir': '/tmp' })
cloudinit.sources.DataSourceAltCloud.CLOUD_INFO_FILE = \
'no such file'
- cloudinit.sources.DataSourceAltCloud.RETRY_TIMES = 1
- cloudinit.sources.DataSourceAltCloud.SLEEP_SECS = 1
def tearDown(self):
# Reset
@@ -218,55 +238,41 @@ class TestDataSouceAltCloud_get_data_no_cloud_info_file(TestCase):
'/etc/sysconfig/cloud-info'
cloudinit.sources.DataSourceAltCloud.CMD_DMI_SYSTEM = \
['dmidecode', '--string', 'system-product-name']
- cloudinit.sources.DataSourceAltCloud.RETRY_TIMES = 3
- cloudinit.sources.DataSourceAltCloud.SLEEP_SECS = 3
- def test_get_data_RHEV_cloud_file(self):
+ def test_rhev_no_cloud_file(self):
'''Test No cloud info file module get_data() forcing RHEV '''
cloudinit.sources.DataSourceAltCloud.CMD_DMI_SYSTEM = \
['echo', 'RHEV Hypervisor']
- ds = DataSourceAltCloud({}, None, self.paths)
- ds.user_data_rhevm = lambda : True
- self.assertEquals(True, ds.get_data())
+ dsrc = DataSourceAltCloud({}, None, self.paths)
+ dsrc.user_data_rhevm = lambda : True
+ self.assertEquals(True, dsrc.get_data())
- def test_get_data_VSPHERE_cloud_file(self):
+ def test_vsphere_no_cloud_file(self):
'''Test No cloud info file module get_data() forcing VSPHERE '''
cloudinit.sources.DataSourceAltCloud.CMD_DMI_SYSTEM = \
['echo', 'VMware Virtual Platform']
- ds = DataSourceAltCloud({}, None, self.paths)
- ds.user_data_vsphere = lambda : True
- self.assertEquals(True, ds.get_data())
+ dsrc = DataSourceAltCloud({}, None, self.paths)
+ dsrc.user_data_vsphere = lambda : True
+ self.assertEquals(True, dsrc.get_data())
- def test_failure_get_data_VSPHERE_cloud_file(self):
+ def test_failure_no_cloud_file(self):
'''Test No cloud info file module get_data() forcing unrecognized '''
cloudinit.sources.DataSourceAltCloud.CMD_DMI_SYSTEM = \
['echo', 'Unrecognized Platform']
- ds = DataSourceAltCloud({}, None, self.paths)
- self.assertEquals(False, ds.get_data())
+ dsrc = DataSourceAltCloud({}, None, self.paths)
+ self.assertEquals(False, dsrc.get_data())
-class TestDataSouceAltCloud_user_data_rhevm(TestCase):
+class TestUserDataRhevm(TestCase):
'''
Test to exercise method: DataSourceAltCloud.user_data_rhevm()
'''
def setUp(self):
''' Set up '''
self.paths = helpers.Paths({ 'cloud_dir': '/tmp' })
- self.mount_dir = '/tmp/cloudinit_test_media'
-
- cloudinit.sources.DataSourceAltCloud.CLOUD_INFO_FILE = \
- '/tmp/cloudinit_test_etc_sysconfig_cloud-info'
- cloudinit.sources.DataSourceAltCloud.RETRY_TIMES = 1
- cloudinit.sources.DataSourceAltCloud.SLEEP_SECS = 1
-
- try:
- os.mkdir(self.mount_dir)
- except OSError, (errno, strerror):
- # Ignore OSError: [Errno 17] File exists:
- if errno is not 17:
- raise
+ self.mount_dir = tempfile.mkdtemp()
_write_user_data_files(self.mount_dir, 'test user data')
@@ -275,64 +281,55 @@ class TestDataSouceAltCloud_user_data_rhevm(TestCase):
_remove_user_data_files(self.mount_dir)
+ # Attempt to remove the temp dir ignoring errors
+ try:
+ shutil.rmtree(self.mount_dir)
+ except OSError:
+ pass
+
cloudinit.sources.DataSourceAltCloud.CLOUD_INFO_FILE = \
'/etc/sysconfig/cloud-info'
-
cloudinit.sources.DataSourceAltCloud.CMD_PROBE_FLOPPY = \
['/sbin/modprobe', 'floppy']
- cloudinit.sources.DataSourceAltCloud.RETRY_TIMES = 3
- cloudinit.sources.DataSourceAltCloud.SLEEP_SECS = 3
- def test_user_data_rhevm(self):
+ def test_mount_cb_fails(self):
'''Test user_data_rhevm() where mount_cb fails'''
cloudinit.sources.DataSourceAltCloud.CMD_PROBE_FLOPPY = \
['echo', 'modprobe floppy']
- ds = DataSourceAltCloud({}, None, self.paths)
+ dsrc = DataSourceAltCloud({}, None, self.paths)
- self.assertEquals(False, ds.user_data_rhevm())
+ self.assertEquals(False, dsrc.user_data_rhevm())
- def test_user_data_rhevm_modprobe_fails(self):
+ def test_modprobe_fails(self):
'''Test user_data_rhevm() where modprobe fails. '''
cloudinit.sources.DataSourceAltCloud.CMD_PROBE_FLOPPY = \
['ls', 'modprobe floppy']
- ds = DataSourceAltCloud({}, None, self.paths)
+ dsrc = DataSourceAltCloud({}, None, self.paths)
- self.assertEquals(False, ds.user_data_rhevm())
+ self.assertEquals(False, dsrc.user_data_rhevm())
- def test_user_data_rhevm_no_modprobe_cmd(self):
+ def test_no_modprobe_cmd(self):
'''Test user_data_rhevm() with no modprobe command. '''
cloudinit.sources.DataSourceAltCloud.CMD_PROBE_FLOPPY = \
['bad command', 'modprobe floppy']
- ds = DataSourceAltCloud({}, None, self.paths)
+ dsrc = DataSourceAltCloud({}, None, self.paths)
- self.assertEquals(False, ds.user_data_rhevm())
+ self.assertEquals(False, dsrc.user_data_rhevm())
-class TestDataSouceAltCloud_user_data_vsphere(TestCase):
+class TestUserDataVsphere(TestCase):
'''
Test to exercise method: DataSourceAltCloud.user_data_vsphere()
'''
def setUp(self):
''' Set up '''
self.paths = helpers.Paths({ 'cloud_dir': '/tmp' })
- self.mount_dir = '/tmp/cloudinit_test_media'
-
- cloudinit.sources.DataSourceAltCloud.CLOUD_INFO_FILE = \
- '/tmp/cloudinit_test_etc_sysconfig_cloud-info'
- cloudinit.sources.DataSourceAltCloud.RETRY_TIMES = 1
- cloudinit.sources.DataSourceAltCloud.SLEEP_SECS = 1
-
- try:
- os.mkdir(self.mount_dir)
- except OSError, (errno, strerror):
- # Ignore OSError: [Errno 17] File exists:
- if errno is not 17:
- raise
+ self.mount_dir = tempfile.mkdtemp()
_write_user_data_files(self.mount_dir, 'test user data')
@@ -341,29 +338,32 @@ class TestDataSouceAltCloud_user_data_vsphere(TestCase):
_remove_user_data_files(self.mount_dir)
+ # Attempt to remove the temp dir ignoring errors
+ try:
+ shutil.rmtree(self.mount_dir)
+ except OSError:
+ pass
+
cloudinit.sources.DataSourceAltCloud.CLOUD_INFO_FILE = \
'/etc/sysconfig/cloud-info'
- cloudinit.sources.DataSourceAltCloud.RETRY_TIMES = 3
- cloudinit.sources.DataSourceAltCloud.SLEEP_SECS = 3
def test_user_data_vsphere(self):
'''Test user_data_vsphere() where mount_cb fails'''
- cloudinit.sources.DataSourceAltCloud.MEDIA_DIR = \
- '/tmp/cloudinit_test_media'
+ cloudinit.sources.DataSourceAltCloud.MEDIA_DIR = self.mount_dir
- ds = DataSourceAltCloud({}, None, self.paths)
+ dsrc = DataSourceAltCloud({}, None, self.paths)
- self.assertEquals(False, ds.user_data_vsphere())
+ self.assertEquals(False, dsrc.user_data_vsphere())
-class TestDataSouceAltCloud_read_user_data_callback(TestCase):
+class TestReadUserDataCallback(TestCase):
'''
Test to exercise method: DataSourceAltCloud.read_user_data_callback()
'''
def setUp(self):
''' Set up '''
self.paths = helpers.Paths({ 'cloud_dir': '/tmp' })
- self.mount_dir = '/tmp/cloudinit_test_media'
+ self.mount_dir = tempfile.mkdtemp()
_write_user_data_files(self.mount_dir, 'test user data')
@@ -372,13 +372,20 @@ class TestDataSouceAltCloud_read_user_data_callback(TestCase):
_remove_user_data_files(self.mount_dir)
- def test_read_user_data_callback_both(self):
+ # Attempt to remove the temp dir ignoring errors
+ try:
+ shutil.rmtree(self.mount_dir)
+ except OSError:
+ pass
+
+
+ def test_callback_both(self):
'''Test read_user_data_callback() with both files'''
self.assertEquals('test user data',
read_user_data_callback(self.mount_dir))
- def test_read_user_data_callback_dc(self):
+ def test_callback_dc(self):
'''Test read_user_data_callback() with only DC file'''
_remove_user_data_files(self.mount_dir,
@@ -388,7 +395,7 @@ class TestDataSouceAltCloud_read_user_data_callback(TestCase):
self.assertEquals('test user data',
read_user_data_callback(self.mount_dir))
- def test_read_user_data_callback_non_dc(self):
+ def test_callback_non_dc(self):
'''Test read_user_data_callback() with only non-DC file'''
_remove_user_data_files(self.mount_dir,
@@ -398,7 +405,7 @@ class TestDataSouceAltCloud_read_user_data_callback(TestCase):
self.assertEquals('test user data',
read_user_data_callback(self.mount_dir))
- def test_read_user_data_callback_none(self):
+ def test_callback_none(self):
'''Test read_user_data_callback() no files are found'''
_remove_user_data_files(self.mount_dir)