e2e: automate pre-teardown leak checks

tests/e2e/iothub_e2e/aio/test_c2d.py

@@ -21,18 +21,17 @@ class TestReceiveC2d(object):
     @pytest.mark.it("Can receive C2D")
     @pytest.mark.quicktest_suite
     async def test_receive_c2d(self, client, service_helper, leak_tracker):
-        leak_tracker.set_initial_object_list()
         event_loop = asyncio.get_running_loop()
 
         message = json.dumps(get_random_dict())
 
-        received_message = None
+        received_message_data = None
         received = asyncio.Event()
 
         async def handle_on_message_received(message):
-            nonlocal received_message, received
+            nonlocal received_message_data
             logger.info("received {}".format(message))
-            received_message = message
+            received_message_data = message.data.decode("utf-8")
             event_loop.call_soon_threadsafe(received.set)
 
         client.on_message_received = handle_on_message_received
@@ -42,7 +41,4 @@ async def handle_on_message_received(message):
         await asyncio.wait_for(received.wait(), timeout=const.E2E_TIMEOUT)
         assert received.is_set()
 
-        assert received_message.data.decode("utf-8") == message
-
-        received_message = None  # so this isn't tagged as a leak
-        leak_tracker.check_for_leaks()
+        assert received_message_data == message

tests/e2e/iothub_e2e/aio/test_connect_disconnect.py

@@ -21,8 +21,6 @@ class TestConnectDisconnect(object):
     async def test_connect_disconnect(self, brand_new_client, leak_tracker):
         client = brand_new_client
 
-        leak_tracker.set_initial_object_list()
-
         assert client
         logger.info("connecting")
         await client.connect()
@@ -34,8 +32,6 @@ async def test_connect_disconnect(self, brand_new_client, leak_tracker):
         await client.connect()
         assert client.connected
 
-        leak_tracker.check_for_leaks()
-
     @pytest.mark.it(
         "Can do a manual connect in the `on_connection_state_change` call that is notifying the user about a disconnect."
     )
@@ -53,7 +49,6 @@ async def test_connect_in_the_middle_of_disconnect(
         client = brand_new_client
         assert client
 
-        leak_tracker.set_initial_object_list()
         event_loop = asyncio.get_running_loop()
 
         reconnected_event = asyncio.Event()
@@ -98,9 +93,6 @@ async def handle_on_connection_state_change():
         event = await service_helper.wait_for_eventhub_arrival(random_message.message_id)
         assert event
 
-        random_message = None  # so this isn't flagged as a leak
-        leak_tracker.check_for_leaks()
-
     @pytest.mark.it(
         "Can do a manual disconnect in the `on_connection_state_change` call that is notifying the user about a connect."
     )
@@ -127,7 +119,6 @@ async def test_disconnect_in_the_middle_of_connect(
         assert client
         disconnect_on_next_connect_event = False
 
-        leak_tracker.set_initial_object_list()
         event_loop = asyncio.get_running_loop()
 
         disconnected_event = asyncio.Event()
@@ -178,9 +169,6 @@ async def handle_on_connection_state_change():
         event = await service_helper.wait_for_eventhub_arrival(random_message.message_id)
         assert event
 
-        random_message = None  # So this doesn't get flagged as a leak.
-        leak_tracker.check_for_leaks()
-
     # TODO: Add connect/disconnect stress, multiple times with connect inside disconnect and disconnect inside connect.
 
 
@@ -194,7 +182,6 @@ async def test_disconnect_on_drop_outgoing(self, client, dropper, leak_tracker):
         This test verifies that the client will disconnect (eventually) if the network starts
         dropping packets
         """
-        leak_tracker.set_initial_object_list()
 
         await client.connect()
         assert client.connected
@@ -211,15 +198,12 @@ async def test_disconnect_on_drop_outgoing(self, client, dropper, leak_tracker):
             lambda: client.connected, timeout=const.E2E_TIMEOUT
         )
 
-        leak_tracker.check_for_leaks()
-
     @pytest.mark.it("disconnects when network rejects all outgoing packets")
     async def test_disconnect_on_reject_outgoing(self, client, dropper, leak_tracker):
         """
         This test verifies that the client will disconnect (eventually) if the network starts
         rejecting packets
         """
-        leak_tracker.set_initial_object_list()
 
         await client.connect()
         assert client.connected
@@ -235,5 +219,3 @@ async def test_disconnect_on_reject_outgoing(self, client, dropper, leak_tracker
         await wait_helpers.async_wait_for_condition(
             lambda: client.connected, timeout=const.E2E_TIMEOUT
         )
-
-        leak_tracker.check_for_leaks()

tests/e2e/iothub_e2e/aio/test_connect_disconnect_stress.py

@@ -20,21 +20,17 @@ class TestConnectDisconnectStress(object):
     async def test_non_overlapped_connect_disconnect_stress(
         self, client, iteration_count, leak_tracker
     ):
-        leak_tracker.set_initial_object_list()
 
         for _ in range(iteration_count):
             await client.connect()
             await client.disconnect()
 
-        leak_tracker.check_for_leaks()
-
     @pytest.mark.parametrize("iteration_count", [20, 250])
     @pytest.mark.it("Can do many overlapped connects and disconnects")
     @pytest.mark.timeout(600)
     async def test_overlapped_connect_disconnect_stress(
         self, client, iteration_count, leak_tracker
     ):
-        leak_tracker.set_initial_object_list()
 
         futures = []
         for _ in range(iteration_count):
@@ -46,15 +42,12 @@ async def test_overlapped_connect_disconnect_stress(
         finally:
             await task_cleanup.cleanup_tasks(futures)
 
-        leak_tracker.check_for_leaks()
-
     @pytest.mark.parametrize("iteration_count", [20, 500])
     @pytest.mark.it("Can do many overlapped random connects and disconnects")
     @pytest.mark.timeout(600)
     async def test_overlapped_random_connect_disconnect_stress(
         self, client, iteration_count, leak_tracker
     ):
-        leak_tracker.set_initial_object_list()
 
         futures = []
         for _ in range(iteration_count):
@@ -67,5 +60,3 @@ async def test_overlapped_random_connect_disconnect_stress(
             await asyncio.gather(*futures)
         finally:
             await task_cleanup.cleanup_tasks(futures)
-
-        leak_tracker.check_for_leaks()

tests/e2e/iothub_e2e/aio/test_methods.py

@@ -36,9 +36,9 @@ async def test_handle_method_call(
         service_helper,
         leak_tracker,
     ):
-        leak_tracker.set_initial_object_list()
 
-        actual_request = None
+        actual_request_name = None
+        actual_request_payload = None
 
         if include_request_payload:
             request_payload = get_random_dict()
@@ -51,9 +51,10 @@ async def test_handle_method_call(
             response_payload = None
 
         async def handle_on_method_request_received(request):
-            nonlocal actual_request
+            nonlocal actual_request_name, actual_request_payload
             logger.info("Method request for {} received".format(request.name))
-            actual_request = request
+            actual_request_name = request.name
+            actual_request_payload = request.payload
             logger.info("Sending response")
             await client.send_method_response(
                 MethodResponse.create_from_method_request(
@@ -67,15 +68,12 @@ async def handle_on_method_request_received(request):
         method_response = await service_helper.invoke_method(method_name, request_payload)
 
         # verify that the method request arrived correctly
-        assert actual_request.name == method_name
+        assert actual_request_name == method_name
         if request_payload:
-            assert actual_request.payload == request_payload
+            assert actual_request_payload == request_payload
         else:
-            assert not actual_request.payload
+            assert not actual_request_payload
 
         # and make sure the response came back successfully
         assert method_response.status == method_response_status
         assert method_response.payload == response_payload
-
-        actual_request = None  # so this isn't tagged as a leak
-        leak_tracker.check_for_leaks()

tests/e2e/iothub_e2e/aio/test_sas_renewal.py

@@ -24,7 +24,6 @@ class TestSasRenewal(object):
     @pytest.mark.parametrize(*parametrize.connection_retry_disabled_and_enabled)
     @pytest.mark.parametrize(*parametrize.auto_connect_disabled_and_enabled)
     async def test_sas_renews(self, client, service_helper, random_message, leak_tracker):
-        leak_tracker.set_initial_object_list()
         event_loop = asyncio.get_running_loop()
 
         connected_event = asyncio.Event()
@@ -80,6 +79,3 @@ async def handle_on_connection_state_change():
         # TODO incoming_event_queue.get should check thread future
         event = await service_helper.wait_for_eventhub_arrival(random_message.message_id)
         assert json.dumps(event.message_body) == random_message.data
-
-        random_message = None  # so this isn't flagged as a leak
-        leak_tracker.check_for_leaks()

tests/e2e/iothub_e2e/aio/test_send_message.py

@@ -20,22 +20,16 @@ class TestSendMessage(object):
     @pytest.mark.quicktest_suite
     async def test_send_simple_message(self, client, random_message, service_helper, leak_tracker):
 
-        leak_tracker.set_initial_object_list()
-
         await client.send_message(random_message)
 
         event = await service_helper.wait_for_eventhub_arrival(random_message.message_id)
         assert event.system_properties["message-id"] == random_message.message_id
         assert json.dumps(event.message_body) == random_message.data
 
-        leak_tracker.check_for_leaks()
-
     @pytest.mark.it("Connects the transport if necessary")
     @pytest.mark.quicktest_suite
     async def test_connect_if_necessary(self, client, random_message, service_helper, leak_tracker):
 
-        leak_tracker.set_initial_object_list()
-
         await client.disconnect()
         assert not client.connected
 
@@ -45,11 +39,8 @@ async def test_connect_if_necessary(self, client, random_message, service_helper
         event = await service_helper.wait_for_eventhub_arrival(random_message.message_id)
         assert json.dumps(event.message_body) == random_message.data
 
-        leak_tracker.check_for_leaks()
-
     @pytest.mark.it("Raises correct exception for un-serializable payload")
     async def test_bad_payload_raises(self, client, leak_tracker):
-        leak_tracker.set_initial_object_list()
 
         # There's no way to serialize a function.
         def thing_that_cant_serialize():
@@ -59,12 +50,8 @@ def thing_that_cant_serialize():
             await client.send_message(thing_that_cant_serialize)
         assert isinstance(e_info.value.__cause__, TypeError)
 
-        # TODO: investigate leak
-        # leak_tracker.check_for_leaks()
-
     @pytest.mark.it("Can send a JSON-formatted string that isn't wrapped in a Message object")
     async def test_sends_json_string(self, client, service_helper, leak_tracker):
-        leak_tracker.set_initial_object_list()
 
         message = json.dumps(dev_utils.get_random_dict())
 
@@ -76,11 +63,8 @@ async def test_sends_json_string(self, client, service_helper, leak_tracker):
         )
         assert json.dumps(event.message_body) == message
 
-        leak_tracker.check_for_leaks()
-
     @pytest.mark.it("Can send a random string that isn't wrapped in a Message object")
     async def test_sends_random_string(self, client, service_helper, leak_tracker):
-        leak_tracker.set_initial_object_list()
 
         message = dev_utils.get_random_string(16)
 
@@ -91,8 +75,6 @@ async def test_sends_random_string(self, client, service_helper, leak_tracker):
         )
         assert event.message_body == message
 
-        leak_tracker.check_for_leaks()
-
 
 @pytest.mark.dropped_connection
 @pytest.mark.describe("Client send_message method with dropped connections")
@@ -103,7 +85,6 @@ class TestSendMessageDroppedConnection(object):
     async def test_sends_if_drop_before_sending(
         self, client, random_message, dropper, service_helper, leak_tracker
     ):
-        leak_tracker.set_initial_object_list()
 
         assert client.connected
 
@@ -132,14 +113,11 @@ async def test_sends_if_drop_before_sending(
 
         logger.info("Success")
 
-        leak_tracker.check_for_leaks()
-
     @pytest.mark.it("Sends if connection rejects send")
     @pytest.mark.uses_iptables
     async def test_sends_if_reject_before_sending(
         self, client, random_message, dropper, service_helper, leak_tracker
     ):
-        leak_tracker.set_initial_object_list()
 
         assert client.connected
 
@@ -168,8 +146,6 @@ async def test_sends_if_reject_before_sending(
 
         logger.info("Success")
 
-        leak_tracker.check_for_leaks()
-
 
 @pytest.mark.describe("Client send_message with reconnect disabled")
 @pytest.mark.keep_alive(5)
@@ -186,20 +162,16 @@ async def reconnect_after_test(self, dropper, client):
     async def test_send_message_retry_disabled(
         self, client, random_message, service_helper, leak_tracker
     ):
-        leak_tracker.set_initial_object_list()
 
         await client.send_message(random_message)
 
         event = await service_helper.wait_for_eventhub_arrival(random_message.message_id)
         assert json.dumps(event.message_body) == random_message.data
 
-        leak_tracker.check_for_leaks()
-
     @pytest.mark.it("Automatically connects if transport manually disconnected before sending")
     async def test_connect_if_necessary_retry_disabled(
         self, client, random_message, service_helper, leak_tracker
     ):
-        leak_tracker.set_initial_object_list()
 
         await client.disconnect()
         assert not client.connected
@@ -210,14 +182,11 @@ async def test_connect_if_necessary_retry_disabled(
         event = await service_helper.wait_for_eventhub_arrival(random_message.message_id)
         assert json.dumps(event.message_body) == random_message.data
 
-        leak_tracker.check_for_leaks()
-
     @pytest.mark.it("Automatically connects if transport automatically disconnected before sending")
     @pytest.mark.uses_iptables
     async def test_connects_after_automatic_disconnect_retry_disabled(
         self, client, random_message, dropper, service_helper, leak_tracker
     ):
-        leak_tracker.set_initial_object_list()
 
         assert client.connected
 
@@ -234,14 +203,10 @@ async def test_connects_after_automatic_disconnect_retry_disabled(
         event = await service_helper.wait_for_eventhub_arrival(random_message.message_id)
         assert json.dumps(event.message_body) == random_message.data
 
-        leak_tracker.check_for_leaks()
-
     @pytest.mark.it("Fails if connection disconnects before sending")
     @pytest.mark.uses_iptables
-    async def test_fails_if_disconnect_before_sending(
-        self, client, random_message, dropper, leak_tracker
-    ):
-        leak_tracker.set_initial_object_list()
+    # TODO: Re-enable leak tracking after the MQTT cancellation refactor.
+    async def test_fails_if_disconnect_before_sending(self, client, random_message, dropper):
 
         assert client.connected
 
@@ -255,16 +220,12 @@ async def test_fails_if_disconnect_before_sending(
         with pytest.raises(OperationCancelled):
             await asyncio.wait_for(send_task, timeout=const.E2E_TIMEOUT)
 
-        random_message = None  # so this doesn't get tagged as a leak
-        # TODO: investigate leak
-        # leak_tracker.check_for_leaks()
-
     @pytest.mark.it("Fails if connection drops before sending")
     @pytest.mark.uses_iptables
+    # TODO: Re-enable leak tracking after the MQTT cancellation refactor.
     async def test_fails_if_drop_before_sending_retry_disabled(
-        self, client, random_message, dropper, leak_tracker
+        self, client, random_message, dropper
     ):
-        leak_tracker.set_initial_object_list()
 
         assert client.connected
 
@@ -273,7 +234,3 @@ async def test_fails_if_drop_before_sending_retry_disabled(
             await client.send_message(random_message)
 
         assert not client.connected
-
-        random_message = None  # so this doesn't get tagged as a leak
-        # TODO: investigate leak
-        # leak_tracker.check_for_leaks()