handle UNC paths for windows (#16005)

.github/workflows/test-windows-unc.yaml

@@ -0,0 +1,53 @@
+name: Test Windows UNC Paths
+
+on:
+  workflow_dispatch: # Allow manual triggering
+  pull_request:
+    paths:
+      - "src/prefect/utilities/filesystem.py"
+      - "scripts/test_unc_paths.py"
+      - ".github/workflows/test-windows-unc.yaml"
+      - "requirements.txt"
+      - "requirements-client.txt"
+
+jobs:
+  test-unc-paths:
+    runs-on: windows-latest
+
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+          fetch-depth: 0
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install -U pip
+          pip install -e .
+
+      - name: Create test UNC path and run flow
+        shell: pwsh
+        run: |
+          # Create a test directory
+          New-Item -ItemType Directory -Path "C:\ShareTest" -Force
+
+          # Create network share
+          New-SmbShare -Name "PrefectTest" -Path "C:\ShareTest" -FullAccess "Everyone"
+
+          # Run the test script from the current directory
+          # This will create and test flows in the UNC path
+          python scripts/test_unc_paths.py
+
+        env:
+          PYTHONPATH: ${{ github.workspace }}
+
+      - name: Cleanup
+        if: always()
+        shell: pwsh
+        run: |
+          Remove-SmbShare -Name "PrefectTest" -Force

scripts/test_unc_paths.py

@@ -0,0 +1,66 @@
+import os
+import sys
+from pathlib import Path
+
+from prefect import flow
+from prefect.flows import Flow
+
+
+def setup_unc_share(base_path: Path) -> Path:
+    """
+    Creates a test UNC path and returns it.
+    Requires admin privileges on Windows.
+    """
+    if os.name != "nt":
+        print("This script only works on Windows")
+        sys.exit(1)
+
+    # Create a test directory structure in the share
+    unc_path = Path(r"\\localhost\PrefectTest")
+
+    # Create a src directory in the share for our test flow
+    src_dir = unc_path / "src"
+    src_dir.mkdir(parents=True, exist_ok=True)
+
+    return unc_path
+
+
+def create_test_flow_file(path: Path):
+    """Create a test flow file in the given path"""
+    flow_code = """
+from prefect import flow
+
+@flow
+def remote_test_flow(name: str = "remote"):
+    print(f"Hello from {name} in remote flow!")
+    return "Remote Success!"
+"""
+    # Create the flow file in src/app.py
+    flow_file = path / "src" / "app.py"
+    flow_file.write_text(flow_code)
+    return flow_file
+
+
+if __name__ == "__main__":
+    try:
+        # Setup UNC share
+        unc_path = setup_unc_share(Path.cwd())
+        print(f"Created UNC path structure at: {unc_path}")
+
+        # Create a test flow file in the share
+        flow_file = create_test_flow_file(unc_path)
+        print(f"Created test flow file at: {flow_file}")
+
+        # Try to load and run flow from UNC path
+        print("Attempting to load flow from UNC path...")
+        remote_flow = flow.from_source(
+            source=unc_path, entrypoint="src/app.py:remote_test_flow"
+        )
+
+        print("Testing if flow was loaded correctly...")
+        assert isinstance(remote_flow, Flow), "Flow was not loaded correctly"
+        print("Flow loaded successfully!")
+
+    except Exception as e:
+        print(f"Error: {type(e).__name__}: {e}")
+        raise

src/prefect/utilities/filesystem.py

@@ -7,7 +7,7 @@
 import threading
 from contextlib import contextmanager
 from pathlib import Path, PureWindowsPath
-from typing import Optional, Union
+from typing import Optional, Union, cast
 
 import fsspec
 import pathspec
@@ -22,8 +22,8 @@ def create_default_ignore_file(path: str) -> bool:
     Creates default ignore file in the provided path if one does not already exist; returns boolean specifying
     whether a file was created.
     """
-    path = pathlib.Path(path)
-    ignore_file = path / ".prefectignore"
+    _path = pathlib.Path(path)
+    ignore_file = _path / ".prefectignore"
     if ignore_file.exists():
         return False
     default_file = pathlib.Path(prefect.__module_path__) / ".prefectignore"
@@ -54,20 +54,47 @@ def filter_files(
 chdir_lock = threading.Lock()
 
 
+def _normalize_path(path: Union[str, Path]) -> str:
+    """
+    Normalize a path, handling UNC paths on Windows specially.
+    """
+    path = Path(path)
+
+    # Handle UNC paths on Windows differently
+    if os.name == "nt" and str(path).startswith("\\\\"):
+        # For UNC paths, use absolute() instead of resolve()
+        # to avoid the Windows path resolution issues
+        return str(path.absolute())
+    else:
+        try:
+            # For non-UNC paths, try resolve() first
+            return str(path.resolve())
+        except OSError:
+            # Fallback to absolute() if resolve() fails
+            return str(path.absolute())
+
+
 @contextmanager
 def tmpchdir(path: str):
     """
-    Change current-working directories for the duration of the context
+    Change current-working directories for the duration of the context,
+    with special handling for UNC paths on Windows.
     """
-    path = os.path.abspath(path)
+    path = _normalize_path(path)
+
     if os.path.isfile(path) or (not os.path.exists(path) and not path.endswith("/")):
         path = os.path.dirname(path)
 
     owd = os.getcwd()
 
     with chdir_lock:
         try:
-            os.chdir(path)
+            # On Windows with UNC paths, we need to handle the directory change carefully
+            if os.name == "nt" and path.startswith("\\\\"):
+                # Use os.path.abspath to handle UNC paths
+                os.chdir(os.path.abspath(path))
+            else:
+                os.chdir(path)
             yield path
         finally:
             os.chdir(owd)
@@ -76,7 +103,7 @@ def tmpchdir(path: str):
 def filename(path: str) -> str:
     """Extract the file name from a path with remote file system support"""
     try:
-        of: OpenFile = fsspec.open(path)
+        of: OpenFile = cast(OpenFile, fsspec.open(path))
         sep = of.fs.sep
     except (ImportError, AttributeError):
         sep = "\\" if "\\" in path else "/"
@@ -98,7 +125,7 @@ def is_local_path(path: Union[str, pathlib.Path, OpenFile]):
     else:
         raise TypeError(f"Invalid path of type {type(path).__name__!r}")
 
-    return type(of.fs) == LocalFileSystem
+    return isinstance(of.fs, LocalFileSystem)
 
 
 def to_display_path(