azure/aria/aria-extension-cloudify/src/aria/tests/orchestrator/execution_plugin/test_common.py

   1 # Licensed to the Apache Software Foundation (ASF) under one or more
   2 # contributor license agreements.  See the NOTICE file distributed with
   3 # this work for additional information regarding copyright ownership.
   4 # The ASF licenses this file to You under the Apache License, Version 2.0
   5 # (the "License"); you may not use this file except in compliance with
   6 # the License.  You may obtain a copy of the License at
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 from collections import namedtuple
  17
  18 import requests
  19 import pytest
  20
  21 from aria.modeling import models
  22 from aria.orchestrator import exceptions
  23 from aria.orchestrator.execution_plugin import common
  24
  25
  26 class TestDownloadScript(object):
  27
  28     @pytest.fixture(autouse=True)
  29     def patch_requests(self, mocker):
  30         def _mock_requests_get(url):
  31             response = namedtuple('Response', 'text status_code')
  32             return response(url, self.status_code)
  33         self.status_code = 200
  34         mocker.patch.object(requests, 'get', _mock_requests_get)
  35
  36     def _test_url(self, url):
  37         class Ctx(object):
  38             task = models.Task
  39
  40         script_path = url
  41         result = common.download_script(Ctx, script_path)
  42         with open(result) as f:
  43             assert script_path == f.read()
  44         assert result.endswith('-some_script.py')
  45
  46     def test_http_url(self):
  47         self._test_url('http://localhost/some_script.py')
  48
  49     def test_https_url(self):
  50         self._test_url('https://localhost/some_script.py')
  51
  52     def test_url_status_code_404(self):
  53         self.status_code = 404
  54         with pytest.raises(exceptions.TaskAbortException) as exc_ctx:
  55             self.test_http_url()
  56         exception = exc_ctx.value
  57         assert 'status code: 404' in str(exception)
  58
  59     def test_blueprint_resource(self):
  60         test_script_path = 'my_script.py'
  61
  62         class Ctx(object):
  63             @staticmethod
  64             def download_resource(destination, path):
  65                 assert path == test_script_path
  66                 return destination
  67         result = common.download_script(Ctx, test_script_path)
  68         assert result.endswith(test_script_path)
  69
  70
  71 class TestCreateProcessConfig(object):
  72
  73     def test_plain_command(self):
  74         script_path = 'path'
  75         process = common.create_process_config(
  76             script_path=script_path,
  77             process={},
  78             operation_kwargs={})
  79         assert process['command'] == script_path
  80
  81     def test_command_with_args(self):
  82         script_path = 'path'
  83         process = {'args': [1, 2, 3]}
  84         process = common.create_process_config(
  85             script_path=script_path,
  86             process=process,
  87             operation_kwargs={})
  88         assert process['command'] == '{0} 1 2 3'.format(script_path)
  89
  90     def test_command_prefix(self):
  91         script_path = 'path'
  92         command_prefix = 'prefix'
  93         process = {'command_prefix': command_prefix}
  94         process = common.create_process_config(
  95             script_path=script_path,
  96             process=process,
  97             operation_kwargs={})
  98         assert process['command'] == '{0} {1}'.format(command_prefix, script_path)
  99
 100     def test_command_with_args_and_prefix(self):
 101         script_path = 'path'
 102         command_prefix = 'prefix'
 103         process = {'command_prefix': command_prefix,
 104                    'args': [1, 2, 3]}
 105         process = common.create_process_config(
 106             script_path=script_path,
 107             process=process,
 108             operation_kwargs={})
 109         assert process['command'] == '{0} {1} 1 2 3'.format(command_prefix, script_path)
 110
 111     def test_ctx_is_removed(self):
 112         process = common.create_process_config(
 113             script_path='',
 114             process={},
 115             operation_kwargs={'ctx': 1})
 116         assert 'ctx' not in process['env']
 117
 118     def test_env_passed_explicitly(self):
 119         env = {'one': '1', 'two': '2'}
 120         process = common.create_process_config(
 121             script_path='',
 122             process={'env': env},
 123             operation_kwargs={})
 124         assert process['env'] == env
 125
 126     def test_env_populated_from_operation_kwargs(self):
 127         operation_kwargs = {'one': '1', 'two': '2'}
 128         process = common.create_process_config(
 129             script_path='',
 130             process={},
 131             operation_kwargs=operation_kwargs)
 132         assert process['env'] == operation_kwargs
 133
 134     def test_env_merged_from_operation_kwargs_and_process(self):
 135         operation_kwargs = {'one': '1', 'two': '2'}
 136         env = {'three': '3', 'four': '4'}
 137         process = common.create_process_config(
 138             script_path='',
 139             process={'env': env},
 140             operation_kwargs=operation_kwargs)
 141         assert process['env'] == dict(operation_kwargs.items() + env.items())
 142
 143     def test_process_env_gets_precedence_over_operation_kwargs(self):
 144         operation_kwargs = {'one': 'from_kwargs'}
 145         env = {'one': 'from_env_process'}
 146         process = common.create_process_config(
 147             script_path='',
 148             process={'env': env},
 149             operation_kwargs=operation_kwargs)
 150         assert process['env'] == env
 151
 152     def test_json_env_vars(self, mocker):
 153         mocker.patch.object(common, 'is_windows', lambda: False)
 154         operation_kwargs = {'a_dict': {'key': 'value'},
 155                             'a_list': ['a', 'b', 'c'],
 156                             'a_tuple': (4, 5, 6),
 157                             'a_bool': True}
 158         process = common.create_process_config(
 159             script_path='',
 160             process={},
 161             operation_kwargs=operation_kwargs)
 162         assert process['env'] == {'a_dict': '{"key": "value"}',
 163                                   'a_list': '["a", "b", "c"]',
 164                                   'a_tuple': '[4, 5, 6]',
 165                                   'a_bool': 'true'}
 166
 167     def test_quote_json_env_vars(self):
 168         operation_kwargs = {'one': []}
 169         process = common.create_process_config(
 170             script_path='',
 171             process={},
 172             operation_kwargs=operation_kwargs,
 173             quote_json_env_vars=True)
 174         assert process['env']['one'] == "'[]'"
 175
 176     def test_env_keys_converted_to_string_on_windows(self, mocker):
 177         mocker.patch.object(common, 'is_windows', lambda: True)
 178         env = {u'one': '1'}
 179         process = common.create_process_config(
 180             script_path='',
 181             process={'env': env},
 182             operation_kwargs={})
 183         print type(process['env'].keys()[0])
 184         assert isinstance(process['env'].keys()[0], str)
 185
 186     def test_env_values_quotes_are_escaped_on_windows(self, mocker):
 187         mocker.patch.object(common, 'is_windows', lambda: True)
 188         env = {'one': '"hello"'}
 189         process = common.create_process_config(
 190             script_path='',
 191             process={'env': env},
 192             operation_kwargs={})
 193         assert process['env']['one'] == '\\"hello\\"'