Fix/test containers (#446)

tombosmansibm · web-flow · commit 4e3b51c9cf4c · 2025-07-14T16:46:14.000+02:00
* fix: add test for containers import zip

* fix: add more values for .env

* feature: test for policy attachments

* fix: indentation

* fix: small correction

* fix: update tests

* fix: reorder code

* fix: policy attachents

* documentation: update version

* documentation: update version 2025.7.14.0

* documentation: update python versions

* documentation: update python versions

* documentation: update python versions
diff --git a/README.md b/README.md
@@ -9,7 +9,7 @@ Code for ISVG appliance is brand new (tested with 10.0.1.0 and higher only).
 
 ## Requirements
 
-Python v3.7 and above is required for this package.
+Python v3.9 and above is required for this package.
 
 The following Python Packages are required:
 1. requests - for making REST API calls
diff --git a/docs/changelog.md b/docs/changelog.md
@@ -2,6 +2,12 @@
 
 ## Latest
 
+## 2025.7.14.0
+
+- fix: certificate_databases : python syntax (indentation)
+- fix: policy_attachments.py : idempotency and handle applications correctly
+- build: update python version support (drop 3.7; add 3.11, 3.13)
+
 ## 2025.6.3.0
 
 - feature: base/management_authentication.py - type federation
diff --git a/docs/contributing.md b/docs/contributing.md
@@ -39,6 +39,8 @@ IVIA_ADMIN=admin@local
 IVIA_PW=admin
 IVIA_HOST=<ip address of lmi>
 # IVIA_PORT = 80
+IVIA_SECMASTER=sec_master
+IVIA_SECMASTER_PW=<password for sec_master>
 ````
 
 ## Standards
diff --git a/ibmsecurity/isam/aac/access_control/policy_attachments.py b/ibmsecurity/isam/aac/access_control/policy_attachments.py
@@ -28,11 +28,11 @@ def get(isamAppliance, server, resourceUri, check_mode=False, force=False):
         check_mode=check_mode,
         force=force,
     )
-    resource_id = ret_obj["data"]
+    resource_id = ret_obj.get("data", {})
 
     if resource_id == {}:
         logger.info(
-            f"Resource {server}/{resourceUri} had no match, skipping retrieval."
+            f"Resource {server}{resourceUri} had no match, skipping retrieval."
         )
         return isamAppliance.create_return_object()
     else:
@@ -57,7 +57,7 @@ def get_attachments(isamAppliance, server, resourceUri, check_mode=False, force=
 
     if resource_id == {}:
         logger.info(
-            f"Resource {server}/{resourceUri} had no match, skipping retrieval."
+            f"Resource {server}{resourceUri} had no match, skipping retrieval."
         )
         return isamAppliance.create_return_object()
     else:
@@ -77,10 +77,11 @@ def search(isamAppliance, server, resourceUri, force=False, check_mode=False):
     for obj in ret_obj["data"]:
         if obj["resourceUri"] == resourceUri and obj["server"] == server:
             logger.info(
-                f"Found server/resourceUri {server}/{resourceUri} id: {obj['id']}"
+                f"Found server/resourceUri {server}{resourceUri} id: {obj['id']}"
             )
             return_obj["data"] = obj["id"]
             return_obj["rc"] = 0
+            return return_obj
 
     return return_obj
 
@@ -144,38 +145,55 @@ def config(
       {'name': '<definition name>', 'type': 'definition'}]
     """
     warnings = []
-    if force is False:
-        ret_obj = search(isamAppliance, server, resourceUri)
+    ret_obj = search(isamAppliance, server, resourceUri)
 
-    if force is True or ret_obj["data"] == {}:
-        json_data = {"server": server, "resourceUri": resourceUri}
-        if policyType is not None:
-            if tools.version_compare(isamAppliance.facts["version"], "9.0.6.0") < 0:
-                warnings.append(
-                    f"Appliance at version: {isamAppliance.facts['version']}, policyType: {cache} is not supported. Needs 9.0.6.0 or higher. Ignoring policyType for this call."
-                )
-            else:
-                json_data["type"] = policyType
+    json_data = {"server": server, "resourceUri": resourceUri}
+    if policyType is not None:
+        if tools.version_compare(isamAppliance.facts["version"], "9.0.6.0") < 0:
+            warnings.append(
+                f"Appliance at version: {isamAppliance.facts['version']}, policyType: {policyType} is not supported. Needs 9.0.6.0 or higher. Ignoring policyType for this call."
+            )
+        else:
+            json_data["type"] = policyType
+
+    json_data["policies"] = _convert_policy_name_to_id(isamAppliance, policies)
+    if policyCombiningAlgorithm is not None:
+        json_data["policyCombiningAlgorithm"] = policyCombiningAlgorithm
+    if cache is not None:
+        if tools.version_compare(isamAppliance.facts["version"], "9.0.3.0") < 0:
+            warnings.append(
+                f"Appliance at version: {isamAppliance.facts['version']}, cache: {cache} is not supported. Needs 9.0.3.0 or higher. Ignoring cache for this call."
+            )
+        else:
+            json_data["cache"] = int(cache)
 
-        json_data["policies"] = _convert_policy_name_to_id(isamAppliance, policies)
-        if policyCombiningAlgorithm is not None:
-            json_data["policyCombiningAlgorithm"] = policyCombiningAlgorithm
-        if cache is not None:
-            if tools.version_compare(isamAppliance.facts["version"], "9.0.3.0") < 0:
-                warnings.append(
-                    f"Appliance at version: {isamAppliance.facts['version']}, cache: {cache} is not supported. Needs 9.0.3.0 or higher. Ignoring cache for this call."
-                )
-            else:
-                json_data["cache"] = int(cache)
-        if check_mode is True:
+    if force or ret_obj.get("data", {}) == {}:
+        if check_mode:
             return isamAppliance.create_return_object(changed=True, warnings=warnings)
         else:
             return isamAppliance.invoke_post(
                 "Configure a resource", uri, json_data, warnings=warnings
             )
-
-    return isamAppliance.create_return_object()
-
+    else:
+        # Idempotency check
+        logger.debug(f"Idempotency check for {server}{resourceUri}")
+        curConfig = get(isamAppliance, server, resourceUri)
+        curConfig = curConfig.get("data", {})
+        # Need to get rid of lastmodified and name in each policy
+        _policies = list(map(lambda item: item.pop('lastmodified', None), curConfig.get("policies", [])))
+        _policies = list(map(lambda item: item.pop('name', None), curConfig.get("policies", [])))
+        if tools.json_equals(curConfig, json_data):
+            # No updates needed
+            return isamAppliance.create_return_object()
+        else:
+            if check_mode:
+                return isamAppliance.create_return_object(changed=True, warnings=warnings)
+            else:
+                # delete and add again
+                delete(isamAppliance, server, resourceUri)
+                return isamAppliance.invoke_post(
+                    "Reconfigure a resource", uri, json_data, warnings=warnings
+                )
 
 def update(
     isamAppliance,
@@ -306,17 +324,22 @@ def publish(isamAppliance, server, resourceUri, check_mode=False, force=False):
     """
     ret_obj = get(isamAppliance, server, resourceUri)
 
-    if force or (
-        ret_obj["data"] != {} and (
-            ret_obj["data"]["deployrequired"] or not ret_obj["data"]["deployed"]
-        ) and ret_obj["data"]["policies"] != []
-    ):
+    deploy_required = False
+    if ret_obj.get("data", {}).get("deployrequired", False) or not ret_obj.get("data", {}).get("deployed", False):
+        deploy_required = True
+
+    logger.debug(f"\n\nDeploy required: {deploy_required}\n\n")
+
+    resource_id = ret_obj.get('data', {}).get('id', None)
+    logger.debug(f"\n\nID: {resource_id}\n\n")
+    # if (force or deploy_required) and ret_obj.get("data", {}).get("policies", []) != []:
+    if (force or deploy_required) and resource_id is not None:
         if check_mode:
             return isamAppliance.create_return_object(changed=True)
         else:
             return isamAppliance.invoke_put(
                 "Publish the policy attachments for a resource",
-                f"{uri}/deployment/{ret_obj['data']['id']}",
+                f"{uri}/deployment/{resource_id}",
                 {},
             )
 
@@ -333,7 +356,7 @@ def publish_list(isamAppliance, attachments, check_mode=False, force=False):
     id_list = []
     for attach in attachments:
         ret_obj = get(isamAppliance, attach["server"], attach["resourceUri"])
-        if force is True or ret_obj["data"]["deployrequired"] is True:
+        if force or ret_obj["data"]["deployrequired"]:
             id_list.append(ret_obj["data"]["id"])
     logger.debug(f"Attachments: {id_list}")
 
diff --git a/ibmsecurity/isam/base/container_ext/volume.py b/ibmsecurity/isam/base/container_ext/volume.py
@@ -66,12 +66,13 @@ def import_zip(isamAppliance, filename, volume_name=None, volume_id=None, check_
     import os.path
     logger.debug(f"\n\nTrying to import {filename}\n\n")
 
+    if volume_id is None and volume_name is None:
+        return isamAppliance.create_return_object(warnings=["Volume name or volume id need to be specified"])
+
     if volume_id is None:
         volume_id = search(isamAppliance, volume_name)
         volume_id = volume_id.get('data', None)
-
-    if volume_id is None and volume_name is None:
-        return isamAppliance.create_return_object(warnings=["Volume name or volume id need to be specified"])
+        logger.debug(f"\n\nVolume id is now set to {volume_id}\n\n")
 
     if force or os.path.exists(filename):
         return isamAppliance.invoke_put_files(
@@ -109,6 +110,7 @@ def search(isamAppliance, volume_name, check_mode=False, force=False):
         if obj["name"] == volume_name:
             return_obj["data"] = obj["id"]
             return_obj["rc"] = 0
+            logger.debug(f"\n\nFound {obj['id']} for {volume_name}\n\n")
             break
 
     return return_obj
diff --git a/ibmsecurity/isam/base/ssl_certificates/certificate_databases.py b/ibmsecurity/isam/base/ssl_certificates/certificate_databases.py
@@ -83,6 +83,7 @@ def create(isamAppliance, kdb_name, type='kdb',
     """
     warnings = []
     if force or not _check(isamAppliance, kdb_name):
+        logger.debug("Creating new keystore")
         if check_mode:
             return isamAppliance.create_return_object(changed=True)
         else:
@@ -126,16 +127,22 @@ def create(isamAppliance, kdb_name, type='kdb',
                    json_data.pop("rfs_port", None)
                    json_data.pop("rfs_auth", None)
 
-               retObj = isamAppliance.invoke_post( f"Creating certificate database {kdb_name}",
+            retObj = isamAppliance.invoke_post( f"Creating certificate database {kdb_name}",
                   "/isam/ssl_certificates",
                   json_data,
                   warnings=warnings,
                   ignore_error=True
                )
-               if retObj.get("rc", 0) == 400:
+            if retObj.get("rc", 0) == 400:
                    warnings.append(f"Invalid type (you need to install an extension to support network hsm {type})")
                    return isamAppliance.create_return_object(warnings=warnings)
-               else:
+            elif retObj.get("rc", 0) > 400 and retObj.get("rc", 0) < 500:
+                   warnings.append(f"HTTP error !")
+                   return isamAppliance.create_return_object(warnings=warnings)
+            elif retObj.get("rc", 0) >= 500 and retObj.get("rc", 0) < 600:
+                   warnings.append(f"Server error !")
+                   return isamAppliance.create_return_object(warnings=warnings)
+            else:
                    return retObj
 
     return isamAppliance.create_return_object()
@@ -243,7 +250,7 @@ def set(isamAppliance, cert_id, description=None, type="kdb", check_mode=False,
     warnings = []
     desc_match = True  # This will remain True even when cert db is not found!
 
-    if type == "kdb":
+    if type in ("p12", "kdb"):
         if not force:
             if description is None:
                 desc_match = True
@@ -302,11 +309,12 @@ def _check(isamAppliance, id):
     Check if certificate database already exists
     """
     ret_obj = get_all(isamAppliance)
-
+    logger.debug(f"\nCHECKING IF {id} EXISTS")
     for certdb in ret_obj['data']:
+        logger.info("Cert databases id " + certdb['id'])
         if certdb['id'] == id:
             return True
-
+    logger.debug(f"\n{id} DOES NOT EXIST\n")
     return False
 
 
diff --git a/pyproject.toml b/pyproject.toml
@@ -7,13 +7,13 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ibmsecurity"
-version = "2025.6.3.0"
+version = "2025.7.14.0"
 authors = [
   { name="IBM", email="secorch@wwpdl.vnet.ibm.com" },
 ]
 description = "Idempotent functions for IBM Security Appliance REST APIs"
 readme = "README.md"
-requires-python = ">=3.7"
+requires-python = ">=3.9"
 license = "Apache-2.0"
 classifiers = [
     "Programming Language :: Python :: 3",
diff --git a/setup.py b/setup.py
@@ -5,16 +5,19 @@
     packages=find_packages(),
     # Date of release used for version - please be sure to use YYYY.MM.DD.seq#, MM and DD should be two digits e.g. 2017.02.05.0
     # seq# will be zero unless there are multiple release on a given day - then increment by one for additional release for that date
-    version="2025.6.3.0",
+    version="2025.7.14.0",
     description="Idempotent functions for IBM Security Appliance REST APIs",
     author="IBM",
     author_email="secorch@wwpdl.vnet.ibm.com",
     url="",
     classifiers=[
         "Programming Language :: Python",
-        "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
         "Programming Language :: Python :: 3.9",
+        "Programming Language :: Python :: 3.10",
+        "Programming Language :: Python :: 3.11",
+        "Programming Language :: Python :: 3.12",
+        "Programming Language :: Python :: 3.13",
         "License :: OSI Approved :: Apache Software License",
         "Operating System :: OS Independent",
         "Development Status :: 5 - Production/Stable",
diff --git a/test/files/oidcop.zip b/test/files/oidcop.zip
diff --git a/test/test_aac_access_control.py b/test/test_aac_access_control.py
diff --git a/test/test_base_certificate_databases.py b/test/test_base_certificate_databases.py
diff --git a/test/test_base_containers.py b/test/test_base_containers.py
