test: refactor test_webserver_graceful_shutdown to use pytest.mark.parametrize

Author: tumidi

tests/questionpy_sdk/webserver/test_webserver.py

@@ -189,37 +189,37 @@ class PackageQuestion(Question):
     return QuestionTypeWrapper(PackageQuestion, package)
 
 
-async def _assert_graceful_shutdown(pkg_location: PackageLocation, storage_path: Path, port: int) -> None:
-    async with WebServer(pkg_location, state_storage_path=storage_path, port=port):
-        await asyncio.sleep(0)
-
-    pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task() and not t.done()]
-    if pending:
-        pending_tasks_display = "\n".join(f"  - {t}" for t in pending)
-        for task in pending:
-            task.cancel()
-        await asyncio.gather(*pending, return_exceptions=True)
-        pytest.fail(f"Pending tasks after shutdown:\n{pending_tasks_display}")
+@pytest.fixture
+def function_pkg_location() -> FunctionPackageLocation:
+    return FunctionPackageLocation.from_function(_pkg_init)
 
 
-@pytest.mark.asyncio(loop_scope="function")
-async def test_webserver_shutdown_function(tmp_path: Path, port: int) -> None:
-    pkg_location = FunctionPackageLocation.from_function(_pkg_init)
+@pytest.fixture
+def dir_pkg_location(source_path: Path) -> DirPackageLocation:
+    with DirPackageBuilder(PackageSource(source_path)) as builder:
+        builder.write_package()
+    return DirPackageLocation(source_path / DIST_DIR)
 
-    await _assert_graceful_shutdown(pkg_location, tmp_path, port)
 
+@pytest.fixture
+def zip_pkg_location(qpy_pkg_path: Path) -> ZipPackageLocation:
+    return ZipPackageLocation(qpy_pkg_path, calculate_hash(qpy_pkg_path))
 
-@pytest.mark.asyncio(loop_scope="function")
-async def test_webserver_shutdown_dir(tmp_path: Path, port: int, source_path: Path) -> None:
-    with DirPackageBuilder(PackageSource(source_path)) as builder:
-        builder.write_package()
-    pkg_location = DirPackageLocation(source_path / DIST_DIR)
 
-    await _assert_graceful_shutdown(pkg_location, tmp_path, port)
+@pytest.fixture(params=["function_pkg_location", "dir_pkg_location", "zip_pkg_location"])
+def pkg_location(request: pytest.FixtureRequest) -> PackageLocation:
+    return request.getfixturevalue(request.param)
 
 
 @pytest.mark.asyncio(loop_scope="function")
-async def test_webserver_shutdown_zip(tmp_path: Path, port: int, qpy_pkg_path: Path) -> None:
-    pkg_location = ZipPackageLocation(qpy_pkg_path, calculate_hash(qpy_pkg_path))
+async def test_webserver_graceful_shutdown(pkg_location: PackageLocation, tmp_path: Path, port: int) -> None:
+    async with WebServer(pkg_location, state_storage_path=tmp_path, port=port):
+        await asyncio.sleep(0)
 
-    await _assert_graceful_shutdown(pkg_location, tmp_path, port)
+    pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task() and not t.done()]
+    if pending:
+        pending_tasks_display = "\n".join(f"  - {t}" for t in pending)
+        for task in pending:
+            task.cancel()
+        await asyncio.gather(*pending, return_exceptions=True)
+        pytest.fail(f"Pending tasks after shutdown:\n{pending_tasks_display}")