Update test preparations and idempotence

plugins/modules/cm_kerberos.py

@@ -15,19 +15,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from ansible_collections.cloudera.cluster.plugins.module_utils.cm_utils import (
-    ClouderaManagerModule,
-    ClouderaManagerMutableModule,
-    resolve_parameter_updates,
-)
-from cm_client.rest import ApiException
-from cm_client import (
-    ClouderaManagerResourceApi,
-    ApiConfigList,
-    ApiConfig,
-)
-import re
-
 DOCUMENTATION = r"""
 module: cm_kerberos
 short_description: Manage and configure Kerberos Authentication for CDP
@@ -154,7 +141,6 @@
 """
 
 EXAMPLES = r"""
----
 - name: Enable Kerberos
   cloudera.cluster.cm_kerberos:
     host: example.cloudera.com
@@ -178,7 +164,6 @@
 """
 
 RETURN = r"""
----
 cm_config:
   description:
     - Cloudera Manager Server configurations with Kerberos settings where available.
@@ -250,6 +235,19 @@
       returned: when supported
 """
 
+import re
+
+from ansible_collections.cloudera.cluster.plugins.module_utils.cm_utils import (
+    ClouderaManagerMutableModule,
+    resolve_parameter_updates,
+)
+
+from cm_client import (
+    ClouderaManagerResourceApi,
+    ApiConfigList,
+    ApiConfig,
+)
+
 
 class ClouderaManagerKerberos(ClouderaManagerMutableModule):
     def __init__(self, module):

tests/unit/conftest.py

@@ -26,7 +26,7 @@
 import sys
 import yaml
 
-from collections.abc import Generator
+from collections.abc import Generator, Callable
 from pathlib import Path
 from time import sleep
 
@@ -117,7 +117,7 @@ def fail_json(*args, **kwargs):
 
 
 @pytest.fixture
-def module_args():
+def module_args() -> Callable[[dict], None]:
     """Prepare module arguments"""
 
     def prep_args(args=dict()):
@@ -128,7 +128,7 @@ def prep_args(args=dict()):
 
 
 @pytest.fixture
-def yaml_args():
+def yaml_args() -> Callable[[dict], None]:
     """Prepare module arguments from YAML"""
 
     def prep_args(args: str = ""):
@@ -139,7 +139,7 @@ def prep_args(args: str = ""):
 
 
 @pytest.fixture(scope="session")
-def conn():
+def conn() -> dict:
     conn = dict(username=os.getenv("CM_USERNAME"), password=os.getenv("CM_PASSWORD"))
 
     if os.getenv("CM_HOST", None):

tests/unit/plugins/modules/cm_kerberos/test_cm_kerberos.py

@@ -16,23 +16,118 @@
 
 from __future__ import absolute_import, division, print_function
 
+from cm_client.api_client import ApiClient
+
 __metaclass__ = type
+
 import os
 import logging
 import pytest
+import re
 
 from pathlib import Path
 
+from cm_client.rest import ApiException
+from cm_client import (
+    ClouderaManagerResourceApi,
+    ApiConfigList,
+    ApiConfig,
+)
+
+
 from ansible_collections.cloudera.cluster.plugins.modules import cm_kerberos
 from ansible_collections.cloudera.cluster.tests.unit import (
     AnsibleExitJson,
     AnsibleFailJson,
+    wait_for_command,
 )
 
 LOG = logging.getLogger(__name__)
 
 
-def test_pytest_enable_kerberos(module_args, conn, request):
+@pytest.fixture(scope="function")
+def krb_disabled(cm_api_client, request) -> None:
+    """
+    Disable any existing Kerberos setup on the target Cloudera on Premise deployment.
+
+    This fixture does not restore any prior configurations.
+    """
+
+    cm_api = ClouderaManagerResourceApi(cm_api_client)
+
+    cm_api.delete_credentials_command()
+
+    reset_params = dict(
+        krb_enc_types="aes256-cts",
+        security_realm="HADOOP.COM",
+        kdc_type="MIT KDC",
+        kdc_admin_host="",
+        kdc_host="",
+        krb_auth_enable=False,
+        ad_account_prefix="",
+        ad_kdc_domain="ou=hadoop,DC=hadoop,DC=com",
+        ad_delete_on_regenerate=False,
+        ad_set_encryption_types=False,
+        kdc_account_creation_host_override="",
+        gen_keytab_script="",
+    )
+
+    body = ApiConfigList(
+        items=[ApiConfig(name=k, value=v) for k, v in reset_params.items()]
+    )
+
+    cm_api.update_config(
+        message=f"{Path(request.node.parent.name).stem}::{request.node.name}::cleared",
+        body=body,
+    )
+
+
+# TODO Should parameterize with a marker
+@pytest.fixture(scope="function")
+def krb_freeipa(cm_api_client, request, krb_disabled) -> None:
+    """
+    Reset any existing Kerberos setup on the target Cloudera on Premise deployment.
+
+    This fixture does not restore any prior configurations.
+    """
+
+    cm_api = ClouderaManagerResourceApi(cm_api_client)
+
+    setup_params = dict(
+        krb_enc_types="aes256-cts aes128-cts rc4-hmac",
+        security_realm="HADOOP.COM",
+        kdc_type="Red Hat IPA",
+        kdc_admin_host=os.getenv("KDC_HOST"),
+        kdc_host=os.getenv("KDC_HOST"),
+    )
+
+    body = ApiConfigList(
+        items=[ApiConfig(name=k, value=v) for k, v in setup_params.items()]
+    )
+
+    cm_api.update_config(
+        message=f"{Path(request.node.parent.name).stem}::{request.node.name}::enabled",
+        body=body,
+    )
+
+    cmd = cm_api.import_admin_credentials(
+        username=os.getenv("KDC_ADMIN_USER"),
+        password=os.getenv("KDC_ADMIN_PASSWORD"),
+    )
+
+    try:
+        wait_for_command(
+            api_client=cm_api_client,
+            command=cmd,
+        )
+    except Exception as e:
+        if re.search("user with name", str(e)):
+            LOG.info("Reusing existing KDC user for Cloudera Manager")
+        else:
+            raise e
+
+
+def test_pytest_enable_kerberos(module_args, conn, krb_disabled, request):
 
     if os.getenv("KDC_ADMIN_USER", None):
         conn.update(kdc_admin_user=os.getenv("KDC_ADMIN_USER"))
@@ -60,8 +155,14 @@ def test_pytest_enable_kerberos(module_args, conn, request):
 
     assert e.value.changed == True
 
+    # Idempotency
+    with pytest.raises(AnsibleExitJson) as e:
+        cm_kerberos.main()
 
-def test_enable_invalid_admin_password(module_args, conn, request):
+    assert e.value.changed == False
+
+
+def test_enable_invalid_admin_password(module_args, conn, krb_disabled, request):
 
     if os.getenv("KDC_ADMIN_USER", None):
         conn.update(kdc_admin_user=os.getenv("KDC_ADMIN_USER"))
@@ -85,22 +186,27 @@ def test_enable_invalid_admin_password(module_args, conn, request):
     with pytest.raises(
         AnsibleFailJson,
         match="Error during Import KDC Account Manager Credentials command",
-    ) as e:
+    ):
         cm_kerberos.main()
-        print("At end")
 
 
-def test_pytest_disable_kerberos(module_args, conn):
+def test_pytest_disable_kerberos(module_args, conn, krb_freeipa):
 
     module_args({**conn, "state": "absent"})
 
     with pytest.raises(AnsibleExitJson) as e:
         cm_kerberos.main()
 
-    # assert e.value.changed == True
+    assert e.value.changed == True
 
+    # Idempotency
+    with pytest.raises(AnsibleExitJson) as e:
+        cm_kerberos.main()
+
+    assert e.value.changed == False
 
-def test_force_enable_kerberos(module_args, conn, request):
+
+def test_force_enable_kerberos(module_args, conn, krb_freeipa, request):
 
     if os.getenv("KDC_ADMIN_USER", None):
         conn.update(kdc_admin_user=os.getenv("KDC_ADMIN_USER"))
@@ -112,20 +218,6 @@ def test_force_enable_kerberos(module_args, conn, request):
         conn.update(kdc_admin_host=os.getenv("KDC_HOST"))
         conn.update(kdc_host=os.getenv("KDC_HOST"))
 
-    # Ensure Kerberos is enabled
-    module_args(
-        {
-            **conn,
-            "kdc_type": "Red Hat IPA",
-            "krb_enc_types": ["aes256-cts", "aes128-cts", "rc4-hmac"],
-            "security_realm": "CLDR.INTERNAL",
-        }
-    )
-
-    with pytest.raises(AnsibleExitJson) as e:
-        cm_kerberos.main()
-
-    # Add force to module call
     module_args(
         {
             **conn,
@@ -136,5 +228,8 @@ def test_force_enable_kerberos(module_args, conn, request):
             "message": f"{Path(request.node.parent.name).stem}::{request.node.name}",
         }
     )
+
     with pytest.raises(AnsibleExitJson) as e:
         cm_kerberos.main()
+
+    assert e.value.changed == True