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. 

1 files changed, 31 insertions, 31 deletions

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