Merge branch 'main' into ct-client-cleanup

Author: cartertinney

dev_utils/dev_utils/service_helper.py

@@ -14,25 +14,23 @@ 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._owns_executor = executor is None
         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,
@@ -41,8 +39,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,
@@ -55,28 +52,25 @@ 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):
-        await self._event_loop.run_in_executor(self._executor, self._inner_object.shutdown)
+        await self._run_in_executor(self._inner_object.shutdown)
 
         if self._owns_executor:
             self._executor.shutdown(wait=True)
+        return

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()