diff options
Diffstat (limited to 'azure/aria/aria-extension-cloudify/src/aria/tests/orchestrator/execution_plugin/test_common.py')
-rw-r--r-- | azure/aria/aria-extension-cloudify/src/aria/tests/orchestrator/execution_plugin/test_common.py | 193 |
1 files changed, 193 insertions, 0 deletions
diff --git a/azure/aria/aria-extension-cloudify/src/aria/tests/orchestrator/execution_plugin/test_common.py b/azure/aria/aria-extension-cloudify/src/aria/tests/orchestrator/execution_plugin/test_common.py new file mode 100644 index 0000000..dd1e9fb --- /dev/null +++ b/azure/aria/aria-extension-cloudify/src/aria/tests/orchestrator/execution_plugin/test_common.py @@ -0,0 +1,193 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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. + +from collections import namedtuple + +import requests +import pytest + +from aria.modeling import models +from aria.orchestrator import exceptions +from aria.orchestrator.execution_plugin import common + + +class TestDownloadScript(object): + + @pytest.fixture(autouse=True) + def patch_requests(self, mocker): + def _mock_requests_get(url): + response = namedtuple('Response', 'text status_code') + return response(url, self.status_code) + self.status_code = 200 + mocker.patch.object(requests, 'get', _mock_requests_get) + + def _test_url(self, url): + class Ctx(object): + task = models.Task + + script_path = url + result = common.download_script(Ctx, script_path) + with open(result) as f: + assert script_path == f.read() + assert result.endswith('-some_script.py') + + def test_http_url(self): + self._test_url('http://localhost/some_script.py') + + def test_https_url(self): + self._test_url('https://localhost/some_script.py') + + def test_url_status_code_404(self): + self.status_code = 404 + with pytest.raises(exceptions.TaskAbortException) as exc_ctx: + self.test_http_url() + exception = exc_ctx.value + assert 'status code: 404' in str(exception) + + def test_blueprint_resource(self): + test_script_path = 'my_script.py' + + class Ctx(object): + @staticmethod + def download_resource(destination, path): + assert path == test_script_path + return destination + result = common.download_script(Ctx, test_script_path) + assert result.endswith(test_script_path) + + +class TestCreateProcessConfig(object): + + def test_plain_command(self): + script_path = 'path' + process = common.create_process_config( + script_path=script_path, + process={}, + operation_kwargs={}) + assert process['command'] == script_path + + def test_command_with_args(self): + script_path = 'path' + process = {'args': [1, 2, 3]} + process = common.create_process_config( + script_path=script_path, + process=process, + operation_kwargs={}) + assert process['command'] == '{0} 1 2 3'.format(script_path) + + def test_command_prefix(self): + script_path = 'path' + command_prefix = 'prefix' + process = {'command_prefix': command_prefix} + process = common.create_process_config( + script_path=script_path, + process=process, + operation_kwargs={}) + assert process['command'] == '{0} {1}'.format(command_prefix, script_path) + + def test_command_with_args_and_prefix(self): + script_path = 'path' + command_prefix = 'prefix' + process = {'command_prefix': command_prefix, + 'args': [1, 2, 3]} + process = common.create_process_config( + script_path=script_path, + process=process, + operation_kwargs={}) + assert process['command'] == '{0} {1} 1 2 3'.format(command_prefix, script_path) + + def test_ctx_is_removed(self): + process = common.create_process_config( + script_path='', + process={}, + operation_kwargs={'ctx': 1}) + assert 'ctx' not in process['env'] + + def test_env_passed_explicitly(self): + env = {'one': '1', 'two': '2'} + process = common.create_process_config( + script_path='', + process={'env': env}, + operation_kwargs={}) + assert process['env'] == env + + def test_env_populated_from_operation_kwargs(self): + operation_kwargs = {'one': '1', 'two': '2'} + process = common.create_process_config( + script_path='', + process={}, + operation_kwargs=operation_kwargs) + assert process['env'] == operation_kwargs + + def test_env_merged_from_operation_kwargs_and_process(self): + operation_kwargs = {'one': '1', 'two': '2'} + env = {'three': '3', 'four': '4'} + process = common.create_process_config( + script_path='', + process={'env': env}, + operation_kwargs=operation_kwargs) + assert process['env'] == dict(operation_kwargs.items() + env.items()) + + def test_process_env_gets_precedence_over_operation_kwargs(self): + operation_kwargs = {'one': 'from_kwargs'} + env = {'one': 'from_env_process'} + process = common.create_process_config( + script_path='', + process={'env': env}, + operation_kwargs=operation_kwargs) + assert process['env'] == env + + def test_json_env_vars(self, mocker): + mocker.patch.object(common, 'is_windows', lambda: False) + operation_kwargs = {'a_dict': {'key': 'value'}, + 'a_list': ['a', 'b', 'c'], + 'a_tuple': (4, 5, 6), + 'a_bool': True} + process = common.create_process_config( + script_path='', + process={}, + operation_kwargs=operation_kwargs) + assert process['env'] == {'a_dict': '{"key": "value"}', + 'a_list': '["a", "b", "c"]', + 'a_tuple': '[4, 5, 6]', + 'a_bool': 'true'} + + def test_quote_json_env_vars(self): + operation_kwargs = {'one': []} + process = common.create_process_config( + script_path='', + process={}, + operation_kwargs=operation_kwargs, + quote_json_env_vars=True) + assert process['env']['one'] == "'[]'" + + def test_env_keys_converted_to_string_on_windows(self, mocker): + mocker.patch.object(common, 'is_windows', lambda: True) + env = {u'one': '1'} + process = common.create_process_config( + script_path='', + process={'env': env}, + operation_kwargs={}) + print type(process['env'].keys()[0]) + assert isinstance(process['env'].keys()[0], str) + + def test_env_values_quotes_are_escaped_on_windows(self, mocker): + mocker.patch.object(common, 'is_windows', lambda: True) + env = {'one': '"hello"'} + process = common.create_process_config( + script_path='', + process={'env': env}, + operation_kwargs={}) + assert process['env']['one'] == '\\"hello\\"' |