test_task_executor.py 文件源码

def test_simple(self):
        task1 = TaskRegistry.create('test_executor_add')
        task1.Input.a = 10
        task1.Input.b = 20

        task2 = TaskRegistry.create('test_executor_add')
        task2.Input.a = 100
        task2.Input.b = 200

        executor1 = TaskExecutor()
        executor1.set_task(task1)
        executor1.execute()

        executor2 = TaskExecutor()
        executor2.set_task(task2)
        executor2.execute()

        self.assertEqual(task1.Output.c, 30)
        self.assertEqual(task2.Output.c, 300)