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
8 # http://www.apache.org/licenses/LICENSE-2.0
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.
24 from aria.orchestrator.execution_plugin import ctx_proxy
27 class TestCtxProxy(object):
29 def test_attribute_access(self, server):
30 response = self.request(server, 'stub_attr', 'some_property')
31 assert response == 'some_value'
33 def test_sugared_attribute_access(self, server):
34 response = self.request(server, 'stub-attr', 'some-property')
35 assert response == 'some_value'
37 def test_dict_prop_access_get_key(self, server):
38 response = self.request(server, 'node', 'properties', 'prop1')
39 assert response == 'value1'
41 def test_dict_prop_access_get_key_nested(self, server):
42 response = self.request(server, 'node', 'properties', 'prop2', 'nested_prop1')
43 assert response == 'nested_value1'
45 def test_dict_prop_access_get_with_list_index(self, server):
46 response = self.request(server, 'node', 'properties', 'prop3', 2, 'value')
47 assert response == 'value_2'
49 def test_dict_prop_access_set(self, server, ctx):
50 self.request(server, 'node', 'properties', 'prop4', 'key', '=', 'new_value')
51 self.request(server, 'node', 'properties', 'prop3', 2, 'value', '=', 'new_value_2')
52 self.request(server, 'node', 'properties', 'prop4', 'some', 'new', 'path', '=',
54 assert ctx.node.properties['prop4']['key'] == 'new_value'
55 assert ctx.node.properties['prop3'][2]['value'] == 'new_value_2'
56 assert ctx.node.properties['prop4']['some']['new']['path'] == 'some_new_value'
58 def test_dict_prop_access_set_with_list_index(self, server, ctx):
59 self.request(server, 'node', 'properties', 'prop3', 2, '=', 'new_value')
60 assert ctx.node.properties['prop3'][2] == 'new_value'
62 def test_illegal_dict_access(self, server):
63 self.request(server, 'node', 'properties', 'prop4', 'key', '=', 'new_value')
64 with pytest.raises(RuntimeError):
65 self.request(server, 'node', 'properties', 'prop4', 'key', '=', 'new_value', 'what')
67 def test_method_invocation(self, server):
68 args = ['[', 'arg1', 'arg2', 'arg3', ']']
69 response_args = self.request(server, 'stub-method', *args)
70 assert response_args == args[1:-1]
72 def test_method_invocation_no_args(self, server):
73 response = self.request(server, 'stub-method', '[', ']')
76 def test_method_return_value(self, server, ctx):
77 response_args = self.request(server, 'node', 'get_prop', '[', 'prop2', ']', 'nested_prop1')
78 assert response_args == 'nested_value1'
80 def test_method_return_value_set(self, server, ctx):
82 server, 'node', 'get_prop', '[', 'prop2', ']', 'nested_prop1', '=', 'new_value')
83 assert ctx.node.properties['prop2']['nested_prop1'] == 'new_value'
85 def test_empty_return_value(self, server):
86 response = self.request(server, 'stub_none')
87 assert response is None
89 def test_client_request_timeout(self, server):
90 with pytest.raises(IOError):
91 ctx_proxy.client._client_request(server.socket_url,
92 args=['stub-sleep', '[', '0.5', ']'],
95 def test_processing_exception(self, server):
96 with pytest.raises(ctx_proxy.client._RequestError):
97 self.request(server, 'property_that_does_not_exist')
99 def test_not_json_serializable(self, server):
100 with pytest.raises(ctx_proxy.client._RequestError):
101 self.request(server, 'logger')
103 def test_no_string_arg(self, server):
104 args = ['[', 1, 2, ']']
105 response = self.request(server, 'stub_method', *args)
106 assert response == args[1:-1]
108 class StubAttribute(object):
109 some_property = 'some_value'
111 class NodeAttribute(object):
112 def __init__(self, properties):
113 self.properties = properties
115 def get_prop(self, name):
116 return self.properties[name]
119 def stub_method(*args):
123 def stub_sleep(seconds):
124 time.sleep(float(seconds))
127 def stub_args(arg1, arg2, arg3='arg3', arg4='arg4', *args, **kwargs):
137 def ctx(self, mocker):
138 class MockCtx(object):
139 INSTRUMENTATION_FIELDS = ()
144 'nested_prop1': 'nested_value1'
147 {'index': 0, 'value': 'value_0'},
148 {'index': 1, 'value': 'value_1'},
149 {'index': 2, 'value': 'value_2'}
156 ctx.stub_method = TestCtxProxy.stub_method
157 ctx.stub_sleep = TestCtxProxy.stub_sleep
158 ctx.stub_args = TestCtxProxy.stub_args
159 ctx.stub_attr = TestCtxProxy.StubAttribute()
160 ctx.node = TestCtxProxy.NodeAttribute(properties)
161 ctx.model = mocker.MagicMock()
165 def server(self, ctx):
166 result = ctx_proxy.server.CtxProxy(ctx)
167 result._close_session = lambda *args, **kwargs: {}
171 def request(self, server, *args):
172 return ctx_proxy.client._client_request(server.socket_url, args, timeout=5)
175 class TestArgumentParsing(object):
177 def test_socket_url_arg(self):
178 self.expected.update(dict(socket_url='sock_url'))
179 ctx_proxy.client.main(['--socket-url', self.expected.get('socket_url')])
181 def test_socket_url_env(self):
182 expected_socket_url = 'env_sock_url'
183 os.environ['CTX_SOCKET_URL'] = expected_socket_url
184 self.expected.update(dict(socket_url=expected_socket_url))
185 ctx_proxy.client.main([])
187 def test_socket_url_missing(self):
188 del os.environ['CTX_SOCKET_URL']
189 with pytest.raises(RuntimeError):
190 ctx_proxy.client.main([])
193 self.expected.update(dict(args=['1', '2', '3']))
194 ctx_proxy.client.main(self.expected.get('args'))
196 def test_timeout(self):
197 self.expected.update(dict(timeout='10'))
198 ctx_proxy.client.main(['--timeout', self.expected.get('timeout')])
199 self.expected.update(dict(timeout='15'))
200 ctx_proxy.client.main(['-t', self.expected.get('timeout')])
202 def test_mixed_order(self):
203 self.expected.update(dict(
204 args=['1', '2', '3'], timeout='20', socket_url='mixed_socket_url'))
205 ctx_proxy.client.main(
206 ['-t', self.expected.get('timeout')] +
207 ['--socket-url', self.expected.get('socket_url')] +
208 self.expected.get('args'))
209 ctx_proxy.client.main(
210 ['-t', self.expected.get('timeout')] +
211 self.expected.get('args') +
212 ['--socket-url', self.expected.get('socket_url')])
213 ctx_proxy.client.main(
214 self.expected.get('args') +
215 ['-t', self.expected.get('timeout')] +
216 ['--socket-url', self.expected.get('socket_url')])
218 def test_json_args(self):
219 args = ['@1', '@[1,2,3]', '@{"key":"value"}']
220 expected_args = [1, [1, 2, 3], {'key': 'value'}]
221 self.expected.update(dict(args=expected_args))
222 ctx_proxy.client.main(args)
224 def test_json_arg_prefix(self):
226 expected_args = [1, '@1']
227 self.expected.update(dict(args=expected_args))
228 ctx_proxy.client.main(args + ['--json-arg-prefix', '_'])
230 def test_json_output(self):
231 self.assert_valid_output('string', 'string', '"string"')
232 self.assert_valid_output(1, '1', '1')
233 self.assert_valid_output([1, '2'], "[1, '2']", '[1, "2"]')
234 self.assert_valid_output({'key': 1},
237 self.assert_valid_output(False, 'False', 'false')
238 self.assert_valid_output(True, 'True', 'true')
239 self.assert_valid_output([], '[]', '[]')
240 self.assert_valid_output({}, '{}', '{}')
242 def assert_valid_output(self, response, ex_typed_output, ex_json_output):
243 self.mock_response = response
244 current_stdout = sys.stdout
246 def run(args, expected):
247 output = StringIO.StringIO()
249 ctx_proxy.client.main(args)
250 assert output.getvalue() == expected
253 run([], ex_typed_output)
254 run(['-j'], ex_json_output)
255 run(['--json-output'], ex_json_output)
257 sys.stdout = current_stdout
259 def mock_client_request(self, socket_url, args, timeout):
260 assert socket_url == self.expected.get('socket_url')
261 assert args == self.expected.get('args')
262 assert timeout == int(self.expected.get('timeout'))
263 return self.mock_response
265 @pytest.fixture(autouse=True)
266 def patch_client_request(self, mocker):
267 mocker.patch.object(ctx_proxy.client,
268 ctx_proxy.client._client_request.__name__,
269 self.mock_client_request)
270 mocker.patch.dict('os.environ', {'CTX_SOCKET_URL': 'stub'})
272 @pytest.fixture(autouse=True)
274 self.expected = dict(args=[], timeout=30, socket_url='stub')
275 self.mock_response = None
278 class TestCtxEntryPoint(object):
280 def test_ctx_in_path(self):
281 p = subprocess.Popen(['ctx', '--help'],
282 stdout=subprocess.PIPE,
283 stderr=subprocess.PIPE)