/usr/lib/python2.7/dist-packages/taskflow/tests/unit/test_functor_task.py is in python-taskflow 2.3.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | # -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed 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 taskflow.engines
from taskflow.patterns import linear_flow
from taskflow import task as base
from taskflow import test
def add(a, b):
return a + b
class BunchOfFunctions(object):
def __init__(self, values):
self.values = values
def run_one(self, *args, **kwargs):
self.values.append('one')
def revert_one(self, *args, **kwargs):
self.values.append('revert one')
def run_fail(self, *args, **kwargs):
self.values.append('fail')
raise RuntimeError('Woot!')
five = lambda: 5
multiply = lambda x, y: x * y
class FunctorTaskTest(test.TestCase):
def test_simple(self):
task = base.FunctorTask(add)
self.assertEqual(__name__ + '.add', task.name)
def test_other_name(self):
task = base.FunctorTask(add, name='my task')
self.assertEqual('my task', task.name)
def test_it_runs(self):
values = []
bof = BunchOfFunctions(values)
t = base.FunctorTask
flow = linear_flow.Flow('test')
flow.add(
t(bof.run_one, revert=bof.revert_one),
t(bof.run_fail)
)
self.assertRaisesRegexp(RuntimeError, '^Woot',
taskflow.engines.run, flow)
self.assertEqual(['one', 'fail', 'revert one'], values)
def test_lambda_functors(self):
t = base.FunctorTask
flow = linear_flow.Flow('test')
flow.add(
t(five, provides='five', name='five'),
t(multiply, provides='product', name='product')
)
flow_store = {
'x': 2,
'y': 3
}
result = taskflow.engines.run(flow, store=flow_store)
expected = flow_store.copy()
expected.update({
'five': 5,
'product': 6
})
self.assertDictEqual(expected, result)
|