1# Licensed under the Apache License, Version 2.0 (the "License");
2# you may not use this file except in compliance with the License.
3# You may obtain a copy of the License at
4#
5#    http://www.apache.org/licenses/LICENSE-2.0
6#
7# Unless required by applicable law or agreed to in writing, software
8# distributed under the License is distributed on an "AS IS" BASIS,
9# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
10# implied.
11# See the License for the specific language governing permissions and
12# limitations under the License.
13
14import json
15import tempfile
16from unittest import mock
17
18from oslo_serialization import base64
19import six
20from six.moves.urllib import error
21import testtools
22from testtools import matchers
23import yaml
24
25from heatclient.common import template_utils
26from heatclient.common import utils
27from heatclient import exc
28
29
30class ShellEnvironmentTest(testtools.TestCase):
31
32    template_a = b'{"heat_template_version": "2013-05-23"}'
33
34    def collect_links(self, env, content, url, env_base_url=''):
35        jenv = yaml.safe_load(env)
36        files = {}
37        if url:
38            def side_effect(args):
39                if url == args:
40                    return six.BytesIO(content)
41            with mock.patch('six.moves.urllib.request.urlopen') as mock_url:
42                mock_url.side_effect = side_effect
43                template_utils.resolve_environment_urls(
44                    jenv.get('resource_registry'), files, env_base_url)
45                self.assertEqual(content.decode('utf-8'), files[url])
46        else:
47            template_utils.resolve_environment_urls(
48                jenv.get('resource_registry'), files, env_base_url)
49
50    @mock.patch('six.moves.urllib.request.urlopen')
51    def test_ignore_env_keys(self, mock_url):
52        env_file = '/home/my/dir/env.yaml'
53        env = b'''
54        resource_registry:
55          resources:
56            bar:
57              hooks: pre_create
58              restricted_actions: replace
59        '''
60        mock_url.return_value = six.BytesIO(env)
61        _, env_dict = template_utils.process_environment_and_files(
62            env_file)
63        self.assertEqual(
64            {u'resource_registry': {u'resources': {
65                u'bar': {u'hooks': u'pre_create',
66                         u'restricted_actions': u'replace'}}}},
67            env_dict)
68        mock_url.assert_called_with('file://%s' % env_file)
69
70    @mock.patch('six.moves.urllib.request.urlopen')
71    def test_process_environment_file(self, mock_url):
72
73        env_file = '/home/my/dir/env.yaml'
74        env = b'''
75        resource_registry:
76          "OS::Thingy": "file:///home/b/a.yaml"
77        '''
78        mock_url.side_effect = [six.BytesIO(env), six.BytesIO(self.template_a),
79                                six.BytesIO(self.template_a)]
80
81        files, env_dict = template_utils.process_environment_and_files(
82            env_file)
83        self.assertEqual(
84            {'resource_registry': {
85                'OS::Thingy': 'file:///home/b/a.yaml'}},
86            env_dict)
87        self.assertEqual(self.template_a.decode('utf-8'),
88                         files['file:///home/b/a.yaml'])
89        mock_url.assert_has_calls([
90            mock.call('file://%s' % env_file),
91            mock.call('file:///home/b/a.yaml'),
92            mock.call('file:///home/b/a.yaml')
93        ])
94
95    @mock.patch('six.moves.urllib.request.urlopen')
96    def test_process_environment_relative_file(self, mock_url):
97
98        env_file = '/home/my/dir/env.yaml'
99        env_url = 'file:///home/my/dir/env.yaml'
100        env = b'''
101        resource_registry:
102          "OS::Thingy": a.yaml
103        '''
104
105        mock_url.side_effect = [six.BytesIO(env), six.BytesIO(self.template_a),
106                                six.BytesIO(self.template_a)]
107
108        self.assertEqual(
109            env_url,
110            utils.normalise_file_path_to_url(env_file))
111        self.assertEqual(
112            'file:///home/my/dir',
113            utils.base_url_for_url(env_url))
114
115        files, env_dict = template_utils.process_environment_and_files(
116            env_file)
117
118        self.assertEqual(
119            {'resource_registry': {
120                'OS::Thingy': 'file:///home/my/dir/a.yaml'}},
121            env_dict)
122        self.assertEqual(self.template_a.decode('utf-8'),
123                         files['file:///home/my/dir/a.yaml'])
124        mock_url.assert_has_calls([
125            mock.call(env_url),
126            mock.call('file:///home/my/dir/a.yaml'),
127            mock.call('file:///home/my/dir/a.yaml')
128        ])
129
130    def test_process_multiple_environment_files_container(self):
131
132        env_list_tracker = []
133        env_paths = ['/home/my/dir/env.yaml']
134        files, env = template_utils.process_multiple_environments_and_files(
135            env_paths, env_list_tracker=env_list_tracker,
136            fetch_env_files=False)
137
138        self.assertEqual(env_paths, env_list_tracker)
139        self.assertEqual({}, files)
140        self.assertEqual({}, env)
141
142    @mock.patch('six.moves.urllib.request.urlopen')
143    def test_process_environment_relative_file_up(self, mock_url):
144
145        env_file = '/home/my/dir/env.yaml'
146        env_url = 'file:///home/my/dir/env.yaml'
147        env = b'''
148        resource_registry:
149          "OS::Thingy": ../bar/a.yaml
150        '''
151        mock_url.side_effect = [six.BytesIO(env), six.BytesIO(self.template_a),
152                                six.BytesIO(self.template_a)]
153
154        env_url = 'file://%s' % env_file
155        self.assertEqual(
156            env_url,
157            utils.normalise_file_path_to_url(env_file))
158        self.assertEqual(
159            'file:///home/my/dir',
160            utils.base_url_for_url(env_url))
161
162        files, env_dict = template_utils.process_environment_and_files(
163            env_file)
164
165        self.assertEqual(
166            {'resource_registry': {
167                'OS::Thingy': 'file:///home/my/bar/a.yaml'}},
168            env_dict)
169        self.assertEqual(self.template_a.decode('utf-8'),
170                         files['file:///home/my/bar/a.yaml'])
171        mock_url.assert_has_calls([
172            mock.call(env_url),
173            mock.call('file:///home/my/bar/a.yaml'),
174            mock.call('file:///home/my/bar/a.yaml')
175        ])
176
177    @mock.patch('six.moves.urllib.request.urlopen')
178    def test_process_environment_url(self, mock_url):
179        env = b'''
180        resource_registry:
181            "OS::Thingy": "a.yaml"
182        '''
183        url = 'http://no.where/some/path/to/file.yaml'
184        tmpl_url = 'http://no.where/some/path/to/a.yaml'
185        mock_url.side_effect = [six.BytesIO(env), six.BytesIO(self.template_a),
186                                six.BytesIO(self.template_a)]
187
188        files, env_dict = template_utils.process_environment_and_files(
189            url)
190
191        self.assertEqual({'resource_registry': {'OS::Thingy': tmpl_url}},
192                         env_dict)
193        self.assertEqual(self.template_a.decode('utf-8'), files[tmpl_url])
194        mock_url.assert_has_calls([
195            mock.call(url),
196            mock.call(tmpl_url),
197            mock.call(tmpl_url)
198        ])
199
200    @mock.patch('six.moves.urllib.request.urlopen')
201    def test_process_environment_empty_file(self, mock_url):
202
203        env_file = '/home/my/dir/env.yaml'
204        env = b''
205        mock_url.return_value = six.BytesIO(env)
206
207        files, env_dict = template_utils.process_environment_and_files(
208            env_file)
209
210        self.assertEqual({}, env_dict)
211        self.assertEqual({}, files)
212        mock_url.assert_called_with('file://%s' % env_file)
213
214    def test_no_process_environment_and_files(self):
215        files, env = template_utils.process_environment_and_files()
216        self.assertEqual({}, env)
217        self.assertEqual({}, files)
218
219    @mock.patch('six.moves.urllib.request.urlopen')
220    def test_process_multiple_environments_and_files(self, mock_url):
221
222        env_file1 = '/home/my/dir/env1.yaml'
223        env_file2 = '/home/my/dir/env2.yaml'
224
225        env1 = b'''
226        parameters:
227          "param1": "value1"
228        resource_registry:
229          "OS::Thingy1": "file:///home/b/a.yaml"
230        '''
231        env2 = b'''
232        parameters:
233          "param2": "value2"
234        resource_registry:
235          "OS::Thingy2": "file:///home/b/b.yaml"
236        '''
237
238        mock_url.side_effect = [six.BytesIO(env1),
239                                six.BytesIO(self.template_a),
240                                six.BytesIO(self.template_a),
241                                six.BytesIO(env2),
242                                six.BytesIO(self.template_a),
243                                six.BytesIO(self.template_a)]
244
245        files, env = template_utils.process_multiple_environments_and_files(
246            [env_file1, env_file2])
247        self.assertEqual(
248            {
249                'resource_registry': {
250                    'OS::Thingy1': 'file:///home/b/a.yaml',
251                    'OS::Thingy2': 'file:///home/b/b.yaml'},
252                'parameters': {
253                    'param1': 'value1',
254                    'param2': 'value2'}
255            },
256            env)
257        self.assertEqual(self.template_a.decode('utf-8'),
258                         files['file:///home/b/a.yaml'])
259        self.assertEqual(self.template_a.decode('utf-8'),
260                         files['file:///home/b/b.yaml'])
261        mock_url.assert_has_calls([
262            mock.call('file://%s' % env_file1),
263            mock.call('file:///home/b/a.yaml'),
264            mock.call('file:///home/b/a.yaml'),
265            mock.call('file://%s' % env_file2),
266            mock.call('file:///home/b/b.yaml'),
267            mock.call('file:///home/b/b.yaml')
268        ])
269
270    @mock.patch('six.moves.urllib.request.urlopen')
271    def test_process_multiple_environments_default_resources(self, mock_url):
272
273        env_file1 = '/home/my/dir/env1.yaml'
274        env_file2 = '/home/my/dir/env2.yaml'
275
276        env1 = b'''
277        resource_registry:
278          resources:
279            resource1:
280              "OS::Thingy1": "file:///home/b/a.yaml"
281            resource2:
282              "OS::Thingy2": "file:///home/b/b.yaml"
283        '''
284        env2 = b'''
285        resource_registry:
286          resources:
287            resource1:
288              "OS::Thingy3": "file:///home/b/a.yaml"
289            resource2:
290              "OS::Thingy4": "file:///home/b/b.yaml"
291        '''
292        mock_url.side_effect = [six.BytesIO(env1),
293                                six.BytesIO(self.template_a),
294                                six.BytesIO(self.template_a),
295                                six.BytesIO(self.template_a),
296                                six.BytesIO(self.template_a),
297                                six.BytesIO(env2),
298                                six.BytesIO(self.template_a),
299                                six.BytesIO(self.template_a),
300                                six.BytesIO(self.template_a),
301                                six.BytesIO(self.template_a)]
302
303        files, env = template_utils.process_multiple_environments_and_files(
304            [env_file1, env_file2])
305        self.assertEqual(
306            {
307                'resource_registry': {
308                    'resources': {
309                        'resource1': {
310                            'OS::Thingy1': 'file:///home/b/a.yaml',
311                            'OS::Thingy3': 'file:///home/b/a.yaml'
312                        },
313                        'resource2': {
314                            'OS::Thingy2': 'file:///home/b/b.yaml',
315                            'OS::Thingy4': 'file:///home/b/b.yaml'
316                        }
317                    }
318                }
319            },
320            env)
321        self.assertEqual(self.template_a.decode('utf-8'),
322                         files['file:///home/b/a.yaml'])
323        self.assertEqual(self.template_a.decode('utf-8'),
324                         files['file:///home/b/b.yaml'])
325        mock_url.assert_has_calls([
326            mock.call('file://%s' % env_file1),
327            mock.call('file:///home/b/a.yaml'),
328            mock.call('file:///home/b/b.yaml'),
329            mock.call('file:///home/b/a.yaml'),
330            mock.call('file:///home/b/b.yaml'),
331            mock.call('file://%s' % env_file2),
332            mock.call('file:///home/b/a.yaml'),
333            mock.call('file:///home/b/b.yaml'),
334            mock.call('file:///home/b/a.yaml'),
335            mock.call('file:///home/b/b.yaml'),
336
337        ], any_order=True)
338
339    def test_no_process_multiple_environments_and_files(self):
340        files, env = template_utils.process_multiple_environments_and_files()
341        self.assertEqual({}, env)
342        self.assertEqual({}, files)
343
344    def test_process_multiple_environments_and_files_from_object(self):
345
346        env_object = 'http://no.where/path/to/env.yaml'
347        env1 = b'''
348        parameters:
349          "param1": "value1"
350        resource_registry:
351          "OS::Thingy1": "b/a.yaml"
352        '''
353
354        self.object_requested = False
355
356        def env_path_is_object(object_url):
357            return True
358
359        def object_request(method, object_url):
360            self.object_requested = True
361            self.assertEqual('GET', method)
362            self.assertTrue(object_url.startswith("http://no.where/path/to/"))
363            if object_url == env_object:
364                return env1
365            else:
366                return self.template_a
367
368        files, env = template_utils.process_multiple_environments_and_files(
369            env_paths=[env_object], env_path_is_object=env_path_is_object,
370            object_request=object_request)
371        self.assertEqual(
372            {
373                'resource_registry': {
374                    'OS::Thingy1': 'http://no.where/path/to/b/a.yaml'},
375                'parameters': {'param1': 'value1'}
376            },
377            env)
378        self.assertEqual(self.template_a.decode('utf-8'),
379                         files['http://no.where/path/to/b/a.yaml'])
380
381    @mock.patch('six.moves.urllib.request.urlopen')
382    def test_process_multiple_environments_and_files_tracker(self, mock_url):
383        # Setup
384        env_file1 = '/home/my/dir/env1.yaml'
385
386        env1 = b'''
387        parameters:
388          "param1": "value1"
389        resource_registry:
390          "OS::Thingy1": "file:///home/b/a.yaml"
391        '''
392        mock_url.side_effect = [six.BytesIO(env1),
393                                six.BytesIO(self.template_a),
394                                six.BytesIO(self.template_a)]
395
396        # Test
397        env_file_list = []
398        files, env = template_utils.process_multiple_environments_and_files(
399            [env_file1], env_list_tracker=env_file_list)
400
401        # Verify
402        expected_env = {'parameters': {'param1': 'value1'},
403                        'resource_registry':
404                            {'OS::Thingy1': 'file:///home/b/a.yaml'}
405                        }
406        self.assertEqual(expected_env, env)
407
408        self.assertEqual(self.template_a.decode('utf-8'),
409                         files['file:///home/b/a.yaml'])
410
411        self.assertEqual(['file:///home/my/dir/env1.yaml'], env_file_list)
412        self.assertIn('file:///home/my/dir/env1.yaml', files)
413        self.assertEqual(expected_env,
414                         json.loads(files['file:///home/my/dir/env1.yaml']))
415        mock_url.assert_has_calls([
416            mock.call('file://%s' % env_file1),
417            mock.call('file:///home/b/a.yaml'),
418            mock.call('file:///home/b/a.yaml'),
419
420        ])
421
422    @mock.patch('six.moves.urllib.request.urlopen')
423    def test_process_environment_relative_file_tracker(self, mock_url):
424
425        env_file = '/home/my/dir/env.yaml'
426        env_url = 'file:///home/my/dir/env.yaml'
427        env = b'''
428        resource_registry:
429          "OS::Thingy": a.yaml
430        '''
431        mock_url.side_effect = [six.BytesIO(env),
432                                six.BytesIO(self.template_a),
433                                six.BytesIO(self.template_a)]
434
435        self.assertEqual(
436            env_url,
437            utils.normalise_file_path_to_url(env_file))
438        self.assertEqual(
439            'file:///home/my/dir',
440            utils.base_url_for_url(env_url))
441
442        env_file_list = []
443        files, env = template_utils.process_multiple_environments_and_files(
444            [env_file], env_list_tracker=env_file_list)
445
446        # Verify
447        expected_env = {'resource_registry':
448                        {'OS::Thingy': 'file:///home/my/dir/a.yaml'}}
449        self.assertEqual(expected_env, env)
450
451        self.assertEqual(self.template_a.decode('utf-8'),
452                         files['file:///home/my/dir/a.yaml'])
453        self.assertEqual(['file:///home/my/dir/env.yaml'], env_file_list)
454        self.assertEqual(json.dumps(expected_env),
455                         files['file:///home/my/dir/env.yaml'])
456        mock_url.assert_has_calls([
457            mock.call(env_url),
458            mock.call('file:///home/my/dir/a.yaml'),
459            mock.call('file:///home/my/dir/a.yaml'),
460
461        ])
462
463    @mock.patch('six.moves.urllib.request.urlopen')
464    def test_process_multiple_environments_empty_registry(self, mock_url):
465        # Setup
466        env_file1 = '/home/my/dir/env1.yaml'
467        env_file2 = '/home/my/dir/env2.yaml'
468
469        env1 = b'''
470        resource_registry:
471          "OS::Thingy1": "file:///home/b/a.yaml"
472        '''
473        env2 = b'''
474        resource_registry:
475        '''
476        mock_url.side_effect = [six.BytesIO(env1),
477                                six.BytesIO(self.template_a),
478                                six.BytesIO(self.template_a),
479                                six.BytesIO(env2)]
480
481        # Test
482        env_file_list = []
483        files, env = template_utils.process_multiple_environments_and_files(
484            [env_file1, env_file2], env_list_tracker=env_file_list)
485
486        # Verify
487        expected_env = {
488            'resource_registry': {'OS::Thingy1': 'file:///home/b/a.yaml'}}
489        self.assertEqual(expected_env, env)
490
491        self.assertEqual(self.template_a.decode('utf-8'),
492                         files['file:///home/b/a.yaml'])
493
494        self.assertEqual(['file:///home/my/dir/env1.yaml',
495                          'file:///home/my/dir/env2.yaml'], env_file_list)
496        self.assertIn('file:///home/my/dir/env1.yaml', files)
497        self.assertIn('file:///home/my/dir/env2.yaml', files)
498        self.assertEqual(expected_env,
499                         json.loads(files['file:///home/my/dir/env1.yaml']))
500        mock_url.assert_has_calls([
501            mock.call('file://%s' % env_file1),
502            mock.call('file:///home/b/a.yaml'),
503            mock.call('file:///home/b/a.yaml'),
504            mock.call('file://%s' % env_file2),
505
506        ])
507
508    def test_global_files(self):
509        url = 'file:///home/b/a.yaml'
510        env = '''
511        resource_registry:
512          "OS::Thingy": "%s"
513        ''' % url
514        self.collect_links(env, self.template_a, url)
515
516    def test_nested_files(self):
517        url = 'file:///home/b/a.yaml'
518        env = '''
519        resource_registry:
520          resources:
521            freddy:
522              "OS::Thingy": "%s"
523        ''' % url
524        self.collect_links(env, self.template_a, url)
525
526    def test_http_url(self):
527        url = 'http://no.where/container/a.yaml'
528        env = '''
529        resource_registry:
530          "OS::Thingy": "%s"
531        ''' % url
532        self.collect_links(env, self.template_a, url)
533
534    def test_with_base_url(self):
535        url = 'ftp://no.where/container/a.yaml'
536        env = '''
537        resource_registry:
538          base_url: "ftp://no.where/container/"
539          resources:
540            server_for_me:
541              "OS::Thingy": a.yaml
542        '''
543        self.collect_links(env, self.template_a, url)
544
545    def test_with_built_in_provider(self):
546        env = '''
547        resource_registry:
548          resources:
549            server_for_me:
550              "OS::Thingy": OS::Compute::Server
551        '''
552        self.collect_links(env, self.template_a, None)
553
554    def test_with_env_file_base_url_file(self):
555        url = 'file:///tmp/foo/a.yaml'
556        env = '''
557        resource_registry:
558          resources:
559            server_for_me:
560              "OS::Thingy": a.yaml
561        '''
562        env_base_url = 'file:///tmp/foo'
563        self.collect_links(env, self.template_a, url, env_base_url)
564
565    def test_with_env_file_base_url_http(self):
566        url = 'http://no.where/path/to/a.yaml'
567        env = '''
568        resource_registry:
569          resources:
570            server_for_me:
571              "OS::Thingy": to/a.yaml
572        '''
573        env_base_url = 'http://no.where/path'
574        self.collect_links(env, self.template_a, url, env_base_url)
575
576    def test_unsupported_protocol(self):
577        env = '''
578        resource_registry:
579          "OS::Thingy": "sftp://no.where/dev/null/a.yaml"
580        '''
581        jenv = yaml.safe_load(env)
582        fields = {'files': {}}
583        self.assertRaises(exc.CommandError,
584                          template_utils.get_file_contents,
585                          jenv['resource_registry'],
586                          fields)
587
588
589class TestGetTemplateContents(testtools.TestCase):
590
591    def test_get_template_contents_file(self):
592        with tempfile.NamedTemporaryFile() as tmpl_file:
593            tmpl = (b'{"AWSTemplateFormatVersion" : "2010-09-09",'
594                    b' "foo": "bar"}')
595            tmpl_file.write(tmpl)
596            tmpl_file.flush()
597
598            files, tmpl_parsed = template_utils.get_template_contents(
599                tmpl_file.name)
600            self.assertEqual({"AWSTemplateFormatVersion": "2010-09-09",
601                              "foo": "bar"}, tmpl_parsed)
602            self.assertEqual({}, files)
603
604    def test_get_template_contents_file_empty(self):
605        with tempfile.NamedTemporaryFile() as tmpl_file:
606
607            ex = self.assertRaises(
608                exc.CommandError,
609                template_utils.get_template_contents,
610                tmpl_file.name)
611            self.assertEqual(
612                'Could not fetch template from file://%s' % tmpl_file.name,
613                str(ex))
614
615    def test_get_template_file_nonextant(self):
616        nonextant_file = '/template/dummy/file/path/and/name.yaml'
617        ex = self.assertRaises(
618            error.URLError,
619            template_utils.get_template_contents,
620            nonextant_file)
621        self.assertEqual(
622            "<urlopen error [Errno 2] No such file or directory: '%s'>"
623            % nonextant_file,
624            str(ex))
625
626    def test_get_template_contents_file_none(self):
627        ex = self.assertRaises(
628            exc.CommandError,
629            template_utils.get_template_contents)
630        self.assertEqual(
631            ('Need to specify exactly one of [--template-file, '
632             '--template-url or --template-object] or --existing'),
633            str(ex))
634
635    def test_get_template_contents_file_none_existing(self):
636        files, tmpl_parsed = template_utils.get_template_contents(
637            existing=True)
638        self.assertIsNone(tmpl_parsed)
639        self.assertEqual({}, files)
640
641    def test_get_template_contents_parse_error(self):
642        with tempfile.NamedTemporaryFile() as tmpl_file:
643
644            tmpl = b'{"foo": "bar"'
645            tmpl_file.write(tmpl)
646            tmpl_file.flush()
647
648            ex = self.assertRaises(
649                exc.CommandError,
650                template_utils.get_template_contents,
651                tmpl_file.name)
652            self.assertThat(
653                str(ex),
654                matchers.MatchesRegex(
655                    'Error parsing template file://%s ' % tmpl_file.name))
656
657    @mock.patch('six.moves.urllib.request.urlopen')
658    def test_get_template_contents_url(self, mock_url):
659        tmpl = b'{"AWSTemplateFormatVersion" : "2010-09-09", "foo": "bar"}'
660        url = 'http://no.where/path/to/a.yaml'
661        mock_url.return_value = six.BytesIO(tmpl)
662
663        files, tmpl_parsed = template_utils.get_template_contents(
664            template_url=url)
665        self.assertEqual({"AWSTemplateFormatVersion": "2010-09-09",
666                          "foo": "bar"}, tmpl_parsed)
667        self.assertEqual({}, files)
668        mock_url.assert_called_with(url)
669
670    def test_get_template_contents_object(self):
671        tmpl = '{"AWSTemplateFormatVersion" : "2010-09-09", "foo": "bar"}'
672        url = 'http://no.where/path/to/a.yaml'
673
674        self.object_requested = False
675
676        def object_request(method, object_url):
677            self.object_requested = True
678            self.assertEqual('GET', method)
679            self.assertEqual('http://no.where/path/to/a.yaml', object_url)
680            return tmpl
681
682        files, tmpl_parsed = template_utils.get_template_contents(
683            template_object=url,
684            object_request=object_request)
685
686        self.assertEqual({"AWSTemplateFormatVersion": "2010-09-09",
687                          "foo": "bar"}, tmpl_parsed)
688        self.assertEqual({}, files)
689        self.assertTrue(self.object_requested)
690
691    def test_get_nested_stack_template_contents_object(self):
692        tmpl = ('{"heat_template_version": "2016-04-08",'
693                '"resources": {'
694                '"FooBar": {'
695                '"type": "foo/bar.yaml"}}}')
696        url = 'http://no.where/path/to/a.yaml'
697
698        self.object_requested = False
699
700        def object_request(method, object_url):
701            self.object_requested = True
702            self.assertEqual('GET', method)
703            self.assertTrue(object_url.startswith("http://no.where/path/to/"))
704            if object_url == url:
705                return tmpl
706            else:
707                return '{"heat_template_version": "2016-04-08"}'
708
709        files, tmpl_parsed = template_utils.get_template_contents(
710            template_object=url,
711            object_request=object_request)
712
713        self.assertEqual(files['http://no.where/path/to/foo/bar.yaml'],
714                         '{"heat_template_version": "2016-04-08"}')
715        self.assertTrue(self.object_requested)
716
717    def check_non_utf8_content(self, filename, content):
718        base_url = 'file:///tmp'
719        url = '%s/%s' % (base_url, filename)
720        template = {'resources':
721                    {'one_init':
722                     {'type': 'OS::Heat::CloudConfig',
723                      'properties':
724                      {'cloud_config':
725                       {'write_files':
726                        [{'path': '/tmp/%s' % filename,
727                          'content': {'get_file': url},
728                          'encoding': 'b64'}]}}}}}
729        with mock.patch('six.moves.urllib.request.urlopen') as mock_url:
730            raw_content = base64.decode_as_bytes(content)
731            response = six.BytesIO(raw_content)
732            mock_url.return_value = response
733            files = {}
734            template_utils.resolve_template_get_files(
735                template, files, base_url)
736            self.assertEqual({url: content}, files)
737            mock_url.assert_called_with(url)
738
739    def test_get_zip_content(self):
740        filename = 'heat.zip'
741        content = b'''\
742UEsDBAoAAAAAAEZZWkRbOAuBBQAAAAUAAAAIABwAaGVhdC50eHRVVAkAAxRbDVNYh\
743t9SdXgLAAEE\n6AMAAATpAwAAaGVhdApQSwECHgMKAAAAAABGWVpEWzgLgQUAAAAF\
744AAAACAAYAAAAAAABAAAApIEA\nAAAAaGVhdC50eHRVVAUAAxRbDVN1eAsAAQToAwA\
745ABOkDAABQSwUGAAAAAAEAAQBOAAAARwAAAAAA\n'''
746        # zip has '\0' in stream
747        self.assertIn(b'\0', base64.decode_as_bytes(content))
748        decoded_content = base64.decode_as_bytes(content)
749        if six.PY3:
750            self.assertRaises(UnicodeDecodeError, decoded_content.decode)
751        else:
752            self.assertRaises(
753                UnicodeDecodeError,
754                json.dumps,
755                {'content': decoded_content})
756        self.check_non_utf8_content(
757            filename=filename, content=content)
758
759    def test_get_utf16_content(self):
760        filename = 'heat.utf16'
761        content = b'//4tTkhTCgA=\n'
762        # utf6 has '\0' in stream
763        self.assertIn(b'\0', base64.decode_as_bytes(content))
764        decoded_content = base64.decode_as_bytes(content)
765        if six.PY3:
766            self.assertRaises(UnicodeDecodeError, decoded_content.decode)
767        else:
768            self.assertRaises(
769                UnicodeDecodeError,
770                json.dumps,
771                {'content': decoded_content})
772        self.check_non_utf8_content(filename=filename, content=content)
773
774    def test_get_gb18030_content(self):
775        filename = 'heat.gb18030'
776        content = b'1tDO5wo=\n'
777        # gb18030 has no '\0' in stream
778        self.assertNotIn('\0', base64.decode_as_bytes(content))
779        decoded_content = base64.decode_as_bytes(content)
780        if six.PY3:
781            self.assertRaises(UnicodeDecodeError, decoded_content.decode)
782        else:
783            self.assertRaises(
784                UnicodeDecodeError,
785                json.dumps,
786                {'content': decoded_content})
787        self.check_non_utf8_content(filename=filename, content=content)
788
789
790@mock.patch('six.moves.urllib.request.urlopen')
791class TestTemplateGetFileFunctions(testtools.TestCase):
792
793    hot_template = b'''heat_template_version: 2013-05-23
794resources:
795  resource1:
796    type: OS::type1
797    properties:
798      foo: {get_file: foo.yaml}
799      bar:
800        get_file:
801          'http://localhost/bar.yaml'
802  resource2:
803    type: OS::type1
804    properties:
805      baz:
806      - {get_file: baz/baz1.yaml}
807      - {get_file: baz/baz2.yaml}
808      - {get_file: baz/baz3.yaml}
809      ignored_list: {get_file: [ignore, me]}
810      ignored_dict: {get_file: {ignore: me}}
811      ignored_none: {get_file: }
812    '''
813
814    def test_hot_template(self, mock_url):
815
816        tmpl_file = '/home/my/dir/template.yaml'
817        url = 'file:///home/my/dir/template.yaml'
818        mock_url.side_effect = [six.BytesIO(self.hot_template),
819                                six.BytesIO(b'bar contents'),
820                                six.BytesIO(b'foo contents'),
821                                six.BytesIO(b'baz1 contents'),
822                                six.BytesIO(b'baz2 contents'),
823                                six.BytesIO(b'baz3 contents')]
824
825        files, tmpl_parsed = template_utils.get_template_contents(
826            template_file=tmpl_file)
827
828        self.assertEqual({
829            'heat_template_version': '2013-05-23',
830            'resources': {
831                'resource1': {
832                    'type': 'OS::type1',
833                    'properties': {
834                        'bar': {'get_file': 'http://localhost/bar.yaml'},
835                        'foo': {'get_file': 'file:///home/my/dir/foo.yaml'},
836                    },
837                },
838                'resource2': {
839                    'type': 'OS::type1',
840                    'properties': {
841                        'baz': [
842                            {'get_file': 'file:///home/my/dir/baz/baz1.yaml'},
843                            {'get_file': 'file:///home/my/dir/baz/baz2.yaml'},
844                            {'get_file': 'file:///home/my/dir/baz/baz3.yaml'},
845                        ],
846                        'ignored_list': {'get_file': ['ignore', 'me']},
847                        'ignored_dict': {'get_file': {'ignore': 'me'}},
848                        'ignored_none': {'get_file': None},
849                    },
850                }
851            }
852        }, tmpl_parsed)
853        mock_url.assert_has_calls([
854            mock.call(url),
855            mock.call('http://localhost/bar.yaml'),
856            mock.call('file:///home/my/dir/foo.yaml'),
857            mock.call('file:///home/my/dir/baz/baz1.yaml'),
858            mock.call('file:///home/my/dir/baz/baz2.yaml'),
859            mock.call('file:///home/my/dir/baz/baz3.yaml')
860        ], any_order=True)
861
862    def test_hot_template_outputs(self, mock_url):
863        tmpl_file = '/home/my/dir/template.yaml'
864        url = 'file://%s' % tmpl_file
865        foo_url = 'file:///home/my/dir/foo.yaml'
866        contents = b'''
867heat_template_version: 2013-05-23\n\
868outputs:\n\
869  contents:\n\
870    value:\n\
871      get_file: foo.yaml\n'''
872        mock_url.side_effect = [six.BytesIO(contents),
873                                six.BytesIO(b'foo contents')]
874        files = template_utils.get_template_contents(
875            template_file=tmpl_file)[0]
876        self.assertEqual({foo_url: b'foo contents'}, files)
877        mock_url.assert_has_calls([
878            mock.call(url),
879            mock.call(foo_url)
880        ])
881
882    def test_hot_template_same_file(self, mock_url):
883        tmpl_file = '/home/my/dir/template.yaml'
884        url = 'file://%s' % tmpl_file
885        foo_url = 'file:///home/my/dir/foo.yaml'
886        contents = b'''
887heat_template_version: 2013-05-23\n
888outputs:\n\
889  contents:\n\
890    value:\n\
891      get_file: foo.yaml\n\
892  template:\n\
893    value:\n\
894      get_file: foo.yaml\n'''
895        mock_url.side_effect = [six.BytesIO(contents),
896                                six.BytesIO(b'foo contents')]
897        # asserts that is fetched only once even though it is
898        # referenced in the template twice
899        files = template_utils.get_template_contents(
900            template_file=tmpl_file)[0]
901        self.assertEqual({foo_url: b'foo contents'}, files)
902        mock_url.assert_has_calls([
903            mock.call(url),
904            mock.call(foo_url)
905        ])
906
907
908class TestTemplateTypeFunctions(testtools.TestCase):
909
910    hot_template = b'''heat_template_version: 2013-05-23
911parameters:
912  param1:
913    type: string
914resources:
915  resource1:
916    type: foo.yaml
917    properties:
918      foo: bar
919  resource2:
920    type: OS::Heat::ResourceGroup
921    properties:
922      resource_def:
923        type: spam/egg.yaml
924    '''
925
926    foo_template = b'''heat_template_version: "2013-05-23"
927parameters:
928  foo:
929    type: string
930    '''
931
932    egg_template = b'''heat_template_version: "2013-05-23"
933parameters:
934  egg:
935    type: string
936    '''
937
938    @mock.patch('six.moves.urllib.request.urlopen')
939    def test_hot_template(self, mock_url):
940        tmpl_file = '/home/my/dir/template.yaml'
941        url = 'file:///home/my/dir/template.yaml'
942
943        def side_effect(args):
944            if url == args:
945                return six.BytesIO(self.hot_template)
946            if 'file:///home/my/dir/foo.yaml' == args:
947                return six.BytesIO(self.foo_template)
948            if 'file:///home/my/dir/spam/egg.yaml' == args:
949                return six.BytesIO(self.egg_template)
950        mock_url.side_effect = side_effect
951
952        files, tmpl_parsed = template_utils.get_template_contents(
953            template_file=tmpl_file)
954
955        self.assertEqual(yaml.safe_load(self.foo_template.decode('utf-8')),
956                         json.loads(files.get('file:///home/my/dir/foo.yaml')))
957
958        self.assertEqual(
959            yaml.safe_load(self.egg_template.decode('utf-8')),
960            json.loads(files.get('file:///home/my/dir/spam/egg.yaml')))
961
962        self.assertEqual({
963            u'heat_template_version': u'2013-05-23',
964            u'parameters': {
965                u'param1': {
966                    u'type': u'string'
967                }
968            },
969            u'resources': {
970                u'resource1': {
971                    u'type': u'file:///home/my/dir/foo.yaml',
972                    u'properties': {u'foo': u'bar'}
973                },
974                u'resource2': {
975                    u'type': u'OS::Heat::ResourceGroup',
976                    u'properties': {
977                        u'resource_def': {
978                            u'type': u'file:///home/my/dir/spam/egg.yaml'
979                        }
980                    }
981                }
982            }
983        }, tmpl_parsed)
984
985        mock_url.assert_has_calls([
986            mock.call('file:///home/my/dir/foo.yaml'),
987            mock.call(url),
988            mock.call('file:///home/my/dir/spam/egg.yaml'),
989        ], any_order=True)
990
991
992class TestTemplateInFileFunctions(testtools.TestCase):
993
994    hot_template = b'''heat_template_version: 2013-05-23
995resources:
996  resource1:
997    type: OS::Heat::Stack
998    properties:
999      template: {get_file: foo.yaml}
1000    '''
1001
1002    foo_template = b'''heat_template_version: "2013-05-23"
1003resources:
1004  foo:
1005    type: OS::Type1
1006    properties:
1007      config: {get_file: bar.yaml}
1008    '''
1009
1010    bar_template = b'''heat_template_version: "2013-05-23"
1011parameters:
1012  bar:
1013    type: string
1014    '''
1015
1016    @mock.patch('six.moves.urllib.request.urlopen')
1017    def test_hot_template(self, mock_url):
1018        tmpl_file = '/home/my/dir/template.yaml'
1019        url = 'file:///home/my/dir/template.yaml'
1020        foo_url = 'file:///home/my/dir/foo.yaml'
1021        bar_url = 'file:///home/my/dir/bar.yaml'
1022
1023        def side_effect(args):
1024            if url == args:
1025                return six.BytesIO(self.hot_template)
1026            if foo_url == args:
1027                return six.BytesIO(self.foo_template)
1028            if bar_url == args:
1029                return six.BytesIO(self.bar_template)
1030        mock_url.side_effect = side_effect
1031
1032        files, tmpl_parsed = template_utils.get_template_contents(
1033            template_file=tmpl_file)
1034
1035        self.assertEqual(yaml.safe_load(self.bar_template.decode('utf-8')),
1036                         json.loads(files.get('file:///home/my/dir/bar.yaml')))
1037
1038        self.assertEqual({
1039            u'heat_template_version': u'2013-05-23',
1040            u'resources': {
1041                u'foo': {
1042                    u'type': u'OS::Type1',
1043                    u'properties': {
1044                        u'config': {
1045                            u'get_file': u'file:///home/my/dir/bar.yaml'
1046                        }
1047                    }
1048                }
1049            }
1050        }, json.loads(files.get('file:///home/my/dir/foo.yaml')))
1051
1052        self.assertEqual({
1053            u'heat_template_version': u'2013-05-23',
1054            u'resources': {
1055                u'resource1': {
1056                    u'type': u'OS::Heat::Stack',
1057                    u'properties': {
1058                        u'template': {
1059                            u'get_file': u'file:///home/my/dir/foo.yaml'
1060                        }
1061                    }
1062                }
1063            }
1064        }, tmpl_parsed)
1065
1066        mock_url.assert_has_calls([
1067            mock.call(foo_url),
1068            mock.call(url),
1069            mock.call(bar_url),
1070        ], any_order=True)
1071
1072
1073class TestNestedIncludes(testtools.TestCase):
1074
1075    hot_template = b'''heat_template_version: 2013-05-23
1076parameters:
1077  param1:
1078    type: string
1079resources:
1080  resource1:
1081    type: foo.yaml
1082    properties:
1083      foo: bar
1084  resource2:
1085    type: OS::Heat::ResourceGroup
1086    properties:
1087      resource_def:
1088        type: spam/egg.yaml
1089      with: {get_file: spam/ham.yaml}
1090    '''
1091
1092    egg_template = b'''heat_template_version: 2013-05-23
1093parameters:
1094  param1:
1095    type: string
1096resources:
1097  resource1:
1098    type: one.yaml
1099    properties:
1100      foo: bar
1101  resource2:
1102    type: OS::Heat::ResourceGroup
1103    properties:
1104      resource_def:
1105        type: two.yaml
1106      with: {get_file: three.yaml}
1107    '''
1108
1109    foo_template = b'''heat_template_version: "2013-05-23"
1110parameters:
1111  foo:
1112    type: string
1113    '''
1114
1115    @mock.patch('six.moves.urllib.request.urlopen')
1116    def test_env_nested_includes(self, mock_url):
1117        env_file = '/home/my/dir/env.yaml'
1118        env_url = 'file:///home/my/dir/env.yaml'
1119        env = b'''
1120        resource_registry:
1121          "OS::Thingy": template.yaml
1122        '''
1123        template_url = u'file:///home/my/dir/template.yaml'
1124        foo_url = u'file:///home/my/dir/foo.yaml'
1125        egg_url = u'file:///home/my/dir/spam/egg.yaml'
1126        ham_url = u'file:///home/my/dir/spam/ham.yaml'
1127        one_url = u'file:///home/my/dir/spam/one.yaml'
1128        two_url = u'file:///home/my/dir/spam/two.yaml'
1129        three_url = u'file:///home/my/dir/spam/three.yaml'
1130
1131        def side_effect(args):
1132            if env_url == args:
1133                return six.BytesIO(env)
1134            if template_url == args:
1135                return six.BytesIO(self.hot_template)
1136            if foo_url == args:
1137                return six.BytesIO(self.foo_template)
1138            if egg_url == args:
1139                return six.BytesIO(self.egg_template)
1140            if ham_url == args:
1141                return six.BytesIO(b'ham contents')
1142            if one_url == args:
1143                return six.BytesIO(self.foo_template)
1144            if two_url == args:
1145                return six.BytesIO(self.foo_template)
1146            if three_url == args:
1147                return six.BytesIO(b'three contents')
1148        mock_url.side_effect = side_effect
1149
1150        files, env_dict = template_utils.process_environment_and_files(
1151            env_file)
1152
1153        self.assertEqual(
1154            {'resource_registry': {
1155                'OS::Thingy': template_url}},
1156            env_dict)
1157
1158        self.assertEqual({
1159            u'heat_template_version': u'2013-05-23',
1160            u'parameters': {u'param1': {u'type': u'string'}},
1161            u'resources': {
1162                u'resource1': {
1163                    u'properties': {u'foo': u'bar'},
1164                    u'type': foo_url
1165                },
1166                u'resource2': {
1167                    u'type': u'OS::Heat::ResourceGroup',
1168                    u'properties': {
1169                        u'resource_def': {
1170                            u'type': egg_url},
1171                        u'with': {u'get_file': ham_url}
1172                    }
1173                }
1174            }
1175        }, json.loads(files.get(template_url)))
1176
1177        self.assertEqual(yaml.safe_load(self.foo_template.decode('utf-8')),
1178                         json.loads(files.get(foo_url)))
1179        self.assertEqual({
1180            u'heat_template_version': u'2013-05-23',
1181            u'parameters': {u'param1': {u'type': u'string'}},
1182            u'resources': {
1183                u'resource1': {
1184                    u'properties': {u'foo': u'bar'},
1185                    u'type': one_url},
1186                u'resource2': {
1187                    u'type': u'OS::Heat::ResourceGroup',
1188                    u'properties': {
1189                        u'resource_def': {u'type': two_url},
1190                        u'with': {u'get_file': three_url}
1191                    }
1192                }
1193            }
1194        }, json.loads(files.get(egg_url)))
1195        self.assertEqual(b'ham contents',
1196                         files.get(ham_url))
1197        self.assertEqual(yaml.safe_load(self.foo_template.decode('utf-8')),
1198                         json.loads(files.get(one_url)))
1199        self.assertEqual(yaml.safe_load(self.foo_template.decode('utf-8')),
1200                         json.loads(files.get(two_url)))
1201        self.assertEqual(b'three contents',
1202                         files.get(three_url))
1203        mock_url.assert_has_calls([
1204            mock.call(env_url),
1205            mock.call(template_url),
1206            mock.call(foo_url),
1207            mock.call(egg_url),
1208            mock.call(ham_url),
1209            mock.call(one_url),
1210            mock.call(two_url),
1211            mock.call(three_url),
1212        ], any_order=True)
1213