testlib_refactor: extract dynamic host configuration API requests int…

…o a separate class Jira ticket: CMK-20440 Change-Id: I06b24610c6307ebb82c309028b60c158b0f26a2a
Checkmk · Dec 17, 2024 · aec0cd4 · aec0cd4
1 parent ef3d8f1
commit aec0cd4
Show file tree

Hide file tree

Showing 2 changed files with 54 additions and 51 deletions.
diff --git a/tests/plugins_integration/checks.py b/tests/plugins_integration/checks.py
@@ -658,7 +658,7 @@ def _dcd_connector(test_site_piggyback: Site) -> Iterator[None]:
         "tag_piggyback": "piggyback",
         "tag_address_family": "no-ip",
     }
-    test_site_piggyback.openapi.create_dynamic_host_configuration(
+    test_site_piggyback.openapi.dcd.create(
         dcd_id=dcd_id,
         title="DCD Connector for piggyback hosts",
         host_attributes=host_attributes,
@@ -673,7 +673,7 @@ def _dcd_connector(test_site_piggyback: Site) -> Iterator[None]:
         yield
     finally:
         if not config.skip_cleanup:
-            test_site_piggyback.openapi.delete_dynamic_host_configuration(dcd_id)
+            test_site_piggyback.openapi.dcd.delete(dcd_id)
             test_site_piggyback.openapi.activate_changes_and_wait_for_completion(
                 force_foreign_changes=True
             )
diff --git a/tests/testlib/openapi_session.py b/tests/testlib/openapi_session.py
@@ -120,6 +120,7 @@ def __init__(
         self.broker_connections = BrokerConnectionsAPI(self)
         self.sites = SitesAPI(self)
         self.background_jobs = BackgroundJobsAPI(self)
+        self.dcd = DcdAPI(self)
 
     def set_authentication_header(self, user: str, password: str) -> None:
         self.headers["Authorization"] = f"Bearer {user} {password}"
@@ -297,55 +298,6 @@ def _handle_wait_redirect(
 
             time.sleep(0.5)
 
-    def create_dynamic_host_configuration(
-        self,
-        dcd_id: str,
-        title: str,
-        comment: str = "",
-        disabled: bool = False,
-        restrict_source_hosts: list | None = None,
-        interval: int = 60,
-        host_attributes: dict | None = None,
-        delete_hosts: bool = False,
-        discover_on_creation: bool = True,
-        no_deletion_time_after_init: int = 600,
-        max_cache_age: int = 3600,
-        validity_period: int = 60,
-    ) -> None:
-        """Create a DCD connection via REST API."""
-        resp = self.post(
-            "/domain-types/dcd/collections/all",
-            json={
-                "dcd_id": dcd_id,
-                "title": title,
-                "comment": comment,
-                "disabled": disabled,
-                "site": self.site,
-                "connector_type": "piggyback",
-                "restrict_source_hosts": restrict_source_hosts or [],
-                "interval": interval,
-                "creation_rules": [
-                    {
-                        "folder_path": "/",
-                        "host_attributes": host_attributes or {},
-                        "delete_hosts": delete_hosts,
-                    }
-                ],
-                "discover_on_creation": discover_on_creation,
-                "no_deletion_time_after_init": no_deletion_time_after_init,
-                "max_cache_age": max_cache_age,
-                "validity_period": validity_period,
-            },
-        )
-        if resp.status_code != 200:
-            raise UnexpectedResponse.from_response(resp)
-
-    def delete_dynamic_host_configuration(self, dcd_id: str) -> None:
-        """Delete a DCD connection via REST API."""
-        resp = self.delete(f"/objects/dcd/{dcd_id}")
-        if resp.status_code != 204:
-            raise UnexpectedResponse.from_response(resp)
-
     def create_ldap_connection(
         self,
         ldap_id: str,
@@ -1130,3 +1082,54 @@ def show(self, job_id: str) -> dict[str, Any]:
 
         value: dict[str, Any] = response.json()
         return value
+
+
+class DcdAPI(BaseAPI):
+    def create(
+        self,
+        dcd_id: str,
+        title: str,
+        comment: str = "",
+        disabled: bool = False,
+        restrict_source_hosts: list | None = None,
+        interval: int = 60,
+        host_attributes: dict | None = None,
+        delete_hosts: bool = False,
+        discover_on_creation: bool = True,
+        no_deletion_time_after_init: int = 600,
+        max_cache_age: int = 3600,
+        validity_period: int = 60,
+    ) -> None:
+        """Create a DCD connection via REST API."""
+        resp = self.session.post(
+            "/domain-types/dcd/collections/all",
+            json={
+                "dcd_id": dcd_id,
+                "title": title,
+                "comment": comment,
+                "disabled": disabled,
+                "site": self.session.site,
+                "connector_type": "piggyback",
+                "restrict_source_hosts": restrict_source_hosts or [],
+                "interval": interval,
+                "creation_rules": [
+                    {
+                        "folder_path": "/",
+                        "host_attributes": host_attributes or {},
+                        "delete_hosts": delete_hosts,
+                    }
+                ],
+                "discover_on_creation": discover_on_creation,
+                "no_deletion_time_after_init": no_deletion_time_after_init,
+                "max_cache_age": max_cache_age,
+                "validity_period": validity_period,
+            },
+        )
+        if resp.status_code != 200:
+            raise UnexpectedResponse.from_response(resp)
+
+    def delete(self, dcd_id: str) -> None:
+        """Delete a DCD connection via REST API."""
+        resp = self.session.delete(f"/objects/dcd/{dcd_id}")
+        if resp.status_code != 204:
+            raise UnexpectedResponse.from_response(resp)