chore: Update test infrastructure

dev_utils/dev_utils/service_helper.py

@@ -12,24 +12,22 @@ def __init__(
         iothub_connection_string,
         eventhub_connection_string,
         eventhub_consumer_group,
-        event_loop=None,
         executor=None,
     ):
-        self._event_loop = event_loop or asyncio.get_event_loop()
         self._executor = executor or concurrent.futures.ThreadPoolExecutor()
         self._inner_object = ServiceHelperSync(
             iothub_connection_string, eventhub_connection_string, eventhub_consumer_group
         )
 
+    async def _run_in_executor(self, func, *args):
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(self._executor, func, *args)
+
     def set_identity(self, device_id, module_id):
         return self._inner_object.set_identity(device_id, module_id)
 
     async def set_desired_properties(self, desired_props):
-        return await self._event_loop.run_in_executor(
-            self._executor,
-            self._inner_object.set_desired_properties,
-            desired_props,
-        )
+        return await self._run_in_executor(self._inner_object.set_desired_properties, desired_props)
 
     async def invoke_method(
         self,
@@ -38,8 +36,7 @@ async def invoke_method(
         connect_timeout_in_seconds=None,
         response_timeout_in_seconds=None,
     ):
-        return await self._event_loop.run_in_executor(
-            self._executor,
+        return await self._run_in_executor(
             self._inner_object.invoke_method,
             method_name,
             payload,
@@ -52,25 +49,21 @@ async def send_c2d(
         payload,
         properties,
     ):
-        return await self._event_loop.run_in_executor(
-            self._executor, self._inner_object.send_c2d, payload, properties
-        )
+        return await self._run_in_executor(self._inner_object.send_c2d, payload, properties)
 
     async def wait_for_eventhub_arrival(self, message_id, timeout=60):
-        return await self._event_loop.run_in_executor(
-            self._executor,
+        return await self._run_in_executor(
             self._inner_object.wait_for_eventhub_arrival,
             message_id,
             timeout,
         )
 
     async def get_next_reported_patch_arrival(self, block=True, timeout=240):
-        return await self._event_loop.run_in_executor(
-            self._executor,
+        return await self._run_in_executor(
             self._inner_object.get_next_reported_patch_arrival,
             block,
             timeout,
         )
 
     async def shutdown(self):
-        return await self._event_loop.run_in_executor(self._executor, self._inner_object.shutdown)
+        return await self._run_in_executor(self._inner_object.shutdown)

pytest.ini

@@ -1,4 +1,5 @@
 [pytest]
+asyncio_mode=auto
 testdox_format = plaintext
 addopts = --testdox --timeout 20 --ignore e2e --ignore tests/e2e
 norecursedirs=__pycache__, *.egg-info

requirements_test.txt

@@ -1,5 +1,5 @@
-pytest < 8.0.0
-pytest-asyncio <= 0.16
+pytest>=8,<9
+pytest-asyncio~=1.2
 pytest-mock
 pytest-testdox>=1.1.1
 pytest-cov

tests/e2e/iothub_e2e/aio/conftest.py

@@ -2,7 +2,6 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 import pytest
-import asyncio
 from dev_utils import test_env, ServiceHelper
 import logging
 import datetime
@@ -56,13 +55,6 @@ def pytest_sessionfinish(session, exitstatus):
             print("-----------------------------------")
 
 
-@pytest.fixture(scope="session")
-def event_loop():
-    loop = asyncio.get_event_loop()
-    yield loop
-    loop.close()
-
-
 @pytest.fixture(scope="function")
 async def brand_new_client(device_identity, client_kwargs, service_helper, device_id, module_id):
     service_helper.set_identity(device_id, module_id)
@@ -104,12 +96,11 @@ async def client(brand_new_client):
 
 
 @pytest.fixture(scope="session")
-async def service_helper(event_loop, executor):
+async def service_helper(executor):
     service_helper = ServiceHelper(
         iothub_connection_string=test_env.IOTHUB_CONNECTION_STRING,
         eventhub_connection_string=test_env.EVENTHUB_CONNECTION_STRING,
         eventhub_consumer_group=test_env.EVENTHUB_CONSUMER_GROUP,
-        event_loop=event_loop,
         executor=executor,
     )
     yield service_helper

tests/e2e/iothub_e2e/aio/test_c2d.py

@@ -10,7 +10,6 @@
 logger = logging.getLogger(__name__)
 logger.setLevel(level=logging.INFO)
 
-pytestmark = pytest.mark.asyncio
 
 # TODO: add tests for various application properties
 # TODO: is there a way to call send_c2d so it arrives as an object rather than a JSON string?
@@ -20,8 +19,9 @@
 class TestReceiveC2d(object):
     @pytest.mark.it("Can receive C2D")
     @pytest.mark.quicktest_suite
-    async def test_receive_c2d(self, client, service_helper, event_loop, leak_tracker):
+    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())
 

tests/e2e/iothub_e2e/aio/test_connect_disconnect.py

@@ -9,8 +9,6 @@
 logger = logging.getLogger(__name__)
 logger.setLevel(level=logging.INFO)
 
-pytestmark = pytest.mark.asyncio
-
 
 @pytest.mark.describe("Client object")
 class TestConnectDisconnect(object):
@@ -44,7 +42,7 @@ async def test_connect_disconnect(self, brand_new_client, leak_tracker):
     # see "This assert fails because of initial and secondary disconnects" below
     @pytest.mark.skip(reason="two stage disconnect causes assertion in test code")
     async def test_connect_in_the_middle_of_disconnect(
-        self, brand_new_client, event_loop, service_helper, random_message, leak_tracker
+        self, brand_new_client, service_helper, random_message, leak_tracker
     ):
         """
         Explanation: People will call `connect` inside `on_connection_state_change` handlers.
@@ -54,6 +52,7 @@ async def test_connect_in_the_middle_of_disconnect(
         assert client
 
         leak_tracker.set_initial_object_list()
+        event_loop = asyncio.get_running_loop()
 
         reconnected_event = asyncio.Event()
 
@@ -112,7 +111,6 @@ async def handle_on_connection_state_change():
     async def test_disconnect_in_the_middle_of_connect(
         self,
         brand_new_client,
-        event_loop,
         service_helper,
         random_message,
         first_connect,
@@ -128,6 +126,7 @@ async def test_disconnect_in_the_middle_of_connect(
         disconnect_on_next_connect_event = False
 
         leak_tracker.set_initial_object_list()
+        event_loop = asyncio.get_running_loop()
 
         disconnected_event = asyncio.Event()
 

tests/e2e/iothub_e2e/aio/test_connect_disconnect_stress.py

@@ -10,8 +10,6 @@
 logger = logging.getLogger(__name__)
 logger.setLevel(level=logging.INFO)
 
-pytestmark = pytest.mark.asyncio
-
 
 @pytest.mark.stress
 @pytest.mark.describe("Client object connect/disconnect stress")

tests/e2e/iothub_e2e/aio/test_infrastructure.py

@@ -4,8 +4,6 @@
 import pytest
 import uuid
 
-pytestmark = pytest.mark.asyncio
-
 
 @pytest.mark.describe("ServiceHelper object")
 class TestServiceHelper(object):

tests/e2e/iothub_e2e/aio/test_methods.py

@@ -11,8 +11,6 @@
 logger = logging.getLogger(__name__)
 logger.setLevel(level=logging.INFO)
 
-pytestmark = pytest.mark.asyncio
-
 
 @pytest.fixture
 def method_name():

tests/e2e/iothub_e2e/aio/test_sas_renewal.py

@@ -11,8 +11,6 @@
 logger = logging.getLogger(__name__)
 logger.setLevel(level=logging.INFO)
 
-pytestmark = pytest.mark.asyncio
-
 
 @pytest.mark.skipif(
     test_config.config.auth not in test_config.AUTH_WITH_RENEWING_TOKEN,
@@ -24,10 +22,9 @@ class TestSasRenewal(object):
     @pytest.mark.it("Renews and reconnects before expiry")
     @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, event_loop, service_helper, random_message, leak_tracker
-    ):
+    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()
         disconnected_event = asyncio.Event()

tests/e2e/iothub_e2e/aio/test_send_message.py

@@ -11,8 +11,6 @@
 logger = logging.getLogger(__name__)
 logger.setLevel(level=logging.INFO)
 
-pytestmark = pytest.mark.asyncio
-
 
 @pytest.mark.describe("Client send_message method")
 class TestSendMessage(object):

tests/e2e/iothub_e2e/aio/test_send_message_stress.py

@@ -15,7 +15,6 @@
 logger = logging.getLogger(__name__)
 logger.setLevel(level=logging.INFO)
 
-pytestmark = pytest.mark.asyncio
 
 # Settings that apply to all tests in this module
 TELEMETRY_PAYLOAD_SIZE = 16 * 1024

tests/e2e/iothub_e2e/aio/test_twin.py

@@ -11,8 +11,6 @@
 logger = logging.getLogger(__name__)
 logger.setLevel(level=logging.INFO)
 
-pytestmark = pytest.mark.asyncio
-
 
 # TODO: tests with drop_incoming and reject_incoming
 
@@ -192,10 +190,9 @@ async def test_updates_reported_if_reject_before_sending(
 class TestDesiredProperties(object):
     @pytest.mark.it("Receives a patch for a simple desired property")
     @pytest.mark.quicktest_suite
-    async def test_receives_simple_desired_patch(
-        self, client, event_loop, service_helper, leak_tracker
-    ):
+    async def test_receives_simple_desired_patch(self, client, service_helper, leak_tracker):
         leak_tracker.set_initial_object_list()
+        event_loop = asyncio.get_running_loop()
 
         received_patch = None
         received = asyncio.Event()

tests/e2e/iothub_e2e/aio/test_twin_stress.py

@@ -12,8 +12,6 @@
 logger = logging.getLogger(__name__)
 logger.setLevel(level=logging.INFO)
 
-pytestmark = pytest.mark.asyncio
-
 
 @pytest.fixture
 def toxic():
@@ -159,13 +157,14 @@ async def test_stress_parallel_reported_property_updates(
     )
     @pytest.mark.it("Can receive continuous desired property updates that were sent one-at-a-time")
     async def test_stress_serial_desired_property_updates(
-        self, client, service_helper, toxic, iteration_count, event_loop, leak_tracker
+        self, client, service_helper, toxic, iteration_count, leak_tracker
     ):
         """
         Update desired properties, one at a time, and verify that the desired property arrives
         at the client before the next update.
         """
         leak_tracker.set_initial_object_list()
+        event_loop = asyncio.get_running_loop()
 
         patches = asyncio.Queue()
 
@@ -202,13 +201,14 @@ async def handle_on_patch_received(patch):
         "Can receive continuous desired property updates that may have been sent in parallel"
     )
     async def test_stress_parallel_desired_property_updates(
-        self, client, service_helper, toxic, iteration_count, batch_size, event_loop, leak_tracker
+        self, client, service_helper, toxic, iteration_count, batch_size, leak_tracker
     ):
         """
         Update desired properties in batches. Each batch updates `batch_size` properties,
         with each property being updated in it's own `PATCH`.
         """
         leak_tracker.set_initial_object_list()
+        event_loop = asyncio.get_running_loop()
 
         patches = asyncio.Queue()
 

tests/e2e/iothub_e2e/pytest.ini

@@ -5,7 +5,6 @@ junit_logging=all
 junit_family=xunit2
 junit_log_passing_tests=True
 asyncio_mode=auto
-# --force-testdox to always use testdox format, even when redirecting to file
 addopts=
     --testdox
     --force-testdox

tests/e2e/provisioning_e2e/pytest.ini

@@ -1,2 +1,7 @@
 [pytest]
-addopts = --timeout 30
+timeout=30
+testdox_format=plaintext
+asyncio_mode=auto
+addopts=
+    --testdox
+    --force-testdox

tests/e2e/provisioning_e2e/tests/test_async_certificate_enrollments.py

@@ -29,7 +29,6 @@
 )
 
 
-pytestmark = pytest.mark.asyncio
 logging.basicConfig(level=logging.DEBUG)
 
 

tests/e2e/provisioning_e2e/tests/test_async_symmetric_enrollments.py

@@ -13,7 +13,7 @@
 import os
 import uuid
 
-pytestmark = pytest.mark.asyncio
+
 logging.basicConfig(level=logging.DEBUG)
 
 

tests/unit/common/test_async_adapter.py

@@ -11,7 +11,6 @@
 import azure.iot.device.common.async_adapter as async_adapter
 
 logging.basicConfig(level=logging.DEBUG)
-pytestmark = pytest.mark.asyncio
 
 
 @pytest.fixture

tests/unit/iothub/aio/test_async_clients.py

@@ -44,7 +44,6 @@
     SharedIoTHubModuleClientCreateFromEdgeEnvironmentWithDebugEnvTests,
 )
 
-pytestmark = pytest.mark.asyncio
 logging.basicConfig(level=logging.DEBUG)
 
 

tests/unit/iothub/aio/test_async_handler_manager.py

@@ -16,7 +16,6 @@
 from azure.iot.device.iothub.inbox_manager import InboxManager
 from azure.iot.device.iothub.aio.async_inbox import AsyncClientInbox
 
-pytestmark = pytest.mark.asyncio
 logging.basicConfig(level=logging.DEBUG)
 
 # NOTE ON TEST IMPLEMENTATION:

tests/unit/provisioning/aio/test_async_provisioning_device_client.py

@@ -20,7 +20,6 @@
 
 
 logging.basicConfig(level=logging.DEBUG)
-pytestmark = pytest.mark.asyncio
 
 
 async def create_completed_future(result=None):