Moving bindings utils from plugins to pulp-smash

[noissue]

pulp_smash/api.py

@@ -675,6 +675,16 @@ def poll_spawned_tasks(cfg, call_report, pulp_host=None):
             yield final_task_state
 
 
+def _get_sleep_time(cfg):
+    """Returns the default waiting time for polling tasks.
+
+    :param cfg: A :class:`pulp_smash.config.PulpSmashConfig` object.
+    """
+    if cfg.pulp_version < Version("3"):
+        return 2
+    return 0.3
+
+
 def poll_task(cfg, href, pulp_host=None):
     """Wait for a task and its children to complete. Yield response bodies.
 
@@ -695,10 +705,7 @@ def poll_task(cfg, href, pulp_host=None):
     # An example: Assuming timeout = 1800s, and sleep_time = 0.3s
     # 1800s/0.3s = 6000
 
-    if cfg.pulp_version < Version("3"):
-        sleep_time = 2
-    else:
-        sleep_time = 0.3
+    sleep_time = _get_sleep_time(cfg)
     poll_limit = int(cfg.timeout / sleep_time)
     poll_counter = 0
     json_client = Client(cfg, json_handler, pulp_host=pulp_host)

pulp_smash/pulp3/bindings.py

@@ -0,0 +1,67 @@
+from unittest import TestCase
+from time import sleep
+
+from pulpcore.client.pulpcore import ApiClient, TasksApi
+
+from pulp_smash.api import _get_sleep_time
+from pulp_smash.config import get_config
+
+
+cfg = get_config()
+SLEEP_TIME = _get_sleep_time(cfg)
+configuration = cfg.get_bindings_config()
+pulpcore_client = ApiClient(configuration)
+tasks = TasksApi(pulpcore_client)
+
+
+class PulpTestCase(TestCase):
+    """Pulp customized test case."""
+
+    def doCleanups(self):
+        """
+        Execute all cleanup functions and waits the deletion tasks.
+
+        Normally called for you after tearDown.
+        """
+        output = super().doCleanups()
+        running_tasks = tasks.list(state="running", name__contains="delete")
+        while running_tasks.count:
+            sleep(SLEEP_TIME)
+            running_tasks = tasks.list(
+                state="running", name__contains="delete"
+            )
+        return output
+
+
+class PulpTaskError(Exception):
+    """Exception to describe task errors."""
+
+    def __init__(self, task):
+        """Provide task info to exception."""
+        description = task.error["description"]
+        super().__init__(self, f"Pulp task failed ({description})")
+        self.task = task
+
+
+def monitor_task(task_href):
+    """Polls the Task API until the task is in a completed state.
+
+    Prints the task details and a success or failure message. Exits on failure.
+
+    Args:
+        task_href(str): The href of the task to monitor
+
+    Returns:
+        list[str]: List of hrefs that identify resource created by the task
+
+    """
+    completed = ["completed", "failed", "canceled"]
+    task = tasks.read(task_href)
+    while task.state not in completed:
+        sleep(SLEEP_TIME)
+        task = tasks.read(task_href)
+
+    if task.state == "completed":
+        return task.created_resources
+
+    raise PulpTaskError(task=task)