test_tasks.py 文件源码

def test_infrastructure(self):
        class AbortException(Exception):
            pass

        def task():
            pass

        cases = dict(
            default=dict(
                decorator=tasks.infrastructure,
                expected_infrastructure='task',
            ),
            invoked=dict(
                decorator=tasks.infrastructure(),
                expected_infrastructure='task',
            ),
            with_custom_name=dict(
                decorator=tasks.infrastructure(name='infrastructure'),
                expected_infrastructure='infrastructure',
            ),
        )

        with fab.settings(abort_on_prompts=True, abort_exception=AbortException):
            with mock.patch.object(fab, 'abort', side_effect=AbortException):
                for case, data in cases.items():
                    with self.subTest(case=case):
                        decorator = data['decorator']
                        infrastructure = decorator(task)

                        self.assertTrue(is_task_object(infrastructure.confirm))
                        self.assertTrue(is_task_object(infrastructure.default))

                        fab.execute(infrastructure.confirm)
                        self.assertEqual(data['expected_infrastructure'], fab.env.infrastructure)

                        fab.env.infrastructure = None
                        with mock.patch.object(console, 'confirm', side_effect=[True, False]):
                            fab.execute(infrastructure.default)
                            self.assertEqual(data['expected_infrastructure'], fab.env.infrastructure)
                            with self.assertRaises(AbortException):
                                fab.execute(infrastructure.default)