--- /dev/null
+# 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\\"'
Code Review / multicloud / azure.git / blobdiff