fix: Fixed ZPA Microtenant Update and ZIA Static IP (#174)

CHANGELOG.md

@@ -1,5 +1,16 @@
 # Zscaler Python SDK Changelog
 
+## 0.9.4 (October, 3 2024)
+
+### Notes
+
+- Python Versions: **v3.8, v3.9, v3.10, v3.11**
+
+### Bug Fixes
+
+* Fixed ZPA Microtenant Update method response processing.  ([#173](https://github.com/zscaler/zscaler-sdk-python/pull/173))
+* Fixed ZIA `check_static_ip` text parsing  ([#173](https://github.com/zscaler/zscaler-sdk-python/pull/173))
+
 ## 0.9.3 (September, 16 2024)
 
 ### Notes

docsrc/conf.py

@@ -28,9 +28,9 @@
 html_title = ""
 
 # The short X.Y version
-version = "0.9.3"
+version = "0.9.4"
 # The full version, including alpha/beta/rc tags
-release = "0.9.3"
+release = "0.9.4"
 
 # -- General configuration ---------------------------------------------------
 

docsrc/dev_requirements.txt

@@ -2,7 +2,7 @@ furo==2024.7.18
 pre-commit==3.8.0
 pytest==8.3.3
 python-box==7.2.0
-restfly==1.4.7
+restfly==1.5.0
 requests==2.32.3
 responses==0.25.3
 sphinx==7.4.7

docsrc/zs/guides/release_notes.rst

@@ -6,6 +6,19 @@ Release Notes
 Zscaler Python SDK Changelog
 ----------------------------
 
+0.9.4 (October, 3 2024)
+------------------------
+
+Notes
+^^^^^
+
+- Python Versions: **v3.8, v3.9, v3.10, v3.11**
+
+### Bug Fixes
+
+* Fixed ZPA Microtenant Update method response processing (`173 <https://github.com/zscaler/zscaler-sdk-python/pull/173>`_)
+* Fixed ZIA `check_static_ip` text parsing (`173 <https://github.com/zscaler/zscaler-sdk-python/pull/173>`_)
+
 0.9.3 (September, 16 2024)
 ---------------------------
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "zscaler-sdk-python"
-version = "0.9.3"
+version = "0.9.4"
 description = "Official Python SDK for the Zscaler Products (Beta)"
 authors = ["Zscaler, Inc. <[email protected]>"]
 license = "MIT"
@@ -37,13 +37,13 @@ python-box = ">=7.2.0"
 python-dateutil = "*"
 requests = ">=2.32.3"
 responses = ">=0.25.3"
-restfly = "*"
+restfly = ">=1.5.0"
 six = "*"
 flatdict = "*"
 pyyaml = "*"
 xmltodict = "*"
 yarl = "*"
-pycryptodomex = "*"
+pycryptodomex = ">=3.20.0"
 aenum = "*"
 pydash = ">=8.0.3"
 flake8 = "*"

tests/integration/zpa/test_app_connector_groups.py

@@ -110,6 +110,7 @@ def test_app_connector_group(self, fs):
         # Assert that no errors occurred during the test
         assert len(errors) == 0, f"Errors occurred during the app connector group lifecycle test: {errors}"
 
+
 class TestCustomerVersionProfile:
     def test_version_profiles(self, fs):
         client = MockZPAClient(fs)
@@ -130,11 +131,10 @@ def test_version_profiles(self, fs):
         for profile_name in profile_names:
             try:
                 profiles_by_name = client.connectors.list_version_profiles(search=profile_name)
-                found_profiles = [profile for profile in profiles_by_name if profile.get('name') == profile_name]
+                found_profiles = [profile for profile in profiles_by_name if profile.get("name") == profile_name]
                 assert found_profiles, f"No profiles found with the name {profile_name}"
             except AssertionError as e:
                 errors.append(f"Error retrieving profile by name '{profile_name}': {str(e)}")
 
         # Assert that no errors occurred during the test
         assert not errors, f"Errors occurred during the tests: {errors}"
-

zscaler/__init__.py

@@ -29,7 +29,7 @@
 __contributors__ = [
     "William Guilherme",
 ]
-__version__ = "0.9.3"
+__version__ = "0.9.4"
 
 from zscaler.zdx import ZDXClientHelper  # noqa
 from zscaler.zia import ZIAClientHelper  # noqa

zscaler/zia/__init__.py

@@ -382,13 +382,16 @@ def put(self, path, json=None, params=None):
         formatted_resp = format_json_response(resp, box_attrs=dict())
         return formatted_resp
 
-    def post(self, path, json=None, params=None, data=None, headers=None):
+    def post(self, path, json=None, params=None, data=None, headers=None, parse_json=True):
         should_wait, delay = self.rate_limiter.wait("POST")
         if should_wait:
             time.sleep(delay)
         resp = self.send("POST", path, json, params, data=data, headers=headers)
-        formatted_resp = format_json_response(resp, box_attrs=dict())
-        return formatted_resp
+        if parse_json:
+            formatted_resp = format_json_response(resp, box_attrs=dict())
+            return formatted_resp
+        else:
+            return resp
 
     def delete(self, path, json=None, params=None):
         should_wait, delay = self.rate_limiter.wait("DELETE")

zscaler/zia/locations.py

@@ -43,8 +43,6 @@ def list_locations(self, **kwargs) -> BoxList:
                 Specifies the page size. The default size is 100, but the maximum size is 1000.
             **search (str, optional):
                 The search string used to partially match against a location's name and port attributes.
-            **xff_enabled (bool, optional):
-                Filter based on whether the Enforce XFF Forwarding setting is enabled or disabled for a location.
 
         Returns:
             :obj:`BoxList`: List of configured locations.
@@ -188,7 +186,8 @@ def add_location(self, name: str, **kwargs) -> Box:
             description (str, optional):
                 Additional notes or information regarding the location or sub-location. The description cannot
                 exceed 1024 characters.
-
+            static_location_groups (list): IDs for static location groups.
+            
         Returns:
             :obj:`Box`: The newly created location resource record
 
@@ -693,5 +692,4 @@ def list_cities_by_name(self, **kwargs) -> BoxList:
 
         """
         return BoxList(Iterator(self.rest, "region/search", **kwargs))
-        # data, _ = self.rest.get_paginated_data(path="region/search", params=kwargs)
-        # return BoxList([Box(city) for city in data])
+

zscaler/zia/traffic.py

@@ -523,13 +523,12 @@ def check_static_ip(self, ip_address: str) -> bool:
         payload = {
             "ipAddress": ip_address,
         }
-        response = self.rest.post("staticIP/validate", json=payload)
+        response = self.rest.post("staticIP/validate", json=payload, parse_json=False)
 
         # Check if the status code is 200 and the response body text is "SUCCESS"
         if response.status_code == 200 and response.text.strip().upper() == "SUCCESS":
             return True
         else:
-            # Optionally, you could log response.text or response.status_code here for debugging
             return False
 
     def update_static_ip(self, static_ip_id: str, **kwargs) -> Box:

zscaler/zpa/connectors.py

@@ -590,4 +590,4 @@ def list_version_profiles(self, **kwargs) -> BoxList:
 
         """
         list, _ = self.rest.get_paginated_data(path="/visible/versionProfiles", **kwargs, api_version="v1")
-        return list
+        return list

zscaler/zpa/microtenants.py

@@ -60,10 +60,14 @@ def get_microtenant(self, microtenant_id: str) -> Box:
 
         """
         response = self.rest.get("/microtenants/%s" % (microtenant_id))
+
+        # If the response is not a Box (indicating an error), return None or handle the error
         if isinstance(response, Response):
             status_code = response.status_code
             if status_code != 200:
-                return None
+                raise Exception(f"Failed to retrieve microtenant: {status_code} - {response.text}")
+
+        # Otherwise, return the formatted Box object
         return response
 
     def get_microtenant_summary(self) -> Box:
@@ -189,17 +193,44 @@ def update_microtenant(self, microtenant_id: str, **kwargs) -> Box:
             ... )
 
         """
-        # Set payload to value of existing record
-        payload = {snake_to_camel(k): v for k, v in self.get_microtenant(microtenant_id).items()}
-
-        # Add optional parameters to payload
+        # Retrieve existing microtenant data
+        existing_microtenant_response = self.get_microtenant(microtenant_id)
+
+        # Handle Response object (in case of error)
+        if isinstance(existing_microtenant_response, Response):
+            if existing_microtenant_response.status_code != 200:
+                raise Exception(f"Failed to retrieve microtenant: {existing_microtenant_response.status_code}")
+            # If there's content, parse the response as JSON
+            if existing_microtenant_response.text.strip():
+                existing_microtenant = existing_microtenant_response.json()
+            else:
+                existing_microtenant = {}  # Treat as empty, no error
+        else:
+            existing_microtenant = existing_microtenant_response
+
+        # Prepare the payload
+        payload = {snake_to_camel(k): v for k, v in existing_microtenant.items()}
+
+        # Add optional parameters from kwargs
         for key, value in kwargs.items():
             payload[snake_to_camel(key)] = value
 
-        resp = self.rest.put(f"microtenants/{microtenant_id}", json=payload).status_code
-        if not isinstance(resp, Response):
+        # Send the PUT request to update the microtenant
+        resp = self.rest.put(f"microtenants/{microtenant_id}", json=payload)
+
+        # Handle 204 No Content as success
+        if resp.status_code == 204:
+            print(f"Update successful for microtenant {microtenant_id} (204 No Content).")
+            return self.get_microtenant(microtenant_id)
+
+        # Handle 200 OK as success
+        if resp.status_code == 200:
+            print(f"Update successful for microtenant {microtenant_id} (200 OK).")
             return self.get_microtenant(microtenant_id)
 
+        # Raise exception for any other status code
+        raise Exception(f"Failed to update microtenant: {resp.status_code}, {resp.text}")
+
     def delete_microtenant(self, microtenant_id: str) -> int:
         """
         Delete the specified microtenant.