summaryrefslogtreecommitdiff
path: root/tests/unittests/test_datasource/test_cloudsigma.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unittests/test_datasource/test_cloudsigma.py')
-rw-r--r--tests/unittests/test_datasource/test_cloudsigma.py49
1 files changed, 44 insertions, 5 deletions
diff --git a/tests/unittests/test_datasource/test_cloudsigma.py b/tests/unittests/test_datasource/test_cloudsigma.py
index 3245aba1..eadb3cb7 100644
--- a/tests/unittests/test_datasource/test_cloudsigma.py
+++ b/tests/unittests/test_datasource/test_cloudsigma.py
@@ -1,9 +1,11 @@
# coding: utf-8
-from unittest import TestCase
+import copy
from cloudinit.cs_utils import Cepko
from cloudinit.sources import DataSourceCloudSigma
+from tests.unittests import helpers as test_helpers
+
SERVER_CONTEXT = {
"cpu": 1000,
@@ -19,21 +21,27 @@ SERVER_CONTEXT = {
"smp": 1,
"tags": ["much server", "very performance"],
"uuid": "65b2fb23-8c03-4187-a3ba-8b7c919e8890",
- "vnc_password": "9e84d6cb49e46379"
+ "vnc_password": "9e84d6cb49e46379",
+ "vendor_data": {
+ "location": "zrh",
+ "cloudinit": "#cloud-config\n\n...",
+ }
}
class CepkoMock(Cepko):
- result = SERVER_CONTEXT
+ def __init__(self, mocked_context):
+ self.result = mocked_context
def all(self):
return self
-class DataSourceCloudSigmaTest(TestCase):
+class DataSourceCloudSigmaTest(test_helpers.TestCase):
def setUp(self):
self.datasource = DataSourceCloudSigma.DataSourceCloudSigma("", "", "")
- self.datasource.cepko = CepkoMock()
+ self.datasource.is_running_in_cloudsigma = lambda: True
+ self.datasource.cepko = CepkoMock(SERVER_CONTEXT)
self.datasource.get_data()
def test_get_hostname(self):
@@ -57,3 +65,34 @@ class DataSourceCloudSigmaTest(TestCase):
def test_user_data(self):
self.assertEqual(self.datasource.userdata_raw,
SERVER_CONTEXT['meta']['cloudinit-user-data'])
+
+ def test_encoded_user_data(self):
+ encoded_context = copy.deepcopy(SERVER_CONTEXT)
+ encoded_context['meta']['base64_fields'] = 'cloudinit-user-data'
+ encoded_context['meta']['cloudinit-user-data'] = 'aGkgd29ybGQK'
+ self.datasource.cepko = CepkoMock(encoded_context)
+ self.datasource.get_data()
+
+ self.assertEqual(self.datasource.userdata_raw, b'hi world\n')
+
+ def test_vendor_data(self):
+ self.assertEqual(self.datasource.vendordata_raw,
+ SERVER_CONTEXT['vendor_data']['cloudinit'])
+
+ def test_lack_of_vendor_data(self):
+ stripped_context = copy.deepcopy(SERVER_CONTEXT)
+ del stripped_context["vendor_data"]
+ self.datasource = DataSourceCloudSigma.DataSourceCloudSigma("", "", "")
+ self.datasource.cepko = CepkoMock(stripped_context)
+ self.datasource.get_data()
+
+ self.assertIsNone(self.datasource.vendordata_raw)
+
+ def test_lack_of_cloudinit_key_in_vendor_data(self):
+ stripped_context = copy.deepcopy(SERVER_CONTEXT)
+ del stripped_context["vendor_data"]["cloudinit"]
+ self.datasource = DataSourceCloudSigma.DataSourceCloudSigma("", "", "")
+ self.datasource.cepko = CepkoMock(stripped_context)
+ self.datasource.get_data()
+
+ self.assertIsNone(self.datasource.vendordata_raw)