Write to temp file instead

Author: howieleung

sdk/ai/azure-ai-projects/tests/samples/test_samples.py

@@ -5,34 +5,18 @@
 # ------------------------------------
 import csv, os, pytest, re, inspect, sys
 import importlib.util
-import io
 import unittest.mock as mock
 from azure.core.exceptions import HttpResponseError
 from devtools_testutils.aio import recorded_by_proxy_async
 from devtools_testutils import AzureRecordedTestCase, recorded_by_proxy, RecordedTransport
-from test_base import servicePreparer
+from test_base import servicePreparer, patched_open_crlf_to_lf
 from pytest import MonkeyPatch
 from azure.ai.projects import AIProjectClient
 
 
 class SampleExecutor:
     """Helper class for executing sample files with proper environment setup and credential mocking."""
 
-    class ConvertedFileWrapper(io.BufferedReader):
-        """File wrapper that converts CRLF to LF content while preserving original filename."""
-
-        def __init__(self, file_path, converted_content):
-            # Create BytesIO with converted content
-            self._bytesio = io.BytesIO(converted_content)
-            # Initialize BufferedReader with the BytesIO
-            super().__init__(self._bytesio)
-            # Override name to be the original file path
-            self._name = file_path
-
-        @property
-        def name(self):
-            return self._name
-
     def __init__(
         self, test_instance: "AzureRecordedTestCase", sample_path: str, env_var_mapping: dict[str, str], **kwargs
     ):
@@ -62,43 +46,19 @@ def __init__(
 
         self.module = importlib.util.module_from_spec(spec)
         self.spec = spec
-        self._original_open = open
 
     def _capture_print(self, *args, **kwargs):
         """Capture print calls while still outputting to console."""
         self.print_calls.append(" ".join(str(arg) for arg in args))
         self._original_print(*args, **kwargs)
 
-    def _patched_open(self, *args, **kwargs):
-        """Patch open to convert CRLF to LF for text files."""
-        file_path = args[0] if args else kwargs.get("file")
-        mode = args[1] if len(args) > 1 else kwargs.get("mode", "r")
-
-        # Check if this is binary read mode for text-like files
-        if "r" in mode and "b" in mode and file_path and isinstance(file_path, str):
-            # Check file extension to determine if it's a text file
-            text_extensions = {".txt", ".json", ".jsonl", ".csv", ".md", ".yaml", ".yml", ".xml"}
-            ext = os.path.splitext(file_path)[1].lower()
-            if ext in text_extensions:
-                # Read the original file
-                with self._original_open(file_path, "rb") as f:
-                    content = f.read()
-
-                # Convert CRLF to LF
-                converted_content = content.replace(b"\r\n", b"\n")
-
-                # Return wrapped file-like object with converted content
-                return SampleExecutor.ConvertedFileWrapper(file_path, converted_content)
-
-        return self._original_open(*args, **kwargs)
-
     def execute(self):
         """Execute a synchronous sample with proper mocking and environment setup."""
 
         with (
             MonkeyPatch.context() as mp,
             mock.patch("builtins.print", side_effect=self._capture_print),
-            mock.patch("builtins.open", side_effect=self._patched_open),
+            mock.patch("builtins.open", side_effect=patched_open_crlf_to_lf),
             mock.patch("azure.identity.DefaultAzureCredential") as mock_credential,
         ):
             for var_name, var_value in self.env_vars.items():
@@ -119,7 +79,7 @@ async def execute_async(self):
         with (
             MonkeyPatch.context() as mp,
             mock.patch("builtins.print", side_effect=self._capture_print),
-            mock.patch("builtins.open", side_effect=self._patched_open),
+            mock.patch("builtins.open", side_effect=patched_open_crlf_to_lf),
             mock.patch("azure.identity.aio.DefaultAzureCredential") as mock_credential,
         ):
             for var_name, var_value in self.env_vars.items():

sdk/ai/azure-ai-projects/tests/test_base.py

@@ -7,6 +7,8 @@
 import re
 import functools
 import json
+import os
+import tempfile
 from typing import Optional, Any, Dict, Final
 from azure.ai.projects.models import (
     Connection,
@@ -33,6 +35,9 @@
 from azure.ai.projects import AIProjectClient as AIProjectClient
 from azure.ai.projects.aio import AIProjectClient as AsyncAIProjectClient
 
+# Store reference to built-in open before any mocking occurs
+_BUILTIN_OPEN = open
+
 
 # Load secrets from environment variables
 servicePreparer = functools.partial(
@@ -63,6 +68,60 @@
 DEVELOPER_TIER_TRAINING_TYPE: Final[str] = "developerTier"
 
 
+def patched_open_crlf_to_lf(*args, **kwargs):
+    """
+    Patched open function that converts CRLF to LF for text files.
+
+    This function should be used with mock.patch("builtins.open", side_effect=TestBase.patched_open_crlf_to_lf)
+    to ensure consistent line endings in test files during recording and playback.
+    """
+    # Extract file path - first positional arg or 'file' keyword arg
+    if args:
+        file_path = args[0]
+    elif "file" in kwargs:
+        file_path = kwargs["file"]
+    else:
+        # No file path provided, just pass through
+        return _BUILTIN_OPEN(*args, **kwargs)
+
+    # Extract mode - second positional arg or 'mode' keyword arg
+    if len(args) > 1:
+        mode = args[1]
+    else:
+        mode = kwargs.get("mode", "r")
+
+    # Check if this is binary read mode for text-like files
+    if "r" in mode and "b" in mode and file_path and isinstance(file_path, str):
+        # Check file extension to determine if it's a text file
+        text_extensions = {".txt", ".json", ".jsonl", ".csv", ".md", ".yaml", ".yml", ".xml"}
+        ext = os.path.splitext(file_path)[1].lower()
+        if ext in text_extensions:
+            # Read the original file
+            with _BUILTIN_OPEN(file_path, "rb") as f:
+                content = f.read()
+
+            # Convert CRLF to LF
+            converted_content = content.replace(b"\r\n", b"\n")
+
+            # Only create temp file if conversion was needed
+            if converted_content != content:
+                # Create a temporary file with the converted content
+                temp_fd, temp_path = tempfile.mkstemp(suffix=ext)
+                os.write(temp_fd, converted_content)
+                os.close(temp_fd)
+                # Replace file path with temp path
+                if args:
+                    # File path was passed as positional arg
+                    return _BUILTIN_OPEN(temp_path, *args[1:], **kwargs)
+                else:
+                    # File path was passed as keyword arg
+                    kwargs = kwargs.copy()
+                    kwargs["file"] = temp_path
+                    return _BUILTIN_OPEN(**kwargs)
+
+    return _BUILTIN_OPEN(*args, **kwargs)
+
+
 class TestBase(AzureRecordedTestCase):
 
     test_redteams_params = {