diff options
Diffstat (limited to 'azure/aria/aria-extension-cloudify/src/aria/tests/orchestrator/execution_plugin/test_ctx_proxy_server.py')
| -rw-r--r-- | azure/aria/aria-extension-cloudify/src/aria/tests/orchestrator/execution_plugin/test_ctx_proxy_server.py | 285 |
1 files changed, 285 insertions, 0 deletions
diff --git a/azure/aria/aria-extension-cloudify/src/aria/tests/orchestrator/execution_plugin/test_ctx_proxy_server.py b/azure/aria/aria-extension-cloudify/src/aria/tests/orchestrator/execution_plugin/test_ctx_proxy_server.py new file mode 100644 index 0000000..94b7409 --- /dev/null +++ b/azure/aria/aria-extension-cloudify/src/aria/tests/orchestrator/execution_plugin/test_ctx_proxy_server.py @@ -0,0 +1,285 @@ +# 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. + +import os +import time +import sys +import subprocess +import StringIO + +import pytest + +from aria.orchestrator.execution_plugin import ctx_proxy + + +class TestCtxProxy(object): + + def test_attribute_access(self, server): + response = self.request(server, 'stub_attr', 'some_property') + assert response == 'some_value' + + def test_sugared_attribute_access(self, server): + response = self.request(server, 'stub-attr', 'some-property') + assert response == 'some_value' + + def test_dict_prop_access_get_key(self, server): + response = self.request(server, 'node', 'properties', 'prop1') + assert response == 'value1' + + def test_dict_prop_access_get_key_nested(self, server): + response = self.request(server, 'node', 'properties', 'prop2', 'nested_prop1') + assert response == 'nested_value1' + + def test_dict_prop_access_get_with_list_index(self, server): + response = self.request(server, 'node', 'properties', 'prop3', 2, 'value') + assert response == 'value_2' + + def test_dict_prop_access_set(self, server, ctx): + self.request(server, 'node', 'properties', 'prop4', 'key', '=', 'new_value') + self.request(server, 'node', 'properties', 'prop3', 2, 'value', '=', 'new_value_2') + self.request(server, 'node', 'properties', 'prop4', 'some', 'new', 'path', '=', + 'some_new_value') + assert ctx.node.properties['prop4']['key'] == 'new_value' + assert ctx.node.properties['prop3'][2]['value'] == 'new_value_2' + assert ctx.node.properties['prop4']['some']['new']['path'] == 'some_new_value' + + def test_dict_prop_access_set_with_list_index(self, server, ctx): + self.request(server, 'node', 'properties', 'prop3', 2, '=', 'new_value') + assert ctx.node.properties['prop3'][2] == 'new_value' + + def test_illegal_dict_access(self, server): + self.request(server, 'node', 'properties', 'prop4', 'key', '=', 'new_value') + with pytest.raises(RuntimeError): + self.request(server, 'node', 'properties', 'prop4', 'key', '=', 'new_value', 'what') + + def test_method_invocation(self, server): + args = ['[', 'arg1', 'arg2', 'arg3', ']'] + response_args = self.request(server, 'stub-method', *args) + assert response_args == args[1:-1] + + def test_method_invocation_no_args(self, server): + response = self.request(server, 'stub-method', '[', ']') + assert response == [] + + def test_method_return_value(self, server, ctx): + response_args = self.request(server, 'node', 'get_prop', '[', 'prop2', ']', 'nested_prop1') + assert response_args == 'nested_value1' + + def test_method_return_value_set(self, server, ctx): + self.request( + server, 'node', 'get_prop', '[', 'prop2', ']', 'nested_prop1', '=', 'new_value') + assert ctx.node.properties['prop2']['nested_prop1'] == 'new_value' + + def test_empty_return_value(self, server): + response = self.request(server, 'stub_none') + assert response is None + + def test_client_request_timeout(self, server): + with pytest.raises(IOError): + ctx_proxy.client._client_request(server.socket_url, + args=['stub-sleep', '[', '0.5', ']'], + timeout=0.1) + + def test_processing_exception(self, server): + with pytest.raises(ctx_proxy.client._RequestError): + self.request(server, 'property_that_does_not_exist') + + def test_not_json_serializable(self, server): + with pytest.raises(ctx_proxy.client._RequestError): + self.request(server, 'logger') + + def test_no_string_arg(self, server): + args = ['[', 1, 2, ']'] + response = self.request(server, 'stub_method', *args) + assert response == args[1:-1] + + class StubAttribute(object): + some_property = 'some_value' + + class NodeAttribute(object): + def __init__(self, properties): + self.properties = properties + + def get_prop(self, name): + return self.properties[name] + + @staticmethod + def stub_method(*args): + return args + + @staticmethod + def stub_sleep(seconds): + time.sleep(float(seconds)) + + @staticmethod + def stub_args(arg1, arg2, arg3='arg3', arg4='arg4', *args, **kwargs): + return dict( + arg1=arg1, + arg2=arg2, + arg3=arg3, + arg4=arg4, + args=args, + kwargs=kwargs) + + @pytest.fixture + def ctx(self, mocker): + class MockCtx(object): + INSTRUMENTATION_FIELDS = () + ctx = MockCtx() + properties = { + 'prop1': 'value1', + 'prop2': { + 'nested_prop1': 'nested_value1' + }, + 'prop3': [ + {'index': 0, 'value': 'value_0'}, + {'index': 1, 'value': 'value_1'}, + {'index': 2, 'value': 'value_2'} + ], + 'prop4': { + 'key': 'value' + } + } + ctx.stub_none = None + ctx.stub_method = TestCtxProxy.stub_method + ctx.stub_sleep = TestCtxProxy.stub_sleep + ctx.stub_args = TestCtxProxy.stub_args + ctx.stub_attr = TestCtxProxy.StubAttribute() + ctx.node = TestCtxProxy.NodeAttribute(properties) + ctx.model = mocker.MagicMock() + return ctx + + @pytest.fixture + def server(self, ctx): + result = ctx_proxy.server.CtxProxy(ctx) + result._close_session = lambda *args, **kwargs: {} + yield result + result.close() + + def request(self, server, *args): + return ctx_proxy.client._client_request(server.socket_url, args, timeout=5) + + +class TestArgumentParsing(object): + + def test_socket_url_arg(self): + self.expected.update(dict(socket_url='sock_url')) + ctx_proxy.client.main(['--socket-url', self.expected.get('socket_url')]) + + def test_socket_url_env(self): + expected_socket_url = 'env_sock_url' + os.environ['CTX_SOCKET_URL'] = expected_socket_url + self.expected.update(dict(socket_url=expected_socket_url)) + ctx_proxy.client.main([]) + + def test_socket_url_missing(self): + del os.environ['CTX_SOCKET_URL'] + with pytest.raises(RuntimeError): + ctx_proxy.client.main([]) + + def test_args(self): + self.expected.update(dict(args=['1', '2', '3'])) + ctx_proxy.client.main(self.expected.get('args')) + + def test_timeout(self): + self.expected.update(dict(timeout='10')) + ctx_proxy.client.main(['--timeout', self.expected.get('timeout')]) + self.expected.update(dict(timeout='15')) + ctx_proxy.client.main(['-t', self.expected.get('timeout')]) + + def test_mixed_order(self): + self.expected.update(dict( + args=['1', '2', '3'], timeout='20', socket_url='mixed_socket_url')) + ctx_proxy.client.main( + ['-t', self.expected.get('timeout')] + + ['--socket-url', self.expected.get('socket_url')] + + self.expected.get('args')) + ctx_proxy.client.main( + ['-t', self.expected.get('timeout')] + + self.expected.get('args') + + ['--socket-url', self.expected.get('socket_url')]) + ctx_proxy.client.main( + self.expected.get('args') + + ['-t', self.expected.get('timeout')] + + ['--socket-url', self.expected.get('socket_url')]) + + def test_json_args(self): + args = ['@1', '@[1,2,3]', '@{"key":"value"}'] + expected_args = [1, [1, 2, 3], {'key': 'value'}] + self.expected.update(dict(args=expected_args)) + ctx_proxy.client.main(args) + + def test_json_arg_prefix(self): + args = ['_1', '@1'] + expected_args = [1, '@1'] + self.expected.update(dict(args=expected_args)) + ctx_proxy.client.main(args + ['--json-arg-prefix', '_']) + + def test_json_output(self): + self.assert_valid_output('string', 'string', '"string"') + self.assert_valid_output(1, '1', '1') + self.assert_valid_output([1, '2'], "[1, '2']", '[1, "2"]') + self.assert_valid_output({'key': 1}, + "{'key': 1}", + '{"key": 1}') + self.assert_valid_output(False, 'False', 'false') + self.assert_valid_output(True, 'True', 'true') + self.assert_valid_output([], '[]', '[]') + self.assert_valid_output({}, '{}', '{}') + + def assert_valid_output(self, response, ex_typed_output, ex_json_output): + self.mock_response = response + current_stdout = sys.stdout + + def run(args, expected): + output = StringIO.StringIO() + sys.stdout = output + ctx_proxy.client.main(args) + assert output.getvalue() == expected + + try: + run([], ex_typed_output) + run(['-j'], ex_json_output) + run(['--json-output'], ex_json_output) + finally: + sys.stdout = current_stdout + + def mock_client_request(self, socket_url, args, timeout): + assert socket_url == self.expected.get('socket_url') + assert args == self.expected.get('args') + assert timeout == int(self.expected.get('timeout')) + return self.mock_response + + @pytest.fixture(autouse=True) + def patch_client_request(self, mocker): + mocker.patch.object(ctx_proxy.client, + ctx_proxy.client._client_request.__name__, + self.mock_client_request) + mocker.patch.dict('os.environ', {'CTX_SOCKET_URL': 'stub'}) + + @pytest.fixture(autouse=True) + def defaults(self): + self.expected = dict(args=[], timeout=30, socket_url='stub') + self.mock_response = None + + +class TestCtxEntryPoint(object): + + def test_ctx_in_path(self): + p = subprocess.Popen(['ctx', '--help'], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + p.communicate() + assert not p.wait() |