summaryrefslogtreecommitdiff
path: root/tests/unittests/test_filters/test_launch_index.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unittests/test_filters/test_launch_index.py')
-rw-r--r--tests/unittests/test_filters/test_launch_index.py132
1 files changed, 0 insertions, 132 deletions
diff --git a/tests/unittests/test_filters/test_launch_index.py b/tests/unittests/test_filters/test_launch_index.py
deleted file mode 100644
index 395713e6..00000000
--- a/tests/unittests/test_filters/test_launch_index.py
+++ /dev/null
@@ -1,132 +0,0 @@
-import copy
-
-from .. import helpers
-
-from six.moves import filterfalse
-
-from cloudinit.filters import launch_index
-from cloudinit import user_data as ud
-from cloudinit import util
-
-
-def count_messages(root):
- am = 0
- for m in root.walk():
- if ud.is_skippable(m):
- continue
- am += 1
- return am
-
-
-class TestLaunchFilter(helpers.ResourceUsingTestCase):
-
- def assertCounts(self, message, expected_counts):
- orig_message = copy.deepcopy(message)
- for (index, count) in expected_counts.items():
- index = util.safe_int(index)
- filtered_message = launch_index.Filter(index).apply(message)
- self.assertEqual(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 filterfalse(ud.is_skippable, msg1_msgs)]
- msg2_msgs = [m for m in msg2.walk()]
- msg2_msgs = [m for m in filterfalse(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
- m1_py = m1_msg.get_payload(decode=True)
- m2_py = m2_msg.get_payload(decode=True)
- if m1_py != m2_py:
- 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,
- 'c': 1,
- 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)