summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/data/filter_cloud_multipart.yaml30
-rw-r--r--tests/data/filter_cloud_multipart_1.email11
-rw-r--r--tests/data/filter_cloud_multipart_2.email39
-rw-r--r--tests/data/filter_cloud_multipart_header.email11
-rw-r--r--tests/unittests/helpers.py41
-rw-r--r--tests/unittests/test_filters/test_launch_index.py133
6 files changed, 265 insertions, 0 deletions
diff --git a/tests/data/filter_cloud_multipart.yaml b/tests/data/filter_cloud_multipart.yaml
new file mode 100644
index 00000000..7acc2b9d
--- /dev/null
+++ b/tests/data/filter_cloud_multipart.yaml
@@ -0,0 +1,30 @@
+#cloud-config-archive
+---
+- content: "\n blah: true\n launch-index: 3\n"
+ type: text/cloud-config
+- content: "\n blah: true\n launch-index: 4\n"
+ type: text/cloud-config
+- content: The quick brown fox jumps over the lazy dog
+ filename: b0.txt
+ launch-index: 0
+ type: plain/text
+- content: The quick brown fox jumps over the lazy dog
+ filename: b3.txt
+ launch-index: 3
+ type: plain/text
+- content: The quick brown fox jumps over the lazy dog
+ filename: b2.txt
+ launch-index: 2
+ type: plain/text
+- content: '#!/bin/bash \n echo "stuff"'
+ filename: b2.txt
+ launch-index: 2
+- content: '#!/bin/bash \n echo "stuff"'
+ filename: b2.txt
+ launch-index: 1
+- content: '#!/bin/bash \n echo "stuff"'
+ filename: b2.txt
+ # Use a string to see if conversion works
+ launch-index: "1"
+...
+
diff --git a/tests/data/filter_cloud_multipart_1.email b/tests/data/filter_cloud_multipart_1.email
new file mode 100644
index 00000000..6d93b1f1
--- /dev/null
+++ b/tests/data/filter_cloud_multipart_1.email
@@ -0,0 +1,11 @@
+From nobody Fri Aug 31 17:17:00 2012
+Content-Type: text/plain; charset="us-ascii"
+MIME-Version: 1.0
+Content-Transfer-Encoding: 7bit
+
+
+#cloud-config
+b: c
+launch-index: 2
+
+
diff --git a/tests/data/filter_cloud_multipart_2.email b/tests/data/filter_cloud_multipart_2.email
new file mode 100644
index 00000000..b04068c5
--- /dev/null
+++ b/tests/data/filter_cloud_multipart_2.email
@@ -0,0 +1,39 @@
+From nobody Fri Aug 31 17:43:04 2012
+Content-Type: multipart/mixed; boundary="===============1668325974=="
+MIME-Version: 1.0
+
+--===============1668325974==
+Content-Type: text/cloud-config; charset="us-ascii"
+MIME-Version: 1.0
+Content-Transfer-Encoding: 7bit
+
+
+#cloud-config
+b: c
+launch-index: 2
+
+
+--===============1668325974==
+Content-Type: text/plain; charset="us-ascii"
+MIME-Version: 1.0
+Content-Transfer-Encoding: 7bit
+
+
+#cloud-config-archive
+- content: The quick brown fox jumps over the lazy dog
+ filename: b3.txt
+ launch-index: 3
+ type: plain/text
+
+--===============1668325974==
+Content-Type: text/plain; charset="us-ascii"
+MIME-Version: 1.0
+Content-Transfer-Encoding: 7bit
+
+
+#cloud-config
+b: c
+launch-index: 2
+
+
+--===============1668325974==--
diff --git a/tests/data/filter_cloud_multipart_header.email b/tests/data/filter_cloud_multipart_header.email
new file mode 100644
index 00000000..770f7ef1
--- /dev/null
+++ b/tests/data/filter_cloud_multipart_header.email
@@ -0,0 +1,11 @@
+From nobody Fri Aug 31 17:17:00 2012
+Content-Type: text/plain; charset="us-ascii"
+MIME-Version: 1.0
+Launch-Index: 5
+Content-Transfer-Encoding: 7bit
+
+
+#cloud-config
+b: c
+
+
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)
generated by cgit v1.2.3 (git 2.39.1) at 2025-11-04 02:44:29 +0000