# This file is part of cloud-init. See LICENSE file for license information.
"""Tests of the built-in user data handlers."""
import copy
import errno
import os
import shutil
import tempfile
from textwrap import dedent
from cloudinit.tests.helpers import (
FilesystemMockingTestCase, CiTestCase, mock, skipUnlessJinja)
from cloudinit import handlers
from cloudinit import helpers
from cloudinit import subp
from cloudinit import util
from cloudinit.handlers.cloud_config import CloudConfigPartHandler
from cloudinit.handlers.jinja_template import (
JinjaTemplatePartHandler, convert_jinja_instance_data,
render_jinja_payload)
from cloudinit.handlers.shell_script import ShellScriptPartHandler
from cloudinit.handlers.upstart_job import UpstartJobPartHandler
from cloudinit.settings import (PER_ALWAYS, PER_INSTANCE)
class TestUpstartJobPartHandler(FilesystemMockingTestCase):
mpath = 'cloudinit.handlers.upstart_job.'
def test_upstart_frequency_no_out(self):
c_root = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, c_root)
up_root = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, up_root)
paths = helpers.Paths({
'cloud_dir': c_root,
'upstart_dir': up_root,
})
h = UpstartJobPartHandler(paths)
# No files should be written out when
# the frequency is ! per-instance
h.handle_part('', handlers.CONTENT_START,
None, None, None)
h.handle_part('blah', 'text/upstart-job',
'test.conf', 'blah', frequency=PER_ALWAYS)
h.handle_part('', handlers.CONTENT_END,
None, None, None)
self.assertEqual(0, len(os.listdir(up_root)))
def test_upstart_frequency_single(self):
# files should be written out when frequency is ! per-instance
new_root = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, new_root)
self.patchOS(new_root)
self.patchUtils(new_root)
paths = helpers.Paths({
'upstart_dir': "/etc/upstart",
})
util.ensure_dir("/run")
util.ensure_dir("/etc/upstart")
with mock.patch(self.mpath + 'SUITABLE_UPSTART', return_value=True):
with mock.patch.object(subp, 'subp') as m_subp:
h = UpstartJobPartHandler(paths)
h.handle_part('', handlers.CONTENT_START,
None, None, None)
h.handle_part('blah', 'text/upstart-job',
'test.conf', 'blah', frequency=PER_INSTANCE)
h.handle_part('', handlers.CONTENT_END,
None, None, None)
self.assertEqual(len(os.listdir('/etc/upstart')), 1)
m_subp.assert_called_once_with(
['initctl', 'reload-configuration'], capture=False)
class TestJinjaTemplatePartHandler(CiTestCase):
with_logs = True
mpath = 'cloudinit.handlers.jinja_template.'
def setUp(self):
super(TestJinjaTemplatePartHandler, self).setUp()
self.tmp = self.tmp_dir()
self.run_dir = os.path.join(self.tmp, 'run_dir')
util.ensure_dir(self.run_dir)
self.paths = helpers.Paths({
'cloud_dir': self.tmp, 'run_dir': self.run_dir})
def test_jinja_template_part_handler_defaults(self):
"""On init, paths are saved and subhandler types are empty."""
h = JinjaTemplatePartHandler(self.paths)
self.assertEqual(['## template: jinja'], h.prefixes)
self.assertEqual(3, h.handler_version)
self.assertEqual(self.paths, h.paths)
self.assertEqual({}, h.sub_handlers)
def test_jinja_template_part_handler_looks_up_sub_handler_types(self):
"""When sub_handlers are passed, init lists types of subhandlers."""
script_handler = ShellScriptPartHandler(self.paths)
cloudconfig_handler = CloudConfigPartHandler(self.paths)
h = JinjaTemplatePartHandler(
self.paths, sub_handlers=[script_handler, cloudconfig_handler])
self.assertCountEqual(
['text/cloud-config', 'text/cloud-config-jsonp',
'text/x-shellscript'],
h.sub_handlers)
def test_jinja_template_part_handler_looks_up_subhandler_types(self):
"""When sub_handlers are passed, init lists types of subhandlers."""
script_handler = ShellScriptPartHandler(self.paths)
cloudconfig_handler = CloudConfigPartHandler(self.paths)
h = JinjaTemplatePartHandler(
self.paths, sub_handlers=[script_handler, cloudconfig_handler])
self.assertCountEqual(
['text/cloud-config', 'text/cloud-config-jsonp',
'text/x-shellscript'],
h.sub_handlers)
def test_jinja_template_handle_noop_on_content_signals(self):
"""Perform no part handling when content type is CONTENT_SIGNALS."""
script_handler = ShellScriptPartHandler(self.paths)
h = JinjaTemplatePartHandler(
self.paths, sub_handlers=[script_handler])
with mock.patch.object(script_handler, 'handle_part') as m_handle_part:
h.handle_part(
data='data', ctype=handlers.CONTENT_START, filename='part-1',
payload='## template: jinja\n#!/bin/bash\necho himom',
frequency='freq', headers='headers')
m_handle_part.assert_not_called()
@skipUnlessJinja()
def test_jinja_template_handle_subhandler_v2_with_clean_payload(self):
"""Call version 2 subhandler.handle_part with stripped payload."""
script_handler = ShellScriptPartHandler(self.paths)
self.assertEqual(2, script_handler.handler_version)
# Create required instance-data.json file
instance_json = os.path.join(self.run_dir, 'instance-data.json')
instance_data = {'topkey': 'echo himom'}
util.write_file(instance_json, util.json_dumps(instance_data))
h = JinjaTemplatePartHandler(
self.paths, sub_handlers=[script_handler])
with mock.patch.object(script_handler, 'handle_part') as m_part:
# ctype with leading '!' not in handlers.CONTENT_SIGNALS
h.handle_part(
data='data', ctype="!" + handlers.CONTENT_START,
filename='part01',
payload='## template: jinja \t \n#!/bin/bash\n{{ topkey }}',
frequency='freq', headers='headers')
m_part.assert_called_once_with(
'data', '!__begin__', 'part01', '#!/bin/bash\necho himom', 'freq')
@skipUnlessJinja()
def test_jinja_template_handle_subhandler_v3_with_clean_payload(self):
"""Call version 3 subhandler.handle_part with stripped payload."""
cloudcfg_handler = CloudConfigPartHandler(self.paths)
self.assertEqual(3, cloudcfg_handler.handler_version)
# Create required instance-data.json file
instance_json = os.path.join(self.run_dir, 'instance-data.json')
instance_data = {'topkey': {'sub': 'runcmd: [echo hi]'}}
util.write_file(instance_json, util.json_dumps(instance_data))
h = JinjaTemplatePartHandler(
self.paths, sub_handlers=[cloudcfg_handler])
with mock.patch.object(cloudcfg_handler, 'handle_part') as m_part:
# ctype with leading '!' not in handlers.CONTENT_SIGNALS
h.handle_part(
data='data', ctype="!" + handlers.CONTENT_END,
filename='part01',
payload='## template: jinja\n#cloud-config\n{{ topkey.sub }}',
frequency='freq', headers='headers')
m_part.assert_called_once_with(
'data', '!__end__', 'part01', '#cloud-config\nruncmd: [echo hi]',
'freq', 'headers')
def test_jinja_template_handle_errors_on_missing_instance_data_json(self):
"""If instance-data is absent, raise an error from handle_part."""
script_handler = ShellScriptPartHandler(self.paths)
h = JinjaTemplatePartHandler(
self.paths, sub_handlers=[script_handler])
with self.assertRaises(RuntimeError) as context_manager:
h.handle_part(
data='data', ctype="!" + handlers.CONTENT_START,
filename='part01',
payload='## template: jinja \n#!/bin/bash\necho himom',
frequency='freq', headers='headers')
script_file = os.path.join(script_handler.script_dir, 'part01')
self.assertEqual(
'Cannot render jinja template vars. Instance data not yet present'
' at {}/instance-data.json'.format(
self.run_dir), str(context_manager.exception))
self.assertFalse(
os.path.exists(script_file),
'Unexpected file created %s' % script_file)
def test_jinja_template_handle_errors_on_unreadable_instance_data(self):
"""If instance-data is unreadable, raise an error from handle_part."""
script_handler = ShellScriptPartHandler(self.paths)
instance_json = os.path.join(self.run_dir, 'instance-data.json')
util.write_file(instance_json, util.json_dumps({}))
h = JinjaTemplatePartHandler(
self.paths, sub_handlers=[script_handler])
with mock.patch(self.mpath + 'load_file') as m_load:
with self.assertRaises(RuntimeError) as context_manager:
m_load.side_effect = OSError(errno.EACCES, 'Not allowed')
h.handle_part(
data='data', ctype="!" + handlers.CONTENT_START,
filename='part01',
payload='## template: jinja \n#!/bin/bash\necho himom',
frequency='freq', headers='headers')
script_file = os.path.join(script_handler.script_dir, 'part01')
self.assertEqual(
'Cannot render jinja template vars. No read permission on'
" '{rdir}/instance-data.json'. Try sudo".format(rdir=self.run_dir),
str(context_manager.exception))
self.assertFalse(
os.path.exists(script_file),
'Unexpected file created %s' % script_file)
@skipUnlessJinja()
def test_jinja_template_handle_renders_jinja_content(self):
"""When present, render jinja variables from instance-data.json."""
script_handler = ShellScriptPartHandler(self.paths)
instance_json = os.path.join(self.run_dir, 'instance-data.json')
instance_data = {'topkey': {'subkey': 'echo himom'}}
util.write_file(instance_json, util.json_dumps(instance_data))
h = JinjaTemplatePartHandler(
self.paths, sub_handlers=[script_handler])
h.handle_part(
data='data', ctype="!" + handlers.CONTENT_START,
filename='part01',
payload=(
'## template: jinja \n'
'#!/bin/bash\n'
'{{ topkey.subkey|default("nosubkey") }}'),
frequency='freq', headers='headers')
script_file = os.path.join(script_handler.script_dir, 'part01')
self.assertNotIn(
'Instance data not yet present at {}/instance-data.json'.format(
self.run_dir),
self.logs.getvalue())
self.assertEqual(
'#!/bin/bash\necho himom', util.load_file(script_file))
@skipUnlessJinja()
def test_jinja_template_handle_renders_jinja_content_missing_keys(self):
"""When specified jinja variable is undefined, log a warning."""
script_handler = ShellScriptPartHandler(self.paths)
instance_json = os.path.join(self.run_dir, 'instance-data.json')
instance_data = {'topkey': {'subkey': 'echo himom'}}
util.write_file(instance_json, util.json_dumps(instance_data))
h = JinjaTemplatePartHandler(
self.paths, sub_handlers=[script_handler])
h.handle_part(
data='data', ctype="!" + handlers.CONTENT_START,
filename='part01',
payload='## template: jinja \n#!/bin/bash\n{{ goodtry }}',
frequency='freq', headers='headers')
script_file = os.path.join(script_handler.script_dir, 'part01')
self.assertTrue(
os.path.exists(script_file),
'Missing expected file %s' % script_file)
self.assertIn(
"WARNING: Could not render jinja template variables in file"
" 'part01': 'goodtry'\n",
self.logs.getvalue())
class TestConvertJinjaInstanceData(CiTestCase):
def test_convert_instance_data_hyphens_to_underscores(self):
"""Replace hyphenated keys with underscores in instance-data."""
data = {'hyphenated-key': 'hyphenated-val',
'underscore_delim_key': 'underscore_delimited_val'}
expected_data = {'hyphenated_key': 'hyphenated-val',
'underscore_delim_key': 'underscore_delimited_val'}
self.assertEqual(
expected_data,
convert_jinja_instance_data(data=data))
def test_convert_instance_data_promotes_versioned_keys_to_top_level(self):
"""Any versioned keys are promoted as top-level keys
This provides any cloud-init standardized keys up at a top-level to
allow ease of reference for users. Intsead of v1.availability_zone,
the name availability_zone can be used in templates.
"""
data = {'ds': {'dskey1': 1, 'dskey2': 2},
'v1': {'v1key1': 'v1.1'},
'v2': {'v2key1': 'v2.1'}}
expected_data = copy.deepcopy(data)
expected_data.update({'v1key1': 'v1.1', 'v2key1': 'v2.1'})
converted_data = convert_jinja_instance_data(data=data)
self.assertCountEqual(
['ds', 'v1', 'v2', 'v1key1', 'v2key1'], converted_data.keys())
self.assertEqual(
expected_data,
converted_data)
def test_convert_instance_data_most_recent_version_of_promoted_keys(self):
"""The most-recent versioned key value is promoted to top-level."""
data = {'v1': {'key1': 'old v1 key1', 'key2': 'old v1 key2'},
'v2': {'key1': 'newer v2 key1', 'key3': 'newer v2 key3'},
'v3': {'key1': 'newest v3 key1'}}
expected_data = copy.deepcopy(data)
expected_data.update(
{'key1': 'newest v3 key1', 'key2': 'old v1 key2',
'key3': 'newer v2 key3'})
converted_data = convert_jinja_instance_data(data=data)
self.assertEqual(
expected_data,
converted_data)
def test_convert_instance_data_decodes_decode_paths(self):
"""Any decode_paths provided are decoded by convert_instance_data."""
data = {'key1': {'subkey1': 'aGkgbW9t'}, 'key2': 'aGkgZGFk'}
expected_data = copy.deepcopy(data)
expected_data['key1']['subkey1'] = 'hi mom'
converted_data = convert_jinja_instance_data(
data=data, decode_paths=('key1/subkey1',))
self.assertEqual(
expected_data,
converted_data)
class TestRenderJinjaPayload(CiTestCase):
with_logs = True
@skipUnlessJinja()
def test_render_jinja_payload_logs_jinja_vars_on_debug(self):
"""When debug is True, log jinja varables available."""
payload = (
'## template: jinja\n#!/bin/sh\necho hi from {{ v1.hostname }}')
instance_data = {'v1': {'hostname': 'foo'}, 'instance-id': 'iid'}
expected_log = dedent("""\
DEBUG: Converted jinja variables
{
"hostname": "foo",
"instance_id": "iid",
"v1": {
"hostname": "foo"
}
}
""")
self.assertEqual(
render_jinja_payload(
payload=payload, payload_fn='myfile',
instance_data=instance_data, debug=True),
'#!/bin/sh\necho hi from foo')
self.assertEqual(expected_log, self.logs.getvalue())
@skipUnlessJinja()
def test_render_jinja_payload_replaces_missing_variables_and_warns(self):
"""Warn on missing jinja variables and replace the absent variable."""
payload = (
'## template: jinja\n#!/bin/sh\necho hi from {{ NOTHERE }}')
instance_data = {'v1': {'hostname': 'foo'}, 'instance-id': 'iid'}
self.assertEqual(
render_jinja_payload(
payload=payload, payload_fn='myfile',
instance_data=instance_data),
'#!/bin/sh\necho hi from CI_MISSING_JINJA_VAR/NOTHERE')
expected_log = (
'WARNING: Could not render jinja template variables in file'
" 'myfile': 'NOTHERE'")
self.assertIn(expected_log, self.logs.getvalue())
# vi: ts=4 expandtab