Method _context

t/unit/worker/test_strategy.py:123–149

(self, sig,
                 rate_limits=True, events=True, utc=True, limit=None)

    @contextmanager
    def _context(self, sig,
                 rate_limits=True, events=True, utc=True, limit=None):
        assert sig.type.Strategy
        assert sig.type.Request

        reserved = Mock()
        consumer = Mock()
        # Create a proper mock for task_buckets that supports __getitem__
        task_buckets_mock = Mock()
        task_buckets_mock.__getitem__ = Mock(side_effect=lambda key: None)
        consumer.task_buckets = task_buckets_mock
        if limit:
            bucket = TokenBucket(rate(limit), capacity=1)
            task_buckets_mock.__getitem__.side_effect = (
                lambda key: bucket if key == sig.task else None
            )
        consumer.controller.state.revoked = set()
        consumer.disable_rate_limits = not rate_limits
        consumer.event_dispatcher.enabled = events
        s = sig.type.start_strategy(self.app, consumer, task_reserved=reserved)
        assert s

        message = self.task_message_from_sig(
            self.app, sig, utc=utc, TaskMessage=self.get_message_class(),
        )
        message = self.prepare_message(message)
        yield self.Context(sig, s, reserved, consumer, message)

    def test_when_logging_disabled(self, caplog):
        # Capture logs at any level above `NOTSET`