Add mechanizm to receive pending undelivered messages before tests

concent_api/api-e2e-additional-verification-test.py

@@ -17,6 +17,7 @@
 from api_testing_common import count_fails
 from api_testing_common import create_client_auth_message
 from api_testing_common import create_signed_task_to_compute
+from api_testing_common import receive_pending_messages_for_requestor_and_provider
 from api_testing_common import run_tests
 from api_testing_common import timestamp_to_isoformat
 from golem_messages.factories.helpers import override_timestamp
@@ -116,6 +117,11 @@ def calculate_verification_deadline(
 
 @count_fails
 def test_case_1_test_for_positive_case(cluster_consts: ProtocolConstants, cluster_url: str) -> None:  # pylint: disable=unused-argument
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
     provider_deposit_value = sci_base.get_provider_gntb_balance()
     requestor_deposit_value = sci_base.get_requestor_deposit_value()
@@ -222,6 +228,11 @@ def test_case_1_test_for_positive_case(cluster_consts: ProtocolConstants, cluste
 
 @count_fails
 def test_case_2_test_for_resources_failure_reason(cluster_consts: ProtocolConstants, cluster_url: str) -> None:  # pylint: disable=unused-argument
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
     file_content = 'test'
     file_size = len(file_content)
@@ -248,6 +259,11 @@ def test_case_2_test_for_resources_failure_reason(cluster_consts: ProtocolConsta
 
 @count_fails
 def test_case_3_test_for_invalid_time(cluster_consts: ProtocolConstants, cluster_url: str) -> None:  # pylint: disable=unused-argument
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
 
     file_content = 'test'
@@ -278,6 +294,11 @@ def test_case_3_test_for_invalid_time(cluster_consts: ProtocolConstants, cluster
 
 @count_fails
 def test_case_4_test_for_duplicated_request(cluster_consts: ProtocolConstants, cluster_url: str) -> None:  # pylint: disable=unused-argument
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
 
     result_file_content_1 = 'test'
@@ -324,6 +345,11 @@ def test_case_4_test_for_duplicated_request(cluster_consts: ProtocolConstants, c
 
 @count_fails
 def test_case_5_test_requestor_status_account_negative(cluster_consts: ProtocolConstants, cluster_url: str) -> None:  # pylint: disable=unused-argument
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
 
     result_file_content_1 = 'test'
@@ -359,6 +385,11 @@ def test_case_5_test_requestor_status_account_negative(cluster_consts: ProtocolC
 
 @count_fails
 def test_case_6_test_without_script_src_in(cluster_consts: ProtocolConstants, cluster_url: str) -> None:  # pylint: disable=unused-argument
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
     provider_gntb_balance = sci_base.get_provider_gntb_balance()
     requestor_deposit_value = sci_base.get_requestor_deposit_value()

concent_api/api-e2e-force-accept-or-reject-results-test.py

@@ -18,6 +18,7 @@
 from api_testing_common import create_signed_report_computed_task
 from api_testing_common import create_signed_subtask_results_accepted
 from api_testing_common import create_signed_task_to_compute
+from api_testing_common import receive_pending_messages_for_requestor_and_provider
 from api_testing_common import REPORT_COMPUTED_TASK_SIZE
 from api_testing_common import run_tests
 from api_testing_common import timestamp_to_isoformat
@@ -112,6 +113,11 @@ def _precalculate_subtask_verification_time(minimum_upload_rate: int, concent_me
 @count_fails
 def test_case_2d_requestor_rejects_subtask_results(cluster_consts: ProtocolConstants, cluster_url: str) -> None:
     # Test CASE 2D + 3 + 4B + 5. Requestor sends ForceSubtaskResultsResponse with SubtaskResultsRejected
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
     signed_task_to_compute = create_signed_task_to_compute(
         timestamp=calculate_timestamp(current_time, cluster_consts.concent_messaging_time, cluster_consts.minimum_upload_rate),
@@ -185,6 +191,11 @@ def test_case_2d_requestor_rejects_subtask_results(cluster_consts: ProtocolConst
 def test_case_4b_requestor_accepts_subtaks_results(cluster_consts: ProtocolConstants, cluster_url: str) -> None:
     # Test CASE 4B + 5. Requestor sends ForceSubtaskResultsResponse with SubtaskResultsAccepted
     #  Step 1. Provider sends ForceSubtaskResults
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
     signed_task_to_compute = create_signed_task_to_compute(
         timestamp=calculate_timestamp(current_time, cluster_consts.concent_messaging_time, cluster_consts.minimum_upload_rate),
@@ -258,6 +269,11 @@ def test_case_4b_requestor_accepts_subtaks_results(cluster_consts: ProtocolConst
 @count_fails
 def test_case_2c_wrong_timestamps(cluster_consts: ProtocolConstants, cluster_url: str) -> None:
     # Test CASE 2C - Send ForceSubtaskResults with wrong timestamps
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
     api_request(
         cluster_url,
@@ -300,6 +316,11 @@ def test_case_2c_wrong_timestamps(cluster_consts: ProtocolConstants, cluster_url
 @count_fails
 def test_case_2b_not_enough_funds(cluster_consts: ProtocolConstants, cluster_url: str) -> None:
     #  Test CASE 2B - Send ForceSubtaskResults with not enough amount of funds on account
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
     api_request(
         cluster_url,
@@ -337,6 +358,11 @@ def test_case_2b_not_enough_funds(cluster_consts: ProtocolConstants, cluster_url
 def test_case_2a_send_duplicated_force_subtask_results(cluster_consts: ProtocolConstants, cluster_url: str) -> None:
     #  Test CASE 2A + 2D + 3 - Send ForceSubtaskResults with same task_id as stored by Concent before
     #  Step 1. Send ForceSubtaskResults first time
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
     signed_task_to_compute = create_signed_task_to_compute(
         timestamp=calculate_timestamp(current_time, cluster_consts.concent_messaging_time, cluster_consts.minimum_upload_rate),

concent_api/api-e2e-force-payment.py

@@ -20,6 +20,7 @@
 from api_testing_common import create_signed_report_computed_task
 from api_testing_common import create_signed_subtask_results_accepted
 from api_testing_common import create_signed_task_to_compute
+from api_testing_common import receive_pending_messages_for_requestor_and_provider
 from api_testing_common import run_tests
 from api_testing_common import timestamp_to_isoformat
 from protocol_constants import ProtocolConstants
@@ -49,6 +50,11 @@ def force_payment(
 @count_fails
 def test_case_2d_send_correct_force_payment(cluster_consts: ProtocolConstants, cluster_url: str) -> None:
     # Test CASE 2D - Send correct ForcePayment
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     provider_gntb_balance = sci_base.get_provider_gntb_balance()
     current_time = get_current_utc_timestamp()
     correct_force_payment = force_payment(
@@ -117,6 +123,11 @@ def test_case_2d_send_correct_force_payment(cluster_consts: ProtocolConstants, c
 @count_fails
 def test_case_2c_send_force_payment_with_no_value_to_be_paid(cluster_consts: ProtocolConstants, cluster_url: str) -> None:
     #  Test CASE 2C - Send ForcePayment with no value to be paid
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
     api_request(
         cluster_url,
@@ -170,6 +181,11 @@ def test_case_2c_send_force_payment_with_no_value_to_be_paid(cluster_consts: Pro
 @count_fails
 def test_case_2b_send_force_payment_beyond_payment_time(cluster_consts: ProtocolConstants, cluster_url: str) -> None:
     #  Test CASE 2B - Send ForcePayment beyond payment time
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
     api_request(
         cluster_url,
@@ -229,6 +245,11 @@ def test_case_2_a_force_payment_with_subtask_result_accepted_where_ethereum_acco
     cluster_url: str,
 ) -> None:
     # Test CASE 2A - Send ForcePayment with SubtaskResultsAccepted where ethereum accounts are different
+    receive_pending_messages_for_requestor_and_provider(
+        cluster_url,
+        sci_base,
+        CONCENT_PUBLIC_KEY
+    )
     current_time = get_current_utc_timestamp()
     api_request(
         cluster_url,

concent_api/api_testing_common.py

@@ -466,3 +466,50 @@ def create_signed_report_computed_task(
             provider_private_key,
         )
         return signed_message
+
+
+def receive_all_left_pending_responses(
+    host: str,
+    endpoint: str,
+    client_private_key: bytes,
+    client_public_key: bytes,
+    concent_public_key: bytes,
+    client_name: str,
+) -> None:
+    url = f"{host}/api/v1/{endpoint}/"
+    status_code = 200
+    received_messages = []
+    while status_code == 200:
+        response = requests.post(f"{url}", headers=REQUEST_HEADERS, data=create_client_auth_message(client_private_key, client_public_key, concent_public_key), verify=False)
+        status_code = response.status_code
+        if status_code == 200:
+            if isinstance(response.content, bytes) and response.headers['Content-Type'] == 'application/octet-stream':
+                received_messages.append(load(response.content, client_private_key, concent_public_key).__class__.__name__)
+
+    if len(received_messages) > 0:
+        print(f'Concent had {len(received_messages)} messages which waited for {client_name} to receive. Name of these messages: {received_messages}')
+    else:
+        print(f'Concent had not any pending messages for {client_name}')
+
+
+def receive_pending_messages_for_requestor_and_provider(
+    cluster_url: str,
+    sci_base: SCIBaseTest,
+    concent_public_key: bytes,
+) -> None:
+    receive_all_left_pending_responses(
+        cluster_url,
+        'receive',
+        sci_base.provider_private_key,
+        sci_base.provider_public_key,
+        concent_public_key,
+        'Provider',
+    )
+    receive_all_left_pending_responses(
+        cluster_url,
+        'receive',
+        sci_base.requestor_private_key,
+        sci_base.requestor_public_key,
+        concent_public_key,
+        'Requestor',
+    )