update S3Bucket._resolve_path logic

src/integrations/prefect-aws/prefect_aws/s3.py

@@ -478,21 +478,24 @@ def _resolve_path(self, path: str) -> str:
         A helper function used in write_path to join `self.basepath` and `path`.
 
         Args:
-
             path: Name of the key, e.g. "file1". Each object in your
                 bucket has a unique key (or key name).
-
         """
-        # If bucket_folder provided, it means we won't write to the root dir of
-        # the bucket. So we need to add it on the front of the path.
-        #
-        # AWS object key naming guidelines require '/' for bucket folders.
-        # Get POSIX path to prevent `pathlib` from inferring '\' on Windows OS
-        path = (
-            (Path(self.bucket_folder) / path).as_posix() if self.bucket_folder else path
-        )
+        # Always remove leading slash from path
+        path = path.lstrip("/")
 
-        return path
+        if self.bucket_folder:
+            # Remove leading and trailing slashes from bucket_folder
+            bucket_folder = self.bucket_folder.strip("/")
+            if bucket_folder:
+                # Join bucket_folder and path, ensuring a single slash between them
+                path = f"{bucket_folder}/{path}" if path else bucket_folder
+
+        # Ensure the path uses forward slashes and remove any duplicate slashes
+        path = "/".join(filter(None, path.replace("\\", "/").split("/")))
+
+        # Remove trailing slash, except when path is empty
+        return path.rstrip("/") or path
 
     def _get_s3_client(self) -> boto3.client:
         """

src/integrations/prefect-aws/tests/test_s3.py

@@ -1,6 +1,7 @@
 import io
 import os
 from pathlib import Path, PurePosixPath, PureWindowsPath
+from unittest.mock import Mock
 
 import boto3
 import pytest
@@ -749,8 +750,57 @@ def test_write_path_in_sync_context(s3_bucket):
     assert content == b"hello"
 
 
-def test_resolve_path(s3_bucket):
-    assert s3_bucket._resolve_path("") == ""
+@pytest.mark.parametrize(
+    "bucket_folder, path, expected",
+    [
+        # No bucket folder
+        (None, "", ""),
+        (None, "file.txt", "file.txt"),
+        (None, "/file.txt", "file.txt"),
+        (None, "folder/file.txt", "folder/file.txt"),
+        (None, "/folder/file.txt", "folder/file.txt"),
+        # Simple bucket folder
+        ("prefix", "", "prefix"),
+        ("prefix", "file.txt", "prefix/file.txt"),
+        ("prefix", "/file.txt", "prefix/file.txt"),
+        ("prefix", "folder/file.txt", "prefix/folder/file.txt"),
+        ("prefix", "/folder/file.txt", "prefix/folder/file.txt"),
+        # Bucket folder with trailing slash
+        ("prefix/", "", "prefix"),
+        ("prefix/", "file.txt", "prefix/file.txt"),
+        ("prefix/", "/file.txt", "prefix/file.txt"),
+        ("prefix/", "folder/file.txt", "prefix/folder/file.txt"),
+        ("prefix/", "/folder/file.txt", "prefix/folder/file.txt"),
+        # Bucket folder with leading slash
+        ("/prefix", "", "prefix"),
+        ("/prefix", "file.txt", "prefix/file.txt"),
+        ("/prefix", "/file.txt", "prefix/file.txt"),
+        ("/prefix", "folder/file.txt", "prefix/folder/file.txt"),
+        ("/prefix", "/folder/file.txt", "prefix/folder/file.txt"),
+        # Bucket folder with both leading and trailing slashes
+        ("/prefix/", "", "prefix"),
+        ("/prefix/", "file.txt", "prefix/file.txt"),
+        ("/prefix/", "/file.txt", "prefix/file.txt"),
+        ("/prefix/", "folder/file.txt", "prefix/folder/file.txt"),
+        ("/prefix/", "/folder/file.txt", "prefix/folder/file.txt"),
+        # Multi-level bucket folder
+        ("prefix1/prefix2", "", "prefix1/prefix2"),
+        ("prefix1/prefix2", "file.txt", "prefix1/prefix2/file.txt"),
+        ("prefix1/prefix2", "/file.txt", "prefix1/prefix2/file.txt"),
+        ("prefix1/prefix2", "folder/file.txt", "prefix1/prefix2/folder/file.txt"),
+        ("prefix1/prefix2", "/folder/file.txt", "prefix1/prefix2/folder/file.txt"),
+        # Edge cases
+        ("", "file.txt", "file.txt"),
+        ("/", "file.txt", "file.txt"),
+        ("///", "file.txt", "file.txt"),
+        ("prefix", "///file.txt", "prefix/file.txt"),
+    ],
+)
+def test_resolve_path(bucket_folder, path, expected):
+    s3_bucket = Mock()
+    s3_bucket.bucket_folder = bucket_folder
+    s3_bucket._resolve_path = lambda p: S3Bucket._resolve_path(s3_bucket, p)
+    assert s3_bucket._resolve_path(path) == expected
 
 
 class TestS3Bucket: