[ML][Pipelines] Test: update unstable tests in ci

sdk/ml/azure-ai-ml/azure/ai/ml/_utils/_asset_utils.py

@@ -109,9 +109,11 @@ def is_file_excluded(self, file_path: Union[str, Path]) -> bool:
         file_path = Path(file_path)
         if file_path.is_absolute():
             ignore_dirname = self._path.parent
-            if len(os.path.commonprefix([file_path, ignore_dirname])) != len(str(ignore_dirname)):
+            try:
+                file_path = os.path.relpath(file_path, ignore_dirname)
+            except ValueError:
+                # 2 paths are on different drives
                 return True
-            file_path = os.path.relpath(file_path, ignore_dirname)
 
         file_path = str(file_path)
         norm_file = normalize_file(file_path)

sdk/ml/azure-ai-ml/tests/component/unittests/test_command_component_entity.py

@@ -1,6 +1,7 @@
 import os
 import shutil
 import sys
+import tempfile
 from io import StringIO
 from pathlib import Path
 from unittest.mock import patch
@@ -500,16 +501,16 @@ def test_invalid_component_outputs(self) -> None:
         assert validation_result.passed
 
     def test_component_code_asset_ignoring_pycache(self) -> None:
-        component_yaml = "./tests/test_configs/components/basic_component_code_local_path.yml"
+        component_yaml = "./tests/test_configs/components/helloworld_component.yml"
         component = load_component(component_yaml)
-        # create some files/folders expected to ignore
-        pycache = Path("./tests/test_configs/components/helloworld_components_with_env/__pycache__")
-        try:
-            if not pycache.is_dir():
-                pycache.mkdir()
+        with tempfile.TemporaryDirectory() as temp_dir:
+            # create some files/folders expected to ignore
+            pycache = Path(temp_dir) / "__pycache__"
+            pycache.mkdir()
             expected_exclude = pycache / "a.pyc"
             expected_exclude.touch()
             # resolve and test for ignore_file's is_file_excluded
+            component.code = temp_dir
             with component._resolve_local_code() as code:
                 excluded = []
                 for root, _, files in os.walk(code.path):
@@ -518,9 +519,6 @@ def test_component_code_asset_ignoring_pycache(self) -> None:
                         if code._ignore_file.is_file_excluded(path):
                             excluded.append(path)
                 assert excluded == [str(expected_exclude.absolute())]
-        finally:
-            if pycache.is_dir():
-                shutil.rmtree(pycache)
 
     def test_normalized_arm_id_in_component_dict(self):
         component_dict = {

sdk/ml/azure-ai-ml/tests/internal_utils/unittests/test_asset_utils.py

@@ -1,6 +1,8 @@
 import os
+import shutil
+import tempfile
 from pathlib import Path
-from typing import Callable
+from typing import Callable, Tuple
 
 import pytest
 
@@ -16,18 +18,21 @@
 
 
 @pytest.fixture
-def gitignore_file_directory() -> str:
-    return "./tests/test_configs/storage/gitignore_only/"
+def storage_test_directory() -> str:
+    with tempfile.TemporaryDirectory() as temp_dir:
+        shutil.rmtree(temp_dir)
+        shutil.copytree("./tests/test_configs/storage/", temp_dir)
+        yield temp_dir
 
 
 @pytest.fixture
-def storage_test_directory() -> str:
-    return "./tests/test_configs/storage/"
+def gitignore_file_directory(storage_test_directory: str) -> str:
+    return os.path.join(storage_test_directory, "gitignore_only")
 
 
 @pytest.fixture
-def no_ignore_file_directory() -> str:
-    return "./tests/test_configs/storage/dont_include_us/"
+def no_ignore_file_directory(storage_test_directory: str) -> str:
+    return os.path.join(storage_test_directory, "dont_include_us")
 
 
 @pytest.fixture
@@ -45,36 +50,18 @@ def no_ignore_file(no_ignore_file_directory: str) -> IgnoreFile:
     return IgnoreFile(None)
 
 
-@pytest.fixture
-def target_file_path(storage_test_directory: str) -> os.PathLike:
+def generate_link_file(base_dir: str) -> Tuple[os.PathLike, os.PathLike]:
     target_file_name = "target_file_rand_name.txt"
-    target_file = Path(os.path.join(os.path.abspath(storage_test_directory), target_file_name))
+    target_file = Path(os.path.join(os.path.abspath(base_dir), target_file_name))
     target_file.write_text("some text")
-    target_file = convert_windows_path_to_unix(target_file)
-    yield target_file
-
-    if os.path.exists(target_file):
-        os.remove(target_file)
-
 
-@pytest.fixture
-def link_file_path(
-    storage_test_directory: str, target_file_path: os.PathLike
-) -> os.PathLike:
     link_file_name = "link_file_rand_name.txt"
-    link_file = Path(os.path.join(os.path.abspath(storage_test_directory), link_file_name))
+    link_file = Path(os.path.join(os.path.abspath(base_dir), link_file_name))
 
-    try:
-        os.symlink(target_file_path, link_file)
-    except FileExistsError:
-        pass
+    os.symlink(target_file, link_file)
 
     assert os.path.islink(link_file)
-    link_file = convert_windows_path_to_unix(link_file)
-    yield link_file
-
-    if os.path.exists(link_file):
-        os.remove(link_file)
+    return convert_windows_path_to_unix(target_file), convert_windows_path_to_unix(link_file)
 
 
 @pytest.mark.unittest
@@ -149,24 +136,25 @@ def test_upload_paths_match(self, storage_test_directory: str) -> None:
                 continue
             assert remote_path in local_path
 
-    def test_symlinks_included_in_hash(self, target_file_path: os.PathLike, link_file_path: os.PathLike) -> None:
+    def test_symlinks_included_in_hash(self, storage_test_directory: str) -> None:
         """Confirm that changes in the original file are respected when the symlink is hashed"""
+        target_file_path, link_file_path = generate_link_file(storage_test_directory)
 
         # hash symlink, update original file, hash symlink again and compare hashes
         original_hash = get_object_hash(path=link_file_path, ignore_file=no_ignore_file)
         Path(target_file_path).write_text("some more text")
         updated_hash = get_object_hash(path=link_file_path, ignore_file=no_ignore_file)
         assert original_hash != updated_hash
 
-    def test_symlink_upload_paths(
-        self, storage_test_directory: str, target_file_path: os.PathLike, link_file_path: os.PathLike
-    ) -> None:
+    def test_symlink_upload_paths(self, storage_test_directory: str) -> None:
         """Confirm that symlink name is preserved for upload to storage, but that target file's path is uploaded
 
         e.g given a file ./dir/foo/bar.txt with a symlink ./other_dir/bar_link.txt, we want to upload the contents of ./dir/food/bar.txt at path ./other_dir/bar_link.txt in the remote storage.
         """
+        target_file_path, link_file_path = generate_link_file(storage_test_directory)
+
         source_path = Path(storage_test_directory).resolve()
-        prefix = source_path.name + "/"
+        prefix = "random_prefix/"
         upload_paths_list = []
 
         for root, _, files in os.walk(source_path, followlinks=True):
@@ -175,5 +163,8 @@ def test_symlink_upload_paths(
         local_paths = [i for i, _ in upload_paths_list]
         remote_paths = [j for _, j in upload_paths_list]
 
-        assert target_file_path in local_paths
-        assert any([rp in link_file_path for rp in remote_paths])  # remote file names are relative
+        # When username is too long, temp folder path will be truncated, e.g. longusername -> LONGUS~
+        # so resolve target_file_path to get the full path
+        assert Path(target_file_path).resolve().as_posix() in local_paths
+        # remote file names are relative to root and include the prefix
+        assert prefix + Path(link_file_path).relative_to(storage_test_directory).as_posix() in remote_paths