Repository: spotify/luigi
File: test/interface_test.py (Python)

Code before:
        ret = self._run_interface(detailed_summary=True)

        self.assertEqual(ret.status, LuigiStatusCode.SUCCESS)
        self.assertEqual(ret.scheduling_succeeded, True)
        self.assertEqual(ret.execution_succeeded, True)

    @patch(_summary_dict_module_path())
    def test_ret_code_1(self, fake_summary_dict):
        test_dict = self._empty_summary_dict()
        test_dict['ever_failed'].update([self.task_a])
        # Nothing in failed tasks (only an entry in ever_failed) so, should succeed with retry
        fake_summary_dict.return_value = test_dict
        self.worker.run = Mock(return_value=False)
        ret = self._run_interface(detailed_summary=True)

        self.assertEqual(ret.status, LuigiStatusCode.SUCCESS_WITH_RETRY)
        self.assertEqual(ret.scheduling_succeeded, False)
        self.assertEqual(ret.execution_succeeded, True)

    @patch(_summary_dict_module_path())
    def test_ret_code_2(self, fake_summary_dict):
        test_dict = self._empty_summary_dict()
        test_dict['ever_failed'].update([self.task_a])
        test_dict['failed'].update([self.task_a])
        # Should fail because a task failed
        fake_summary_dict.return_value = test_dict
        self.worker.run = Mock(return_value=False)
        ret = self._run_interface(detailed_summary=True)

        self.assertEqual(ret.status, LuigiStatusCode.FAILED)
        self.assertEqual(ret.scheduling_succeeded, False)
        self.assertEqual(ret.execution_succeeded, False)

    @patch(_summary_dict_module_path())
    def test_ret_code_3(self, fake_summary_dict):
        test_dict = self._empty_summary_dict()
        test_dict['ever_failed'].update([self.task_a])
        test_dict['failed'].update([self.task_a])
        test_dict['scheduling_error'].update([self.task_b])
        # Failed task and also a scheduling error
        fake_summary_dict.return_value = test_dict
        self.worker.add = Mock(side_effect=[True, False])
        self.worker.run = Mock(return_value=False)
        ret = self._run_interface(detailed_summary=True)

        self.assertEqual(ret.status, LuigiStatusCode.FAILED_AND_SCHEDULING_FAILED)
        self.assertEqual(ret.scheduling_succeeded, False)
        self.assertEqual(ret.execution_succeeded, False)

    @patch(_summary_dict_module_path())
    def test_ret_code_4(self, fake_summary_dict):

Review comment (bug):
actually I wonder if isn't nicer to write

fake_summary_dict.return_value = `self._empty_summary_dict().update({'evail_failed': set([self.task_a]), ...})`. And then you lint the line.

it's less likely to mess up if you can skip the local variable `test_dict`. But again only do this if the end result looks nicer/more readable.

Code after:
    @patch(_summary_dict_module_path())
    def test_that_status_is_success(self, fake_summary_dict):
        # Nothing in failed tasks so, should succeed
        fake_summary_dict.return_value = self._create_summary_dict_with()
        luigi_run_result = self._run_interface(detailed_summary=True)
        self.assertEqual(luigi_run_result.status, LuigiStatusCode.SUCCESS)

    @patch(_summary_dict_module_path())
    def test_that_status_is_success_with_retry(self, fake_summary_dict):
        # Nothing in failed tasks (only an entry in ever_failed) so, should succeed with retry
        fake_summary_dict.return_value = self._create_summary_dict_with({
            'ever_failed': [self.task_a]
        })
        luigi_run_result = self._run_interface(detailed_summary=True)
        self.assertEqual(luigi_run_result.status, LuigiStatusCode.SUCCESS_WITH_RETRY)

    @patch(_summary_dict_module_path())
    def test_that_status_is_failed_when_there_is_one_failed_task(self, fake_summary_dict):
        # Should fail because a task failed
        fake_summary_dict.return_value = self._create_summary_dict_with({
            'ever_failed': [self.task_a],
            'failed': [self.task_a]
        })
        luigi_run_result = self._run_interface(detailed_summary=True)
        self.assertEqual(luigi_run_result.status, LuigiStatusCode.FAILED)

    @patch(_summary_dict_module_path())
    def test_that_status_is_failed_with_scheduling_failure(self, fake_summary_dict):
        # Failed task and also a scheduling error
        fake_summary_dict.return_value = self._create_summary_dict_with({
            'ever_failed': [self.task_a],
            'failed': [self.task_a],
            'scheduling_error': [self.task_b]
        })
        luigi_run_result = self._run_interface(detailed_summary=True)
        self.assertEqual(luigi_run_result.status, LuigiStatusCode.FAILED_AND_SCHEDULING_FAILED)

    @patch(_summary_dict_module_path())
    def test_that_status_is_scheduling_failed_with_one_scheduling_error(self, fake_summary_dict):
        # Scheduling error for at least one task
        fake_summary_dict.return_value = self._create_summary_dict_with({
            'scheduling_error': [self.task_b]
        })
        luigi_run_result = self._run_interface(detailed_summary=True)
        self.assertEqual(luigi_run_result.status, LuigiStatusCode.SCHEDULING_FAILED)

    @patch(_summary_dict_module_path())
    def test_that_status_is_not_run_with_one_task_not_run(self, fake_summary_dict):
        # At least one of the tasks was not run
        fake_summary_dict.return_value = self._create_summary_dict_with({
            'not_run': [self.task_a]