wip: review iteration

Author: pedro-psb

pulpcore/pytest_plugin.py

@@ -9,6 +9,7 @@
 import socket
 import ssl
 import subprocess
+import sys
 import threading
 import uuid
 
@@ -993,53 +994,32 @@ def _wget_recursive_download_on_host(url, destination):
 # Tasking related fixtures
 
 
-def _build_dispatch_cmd(pulpcore_bindings, task_group_id, args, kwargs):
-    cid = pulpcore_bindings.client.default_headers.get("Correlation-ID") or str(uuid.uuid4())
-    username = pulpcore_bindings.client.configuration.username
-    return (
-        "from django_guid import set_guid; "
-        "from pulpcore.tasking.tasks import dispatch; "
-        "from pulpcore.app.models import TaskGroup; "
-        "from pulpcore.app.util import get_url, set_current_user; "
-        "from django.contrib.auth import get_user_model; "
-        "User = get_user_model(); "
-        f"user = User.objects.filter(username='{username}').first(); "
-        "set_current_user(user); "
-        f"set_guid({cid!r}); "
-        f"tg = {task_group_id!r} and TaskGroup.objects.filter(pk={task_group_id!r}).first(); "
-        f"task = dispatch(*{args!r}, task_group=tg, **{kwargs!r}); "
-        "print(get_url(task))"
-    )
-
-
 @pytest.fixture(scope="session")
 def dispatch_task(pulpcore_bindings):
     def _dispatch_task(*args, task_group_id=None, **kwargs):
-        commands = _build_dispatch_cmd(pulpcore_bindings, task_group_id, args, kwargs)
-        process = subprocess.run(["pulpcore-manager", "shell", "-c", commands], capture_output=True)
-        assert process.returncode == 0
-        task_href = process.stdout.decode().strip()
-        return task_href
-
-    return _dispatch_task
-
-
-@pytest.fixture(scope="session")
-def dispatch_task_with_stderr(pulpcore_bindings):
-    """Variation of dispatch_task that returns (task_href, stderr_content).
-
-    The stderr can be useful write asserts over log messages.
-    This is required because the pytest caplog fixture can't capture log messages
-    streamed from the pulpcore-manager call.
-    """
+        cid = pulpcore_bindings.client.default_headers.get("Correlation-ID") or str(uuid.uuid4())
+        username = pulpcore_bindings.client.configuration.username
+        commands = (
+            "from django_guid import set_guid; "
+            "from pulpcore.tasking.tasks import dispatch; "
+            "from pulpcore.app.models import TaskGroup; "
+            "from pulpcore.app.util import get_url, set_current_user; "
+            "from django.contrib.auth import get_user_model; "
+            "User = get_user_model(); "
+            f"user = User.objects.filter(username='{username}').first(); "
+            "set_current_user(user); "
+            f"set_guid({cid!r}); "
+            f"tg = {task_group_id!r} and TaskGroup.objects.filter(pk={task_group_id!r}).first(); "
+            f"task = dispatch(*{args!r}, task_group=tg, **{kwargs!r}); "
+            "print(get_url(task))"
+        )
 
-    def _dispatch_task(*args, task_group_id=None, **kwargs):
-        commands = _build_dispatch_cmd(pulpcore_bindings, task_group_id, args, kwargs)
         process = subprocess.run(["pulpcore-manager", "shell", "-c", commands], capture_output=True)
+
         assert process.returncode == 0
         task_href = process.stdout.decode().strip()
-        stderr = process.stderr.decode()
-        return task_href, stderr
+        print(process.stderr.decode(), file=sys.stderr)
+        return task_href
 
     return _dispatch_task
 

pulpcore/tasking/tasks.py

@@ -92,7 +92,6 @@ def _execute_task(task):
             coro = func(*args, **kwargs)
             if immediate:
                 coro = asyncio.wait_for(coro, timeout=IMMEDIATE_TIMEOUT)
-            assert asyncio.iscoroutine(coro), coro
             loop = asyncio.get_event_loop()
             try:
                 loop.run_until_complete(coro)

pulpcore/tests/functional/api/test_tasking.py

@@ -471,25 +471,22 @@ def test_succeeds_on_api_worker(self, pulpcore_bindings, dispatch_task):
         assert task.worker is None
 
     @pytest.mark.parallel
-    def test_executes_on_api_worker_when_no_async(
-        self,
-        pulpcore_bindings,
-        dispatch_task_with_stderr,
-    ):
+    def test_executes_on_api_worker_when_no_async(self, pulpcore_bindings, dispatch_task, capsys):
         """
         GIVEN a task with no resource requirements
         AND the task IS NOT an async function
         WHEN dispatching a task as immediate
         THEN the task completes with no associated worker
         """
         # TODO: on 3.85 this should throw an error
-        task_href, stderr = dispatch_task_with_stderr(
+        task_href = dispatch_task(
             "pulpcore.app.tasks.test.sleep", args=(LT_TIMEOUT,), immediate=True
         )
+        stderr_content = capsys.readouterr().err
         task = pulpcore_bindings.TasksApi.read(task_href)
         assert task.state == "completed"
         assert task.worker is None
-        assert "Support for non-coroutine immediate tasks will be dropped" in stderr
+        assert "Support for non-coroutine immediate tasks will be dropped" in stderr_content
 
     @pytest.mark.parallel
     def test_timeouts_on_api_worker(self, pulpcore_bindings, dispatch_task):
@@ -509,7 +506,7 @@ def test_timeouts_on_api_worker(self, pulpcore_bindings, dispatch_task):
 
 
 @pytest.fixture
-def dispatch_long_task(pulpcore_bindings, dispatch_task):
+def resource_blocker(pulpcore_bindings, dispatch_task):
 
     def wait_until(state, task_href, timeout=10):
         for i in range(timeout):
@@ -520,34 +517,39 @@ def wait_until(state, task_href, timeout=10):
         raise RuntimeError("Timeout waiting for task to transition")
 
     @contextmanager
-    def _dispatch_long_task(required_resources: list[str], duration=5):
+    def _resource_blocker(exclusive_resources: list[str], duration=5):
         task_href = dispatch_task(
             "pulpcore.app.tasks.test.sleep",
             args=(duration,),
-            exclusive_resources=required_resources,
+            exclusive_resources=exclusive_resources,
         )
         wait_until("running", task_href)
         yield
-        task = pulpcore_bindings.TasksApi.read(task_href)
-        if task.state == "running":
+        # Trying to cancel a finished task will return a 409 code.
+        # We can ignore if that's the case, because all we want here is to cut time down.
+        # Otherwise it might be a real error.
+        try:
             pulpcore_bindings.TasksApi.tasks_cancel(task_href, {"state": "canceled"})
+        except ApiException as e:
+            if e.status != 409:
+                raise
 
-    return _dispatch_long_task
+    return _resource_blocker
 
 
 class TestImmediateTaskWithBlockedResource:
 
     @pytest.mark.parallel
     def test_executes_in_task_worker(
-        self, dispatch_long_task, dispatch_task, monitor_task, pulpcore_bindings
+        self, resource_blocker, dispatch_task, monitor_task, pulpcore_bindings
     ):
         """
         GIVEN an async task requiring busy resources
         WHEN dispatching a task as immediate
         THEN the task completes with a worker
         """
         COMMON_RESOURCE = str(uuid4())
-        with dispatch_long_task(required_resources=[COMMON_RESOURCE]):
+        with resource_blocker(exclusive_resources=[COMMON_RESOURCE]):
             task_href = dispatch_task(
                 "pulpcore.app.tasks.test.asleep",
                 args=(LT_TIMEOUT,),
@@ -560,15 +562,15 @@ def test_executes_in_task_worker(
 
     @pytest.mark.parallel
     def test_throws_when_non_deferrable(
-        self, dispatch_long_task, pulpcore_bindings, dispatch_task, monitor_task
+        self, resource_blocker, pulpcore_bindings, dispatch_task, monitor_task
     ):
         """
         GIVEN an async task requiring busy resources
         WHEN dispatching as immediate and not deferrable
         THEN an error is raised
         """
         COMMON_RESOURCE = str(uuid4())
-        with dispatch_long_task(required_resources=[COMMON_RESOURCE]):
+        with resource_blocker(exclusive_resources=[COMMON_RESOURCE]):
             task_href = dispatch_task(
                 "pulpcore.app.tasks.test.asleep",
                 args=(0,),
@@ -582,8 +584,8 @@ def test_throws_when_non_deferrable(
             assert "Resources temporarily unavailable." in task.error["reason"]
 
     @pytest.mark.parallel
-    def test_timeouts_on_task_worker(
-        self, dispatch_long_task, pulpcore_bindings, dispatch_task, monitor_task
+    def test_times_out_on_task_worker(
+        self, resource_blocker, pulpcore_bindings, dispatch_task, monitor_task
     ):
         """
         GIVEN an async task requiring busy resources
@@ -593,7 +595,7 @@ def test_timeouts_on_task_worker(
         """
         COMMON_RESOURCE = str(uuid4())
         with pytest.raises(PulpTaskError) as ctx:
-            with dispatch_long_task(required_resources=[COMMON_RESOURCE]):
+            with resource_blocker(exclusive_resources=[COMMON_RESOURCE]):
                 task_href = dispatch_task(
                     "pulpcore.app.tasks.test.asleep",
                     args=(GT_TIMEOUT,),