Marmelator/pyinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

adapted queue manager mock for new queue manager interface

Browse Source
This commit is contained in:
Matthias Bisping 2022-03-10 13:02:59 +01:00
parent 9aff168f0d
commit ce1217c191
2 changed files with 14 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
pyinfra/test/queue_manager_mock.py
View File

@ -16,17 +16,19 @@ class QueueManagerMock(QueueManager):
def publish_request(self, request): def publish_request(self, request):
self._input_queue.append(request) self._input_queue.append(request)
def publish_response(self, response): def publish_response(self, message, callback):
self._output_queue.append(response) self._output_queue.append(callback(message))
def pull_request(self): def pull_request(self):
return self._input_queue.pop() return self._input_queue.popleft()
def consume(self, callback): def consume(self):
while self._input_queue: while self._input_queue:
request = self._input_queue.popleft() yield self.pull_request()
response = callback(request)
self._output_queue.append(response) def consume_and_publish(self, callback):
for message in self.consume():
self.publish_response(message, callback)
def clear(self): def clear(self):
self._input_queue.clear() self._input_queue.clear()

10
pyinfra/test/unit_tests/consumer_test.py
View File

@ -13,7 +13,7 @@ def callback():
@pytest.fixture(scope="session") @pytest.fixture(scope="session")
def queue_manager() -> QueueManager: def queue_manager() -> QueueManager:
# return QueueManagerMock("input", "output") return QueueManagerMock("input", "output")
return PikaQueueManager("input", "output") return PikaQueueManager("input", "output")
@ -22,10 +22,10 @@ def consumer(queue_manager, callback):
return Consumer(callback, queue_manager) return Consumer(callback, queue_manager)
# def test_consuming_empty_input_queue_does_not_put_anything_on_output_queue(consumer, queue_manager): def test_consuming_empty_input_queue_does_not_put_anything_on_output_queue(consumer, queue_manager):
# queue_manager.clear() queue_manager.clear()
# consumer.consume() consumer.consume()
# assert queue_manager.output_queue.empty() assert queue_manager.output_queue.empty()
def test_consuming_nonempty_input_queue_puts_messages_on_output_queue_in_fifo_order(consumer, queue_manager, callback): def test_consuming_nonempty_input_queue_puts_messages_on_output_queue_in_fifo_order(consumer, queue_manager, callback):