Merge pull request #147 from ChrisMacNaughton/file-ownership

WIP: Add support for testing file ownership

unit_tests/test_zaza_charm_lifecycle_deploy.py

@@ -133,6 +133,9 @@ class TestCharmLifecycleDeploy(ut_utils.BaseTestCase):
         self.assertIsNone(lc_deploy.render_overlay('mybundle.yaml', '/tmp/'))
 
     def test_render_local_overlay(self):
+        self.patch_object(lc_deploy.utils, 'get_charm_config')
+        self.get_charm_config.return_value = {
+            'charm_name': 'mycharm'}
         self.patch_object(lc_deploy.jinja2, 'Environment')
         self.patch_object(lc_deploy, 'get_template', return_value='atemplate')
         self.patch_object(lc_deploy, 'render_template')
@@ -147,6 +150,9 @@ class TestCharmLifecycleDeploy(ut_utils.BaseTestCase):
     def test_render_local_overlay_default(self):
         jenv_mock = mock.MagicMock()
         jenv_mock.from_string.return_value = 'atemplate'
+        self.patch_object(lc_deploy.utils, 'get_charm_config')
+        self.get_charm_config.return_value = {
+            'charm_name': 'mycharm'}
         self.patch_object(lc_deploy.jinja2, 'Environment',
                           return_value=jenv_mock)
         self.patch_object(lc_deploy, 'get_template', return_value=None)

zaza/charm_lifecycle/deploy.py

@@ -88,7 +88,12 @@ def get_template_overlay_context():
     context = {}
     contexts = [
         get_template_context_from_env(),
-        get_charm_config_context()]
+    ]
+    try:
+        contexts.append(get_charm_config_context())
+    except KeyError:
+        pass
+
     for c in contexts:
         context.update(c)
     return context
@@ -191,8 +196,9 @@ def render_local_overlay(target_dir):
     rendered_template_file = os.path.join(
         target_dir,
         os.path.basename(LOCAL_OVERLAY_TEMPLATE_NAME))
-    render_template(template, rendered_template_file)
-    return rendered_template_file
+    if utils.get_charm_config().get('charm_name', None):
+        render_template(template, rendered_template_file)
+        return rendered_template_file
 
 
 def render_overlays(bundle, target_dir):
@@ -205,7 +211,10 @@ def render_overlays(bundle, target_dir):
     :returns: List of rendered overlays
     :rtype: [str, str,...]
     """
-    overlays = [render_local_overlay(target_dir)]
+    overlays = []
+    local_overlay = render_local_overlay(target_dir)
+    if local_overlay:
+        overlays.append(local_overlay)
     rendered_bundle_overlay = render_overlay(bundle, target_dir)
     if rendered_bundle_overlay:
         overlays.append(rendered_bundle_overlay)

zaza/charm_tests/security/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2018 Canonical Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Test security checklist."""

zaza/charm_tests/security/tests.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+
+# Copyright 2018 Canonical Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Encapsulate general security testing."""
+
+import unittest
+
+import zaza.model as model
+import zaza.charm_lifecycle.utils as utils
+
+
+def _make_test_function(application, file_details):
+    def test(self):
+        expected_owner = file_details.get("owner", "root")
+        expected_group = file_details.get("group", "root")
+        expected_mode = file_details.get("mode", "600")
+        for unit in model.get_units(application):
+            unit = unit.entity_id
+            result = model.run_on_unit(
+                unit, 'stat -c "%U %G %a" {}'.format(file_details['path']))
+            ownership = result['Stdout']
+            owner, group, mode = ownership.split()
+            self.assertEqual(expected_owner,
+                             owner,
+                             "Owner is incorrect for {}: {}"
+                             .format(unit, owner))
+            self.assertEqual(expected_group,
+                             group,
+                             "Group is incorrect for {}: {}"
+                             .format(unit, group))
+            self.assertEqual(expected_mode,
+                             mode,
+                             "Mode is incorrect for {}: {}"
+                             .format(unit, mode))
+    return test
+
+
+def _add_tests():
+    def class_decorator(cls):
+        """Add tests based on input yaml to `cls`."""
+        files = utils.get_charm_config('./files.yaml')
+        deployed_applications = model.sync_deployed()
+        for name, attributes in files.items():
+            # Lets make sure to only add tests for deployed applications
+            if name in deployed_applications:
+                for file in attributes['files']:
+                    test_func = _make_test_function(name, file)
+                    setattr(
+                        cls,
+                        'test_{}_{}'.format(name, file['path']),
+                        test_func)
+        return cls
+    return class_decorator
+
+
+class FileOwnershipTest(unittest.TestCase):
+    """Encapsulate File ownership tests."""
+
+    pass
+
+
+FileOwnershipTest = _add_tests()(FileOwnershipTest)

zaza/model.py

@@ -89,6 +89,8 @@ async def deployed():
         # Disconnect from the api server and cleanup.
         await model.disconnect()
 
+sync_deployed = sync_wrapper(deployed)
+
 
 def get_unit_from_name(unit_name, model=None, model_name=None):
     """Return the units that corresponds to the name in the given model.