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

Code before:
        self.assertFalse(self._run_interface(detailed_summary=True).scheduling_succeeded)

    @patch(_summary_dict_module_path())
    def test_ret_code_0(self, fake_summary_dict):
        test_dict = self._empty_summary_dict()
        # Nothing in failed tasks so, should succeed
        fake_summary_dict.return_value = test_dict
        self.worker.run = Mock(return_value=True)
        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)

Review comment (suggestion):
In general can the testing of the scheduling `scheduling_succeeded` and `execution_succeeded` be seperated from testing of the `.status`?

Actually were you able to even completely remove the need for the two booleans? That would be so good if there was only this enum and the tests would be so much easier.

Code after:
        self.worker.run = Mock(return_value=False)
        self.assertFalse(self._run_interface())

    def test_interface_run_with_run_failure_with_detailed_summary_enabled(self):
        self.worker.add = Mock(side_effect=[True, True])
        self.worker.run = Mock(return_value=False)
        self.assertFalse(self._run_interface(detailed_summary=True).scheduling_succeeded)

    @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]
        })