1. Add a helper for tests to use to load resource/data files from

2. Add a set of tests+data that ensure the launch index filtering
   works as expected in the various modes including raw yaml 
   and via mime/email message formats.

2 files changed, 174 insertions, 0 deletions

diff --git a/tests/unittests/helpers.py b/tests/unittests/helpers.py
new file mode 100644
index 00000000..60891191
--- /dev/null
+++ b/tests/unittests/helpers.py
@@ -0,0 +1,41 @@
+import os
+
+from mocker import MockerTestCase
+
+from cloudinit import helpers as ch
+
+
+class ResourceUsingTestCase(MockerTestCase):
+    def __init__(self, methodName="runTest"):
+        MockerTestCase.__init__(self, methodName)
+        self.resource_path = None
+
+    def resourceLocation(self, subname=None):
+        if self.resource_path is None:
+            paths = [
+                os.path.join('tests', 'data'),
+                os.path.join('data'),
+                os.path.join(os.pardir, 'tests', 'data'),
+                os.path.join(os.pardir, 'data'),
+            ]
+            for p in paths:
+                if os.path.isdir(p):
+                    self.resource_path = p
+                    break
+        self.assertTrue((self.resource_path and os.path.isdir(self.resource_path)),
+                        msg="Unable to locate test resource data path!")
+        if not subname:
+            return self.resource_path
+        return os.path.join(self.resource_path, subname)
+
+    def readResource(self, name):
+        where = self.resourceLocation(name)
+        with open(where, 'r') as fh:
+            return fh.read()
+
+    def getCloudPaths(self):
+        cp = ch.Paths({
+            'cloud_dir': self.makeDir(),
+            'templates_dir': self.resourceLocation(),
+        })
+        return cp
diff --git a/tests/unittests/test_filters/test_launch_index.py b/tests/unittests/test_filters/test_launch_index.py
new file mode 100644
index 00000000..c122609a
--- /dev/null
+++ b/tests/unittests/test_filters/test_launch_index.py
@@ -0,0 +1,133 @@
+import os
+import copy
+import sys
+
+import helpers as th
+
+import itertools
+
+from cloudinit import user_data as ud
+from cloudinit import util
+from cloudinit.filters import launch_index
+
+
+def count_messages(root):
+    am = 0
+    for m in root.walk():
+        if ud.is_skippable(m):
+            continue
+        am += 1
+    return am
+
+
+
+class TestLaunchFilter(th.ResourceUsingTestCase):
+
+    def assertCounts(self, message, expected_counts):
+        orig_message = copy.deepcopy(message)
+        for (index, count) in expected_counts.items():
+            filtered_message = launch_index.Filter(util.safe_int(index)).apply(message)
+            self.assertEquals(count_messages(filtered_message), count)
+        # Ensure original message still ok/not modified
+        self.assertTrue(self.equivalentMessage(message, orig_message))
+
+    def equivalentMessage(self, msg1, msg2):
+        msg1_count = count_messages(msg1)
+        msg2_count = count_messages(msg2)
+        if msg1_count != msg2_count:
+            return False
+        # Do some basic payload checking
+        msg1_msgs = [m for m in msg1.walk()]
+        msg1_msgs = [m for m in
+                     itertools.ifilterfalse(ud.is_skippable, msg1_msgs)]
+        msg2_msgs = [m for m in msg2.walk()]
+        msg2_msgs = [m for m in
+                     itertools.ifilterfalse(ud.is_skippable, msg2_msgs)]
+        for i in range(0, len(msg2_msgs)):
+            m1_msg = msg1_msgs[i]
+            m2_msg = msg2_msgs[i]
+            if m1_msg.get_charset() != m2_msg.get_charset():
+                return False
+            if m1_msg.is_multipart() != m2_msg.is_multipart():
+                return False
+            if m1_msg.get_payload(decode=True) != m2_msg.get_payload(decode=True):
+                return False
+        return True
+
+    def testMultiEmailIndex(self):
+        test_data = self.readResource('filter_cloud_multipart_2.email')
+        ud_proc = ud.UserDataProcessor(self.getCloudPaths())
+        message = ud_proc.process(test_data)
+        self.assertTrue(count_messages(message) > 0)
+        # This file should have the following
+        # indexes -> amount mapping in it
+        expected_counts = {
+            3: 1,
+            2: 2,
+            None: 3,
+            -1: 0,
+        }
+        self.assertCounts(message, expected_counts)
+
+    def testHeaderEmailIndex(self):
+        test_data = self.readResource('filter_cloud_multipart_header.email')
+        ud_proc = ud.UserDataProcessor(self.getCloudPaths())
+        message = ud_proc.process(test_data)
+        self.assertTrue(count_messages(message) > 0)
+        # This file should have the following
+        # indexes -> amount mapping in it
+        expected_counts = {
+            5: 1,
+            -1: 0,
+            None: 1,
+        }
+        self.assertCounts(message, expected_counts)
+    
+    def testConfigEmailIndex(self):
+        test_data = self.readResource('filter_cloud_multipart_1.email')
+        ud_proc = ud.UserDataProcessor(self.getCloudPaths())
+        message = ud_proc.process(test_data)
+        self.assertTrue(count_messages(message) > 0)
+        # This file should have the following
+        # indexes -> amount mapping in it
+        expected_counts = {
+            2: 1,
+            -1: 0,
+            None: 1,
+        }
+        self.assertCounts(message, expected_counts)
+    
+    def testNoneIndex(self):
+        test_data = self.readResource('filter_cloud_multipart.yaml')
+        ud_proc = ud.UserDataProcessor(self.getCloudPaths())
+        message = ud_proc.process(test_data)
+        start_count = count_messages(message)
+        self.assertTrue(start_count > 0)
+        filtered_message = launch_index.Filter(None).apply(message)
+        self.assertTrue(self.equivalentMessage(message, filtered_message))
+    
+    def testIndexes(self):
+        test_data = self.readResource('filter_cloud_multipart.yaml')
+        ud_proc = ud.UserDataProcessor(self.getCloudPaths())
+        message = ud_proc.process(test_data)
+        start_count = count_messages(message)
+        self.assertTrue(start_count > 0)
+        # This file should have the following
+        # indexes -> amount mapping in it
+        expected_counts = {
+            2:2,
+            3:2,
+            1:2,
+            0:1,
+            4:1,
+            7:0,
+            -1:0,
+            100:0,
+            # None should just give all back
+            None: start_count,
+            # Non ints should be ignored
+            'c': start_count,
+            # Strings should be converted
+            '1': 2,
+        }
+        self.assertCounts(message, expected_counts)