← Back to branch summary

~ubuntu-branches/ubuntu/utopic/heat/utopic-updates

« back to all changes in this revision

Viewing changes to contrib/heat_barbican/heat_barbican/tests/test_secret.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2014-09-08 09:40:59 UTC
  • mfrom: (1.1.16)
  • Revision ID: package-import@ubuntu.com-20140908094059-pzysrm0uy4senjez
Tags: 2014.2~b3-0ubuntu1
New upstream version. 

Show diffs side-by-side

added added

removed removed

Lines of Context:
contrib/heat_barbican/heat_barbican/tests/test_secret.py
 
1
#
 
2
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
3
#    not use this file except in compliance with the License. You may obtain
 
4
#    a copy of the License at
 
5
#
 
6
#         http://www.apache.org/licenses/LICENSE-2.0
 
7
#
 
8
#    Unless required by applicable law or agreed to in writing, software
 
9
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
10
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
11
#    License for the specific language governing permissions and limitations
 
12
#    under the License.
 
13
 
 
14
import mock
 
15
import six
 
16
 
 
17
from heat.common import exception
 
18
from heat.common import template_format
 
19
from heat.engine import resource
 
20
from heat.engine import rsrc_defn
 
21
from heat.engine import scheduler
 
22
from heat.tests.common import HeatTestCase
 
23
from heat.tests import utils
 
24
 
 
25
from .. import client  # noqa
 
26
from ..resources import secret  # noqa
 
27
 
 
28
stack_template = '''
 
29
heat_template_version: 2013-05-23
 
30
description: Test template
 
31
resources:
 
32
  secret:
 
33
    type: OS::Barbican::Secret
 
34
    properties:
 
35
      name: foobar-secret
 
36
'''
 
37
 
 
38
 
 
39
class TestSecret(HeatTestCase):
 
40
 
 
41
    def setUp(self):
 
42
        super(TestSecret, self).setUp()
 
43
        utils.setup_dummy_db()
 
44
        self.ctx = utils.dummy_context()
 
45
 
 
46
        self.patcher_client = mock.patch.object(secret.Secret, 'barbican')
 
47
        mock_client = self.patcher_client.start()
 
48
        self.barbican = mock_client.return_value
 
49
 
 
50
        self._register_resources()
 
51
        self.stack = utils.parse_stack(template_format.parse(stack_template))
 
52
        self.stack.validate()
 
53
        resource_defns = self.stack.t.resource_definitions(self.stack)
 
54
        self.res_template = resource_defns['secret']
 
55
        self.res = self._create_resource('foo', self.res_template, self.stack)
 
56
 
 
57
    def tearDown(self):
 
58
        super(TestSecret, self).tearDown()
 
59
        self.patcher_client.stop()
 
60
 
 
61
    def _register_resources(self):
 
62
        for res_name, res_class in six.iteritems(secret.resource_mapping()):
 
63
            resource._register_class(res_name, res_class)
 
64
 
 
65
    def _create_resource(self, name, snippet, stack):
 
66
        res = secret.Secret(name, snippet, stack)
 
67
        self.barbican.secrets.store.return_value = name + '_id'
 
68
        scheduler.TaskRunner(res.create)()
 
69
        return res
 
70
 
 
71
    def test_create_secret(self):
 
72
        expected_state = (self.res.CREATE, self.res.COMPLETE)
 
73
        self.assertEqual(expected_state, self.res.state)
 
74
        args = self.barbican.secrets.store.call_args[1]
 
75
        self.assertEqual('foobar-secret', args['name'])
 
76
 
 
77
    def test_attributes(self):
 
78
        mock_secret = mock.Mock()
 
79
        mock_secret.status = 'test-status'
 
80
        self.barbican.secrets.get.return_value = mock_secret
 
81
        self.barbican.secrets.decrypt.return_value = 'foo'
 
82
 
 
83
        self.assertEqual('test-status', self.res.FnGetAtt('status'))
 
84
        self.assertEqual('foo', self.res.FnGetAtt('decrypted_payload'))
 
85
 
 
86
    @mock.patch.object(client, 'barbican_client', new=mock.Mock())
 
87
    def test_attributes_handles_exceptions(self):
 
88
        client.barbican_client.HTTPClientError = Exception
 
89
        self.barbican.secrets.get.side_effect = Exception('boom')
 
90
        self.assertRaises(client.barbican_client.HTTPClientError,
 
91
                          self.res.FnGetAtt, 'order_ref')
 
92
 
 
93
    def test_create_secret_sets_resource_id(self):
 
94
        self.assertEqual('foo_id', self.res.resource_id)
 
95
 
 
96
    def test_create_secret_with_plain_text(self):
 
97
        content_type = 'text/plain'
 
98
        props = {
 
99
            'name': 'secret',
 
100
            'payload': 'foobar',
 
101
            'payload_content_type': content_type,
 
102
        }
 
103
        defn = rsrc_defn.ResourceDefinition('secret',
 
104
                                            'OS::Barbican::Secret',
 
105
                                            props)
 
106
        res = self._create_resource(defn.name, defn, self.stack)
 
107
 
 
108
        args = self.barbican.secrets.store.call_args[1]
 
109
        self.assertEqual('foobar', args[res.PAYLOAD])
 
110
        self.assertEqual(content_type, args[res.PAYLOAD_CONTENT_TYPE])
 
111
 
 
112
    def test_create_secret_with_octet_stream(self):
 
113
        content_type = 'application/octet-stream'
 
114
        props = {
 
115
            'name': 'secret',
 
116
            'payload': 'foobar',
 
117
            'payload_content_type': content_type,
 
118
        }
 
119
        defn = rsrc_defn.ResourceDefinition('secret',
 
120
                                            'OS::Barbican::Secret',
 
121
                                            props)
 
122
        res = self._create_resource(defn.name, defn, self.stack)
 
123
 
 
124
        args = self.barbican.secrets.store.call_args[1]
 
125
        self.assertEqual('foobar', args[res.PAYLOAD])
 
126
        self.assertEqual(content_type, args[res.PAYLOAD_CONTENT_TYPE])
 
127
 
 
128
    def test_create_secret_other_content_types_not_allowed(self):
 
129
        props = {
 
130
            'name': 'secret',
 
131
            'payload_content_type': 'not/allowed',
 
132
        }
 
133
        defn = rsrc_defn.ResourceDefinition('secret',
 
134
                                            'OS::Barbican::Secret',
 
135
                                            props)
 
136
        self.assertRaises(exception.ResourceFailure,
 
137
                          self._create_resource, defn.name, defn,
 
138
                          self.stack)
 
139
 
 
140
    def test_validate_payload_and_content_type(self):
 
141
        props = {'payload_content_type': 'text/plain'}
 
142
        defn = rsrc_defn.ResourceDefinition('nopayload',
 
143
                                            'OS::Barbican::Secret',
 
144
                                            props)
 
145
        res = self._create_resource(defn.name, defn, self.stack)
 
146
        exc = self.assertRaises(exception.StackValidationFailed, res.validate)
 
147
        self.assertIn('payload', six.text_type(exc))
 
148
        self.assertIn('payload_content_type', six.text_type(exc))
 
149
 
 
150
        defn = rsrc_defn.ResourceDefinition('notype', 'OS::Barbican::Secret',
 
151
                                            {'payload': 'foo'})
 
152
        res = self._create_resource(defn.name, defn, self.stack)
 
153
        exc = self.assertRaises(exception.StackValidationFailed, res.validate)
 
154
        self.assertIn('payload', six.text_type(exc))
 
155
        self.assertIn('payload_content_type', six.text_type(exc))
 
156
 
 
157
    def test_delete_secret(self):
 
158
        self.assertEqual('foo_id', self.res.resource_id)
 
159
 
 
160
        mock_delete = self.barbican.secrets.delete
 
161
        scheduler.TaskRunner(self.res.delete)()
 
162
 
 
163
        mock_delete.assert_called_once_with('foo_id')
 
164
 
 
165
    @mock.patch.object(client, 'barbican_client', new=mock.Mock())
 
166
    def test_handle_delete_ignores_not_found_errors(self):
 
167
        client.barbican_client.HTTPClientError = Exception
 
168
        exc = client.barbican_client.HTTPClientError('Not Found.')
 
169
        self.barbican.secrets.delete.side_effect = exc
 
170
        scheduler.TaskRunner(self.res.delete)()
 
171
        self.assertTrue(self.barbican.secrets.delete.called)
 
172
 
 
173
    @mock.patch.object(client, 'barbican_client', new=mock.Mock())
 
174
    def test_handle_delete_raises_resource_failure_on_error(self):
 
175
        client.barbican_client.HTTPClientError = Exception
 
176
        exc = client.barbican_client.HTTPClientError('Boom.')
 
177
        self.barbican.secrets.delete.side_effect = exc
 
178
        exc = self.assertRaises(exception.ResourceFailure,
 
179
                                scheduler.TaskRunner(self.res.delete))
 
180
        self.assertIn('Boom.', six.text_type(exc))