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

Code before:
#
# 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 sys

import luigi
import luigi.date_interval
import luigi.notifications
from luigi.interface import _WorkerSchedulerFactory
from luigi.worker import Worker
from luigi.interface import core
from luigi.execution_summary import LuigiStatusCode

from mock import Mock, patch, MagicMock
from helpers import LuigiTestCase, with_config

luigi.notifications.DEBUG = True


# Using this updatable dict to make inline dict updates for beautiful tests
class UpdatableDict(dict):
    def update(self, *args):
        dict.update(self, *args)
        return self


class InterfaceTest(LuigiTestCase):

    def setUp(self):
        self.worker = Worker()

        self.worker_scheduler_factory = _WorkerSchedulerFactory()
        self.worker_scheduler_factory.create_worker = Mock(return_value=self.worker)
        self.worker_scheduler_factory.create_local_scheduler = Mock()
        super(InterfaceTest, self).setUp()

        class NoOpTask(luigi.Task):
            param = luigi.Parameter()

        self.task_a = NoOpTask("a")
        self.task_b = NoOpTask("b")

    def _create_empty_summary_dict(self):
        return UpdatableDict({
            'completed': set(),
            'already_done': set(),

Review comment (bug):
Interesting idea, but this is going to be confusing as you later do `.update()` and *do* get a return value (most readers aren't used to that).


What about instead of 

         fake_summary_dict.return_value = self._create_empty_summary_dict().update({
            'ever_failed': [self.task_a]
        })

You can do

         fake_summary_dict.return_value = self._create_summary_dict_with({
            'ever_failed': [self.task_a]
        })

Code after:
#
# 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 sys

import luigi
import luigi.date_interval
import luigi.notifications
from luigi.interface import _WorkerSchedulerFactory
from luigi.worker import Worker
from luigi.interface import core
from luigi.execution_summary import LuigiStatusCode

from mock import Mock, patch, MagicMock
from helpers import LuigiTestCase, with_config

luigi.notifications.DEBUG = True


class InterfaceTest(LuigiTestCase):

    def setUp(self):
        self.worker = Worker()

        self.worker_scheduler_factory = _WorkerSchedulerFactory()
        self.worker_scheduler_factory.create_worker = Mock(return_value=self.worker)
        self.worker_scheduler_factory.create_local_scheduler = Mock()
        super(InterfaceTest, self).setUp()

        class NoOpTask(luigi.Task):
            param = luigi.Parameter()

        self.task_a = NoOpTask("a")
        self.task_b = NoOpTask("b")

    def _create_summary_dict_with(self, updates={}):
        summary_dict = {
            'completed': set(),
            'already_done': set(),
            'ever_failed': set(),
            'failed': set(),
            'scheduling_error': set(),
            'still_pending_ext': set(),
            'still_pending_not_ext': set(),
            'run_by_other_worker': set(),
            'upstream_failure': set(),