1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
|
# 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 pytest
from aria import extension
from aria.orchestrator.workflows import api
from aria.orchestrator.workflows.core import engine, graph_compiler
from aria.orchestrator.workflows.executor import process
from aria.orchestrator import workflow, operation
import tests
from tests import mock
from tests import storage
def test_decorate_extension(context, executor):
arguments = {'arg1': 1, 'arg2': 2}
def get_node(ctx):
return ctx.model.node.get_by_name(mock.models.DEPENDENCY_NODE_NAME)
node = get_node(context)
interface_name = 'test_interface'
operation_name = 'operation'
interface = mock.models.create_interface(
context.service,
interface_name,
operation_name,
operation_kwargs=dict(function='{0}.{1}'.format(__name__, _mock_operation.__name__),
arguments=arguments)
)
node.interfaces[interface.name] = interface
context.model.node.update(node)
@workflow
def mock_workflow(ctx, graph):
node = get_node(ctx)
task = api.task.OperationTask(
node,
interface_name=interface_name,
operation_name=operation_name,
arguments=arguments)
graph.add_tasks(task)
return graph
graph = mock_workflow(ctx=context) # pylint: disable=no-value-for-parameter
graph_compiler.GraphCompiler(context, executor.__class__).compile(graph)
eng = engine.Engine({executor.__class__: executor})
eng.execute(context)
out = get_node(context).attributes.get('out').value
assert out['wrapper_arguments'] == arguments
assert out['function_arguments'] == arguments
@extension.process_executor
class MockProcessExecutorExtension(object):
def decorate(self):
def decorator(function):
def wrapper(ctx, **operation_arguments):
with ctx.model.instrument(ctx.model.node.model_cls.attributes):
ctx.node.attributes['out'] = {'wrapper_arguments': operation_arguments}
function(ctx=ctx, **operation_arguments)
return wrapper
return decorator
@operation
def _mock_operation(ctx, **operation_arguments):
ctx.node.attributes['out']['function_arguments'] = operation_arguments
@pytest.fixture
def executor():
result = process.ProcessExecutor(python_path=[tests.ROOT_DIR])
try:
yield result
finally:
result.close()
@pytest.fixture
def context(tmpdir):
result = mock.context.simple(str(tmpdir))
yield result
storage.release_sqlite_storage(result.model)
|
