cc_snap: Add new module to install and configure snapd and snap packages.

Support installing and configuring snaps on ubuntu systems. Now,
cloud-config files can provide a list or dictionary of snap:assertions
which will be allow configuration of snapd on a system via 'snap ack'
calls. The snap:commands configuration option supports arbitrary system
commands intended to interact with snappy's cli. This allows users to run
arbitrary snappy commands to create users, download, install and
configure snap packages and snapd.

This branch also deprecates old snappy and snap_config modules leaving
warnings in documentation and runtime for consumers of these modules.
Deprecated snap* modules will be dropped in cloud-init v.18.2 release.

1 files changed, 533 insertions, 0 deletions

diff --git a/cloudinit/config/tests/test_snap.py b/cloudinit/config/tests/test_snap.py
new file mode 100644
index 00000000..c2dd6afe
--- /dev/null
+++ b/cloudinit/config/tests/test_snap.py
@@ -0,0 +1,533 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+import re
+from six import StringIO
+
+from cloudinit.config.cc_snap import (
+    ASSERTIONS_FILE, add_assertions, handle, prepend_snap_commands,
+    maybe_install_squashfuse, run_commands, schema)
+from cloudinit.config.schema import validate_cloudconfig_schema
+from cloudinit import util
+from cloudinit.tests.helpers import CiTestCase, mock, wrap_and_call
+
+
+SYSTEM_USER_ASSERTION = """\
+type: system-user
+authority-id: LqvZQdfyfGlYvtep4W6Oj6pFXP9t1Ksp
+brand-id: LqvZQdfyfGlYvtep4W6Oj6pFXP9t1Ksp
+email: foo@bar.com
+password: $6$E5YiAuMIPAwX58jG$miomhVNui/vf7f/3ctB/f0RWSKFxG0YXzrJ9rtJ1ikvzt
+series:
+- 16
+since: 2016-09-10T16:34:00+03:00
+until: 2017-11-10T16:34:00+03:00
+username: baz
+sign-key-sha3-384: RuVvnp4n52GilycjfbbTCI3_L8Y6QlIE75wxMc0KzGV3AUQqVd9GuXoj
+
+AcLBXAQAAQoABgUCV/UU1wAKCRBKnlMoJQLkZVeLD/9/+hIeVywtzsDA3oxl+P+u9D13y9s6svP
+Jd6Wnf4FTw6sq1GjBE4ZA7lrwSaRCUJ9Vcsvf2q9OGPY7mOb2TBxaDe0PbUMjrSrqllSSQwhpNI
+zG+NxkkKuxsUmLzFa+k9m6cyojNbw5LFhQZBQCGlr3JYqC0tIREq/UsZxj+90TUC87lDJwkU8GF
+s4CR+rejZj4itIcDcVxCSnJH6hv6j2JrJskJmvObqTnoOlcab+JXdamXqbldSP3UIhWoyVjqzkj
++to7mXgx+cCUA9+ngNCcfUG+1huGGTWXPCYkZ78HvErcRlIdeo4d3xwtz1cl/w3vYnq9og1XwsP
+Yfetr3boig2qs1Y+j/LpsfYBYncgWjeDfAB9ZZaqQz/oc8n87tIPZDJHrusTlBfop8CqcM4xsKS
+d+wnEY8e/F24mdSOYmS1vQCIDiRU3MKb6x138Ud6oHXFlRBbBJqMMctPqWDunWzb5QJ7YR0I39q
+BrnEqv5NE0G7w6HOJ1LSPG5Hae3P4T2ea+ATgkb03RPr3KnXnzXg4TtBbW1nytdlgoNc/BafE1H
+f3NThcq9gwX4xWZ2PAWnqVPYdDMyCtzW3Ck+o6sIzx+dh4gDLPHIi/6TPe/pUuMop9CBpWwez7V
+v1z+1+URx6Xlq3Jq18y5pZ6fY3IDJ6km2nQPMzcm4Q=="""
+
+ACCOUNT_ASSERTION = """\
+type: account-key
+authority-id: canonical
+revision: 2
+public-key-sha3-384: BWDEoaqyr25nF5SNCvEv2v7QnM9QsfCc0PBMYD_i2NGSQ32EF2d4D0
+account-id: canonical
+name: store
+since: 2016-04-01T00:00:00.0Z
+body-length: 717
+sign-key-sha3-384: -CvQKAwRQ5h3Ffn10FILJoEZUXOv6km9FwA80-Rcj-f-6jadQ89VRswH
+
+AcbBTQRWhcGAARAA0KKYYQWuHOrsFVi4p4l7ZzSvX7kLgJFFeFgOkzdWKBTHEnsMKjl5mefFe9j
+qe8NlmJdfY7BenP7XeBtwKp700H/t9lLrZbpTNAPHXYxEWFJp5bPqIcJYBZ+29oLVLN1Tc5X482
+vCiDqL8+pPYqBrK2fNlyPlNNSum9wI70rDDL4r6FVvr+osTnGejibdV8JphWX+lrSQDnRSdM8KJ
+UM43vTgLGTi9W54oRhsA2OFexRfRksTrnqGoonCjqX5wO3OFSaMDzMsO2MJ/hPfLgDqw53qjzuK
+Iec9OL3k5basvu2cj5u9tKwVFDsCKK2GbKUsWWpx2KTpOifmhmiAbzkTHbH9KaoMS7p0kJwhTQG
+o9aJ9VMTWHJc/NCBx7eu451u6d46sBPCXS/OMUh2766fQmoRtO1OwCTxsRKG2kkjbMn54UdFULl
+VfzvyghMNRKIezsEkmM8wueTqGUGZWa6CEZqZKwhe/PROxOPYzqtDH18XZknbU1n5lNb7vNfem9
+2ai+3+JyFnW9UhfvpVF7gzAgdyCqNli4C6BIN43uwoS8HkykocZS/+Gv52aUQ/NZ8BKOHLw+7an
+Q0o8W9ltSLZbEMxFIPSN0stiZlkXAp6DLyvh1Y4wXSynDjUondTpej2fSvSlCz/W5v5V7qA4nIc
+vUvV7RjVzv17ut0AEQEAAQ==
+
+AcLDXAQAAQoABgUCV83k9QAKCRDUpVvql9g3IBT8IACKZ7XpiBZ3W4lqbPssY6On81WmxQLtvsM
+WTp6zZpl/wWOSt2vMNUk9pvcmrNq1jG9CuhDfWFLGXEjcrrmVkN3YuCOajMSPFCGrxsIBLSRt/b
+nrKykdLAAzMfG8rP1d82bjFFiIieE+urQ0Kcv09Jtdvavq3JT1Tek5mFyyfhHNlQEKOzWqmRWiL
+3c3VOZUs1ZD8TSlnuq/x+5T0X0YtOyGjSlVxk7UybbyMNd6MZfNaMpIG4x+mxD3KHFtBAC7O6kL
+eX3i6j5nCY5UABfA3DZEAkWP4zlmdBEOvZ9t293NaDdOpzsUHRkoi0Zez/9BHQ/kwx/uNc2WqrY
+inCmu16JGNeXqsyinnLl7Ghn2RwhvDMlLxF6RTx8xdx1yk6p3PBTwhZMUvuZGjUtN/AG8BmVJQ1
+rsGSRkkSywvnhVJRB2sudnrMBmNS2goJbzSbmJnOlBrd2WsV0T9SgNMWZBiov3LvU4o2SmAb6b+
+rYwh8H5QHcuuYJuxDjFhPswIp6Wes5T6hUicf3SWtObcDS4HSkVS4ImBjjX9YgCuFy7QdnooOWE
+aPvkRw3XCVeYq0K6w9GRsk1YFErD4XmXXZjDYY650MX9v42Sz5MmphHV8jdIY5ssbadwFSe2rCQ
+6UX08zy7RsIb19hTndE6ncvSNDChUR9eEnCm73eYaWTWTnq1cxdVP/s52r8uss++OYOkPWqh5nO
+haRn7INjH/yZX4qXjNXlTjo0PnHH0q08vNKDwLhxS+D9du+70FeacXFyLIbcWllSbJ7DmbumGpF
+yYbtj3FDDPzachFQdIG3lSt+cSUGeyfSs6wVtc3cIPka/2Urx7RprfmoWSI6+a5NcLdj0u2z8O9
+HxeIgxDpg/3gT8ZIuFKePMcLDM19Fh/p0ysCsX+84B9chNWtsMSmIaE57V+959MVtsLu7SLb9gi
+skrju0pQCwsu2wHMLTNd1f3PTHmrr49hxetTus07HSQUApMtAGKzQilF5zqFjbyaTd4xgQbd+PK
+CjFyzQTDOcUhXpuUGt/IzlqiFfsCsmbj2K4KdSNYMlqIgZ3Azu8KvZLIhsyN7v5vNIZSPfEbjde
+ClU9r0VRiJmtYBUjcSghD9LWn+yRLwOxhfQVjm0cBwIt5R/yPF/qC76yIVuWUtM5Y2/zJR1J8OF
+qWchvlImHtvDzS9FQeLyzJAOjvZ2CnWp2gILgUz0WQdOk1Dq8ax7KS9BQ42zxw9EZAEPw3PEFqR
+IQsRTONp+iVS8YxSmoYZjDlCgRMWUmawez/Fv5b9Fb/XkO5Eq4e+KfrpUujXItaipb+tV8h5v3t
+oG3Ie3WOHrVjCLXIdYslpL1O4nadqR6Xv58pHj6k"""
+
+
+class FakeCloud(object):
+    def __init__(self, distro):
+        self.distro = distro
+
+
+class TestAddAssertions(CiTestCase):
+
+    with_logs = True
+
+    def setUp(self):
+        super(TestAddAssertions, self).setUp()
+        self.tmp = self.tmp_dir()
+
+    @mock.patch('cloudinit.config.cc_snap.util.subp')
+    def test_add_assertions_on_empty_list(self, m_subp):
+        """When provided with an empty list, add_assertions does nothing."""
+        add_assertions([])
+        self.assertEqual('', self.logs.getvalue())
+        m_subp.assert_not_called()
+
+    def test_add_assertions_on_non_list_or_dict(self):
+        """When provided an invalid type, add_assertions raises an error."""
+        with self.assertRaises(TypeError) as context_manager:
+            add_assertions(assertions="I'm Not Valid")
+        self.assertEqual(
+            "assertion parameter was not a list or dict: I'm Not Valid",
+            str(context_manager.exception))
+
+    @mock.patch('cloudinit.config.cc_snap.util.subp')
+    def test_add_assertions_adds_assertions_as_list(self, m_subp):
+        """When provided with a list, add_assertions adds all assertions."""
+        self.assertEqual(
+            ASSERTIONS_FILE, '/var/lib/cloud/instance/snapd.assertions')
+        assert_file = self.tmp_path('snapd.assertions', dir=self.tmp)
+        assertions = [SYSTEM_USER_ASSERTION, ACCOUNT_ASSERTION]
+        wrap_and_call(
+            'cloudinit.config.cc_snap',
+            {'ASSERTIONS_FILE': {'new': assert_file}},
+            add_assertions, assertions)
+        self.assertIn(
+            'Importing user-provided snap assertions', self.logs.getvalue())
+        self.assertIn(
+            'sertions', self.logs.getvalue())
+        self.assertEqual(
+            [mock.call(['snap', 'ack', assert_file], capture=True)],
+            m_subp.call_args_list)
+        compare_file = self.tmp_path('comparison', dir=self.tmp)
+        util.write_file(compare_file, '\n'.join(assertions).encode('utf-8'))
+        self.assertEqual(
+            util.load_file(compare_file), util.load_file(assert_file))
+
+    @mock.patch('cloudinit.config.cc_snap.util.subp')
+    def test_add_assertions_adds_assertions_as_dict(self, m_subp):
+        """When provided with a dict, add_assertions adds all assertions."""
+        self.assertEqual(
+            ASSERTIONS_FILE, '/var/lib/cloud/instance/snapd.assertions')
+        assert_file = self.tmp_path('snapd.assertions', dir=self.tmp)
+        assertions = {'00': SYSTEM_USER_ASSERTION, '01': ACCOUNT_ASSERTION}
+        wrap_and_call(
+            'cloudinit.config.cc_snap',
+            {'ASSERTIONS_FILE': {'new': assert_file}},
+            add_assertions, assertions)
+        self.assertIn(
+            'Importing user-provided snap assertions', self.logs.getvalue())
+        self.assertIn(
+            "DEBUG: Snap acking: ['type: system-user', 'authority-id: Lqv",
+            self.logs.getvalue())
+        self.assertIn(
+            "DEBUG: Snap acking: ['type: account-key', 'authority-id: canonic",
+            self.logs.getvalue())
+        self.assertEqual(
+            [mock.call(['snap', 'ack', assert_file], capture=True)],
+            m_subp.call_args_list)
+        compare_file = self.tmp_path('comparison', dir=self.tmp)
+        combined = '\n'.join(assertions.values())
+        util.write_file(compare_file, combined.encode('utf-8'))
+        self.assertEqual(
+            util.load_file(compare_file), util.load_file(assert_file))
+
+
+class TestPrepentSnapCommands(CiTestCase):
+
+    with_logs = True
+
+    def test_prepend_snap_commands_errors_on_neither_string_nor_list(self):
+        """Raise an error for each command which is not a string or list."""
+        orig_commands = ['ls', 1, {'not': 'gonna work'}, ['snap', 'list']]
+        with self.assertRaises(TypeError) as context_manager:
+            prepend_snap_commands(orig_commands)
+        self.assertEqual(
+            "Invalid snap config. These commands are not a string or list:\n"
+            "1\n{'not': 'gonna work'}",
+            str(context_manager.exception))
+
+    def test_prepend_snap_commands_warns_on_non_snap_string_commands(self):
+        """Warn on each non-snap for commands of type string."""
+        orig_commands = ['ls', 'snap list', 'touch /blah', 'snap install x']
+        fixed_commands = prepend_snap_commands(orig_commands)
+        self.assertEqual(
+            'WARNING: Non-snap commands in snap config:\n'
+            'ls\ntouch /blah\n',
+            self.logs.getvalue())
+        self.assertEqual(orig_commands, fixed_commands)
+
+    def test_prepend_snap_commands_prepends_on_non_snap_list_commands(self):
+        """Prepend 'snap' for each non-snap command of type list."""
+        orig_commands = [['ls'], ['snap', 'list'], ['snapa', '/blah'],
+                         ['snap', 'install', 'x']]
+        expected = [['snap', 'ls'], ['snap', 'list'],
+                    ['snap', 'snapa', '/blah'],
+                    ['snap', 'install', 'x']]
+        fixed_commands = prepend_snap_commands(orig_commands)
+        self.assertEqual('', self.logs.getvalue())
+        self.assertEqual(expected, fixed_commands)
+
+    def test_prepend_snap_commands_removes_first_item_when_none(self):
+        """Remove the first element of a non-snap command when it is None."""
+        orig_commands = [[None, 'ls'], ['snap', 'list'],
+                         [None, 'touch', '/blah'],
+                         ['snap', 'install', 'x']]
+        expected = [['ls'], ['snap', 'list'],
+                    ['touch', '/blah'],
+                    ['snap', 'install', 'x']]
+        fixed_commands = prepend_snap_commands(orig_commands)
+        self.assertEqual('', self.logs.getvalue())
+        self.assertEqual(expected, fixed_commands)
+
+
+class TestRunCommands(CiTestCase):
+
+    with_logs = True
+
+    def setUp(self):
+        super(TestRunCommands, self).setUp()
+        self.tmp = self.tmp_dir()
+
+    @mock.patch('cloudinit.config.cc_snap.util.subp')
+    def test_run_commands_on_empty_list(self, m_subp):
+        """When provided with an empty list, run_commands does nothing."""
+        run_commands([])
+        self.assertEqual('', self.logs.getvalue())
+        m_subp.assert_not_called()
+
+    def test_run_commands_on_non_list_or_dict(self):
+        """When provided an invalid type, run_commands raises an error."""
+        with self.assertRaises(TypeError) as context_manager:
+            run_commands(commands="I'm Not Valid")
+        self.assertEqual(
+            "commands parameter was not a list or dict: I'm Not Valid",
+            str(context_manager.exception))
+
+    def test_run_command_logs_commands_and_exit_codes_to_stderr(self):
+        """All exit codes are logged to stderr."""
+        outfile = self.tmp_path('output.log', dir=self.tmp)
+
+        cmd1 = 'echo "HI" >> %s' % outfile
+        cmd2 = 'bogus command'
+        cmd3 = 'echo "MOM" >> %s' % outfile
+        commands = [cmd1, cmd2, cmd3]
+
+        mock_path = 'cloudinit.config.cc_snap.sys.stderr'
+        with mock.patch(mock_path, new_callable=StringIO) as m_stderr:
+            with self.assertRaises(RuntimeError) as context_manager:
+                run_commands(commands=commands)
+
+        self.assertIsNotNone(
+            re.search(r'bogus: (command )?not found',
+                      str(context_manager.exception)),
+            msg='Expected bogus command not found')
+        expected_stderr_log = '\n'.join([
+            'Begin run command: {cmd}'.format(cmd=cmd1),
+            'End run command: exit(0)',
+            'Begin run command: {cmd}'.format(cmd=cmd2),
+            'ERROR: End run command: exit(127)',
+            'Begin run command: {cmd}'.format(cmd=cmd3),
+            'End run command: exit(0)\n'])
+        self.assertEqual(expected_stderr_log, m_stderr.getvalue())
+
+    def test_run_command_as_lists(self):
+        """When commands are specified as a list, run them in order."""
+        outfile = self.tmp_path('output.log', dir=self.tmp)
+
+        cmd1 = 'echo "HI" >> %s' % outfile
+        cmd2 = 'echo "MOM" >> %s' % outfile
+        commands = [cmd1, cmd2]
+        mock_path = 'cloudinit.config.cc_snap.sys.stderr'
+        with mock.patch(mock_path, new_callable=StringIO):
+            run_commands(commands=commands)
+
+        self.assertIn(
+            'DEBUG: Running user-provided snap commands',
+            self.logs.getvalue())
+        self.assertEqual('HI\nMOM\n', util.load_file(outfile))
+        self.assertIn(
+            'WARNING: Non-snap commands in snap config:', self.logs.getvalue())
+
+    def test_run_command_dict_sorted_as_command_script(self):
+        """When commands are a dict, sort them and run."""
+        outfile = self.tmp_path('output.log', dir=self.tmp)
+        cmd1 = 'echo "HI" >> %s' % outfile
+        cmd2 = 'echo "MOM" >> %s' % outfile
+        commands = {'02': cmd1, '01': cmd2}
+        mock_path = 'cloudinit.config.cc_snap.sys.stderr'
+        with mock.patch(mock_path, new_callable=StringIO):
+            run_commands(commands=commands)
+
+        expected_messages = [
+            'DEBUG: Running user-provided snap commands']
+        for message in expected_messages:
+            self.assertIn(message, self.logs.getvalue())
+        self.assertEqual('MOM\nHI\n', util.load_file(outfile))
+
+
+class TestSchema(CiTestCase):
+
+    with_logs = True
+
+    def test_schema_warns_on_snap_not_as_dict(self):
+        """If the snap configuration is not a dict, emit a warning."""
+        validate_cloudconfig_schema({'snap': 'wrong type'}, schema)
+        self.assertEqual(
+            "WARNING: Invalid config:\nsnap: 'wrong type' is not of type"
+            " 'object'\n",
+            self.logs.getvalue())
+
+    @mock.patch('cloudinit.config.cc_snap.run_commands')
+    def test_schema_disallows_unknown_keys(self, _):
+        """Unknown keys in the snap configuration emit warnings."""
+        validate_cloudconfig_schema(
+            {'snap': {'commands': ['ls'], 'invalid-key': ''}}, schema)
+        self.assertIn(
+            'WARNING: Invalid config:\nsnap: Additional properties are not'
+            " allowed ('invalid-key' was unexpected)",
+            self.logs.getvalue())
+
+    def test_warn_schema_requires_either_commands_or_assertions(self):
+        """Warn when snap configuration lacks both commands and assertions."""
+        validate_cloudconfig_schema(
+            {'snap': {}}, schema)
+        self.assertIn(
+            'WARNING: Invalid config:\nsnap: {} does not have enough'
+            ' properties',
+            self.logs.getvalue())
+
+    @mock.patch('cloudinit.config.cc_snap.run_commands')
+    def test_warn_schema_commands_is_not_list_or_dict(self, _):
+        """Warn when snap:commands config is not a list or dict."""
+        validate_cloudconfig_schema(
+            {'snap': {'commands': 'broken'}}, schema)
+        self.assertEqual(
+            "WARNING: Invalid config:\nsnap.commands: 'broken' is not of type"
+            " 'object', 'array'\n",
+            self.logs.getvalue())
+
+    @mock.patch('cloudinit.config.cc_snap.run_commands')
+    def test_warn_schema_when_commands_is_empty(self, _):
+        """Emit warnings when snap:commands is an empty list or dict."""
+        validate_cloudconfig_schema(
+            {'snap': {'commands': []}}, schema)
+        validate_cloudconfig_schema(
+            {'snap': {'commands': {}}}, schema)
+        self.assertEqual(
+            "WARNING: Invalid config:\nsnap.commands: [] is too short\n"
+            "WARNING: Invalid config:\nsnap.commands: {} does not have enough"
+            " properties\n",
+            self.logs.getvalue())
+
+    @mock.patch('cloudinit.config.cc_snap.run_commands')
+    def test_schema_when_commands_are_list_or_dict(self, _):
+        """No warnings when snap:commands are either a list or dict."""
+        validate_cloudconfig_schema(
+            {'snap': {'commands': ['valid']}}, schema)
+        validate_cloudconfig_schema(
+            {'snap': {'commands': {'01': 'also valid'}}}, schema)
+        self.assertEqual('', self.logs.getvalue())
+
+    @mock.patch('cloudinit.config.cc_snap.add_assertions')
+    def test_warn_schema_assertions_is_not_list_or_dict(self, _):
+        """Warn when snap:assertions config is not a list or dict."""
+        validate_cloudconfig_schema(
+            {'snap': {'assertions': 'broken'}}, schema)
+        self.assertEqual(
+            "WARNING: Invalid config:\nsnap.assertions: 'broken' is not of"
+            " type 'object', 'array'\n",
+            self.logs.getvalue())
+
+    @mock.patch('cloudinit.config.cc_snap.add_assertions')
+    def test_warn_schema_when_assertions_is_empty(self, _):
+        """Emit warnings when snap:assertions is an empty list or dict."""
+        validate_cloudconfig_schema(
+            {'snap': {'assertions': []}}, schema)
+        validate_cloudconfig_schema(
+            {'snap': {'assertions': {}}}, schema)
+        self.assertEqual(
+            "WARNING: Invalid config:\nsnap.assertions: [] is too short\n"
+            "WARNING: Invalid config:\nsnap.assertions: {} does not have"
+            " enough properties\n",
+            self.logs.getvalue())
+
+    @mock.patch('cloudinit.config.cc_snap.add_assertions')
+    def test_schema_when_assertions_are_list_or_dict(self, _):
+        """No warnings when snap:assertions are a list or dict."""
+        validate_cloudconfig_schema(
+            {'snap': {'assertions': ['valid']}}, schema)
+        validate_cloudconfig_schema(
+            {'snap': {'assertions': {'01': 'also valid'}}}, schema)
+        self.assertEqual('', self.logs.getvalue())
+
+
+class TestHandle(CiTestCase):
+
+    with_logs = True
+
+    def setUp(self):
+        super(TestHandle, self).setUp()
+        self.tmp = self.tmp_dir()
+
+    @mock.patch('cloudinit.config.cc_snap.run_commands')
+    @mock.patch('cloudinit.config.cc_snap.add_assertions')
+    @mock.patch('cloudinit.config.cc_snap.validate_cloudconfig_schema')
+    def test_handle_no_config(self, m_schema, m_add, m_run):
+        """When no snap-related configuration is provided, nothing happens."""
+        cfg = {}
+        handle('snap', cfg=cfg, cloud=None, log=self.logger, args=None)
+        self.assertIn(
+            "DEBUG: Skipping module named snap, no 'snap' key in config",
+            self.logs.getvalue())
+        m_schema.assert_not_called()
+        m_add.assert_not_called()
+        m_run.assert_not_called()
+
+    @mock.patch('cloudinit.config.cc_snap.run_commands')
+    @mock.patch('cloudinit.config.cc_snap.add_assertions')
+    @mock.patch('cloudinit.config.cc_snap.maybe_install_squashfuse')
+    def test_handle_skips_squashfuse_when_unconfigured(self, m_squash, m_add,
+                                                       m_run):
+        """When squashfuse_in_container is unset, don't attempt to install."""
+        handle(
+            'snap', cfg={'snap': {}}, cloud=None, log=self.logger, args=None)
+        handle(
+            'snap', cfg={'snap': {'squashfuse_in_container': None}},
+            cloud=None, log=self.logger, args=None)
+        handle(
+            'snap', cfg={'snap': {'squashfuse_in_container': False}},
+            cloud=None, log=self.logger, args=None)
+        self.assertEqual([], m_squash.call_args_list)  # No calls
+        # snap configuration missing assertions and commands will default to []
+        self.assertIn(mock.call([]), m_add.call_args_list)
+        self.assertIn(mock.call([]), m_run.call_args_list)
+
+    @mock.patch('cloudinit.config.cc_snap.maybe_install_squashfuse')
+    def test_handle_tries_to_install_squashfuse(self, m_squash):
+        """If squashfuse_in_container is True, try installing squashfuse."""
+        cfg = {'snap': {'squashfuse_in_container': True}}
+        mycloud = FakeCloud(None)
+        handle('snap', cfg=cfg, cloud=mycloud, log=self.logger, args=None)
+        self.assertEqual(
+            [mock.call(mycloud)], m_squash.call_args_list)
+
+    def test_handle_runs_commands_provided(self):
+        """If commands are specified as a list, run them."""
+        outfile = self.tmp_path('output.log', dir=self.tmp)
+
+        cfg = {
+            'snap': {'commands': ['echo "HI" >> %s' % outfile,
+                                  'echo "MOM" >> %s' % outfile]}}
+        handle('snap', cfg=cfg, cloud=None, log=self.logger, args=None)
+        self.assertEqual('HI\nMOM\n', util.load_file(outfile))
+
+    @mock.patch('cloudinit.config.cc_snap.util.subp')
+    def test_handle_adds_assertions(self, m_subp):
+        """Any configured snap assertions are provided to add_assertions."""
+        assert_file = self.tmp_path('snapd.assertions', dir=self.tmp)
+        compare_file = self.tmp_path('comparison', dir=self.tmp)
+        cfg = {
+            'snap': {'assertions': [SYSTEM_USER_ASSERTION, ACCOUNT_ASSERTION]}}
+        wrap_and_call(
+            'cloudinit.config.cc_snap',
+            {'ASSERTIONS_FILE': {'new': assert_file}},
+            handle, 'snap', cfg=cfg, cloud=None, log=self.logger, args=None)
+        content = '\n'.join(cfg['snap']['assertions'])
+        util.write_file(compare_file, content.encode('utf-8'))
+        self.assertEqual(
+            util.load_file(compare_file), util.load_file(assert_file))
+
+    @mock.patch('cloudinit.config.cc_snap.util.subp')
+    def test_handle_validates_schema(self, m_subp):
+        """Any provided configuration is runs validate_cloudconfig_schema."""
+        assert_file = self.tmp_path('snapd.assertions', dir=self.tmp)
+        cfg = {'snap': {'invalid': ''}}  # Generates schema warning
+        wrap_and_call(
+            'cloudinit.config.cc_snap',
+            {'ASSERTIONS_FILE': {'new': assert_file}},
+            handle, 'snap', cfg=cfg, cloud=None, log=self.logger, args=None)
+        self.assertEqual(
+            "WARNING: Invalid config:\nsnap: Additional properties are not"
+            " allowed ('invalid' was unexpected)\n",
+            self.logs.getvalue())
+
+
+class TestMaybeInstallSquashFuse(CiTestCase):
+
+    with_logs = True
+
+    def setUp(self):
+        super(TestMaybeInstallSquashFuse, self).setUp()
+        self.tmp = self.tmp_dir()
+
+    @mock.patch('cloudinit.config.cc_snap.util.is_container')
+    def test_maybe_install_squashfuse_skips_non_containers(self, m_container):
+        """maybe_install_squashfuse does nothing when not on a container."""
+        m_container.return_value = False
+        maybe_install_squashfuse(cloud=FakeCloud(None))
+        self.assertEqual([mock.call()], m_container.call_args_list)
+        self.assertEqual('', self.logs.getvalue())
+
+    @mock.patch('cloudinit.config.cc_snap.util.is_container')
+    def test_maybe_install_squashfuse_raises_install_errors(self, m_container):
+        """maybe_install_squashfuse logs and raises package install errors."""
+        m_container.return_value = True
+        distro = mock.MagicMock()
+        distro.update_package_sources.side_effect = RuntimeError(
+            'Some apt error')
+        with self.assertRaises(RuntimeError) as context_manager:
+            maybe_install_squashfuse(cloud=FakeCloud(distro))
+        self.assertEqual('Some apt error', str(context_manager.exception))
+        self.assertIn('Package update failed\nTraceback', self.logs.getvalue())
+
+    @mock.patch('cloudinit.config.cc_snap.util.is_container')
+    def test_maybe_install_squashfuse_raises_update_errors(self, m_container):
+        """maybe_install_squashfuse logs and raises package update errors."""
+        m_container.return_value = True
+        distro = mock.MagicMock()
+        distro.update_package_sources.side_effect = RuntimeError(
+            'Some apt error')
+        with self.assertRaises(RuntimeError) as context_manager:
+            maybe_install_squashfuse(cloud=FakeCloud(distro))
+        self.assertEqual('Some apt error', str(context_manager.exception))
+        self.assertIn('Package update failed\nTraceback', self.logs.getvalue())
+
+    @mock.patch('cloudinit.config.cc_snap.util.is_container')
+    def test_maybe_install_squashfuse_happy_path(self, m_container):
+        """maybe_install_squashfuse logs and raises package install errors."""
+        m_container.return_value = True
+        distro = mock.MagicMock()  # No errors raised
+        maybe_install_squashfuse(cloud=FakeCloud(distro))
+        self.assertEqual(
+            [mock.call()], distro.update_package_sources.call_args_list)
+        self.assertEqual(
+            [mock.call(['squashfuse'])],
+            distro.install_packages.call_args_list)
+
+# vi: ts=4 expandtab