fix(review): address PR #168 Copilot review comments

- assets.py replace(): use per-request header copy (no Content-Type) instead
  of mutating shared client headers dict; wrap file open in context manager
  to prevent FD leak
- assets.py update(): use per-request header copy with explicit
  Content-Type: application/json, robust against prior upload/replace calls
- capture.py retry: close old FD before reopening on transient retry;
  use tuple(val[2:]) to handle both 2-tuple and 3-tuple file specs without
  crashing on missing index
- setup.py: remove references to non-existent .env.example in docstring
  and runtime error message
- test_assets_unit.py: fix scan-status unit tests to use correct query param
  name (include_asset_scan_status) and lowercase string value (true),
  matching the integration suite and CMA API wire format

Author: aniket-shikhare-cstk

.talismanrc

@@ -406,3 +406,7 @@ fileignoreconfig:
 - filename: tests/mock/assets/test_assets_mock.py
   checksum: 12c9091bb88c0c12712f046b29fb4a9dce3b95ecc45f4ea46bbc3fd4529742a0
 version: "1.0"
+fileignoreconfig:
+- filename: tests/integration/framework/capture.py
+  checksum: c9680c4ee3e1def0765fd89767fa1383aee5d195222fa021a19618607373047c
+version: "1.0"

contentstack_management/assets/assets.py

@@ -174,12 +174,11 @@ def replace(self, file_path):
         """
 
         url = f"assets/{self.asset_uid}"
-        # Let requests build the multipart body and set Content-Type WITH a boundary.
-        # Setting a bare "multipart/form-data" (no boundary) makes the API reject the
-        # upload with 422 "Please send a valid multipart/form-data payload".
-        self.client.headers.pop("Content-Type", None)
-        files = {"asset": open(f"{file_path}",'rb')}
-        return self.client.put(url, headers = self.client.headers, params = self.params, files = files)
+        # Omit Content-Type so requests can set it with the correct multipart boundary.
+        # Use a per-request header copy to avoid mutating the shared client headers dict.
+        request_headers = {k: v for k, v in self.client.headers.items() if k.lower() != "content-type"}
+        with open(file_path, 'rb') as fh:
+            return self.client.put(url, headers=request_headers, params=self.params, files={"asset": fh})
 
     def generate(self, data):
         """
@@ -410,10 +409,11 @@ def update(self, data):
         if self.asset_uid is None or '':
             raise Exception(ASSET_UID_REQUIRED)
         url = f"assets/{self.asset_uid}"
-        # Updating an asset's title/description sends a JSON body, so it must use
-        # application/json. Forcing multipart/form-data here makes the API reject
-        # the request with 422 "Please send a valid multipart/form-data payload".
-        return self.client.put(url, headers = self.client.headers, params = self.params, data = data)
+        # Use a per-request header copy with Content-Type explicitly set to application/json
+        # so this call is not affected by whatever a preceding upload/replace left in
+        # the shared headers dict.
+        request_headers = {**self.client.headers, "Content-Type": "application/json"}
+        return self.client.put(url, headers=request_headers, params=self.params, data=data)
 
     def publish(self, data):
         """

tests/integration/framework/capture.py

@@ -136,9 +136,23 @@ def _request_with_retry(method, url, kwargs):
             files = kwargs.get("files")
             if files:
                 for key, val in list(files.items()):
-                    name = getattr(val[1] if isinstance(val, (tuple, list)) else val, "name", None)
-                    if name and os.path.exists(name):
-                        kwargs["files"][key] = (val[0], open(name, "rb"), val[2]) if isinstance(val, (tuple, list)) else open(name, "rb")
+                    if isinstance(val, (tuple, list)):
+                        fh = val[1]
+                        name = getattr(fh, "name", None)
+                        if name and os.path.exists(name):
+                            try:
+                                fh.close()
+                            except Exception:
+                                pass
+                            kwargs["files"][key] = (val[0], open(name, "rb")) + tuple(val[2:])
+                    else:
+                        name = getattr(val, "name", None)
+                        if name and os.path.exists(name):
+                            try:
+                                val.close()
+                            except Exception:
+                                pass
+                            kwargs["files"][key] = open(name, "rb")
     raise last_exc
 
 

tests/integration/framework/setup.py

@@ -11,7 +11,7 @@
 All calls go through the SDK so the request-capture layer records them. Uses the
 SDK exactly as a customer would — exercising the real artifact under test.
 
-Env vars (see .env.example):
+Env vars (set in tests/integration/.env):
   Required: EMAIL, PASSWORD, HOST, ORGANIZATION
   Optional: MFA_SECRET, DELETE_DYNAMIC_RESOURCES (default 'true')
 """
@@ -47,7 +47,7 @@ def validate_environment() -> None:
     if missing:
         raise RuntimeError(
             f"Missing required environment variables: {', '.join(missing)}. "
-            f"See tests/integration/.env.example."
+            f"Create tests/integration/.env with these variables set."
         )
 
 

tests/unit/assets/test_assets_unit.py

@@ -261,49 +261,49 @@ def test_delete_folder(self):
 
     def test_fetch_includes_scan_status_param(self):
         asset = self.client.stack(api_key).assets(asset_uid)
-        asset.add_param("_asset_scan_status", True)
+        asset.add_param("include_asset_scan_status", "true")
         response = asset.fetch()
-        self.assertIn("_asset_scan_status=True", response.request.url)
+        self.assertIn("include_asset_scan_status=true", response.request.url)
         self.assertEqual(response.request.method, "GET")
 
     def test_find_includes_scan_status_param(self):
         asset = self.client.stack(api_key).assets()
-        asset.add_param("_asset_scan_status", True)
+        asset.add_param("include_asset_scan_status", "true")
         response = asset.find()
-        self.assertIn("_asset_scan_status=True", response.request.url)
+        self.assertIn("include_asset_scan_status=true", response.request.url)
         self.assertEqual(response.request.method, "GET")
 
     def test_upload_includes_scan_status_param(self):
         file_path = "tests/resources/mock_assets/chaat.jpeg"
         asset = self.client.stack(api_key).assets()
-        asset.add_param("_asset_scan_status", True)
+        asset.add_param("include_asset_scan_status", "true")
         response = asset.upload(file_path)
-        self.assertIn("_asset_scan_status=True", response.request.url)
+        self.assertIn("include_asset_scan_status=true", response.request.url)
         self.assertEqual(response.request.method, "POST")
 
     def test_fetch_without_scan_status_param_field_absent(self):
         response = self.client.stack(api_key).assets(asset_uid).fetch()
-        self.assertNotIn("_asset_scan_status", response.request.url)
+        self.assertNotIn("include_asset_scan_status", response.request.url)
         self.assertEqual(response.request.method, "GET")
 
     def test_find_without_scan_status_param_field_absent(self):
         response = self.client.stack(api_key).assets().find()
-        self.assertNotIn("_asset_scan_status", response.request.url)
+        self.assertNotIn("include_asset_scan_status", response.request.url)
         self.assertEqual(response.request.method, "GET")
 
     def test_upload_without_scan_status_param_field_absent(self):
         file_path = "tests/resources/mock_assets/chaat.jpeg"
         response = self.client.stack(api_key).assets().upload(file_path)
-        self.assertNotIn("_asset_scan_status", response.request.url)
+        self.assertNotIn("include_asset_scan_status", response.request.url)
         self.assertEqual(response.request.method, "POST")
 
     def test_scan_status_param_coexists_with_other_params(self):
         asset = self.client.stack(api_key).assets(asset_uid)
         asset.add_param("locale", "en-us")
-        asset.add_param("_asset_scan_status", True)
+        asset.add_param("include_asset_scan_status", "true")
         response = asset.fetch()
         self.assertIn("locale=en-us", response.request.url)
-        self.assertIn("_asset_scan_status=True", response.request.url)
+        self.assertIn("include_asset_scan_status=true", response.request.url)
         self.assertEqual(response.request.method, "GET")
 
     def test_publish_includes_api_version_header(self):