(self, warn)
| 188 | |
| 189 | @patch('celery.worker.consumer.consumer.warn') |
| 190 | def test_receive_message_unknown(self, warn): |
| 191 | c = self.LoopConsumer() |
| 192 | c.blueprint.state = RUN |
| 193 | c.steps.pop() |
| 194 | channel = Mock(name='.channeol') |
| 195 | m = create_message(channel, unknown={'baz': '!!!'}) |
| 196 | |
| 197 | callback = self._get_on_message(c) |
| 198 | callback(m) |
| 199 | warn.assert_called() |
| 200 | |
| 201 | @patch('celery.worker.strategy.to_timestamp') |
| 202 | def test_receive_message_eta_OverflowError(self, to_timestamp): |
nothing calls this directly
no test coverage detected