Add cms_session fixture and update zk_auto fixture to use cluster hosts

Author: wmudge

tests/unit/conftest.py

@@ -204,15 +204,83 @@ def cm_api_client(conn) -> ApiClient:
 
 
 @pytest.fixture(scope="session")
-def base_cluster(cm_api_client, request) -> Generator[ApiCluster]:
+def cms_session(cm_api_client) -> Generator[ApiService]:
+    """
+    Provisions the Cloudera Manager Service. If the Cloudera Manager Service
+    is present, will read and yield this reference. Otherwise, will
+    yield a new Cloudera Manager Service, deleting it after use.
+
+    If it does create a new Cloudera Manager Service, it will do so on the
+    first available host and will auto-configure the following roles:
+        - HOSTMONITOR
+        - SERVICEMONITOR
+        - EVENTSERVER
+        - ALERTPUBLISHER
+
+    It starts this Cloudera Manager Service, yields, and will remove this
+    service if it has created it.
+
+    Args:
+        cm_api_client (ApiClient): CM API client
+
+    Yields:
+        Generator[ApiService]: Cloudera Manager Service
+    """
+
+    cms_api = MgmtServiceResourceApi(cm_api_client)
+
+    try:
+        # Return if the Cloudera Manager Service is already present
+        yield cms_api.read_service()
+
+        # Do nothing on teardown
+        return
+    except ApiException as ae:
+        if ae.status != 404 or "Cannot find management service." not in str(ae.body):
+            raise Exception(str(ae))
+
+    service_api = MgmtServiceResourceApi(cm_api_client)
+    host_api = HostsResourceApi(cm_api_client)
+
+    host = next((h for h in host_api.read_hosts().items if not h.cluster_ref), None)
+
+    if host is None:
+        raise Exception("No available hosts to assign Cloudera Manager Service roles")
+
+    name = "pytest-" + "".join(random.choices(string.ascii_lowercase, k=6))
+
+    service_api.setup_cms(
+        body=ApiService(
+            name=name,
+            type="MGMT",
+            roles=[
+                ApiRole(type="HOSTMONITOR"),
+                ApiRole(type="SERVICEMONITOR"),
+                ApiRole(type="EVENTSERVER"),
+                ApiRole(type="ALERTPUBLISHER"),
+            ],
+        )
+    )
+    service_api.auto_configure()
+
+    monitor_command(cm_api_client, service_api.start_command())
+
+    # Return the newly-minted CMS
+    yield service_api.read_service()
+
+    # Delete the created CMS
+    cms_api.delete_cms()
+
+
+@pytest.fixture(scope="session")
+def base_cluster(cm_api_client, cms_session) -> Generator[ApiCluster]:
     """Provision a Cloudera on premise base cluster for the session.
        If the variable 'CM_CLUSTER' is present, will attempt to read and yield
        a reference to this cluster. Otherwise, will yield a new base cluster
        with a single host, deleting the cluster once completed.
 
     Args:
         cm_api_client (ApiClient): CM API client
-        request (FixtureRequest): Fixture request
 
     Raises:
         Exception: _description_
@@ -236,7 +304,7 @@ def base_cluster(cm_api_client, request) -> Generator[ApiCluster]:
             )
 
         name = (
-            Path(request.fixturename).stem
+            cms_session.name
             + "_"
             + "".join(random.choices(string.ascii_lowercase, k=6))
         )
@@ -314,19 +382,19 @@ def zk_auto(cm_api_client, base_cluster, request) -> Generator[ApiService]:
     service_api = ServicesResourceApi(cm_api_client)
     host_api = HostsResourceApi(cm_api_client)
 
-    host = next((h for h in host_api.read_hosts().items if not h.cluster_ref), None)
+    host = next((h for h in host_api.read_hosts().items if h.cluster_ref), None)
 
     if host is None:
         raise NoHostsFoundException(
-            "No available hosts to assign Cloudera Manager Service roles"
+            "No available hosts to assign ZooKeeper service roles"
         )
 
     payload = ApiService(
-        name=f"zk-{Path(request.node.parent.name).stem}",
+        name="-".join(["zk", request.node.name]),
         type="ZOOKEEPER",
         roles=[
             ApiRole(
-                type="ZOOKEEPER",
+                type="SERVER",
                 host_ref=ApiHostRef(host.host_id, host.hostname),
             ),
         ],
@@ -779,7 +847,7 @@ def handle_commands(api_client: ApiClient, commands: ApiBulkCommandList):
 
 
 def monitor_command(
-    api_client: ApiClient, command: ApiCommand, polling: int = 10, delay: int = 15
+    api_client: ApiClient, command: ApiCommand, polling: int = 120, delay: int = 10
 ):
     poll_count = 0
     while command.active: