test(iast): give iast_packages fixtures private filesystem state (#19663)

APPSEC-69623

These flask endpoints are hit by both `test_packages_patched` and `test_packages_not_patched`, which run in different xdist workers sharing a cwd and a HOME. A fixed path lets one request's cleanup delete state another request is still using.

**Fixes the flake** (7 quarantined tests, top impact on main). `pkg_virtualenv` built the env at `os.getcwd()/<name>`, so a concurrent `rmtree` broke creation: `[Errno 2] No such file or directory: .../myenv/lib/python3.12/site-packages/pip/_vendor/pygments/formatters`. Now a per-request `TemporaryDirectory`.

**Same bug, not yet caught.** `pkg_importlib_resources`, `pkg_platformdirs`, `pkg_openpyxl`, `pkg_pandas`, `pkg_zipp`, `pkg_iniconfig` shared `data/`, `~/.local/share/<app>`, `example.xlsx`, `example.csv`, `example.zip`, `example.ini`. None have failed in 30 days — they write in milliseconds, where a venv takes ~7s — but the race is the same.

Notes:
- `pkg_importlib_resources` stays under the cwd so `resources.files()` can resolve it, hence the cache invalidation and the `sys.modules` cleanup.
- `pkg_platformdirs` can't be relocated (the expected output embeds the path), so it uses `exist_ok` and tolerates a concurrent removal.
- `pkg_zipp` now reports member names instead of full paths, so the assertion no longer depends on the archive's location; expected string updated.

Reviewed by codex; findings applied. Verified via attempt-to-fix on the 7 keys.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: christophe.papazian <christophe.papazian@datadoghq.com>

Author: christophe-papazian

tests/appsec/iast_packages/packages/pkg_importlib_resources.py

@@ -4,8 +4,11 @@
 https://pypi.org/project/importlib-resources/
 """
 
+import importlib
 import os
 import shutil
+import sys
+import uuid
 
 from flask import Blueprint
 from flask import request
@@ -25,15 +28,16 @@ def pkg_importlib_resources_view():
     try:
         resource_name = request.args.get("package_param", "default.txt")
 
-        # Ensure the data directory and file exist
-        data_dir = "data"
+        # Unique per request: xdist workers share a cwd, so under a fixed name one request's
+        # cleanup deletes the file another is still reading. It has to stay under the cwd for
+        # resources.files() to resolve it as a namespace package, hence the cache invalidation.
+        data_dir = f"data_{uuid.uuid4().hex}"
         file_path = os.path.join(data_dir, resource_name)
 
-        if not os.path.exists(data_dir):
-            os.makedirs(data_dir)
-        if not os.path.exists(file_path):
-            with open(file_path, "w") as f:
-                f.write("This is the default content of the file.")
+        os.makedirs(data_dir, exist_ok=True)
+        with open(file_path, "w") as f:
+            f.write("This is the default content of the file.")
+        importlib.invalidate_caches()
 
         try:
             content = resources.files(data_dir).joinpath(resource_name).read_text()
@@ -45,10 +49,9 @@ def pkg_importlib_resources_view():
     except Exception as e:
         response.result1 = f"Error: {str(e)}"
     finally:
-        if data_dir and os.path.exists(data_dir):
-            try:
-                shutil.rmtree(data_dir)
-            except Exception:
-                pass
+        if data_dir:
+            # resources.files() leaves a namespace module behind for each unique name.
+            sys.modules.pop(data_dir, None)
+            shutil.rmtree(data_dir, ignore_errors=True)
 
     return response.json()

tests/appsec/iast_packages/packages/pkg_iniconfig.py

@@ -5,6 +5,7 @@
 """
 
 import os
+import tempfile
 
 from flask import Blueprint
 from flask import jsonify
@@ -25,20 +26,21 @@ def pkg_iniconfig_view():
     try:
         value = request.args.get("package_param", "test1234")
         ini_content = f"[section]\nkey={value}"
-        ini_path = "example.ini"
 
-        try:
-            with open(ini_path, "w") as f:
-                f.write(ini_content)
+        # Private directory per request: xdist workers share a cwd, so under a fixed name one
+        # request's cleanup removes the file another request is still reading.
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            ini_path = os.path.join(tmp_dir, "example.ini")
 
-            config = iniconfig.IniConfig(ini_path)
-            parsed_data = {section.name: list(section.items()) for section in config}
-            result_output = f"Parsed INI data: {parsed_data}"
+            try:
+                with open(ini_path, "w") as f:
+                    f.write(ini_content)
 
-            if os.path.exists(ini_path):
-                os.remove(ini_path)
-        except Exception as e:
-            result_output = f"Error: {str(e)}"
+                config = iniconfig.IniConfig(ini_path)
+                parsed_data = {section.name: list(section.items()) for section in config}
+                result_output = f"Parsed INI data: {parsed_data}"
+            except Exception as e:
+                result_output = f"Error: {str(e)}"
 
         response.result1 = result_output
     except Exception as e:

tests/appsec/iast_packages/packages/pkg_openpyxl.py

@@ -5,6 +5,7 @@
 """
 
 import os
+import tempfile
 
 from flask import Blueprint
 from flask import request
@@ -31,17 +32,16 @@ def pkg_openpyxl_view():
         # Write the parameter value to the first cell
         ws["A1"] = param_value
 
-        # Save the workbook to a file
-        file_path = "example.xlsx"
-        wb.save(file_path)
+        # Private directory per request: xdist workers share a cwd, so under a fixed name one
+        # request's cleanup removes the file another request is still reading.
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            file_path = os.path.join(tmp_dir, "example.xlsx")
+            wb.save(file_path)
 
-        # Read back the value from the file to ensure it was written correctly
-        wb_read = openpyxl.load_workbook(file_path)
-        ws_read = wb_read.active
-        read_value = ws_read["A1"].value
-
-        # Clean up the created file
-        os.remove(file_path)
+            # Read back the value from the file to ensure it was written correctly
+            wb_read = openpyxl.load_workbook(file_path)
+            ws_read = wb_read.active
+            read_value = ws_read["A1"].value
 
         result_output = f"Written value: {read_value}"
 

tests/appsec/iast_packages/packages/pkg_pandas.py

@@ -5,6 +5,7 @@
 """
 
 import os
+import tempfile
 
 from flask import Blueprint
 from flask import request
@@ -27,16 +28,15 @@ def pkg_pandas_view():
         # Create a DataFrame
         df = pd.DataFrame({"Column1": [param_value]})
 
-        # Save the DataFrame to a CSV file
-        file_path = "example.csv"
-        df.to_csv(file_path, index=False)
+        # Private directory per request: xdist workers share a cwd, so under a fixed name one
+        # request's cleanup removes the file another request is still reading.
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            file_path = os.path.join(tmp_dir, "example.csv")
+            df.to_csv(file_path, index=False)
 
-        # Read back the value from the file to ensure it was written correctly
-        df_read = pd.read_csv(file_path)
-        read_value = df_read.iloc[0]["Column1"]
-
-        # Clean up the created file
-        os.remove(file_path)
+            # Read back the value from the file to ensure it was written correctly
+            df_read = pd.read_csv(file_path)
+            read_value = df_read.iloc[0]["Column1"]
 
         result_output = f"Written value: {read_value}"
 

tests/appsec/iast_packages/packages/pkg_platformdirs.py

@@ -4,6 +4,7 @@
 https://pypi.org/project/platformdirs/
 """
 
+import contextlib
 import os
 
 from flask import Blueprint
@@ -27,14 +28,14 @@ def pkg_platformdirs_view():
         # Get the user data directory for the application
         data_dir = user_data_dir(app_name)
 
-        # Create the directory if it doesn't exist
-        if not os.path.exists(data_dir):
-            os.makedirs(data_dir)
+        # The path derives from the app name, so every xdist worker shares it and two requests
+        # can create and remove it concurrently.
+        os.makedirs(data_dir, exist_ok=True)
 
         result_output = f"User data directory for {app_name}: {data_dir}"
 
-        # Clean up the created directory
-        if os.path.exists(data_dir):
+        # Clean up the created directory; another worker may have removed it already.
+        with contextlib.suppress(FileNotFoundError):
             os.rmdir(data_dir)
 
         response.result1 = result_output

tests/appsec/iast_packages/packages/pkg_virtualenv.py

@@ -5,7 +5,7 @@
 """
 
 import os
-import shutil
+import tempfile
 
 from flask import Blueprint
 from flask import request
@@ -24,24 +24,23 @@ def pkg_virtualenv_view():
 
     try:
         env_name = request.args.get("package_param", "default-env")
-        env_path = os.path.join(os.getcwd(), env_name)
-
-        try:
-            # Create a virtual environment
-            virtualenv.cli_run([env_path])
-            result_output = "Virtual environment created at replaced_path"
-
-            # Optionally, list the contents of the virtual environment's bin/Scripts directory
-            _ = os.path.join(env_path, "bin" if os.name != "nt" else "Scripts")
-            result_output += "\nContents of replaced_path: replaced_contents"
-
-        except Exception as e:
-            result_output = f"Error: {str(e)}"
-
-        finally:
-            # Clean up the created virtual environment
-            if os.path.exists(env_path):
-                shutil.rmtree(env_path)
+        # Private directory per request: xdist workers share a cwd, so under a fixed path one
+        # request's cleanup deletes the environment another request is still creating.
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            # basename so the env cannot land outside tmp_dir and escape its cleanup.
+            env_path = os.path.join(tmp_dir, os.path.basename(env_name) or "env")
+
+            try:
+                # Create a virtual environment
+                virtualenv.cli_run([env_path])
+                result_output = "Virtual environment created at replaced_path"
+
+                # Optionally, list the contents of the virtual environment's bin/Scripts directory
+                _ = os.path.join(env_path, "bin" if os.name != "nt" else "Scripts")
+                result_output += "\nContents of replaced_path: replaced_contents"
+
+            except Exception as e:
+                result_output = f"Error: {str(e)}"
 
         response.result1 = result_output
     except Exception as e:

tests/appsec/iast_packages/packages/pkg_zipp.py

@@ -5,6 +5,7 @@
 """
 
 import os
+import tempfile
 import zipfile
 
 from flask import Blueprint
@@ -25,21 +26,23 @@ def pkg_zipp_view():
     try:
         zip_param = request.args.get("package_param", "example.zip")
 
-        try:
-            # Create an example zip file
-            with zipfile.ZipFile(zip_param, "w") as zip_file:
-                zip_file.writestr("example.txt", "This is an example file.")
-
-            # Read the contents of the zip file using zipp
-            zip_path = zipp.Path(zip_param)
-            contents = [str(file) for file in zip_path.iterdir()]
-            result_output = f"Contents of {zip_param}: {contents}"
-
-            # Clean up the created zip file
-            if os.path.exists(zip_param):
-                os.remove(zip_param)
-        except Exception as e:
-            result_output = f"Error: {str(e)}"
+        # Private directory per request: xdist workers share a cwd, so under a fixed name one
+        # request's cleanup removes the archive another request is still reading.
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            zip_file_path = os.path.join(tmp_dir, os.path.basename(zip_param))
+
+            try:
+                # Create an example zip file
+                with zipfile.ZipFile(zip_file_path, "w") as zip_file:
+                    zip_file.writestr("example.txt", "This is an example file.")
+
+                # Read the contents of the zip file using zipp. Report member names rather than
+                # full paths so the result does not depend on where the archive lives.
+                zip_path = zipp.Path(zip_file_path)
+                contents = [file.name for file in zip_path.iterdir()]
+                result_output = f"Contents of {os.path.basename(zip_file_path)}: {contents}"
+            except Exception as e:
+                result_output = f"Error: {str(e)}"
 
         response.result1 = result_output
     except Exception as e:

tests/appsec/iast_packages/test_packages.py

@@ -725,7 +725,7 @@ def create_venv(self):
         "zipp",
         "3.18.2",
         "example.zip",
-        "Contents of example.zip: ['example.zip/example.txt']",
+        "Contents of example.zip: ['example.txt']",
         "",
     ),
     ## Skip due to typing-extensions added to the denylist