From d0e7898d4d0aef452fbfdd9b51fa2da6437397ff Mon Sep 17 00:00:00 2001 From: Vlastimil Holer Date: Thu, 5 Sep 2013 17:55:53 +0200 Subject: PEP8 and Pylint fixes. Move context.sh "parser" into separate function. Fix fetching user specified dsmode (from context). Rename context_sh->context. Reuse unittests.helpers.populate_dir. --- tests/unittests/test_datasource/test_opennebula.py | 62 +++++++++++----------- 1 file changed, 31 insertions(+), 31 deletions(-) (limited to 'tests/unittests') diff --git a/tests/unittests/test_datasource/test_opennebula.py b/tests/unittests/test_datasource/test_opennebula.py index 66a38e31..4b82a49c 100644 --- a/tests/unittests/test_datasource/test_opennebula.py +++ b/tests/unittests/test_datasource/test_opennebula.py @@ -1,8 +1,9 @@ -import os - from cloudinit.sources import DataSourceOpenNebula as ds from cloudinit import util from mocker import MockerTestCase +from tests.unittests.helpers import populate_dir + +import os TEST_VARS = { 'var1': 'single', @@ -13,7 +14,11 @@ TEST_VARS = { 'var6': "'multi\nline\n'", 'var7': 'single\\t', 'var8': 'double\\tword', - 'var9': 'multi\\t\nline\n'} + 'var9': 'multi\\t\nline\n', + 'var10': '\\', # expect \ + 'var11': '\'', # expect ' + 'var12': '$', # expect $ +} USER_DATA = '#cloud-config\napt_upgrade: true' SSH_KEY = 'ssh-rsa AAAAB3NzaC1....sIkJhq8wdX+4I3A4cYbYP ubuntu@server-460-%i' @@ -38,17 +43,15 @@ class TestOpenNebulaDataSource(MockerTestCase): my_d = os.path.join(self.tmp, 'non-contextdisk') self.assertRaises(ds.NonContextDiskDir, ds.read_context_disk_dir, my_d) - def test_seed_dir_bad_context_sh(self): - my_d = os.path.join(self.tmp, 'bad-context-sh') + def test_seed_dir_bad_context(self): + my_d = os.path.join(self.tmp, 'bad-context') os.mkdir(my_d) - with open(os.path.join(my_d, "context.sh"), "w") as fp: - fp.write('/bin/false\n') - fp.close() + open(os.path.join(my_d, "context.sh"), "w").close() self.assertRaises(ds.NonContextDiskDir, ds.read_context_disk_dir, my_d) - def test_context_sh_parser(self): - my_d = os.path.join(self.tmp, 'context-sh-parser') - populate_dir(my_d, TEST_VARS) + def test_context_parser(self): + my_d = os.path.join(self.tmp, 'context-parser') + populate_context_dir(my_d, TEST_VARS) results = ds.read_context_disk_dir(my_d) self.assertTrue('metadata' in results) @@ -59,7 +62,7 @@ class TestOpenNebulaDataSource(MockerTestCase): for c in range(4): for k in ('SSH_KEY', 'SSH_PUBLIC_KEY'): my_d = os.path.join(self.tmp, "%s-%i" % (k, c)) - populate_dir(my_d, {k: '\n'.join(public_keys)}) + populate_context_dir(my_d, {k: '\n'.join(public_keys)}) results = ds.read_context_disk_dir(my_d) self.assertTrue('metadata' in results) @@ -72,7 +75,7 @@ class TestOpenNebulaDataSource(MockerTestCase): def test_user_data(self): for k in ('USER_DATA', 'USERDATA'): my_d = os.path.join(self.tmp, k) - populate_dir(my_d, {k: USER_DATA}) + populate_context_dir(my_d, {k: USER_DATA}) results = ds.read_context_disk_dir(my_d) self.assertTrue('userdata' in results) @@ -81,7 +84,7 @@ class TestOpenNebulaDataSource(MockerTestCase): def test_hostname(self): for k in ('HOSTNAME', 'PUBLIC_IP', 'IP_PUBLIC', 'ETH0_IP'): my_d = os.path.join(self.tmp, k) - populate_dir(my_d, {k: PUBLIC_IP}) + populate_context_dir(my_d, {k: PUBLIC_IP}) results = ds.read_context_disk_dir(my_d) self.assertTrue('metadata' in results) @@ -89,14 +92,12 @@ class TestOpenNebulaDataSource(MockerTestCase): self.assertEqual(PUBLIC_IP, results['metadata']['local-hostname']) def test_find_candidates(self): - devs_with_answers = { - "TYPE=iso9660": ["/dev/vdb"], - "LABEL=CDROM": ["/dev/sr0"], - "LABEL=CONTEXT": ["/dev/sdb"], - } - def my_devs_with(criteria): - return devs_with_answers[criteria] + return { + "LABEL=CONTEXT": ["/dev/sdb"], + "LABEL=CDROM": ["/dev/sr0"], + "TYPE=iso9660": ["/dev/vdb"], + }.get(criteria, []) try: orig_find_devs_with = util.find_devs_with @@ -133,16 +134,17 @@ iface eth0 inet static ''') def test_eth0_override(self): - context_sh = { + context = { 'dns': '1.2.3.8', 'eth0_ip': '1.2.3.4', 'eth0_network': '1.2.3.0', 'eth0_mask': '255.255.0.0', 'eth0_gateway': '1.2.3.5', 'eth0_domain': 'example.com', - 'eth0_dns': '1.2.3.6 1.2.3.7'} + 'eth0_dns': '1.2.3.6 1.2.3.7' + } - net = ds.OpenNebulaNetwork(CMD_IP_OUT, context_sh) + net = ds.OpenNebulaNetwork(CMD_IP_OUT, context) self.assertEqual(net.gen_conf(), u'''\ auto lo iface lo inet loopback @@ -158,12 +160,10 @@ iface eth0 inet static ''') -def populate_dir(seed_dir, files): - os.mkdir(seed_dir) - with open(os.path.join(seed_dir, "context.sh"), "w") as fp: - fp.write("# Context variables generated by OpenNebula\n") - for (name, content) in files.iteritems(): - fp.write("%s='%s'\n" % (name.upper(), content.replace(r"'", r"'\''"))) - fp.close() +def populate_context_dir(path, variables): + data = "# Context variables generated by OpenNebula\n" + for (k, v) in variables.iteritems(): + data += ("%s='%s'\n" % (k.upper(), v.replace(r"'", r"'\''"))) + populate_dir(path, {'context.sh': data}) # vi: ts=4 expandtab -- cgit v1.2.3