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

applied black

Browse Source
This commit is contained in:
Matthias Bisping 2022-03-14 14:49:03 +01:00
parent 32740a914f
commit f78a6894bc
3 changed files with 7 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
pyinfra/test/unit_tests/conftest.py
View File

@ -101,6 +101,7 @@ def callback():
def analysis_callback(callback):
def inner(data: bytes):
return callback(data.decode())
return inner

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

@ -21,16 +21,14 @@ def access_callback():
@pytest.mark.parametrize("queue_manager_name", ["mock", "pika"], scope="session")
class TestConsumer:
def test_consuming_empty_input_queue_does_not_put_anything_on_output_queue(self, consumer, queue_manager):
queue_manager.clear()
consumer.consume()
assert queue_manager.output_queue.empty()
def test_consuming_nonempty_input_queue_puts_messages_on_output_queue_in_fifo_order(
self, consumer, queue_manager, callback
self, consumer, queue_manager, callback
):
def produce_items():
return map(str, range(3))
@ -49,17 +47,15 @@ class TestConsumer:
@pytest.mark.parametrize("client_name", ["mock", "s3", "azure"], scope="session")
@pytest.mark.parametrize("response_strategy_name", ["forwarding", "storage"], scope="session")
def test_consuming_nonempty_input_queue_with_visitor_puts_messages_on_output_queue_in_fifo_order(
self, consumer, queue_manager, visitor,
bucket_name, storage
self, consumer, queue_manager, visitor, bucket_name, storage
):
def produce_items():
for i in range(3):
body = {
"dossierId": "folder",
"fileId": f"file{i}",
"targetFileExtension": "in.gz",
"responseFileExtension": "out.gz"
"responseFileExtension": "out.gz",
}
yield f"{i}".encode(), body

5
pyinfra/test/unit_tests/queue_visitor_test.py
View File

@ -14,8 +14,9 @@ def body():
@pytest.mark.parametrize("client_name", ["mock", "azure", "s3"], scope="session")
@pytest.mark.parametrize("response_strategy_name", ["forwarding", "storage"], scope="session")
class TestVisitor:
def test_given_a_input_queue_message_callback_pulls_the_data_from_storage(self, visitor, body, storage, bucket_name):
def test_given_a_input_queue_message_callback_pulls_the_data_from_storage(
self, visitor, body, storage, bucket_name
):
storage.clear_bucket(bucket_name)
storage.put_object(**get_object_descriptor(body), data=gzip.compress(b"content"))
data_received = visitor.load_data(body)