diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/unittests/test_data.py | 31 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_gce.py | 49 |
2 files changed, 74 insertions, 6 deletions
diff --git a/tests/unittests/test_data.py b/tests/unittests/test_data.py index 4f24e2dd..c603bfdb 100644 --- a/tests/unittests/test_data.py +++ b/tests/unittests/test_data.py @@ -494,10 +494,10 @@ c: 4 ]) def test_mime_application_octet_stream(self): - """Mime message of type application/octet-stream is ignored but shows warning.""" + """Mime type application/octet-stream is ignored but shows warning.""" ci = stages.Init() message = MIMEBase("application", "octet-stream") - message.set_payload(b'\xbf\xe6\xb2\xc3\xd3\xba\x13\xa4\xd8\xa1\xcc\xbf') + message.set_payload(b'\xbf\xe6\xb2\xc3\xd3\xba\x13\xa4\xd8\xa1\xcc') encoders.encode_base64(message) ci.datasource = FakeDataSource(message.as_string().encode()) @@ -511,6 +511,33 @@ c: 4 mockobj.assert_called_once_with( ci.paths.get_ipath("cloud_config"), "", 0o600) + def test_cloud_config_archive(self): + non_decodable = b'\x11\xc9\xb4gTH\xee\x12' + data = [{'content': '#cloud-config\npassword: gocubs\n'}, + {'content': '#cloud-config\nlocale: chicago\n'}, + {'content': non_decodable}] + message = b'#cloud-config-archive\n' + util.yaml_dumps(data).encode() + + ci = stages.Init() + ci.datasource = FakeDataSource(message) + + fs = {} + + def fsstore(filename, content, mode=0o0644, omode="wb"): + fs[filename] = content + + # consuming the user-data provided should write 'cloud_config' file + # which will have our yaml in it. + with mock.patch('cloudinit.util.write_file') as mockobj: + mockobj.side_effect = fsstore + ci.fetch() + ci.consume_data() + + cfg = util.load_yaml(fs[ci.paths.get_ipath("cloud_config")]) + self.assertEqual(cfg.get('password'), 'gocubs') + self.assertEqual(cfg.get('locale'), 'chicago') + + class TestUDProcess(helpers.ResourceUsingTestCase): def test_bytes_in_userdata(self): diff --git a/tests/unittests/test_datasource/test_gce.py b/tests/unittests/test_datasource/test_gce.py index 4280abc4..1fb100f7 100644 --- a/tests/unittests/test_datasource/test_gce.py +++ b/tests/unittests/test_datasource/test_gce.py @@ -113,10 +113,6 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): self.assertEqual(GCE_META.get('instance/attributes/user-data'), self.ds.get_userdata_raw()) - # we expect a list of public ssh keys with user names stripped - self.assertEqual(['ssh-rsa AA2..+aRD0fyVw== root@server'], - self.ds.get_public_ssh_keys()) - # test partial metadata (missing user-data in particular) @httpretty.activate def test_metadata_partial(self): @@ -141,3 +137,48 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): decoded = b64decode( GCE_META_ENCODING.get('instance/attributes/user-data')) self.assertEqual(decoded, self.ds.get_userdata_raw()) + + @httpretty.activate + def test_missing_required_keys_return_false(self): + for required_key in ['instance/id', 'instance/zone', + 'instance/hostname']: + meta = GCE_META_PARTIAL.copy() + del meta[required_key] + httpretty.register_uri(httpretty.GET, MD_URL_RE, + body=_new_request_callback(meta)) + self.assertEqual(False, self.ds.get_data()) + httpretty.reset() + + @httpretty.activate + def test_project_level_ssh_keys_are_used(self): + httpretty.register_uri(httpretty.GET, MD_URL_RE, + body=_new_request_callback()) + self.ds.get_data() + + # we expect a list of public ssh keys with user names stripped + self.assertEqual(['ssh-rsa AA2..+aRD0fyVw== root@server'], + self.ds.get_public_ssh_keys()) + + @httpretty.activate + def test_instance_level_ssh_keys_are_used(self): + key_content = 'ssh-rsa JustAUser root@server' + meta = GCE_META.copy() + meta['instance/attributes/sshKeys'] = 'user:{0}'.format(key_content) + + httpretty.register_uri(httpretty.GET, MD_URL_RE, + body=_new_request_callback(meta)) + self.ds.get_data() + + self.assertIn(key_content, self.ds.get_public_ssh_keys()) + + @httpretty.activate + def test_instance_level_keys_replace_project_level_keys(self): + key_content = 'ssh-rsa JustAUser root@server' + meta = GCE_META.copy() + meta['instance/attributes/sshKeys'] = 'user:{0}'.format(key_content) + + httpretty.register_uri(httpretty.GET, MD_URL_RE, + body=_new_request_callback(meta)) + self.ds.get_data() + + self.assertEqual([key_content], self.ds.get_public_ssh_keys()) |