From 4e986fbe81aff5945bb0c3a6b5a19ac70c89ee96 Mon Sep 17 00:00:00 2001 From: Alessandro Cere Date: Wed, 3 Jun 2026 19:46:57 +0800 Subject: [PATCH 1/3] feat: replace s3fs/aiobotocore with boto3-based fsspec backend Replace the problematic s3fs/aiobotocore dependency chain with a self-contained AsyncFileSystem implementation using boto3 directly. This eliminates the well-known version conflicts between s3fs's pinned aiobotocore and boto3 (GitHub issue #77). Key changes: - Add llmeter/s3fs.py: Boto3S3FileSystem (AsyncFileSystem subclass) wrapping synchronous boto3 calls in asyncio.to_thread() - Add registration in llmeter/__init__.py with fallback to s3fs/obstore - Change pyproject.toml: fsspec[http,s3] -> fsspec[http], add entry point - Add tests/test_s3fs.py: 47 unit tests using moto mock_aws - Add tests/integ/test_s3fs_integ.py: 23 integration tests against real S3 - Add S3 fixtures to tests/integ/conftest.py --- llmeter/__init__.py | 36 ++ llmeter/s3fs.py | 445 ++++++++++++++++++++++++ pyproject.toml | 6 +- tests/integ/conftest.py | 67 ++++ tests/integ/test_s3fs_integ.py | 458 ++++++++++++++++++++++++ tests/test_s3fs.py | 618 +++++++++++++++++++++++++++++++++ 6 files changed, 1629 insertions(+), 1 deletion(-) create mode 100644 llmeter/s3fs.py create mode 100644 tests/integ/test_s3fs_integ.py create mode 100644 tests/test_s3fs.py diff --git a/llmeter/__init__.py b/llmeter/__init__.py index 51b4165..663ed63 100644 --- a/llmeter/__init__.py +++ b/llmeter/__init__.py @@ -4,3 +4,39 @@ __version__ = version(__name__) except PackageNotFoundError: __version__ = "0.0.0" + + +def _register_s3_filesystem(): + """Register boto3 S3 backend if no other S3 backend is available.""" + try: + import s3fs # noqa: F401 + + return # s3fs is installed, defer to it + except ImportError: + pass + + try: + from obstore.fsspec import FsspecStore # noqa: F401 + + return # obstore is installed, defer to it + except ImportError: + pass + + try: + import fsspec + + fsspec.register_implementation( + "s3", "llmeter.s3fs.Boto3S3FileSystem", clobber=True + ) + fsspec.register_implementation( + "s3a", "llmeter.s3fs.Boto3S3FileSystem", clobber=True + ) + except Exception: + import logging + + logging.getLogger(__name__).warning( + "Failed to register llmeter S3 filesystem backend", exc_info=True + ) + + +_register_s3_filesystem() diff --git a/llmeter/s3fs.py b/llmeter/s3fs.py new file mode 100644 index 0000000..65958f7 --- /dev/null +++ b/llmeter/s3fs.py @@ -0,0 +1,445 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Async fsspec filesystem for S3 using boto3 directly. + +This module provides a self-contained S3 filesystem implementation that avoids +the s3fs/aiobotocore dependency chain which causes version conflicts with boto3. +It implements fsspec's AsyncFileSystem interface, wrapping synchronous boto3 calls +in asyncio.to_thread() for non-blocking execution. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Any + +try: + import boto3 + from botocore.exceptions import ClientError +except ImportError as e: + raise ImportError( + "boto3 is required for the llmeter S3 filesystem backend. " + "Install it with: pip install boto3" + ) from e + +import fsspec +import fsspec.spec +from fsspec.asyn import AsyncFileSystem + +logger = logging.getLogger(__name__) + + +class S3File(fsspec.spec.AbstractBufferedFile): + """File-like object for S3 backed by in-memory buffer.""" + + def _fetch_range(self, start: int, end: int) -> bytes: + """Download a range of bytes from S3.""" + bucket, key = self.fs._split_path(self.path) + try: + response = self.fs._s3_client.get_object( + Bucket=bucket, + Key=key, + Range=f"bytes={start}-{end - 1}", + ) + return response["Body"].read() + except self.fs._s3_client.exceptions.NoSuchKey: + raise FileNotFoundError(self.path) + except ClientError as e: + code = e.response["Error"]["Code"] + if code in ("403", "AccessDenied"): + raise PermissionError(self.path) from e + if code in ("404", "NoSuchBucket"): + raise FileNotFoundError(self.path) from e + raise OSError( + f"S3 error {code}: {e.response['Error'].get('Message', '')}" + ) from e + + def _initiate_upload(self) -> None: + """Initialize upload state. Buffer is already set by AbstractBufferedFile.""" + pass + + def _upload_chunk(self, final: bool = False) -> None: + """Upload buffered content to S3 via put_object when final=True.""" + if final: + self.buffer.seek(0) + bucket, key = self.fs._split_path(self.path) + data = self.buffer.read() + try: + self.fs._s3_client.put_object(Bucket=bucket, Key=key, Body=data) + except ClientError as e: + code = e.response["Error"]["Code"] + raise OSError( + f"S3 upload error {code}: {e.response['Error'].get('Message', '')}" + ) from e + + +class Boto3S3FileSystem(AsyncFileSystem): + """Async fsspec filesystem for S3 using boto3 directly. + + This implementation wraps synchronous boto3 calls in asyncio.to_thread() + for non-blocking execution. It avoids the s3fs/aiobotocore dependency chain + that causes version conflicts with boto3. + + Parameters + ---------- + session : boto3.Session, optional + Custom boto3 session for credential management. + region_name : str, optional + AWS region for the S3 client. + endpoint_url : str, optional + Custom endpoint URL for S3-compatible services (e.g., MinIO, LocalStack). + """ + + protocol = ("s3", "s3a") + async_impl = True + use_listings_cache = False + + def __init__( + self, + session: Any | None = None, + region_name: str | None = None, + endpoint_url: str | None = None, + **kwargs: Any, + ): + super().__init__(**kwargs) + self._session = session + self._region_name = region_name + self._endpoint_url = endpoint_url + self._client: Any | None = None + + @property + def _s3_client(self): + """Lazy boto3 S3 client creation.""" + if self._client is None: + kwargs: dict[str, Any] = {} + if self._region_name: + kwargs["region_name"] = self._region_name + if self._endpoint_url: + kwargs["endpoint_url"] = self._endpoint_url + + if self._session is not None: + self._client = self._session.client("s3", **kwargs) + else: + self._client = boto3.client("s3", **kwargs) + return self._client + + @staticmethod + def _strip_protocol(path: str) -> str: + """Strip s3:// or s3a:// prefix and normalize.""" + if isinstance(path, list): + return [Boto3S3FileSystem._strip_protocol(p) for p in path] + for proto in ("s3://", "s3a://"): + if path.startswith(proto): + path = path[len(proto) :] + return path.strip("/") + + def _split_path(self, path: str) -> tuple[str, str]: + """Split a normalized path into (bucket, key).""" + path = self._strip_protocol(path) + if "/" in path: + bucket, key = path.split("/", 1) + else: + bucket, key = path, "" + return bucket, key + + # --- Async core methods --- + + async def _cat_file( + self, path: str, start: int | None = None, end: int | None = None, **kwargs: Any + ) -> bytes: + """Read file contents from S3.""" + bucket, key = self._split_path(path) + + def _do_get(): + get_kwargs: dict[str, Any] = {"Bucket": bucket, "Key": key} + if start is not None or end is not None: + s = start or 0 + e = f"{end - 1}" if end else "" + get_kwargs["Range"] = f"bytes={s}-{e}" + try: + response = self._s3_client.get_object(**get_kwargs) + return response["Body"].read() + except self._s3_client.exceptions.NoSuchKey: + raise FileNotFoundError(path) + except ClientError as e: + code = e.response["Error"]["Code"] + if code in ("403", "AccessDenied"): + raise PermissionError(path) from e + if code in ("404", "NoSuchBucket"): + raise FileNotFoundError(path) from e + raise OSError( + f"S3 error {code}: {e.response['Error'].get('Message', '')}" + ) from e + + return await asyncio.to_thread(_do_get) + + async def _pipe_file(self, path: str, value: bytes, **kwargs: Any) -> None: + """Write bytes content to S3.""" + bucket, key = self._split_path(path) + + def _do_put(): + try: + self._s3_client.put_object(Bucket=bucket, Key=key, Body=value) + except ClientError as e: + code = e.response["Error"]["Code"] + raise OSError( + f"S3 upload error {code}: {e.response['Error'].get('Message', '')}" + ) from e + + await asyncio.to_thread(_do_put) + + async def _info(self, path: str, **kwargs: Any) -> dict[str, Any]: + """Get metadata for an S3 path.""" + bucket, key = self._split_path(path) + + def _do_info(): + # Try as file first (head_object) + if key: + try: + response = self._s3_client.head_object(Bucket=bucket, Key=key) + return { + "name": path, + "type": "file", + "size": response["ContentLength"], + "LastModified": response.get("LastModified"), + "ETag": response.get("ETag"), + } + except ClientError as e: + code = e.response["Error"]["Code"] + if code in ("403", "AccessDenied"): + raise PermissionError(path) from e + if code not in ("404", "NoSuchKey"): + raise OSError( + f"S3 error {code}: {e.response['Error'].get('Message', '')}" + ) from e + + # Try as directory (prefix check) + prefix = f"{key}/" if key else "" + response = self._s3_client.list_objects_v2( + Bucket=bucket, Prefix=prefix, MaxKeys=1 + ) + if response.get("KeyCount", 0) > 0: + return { + "name": path, + "type": "directory", + "size": 0, + } + + raise FileNotFoundError(path) + + return await asyncio.to_thread(_do_info) + + async def _exists(self, path: str, **kwargs: Any) -> bool: + """Check if an S3 path exists (as object or non-empty prefix).""" + try: + await self._info(path) + return True + except FileNotFoundError: + return False + + async def _ls(self, path: str, detail: bool = False, **kwargs: Any) -> list: + """List objects and prefixes under an S3 path.""" + bucket, key = self._split_path(path) + + def _do_ls(): + prefix = f"{key}/" if key else "" + results: list[dict[str, Any]] = [] + continuation_token = None + + while True: + list_kwargs: dict[str, Any] = { + "Bucket": bucket, + "Prefix": prefix, + "Delimiter": "/", + } + if continuation_token: + list_kwargs["ContinuationToken"] = continuation_token + + response = self._s3_client.list_objects_v2(**list_kwargs) + + # Common prefixes (directories) + for cp in response.get("CommonPrefixes", []): + name = f"{bucket}/{cp['Prefix'].rstrip('/')}" + results.append({"name": name, "type": "directory", "size": 0}) + + # Objects (files) + for obj in response.get("Contents", []): + obj_key = obj["Key"] + # Skip the prefix itself if it appears as an object + if obj_key == prefix: + continue + name = f"{bucket}/{obj_key}" + results.append( + { + "name": name, + "type": "file", + "size": obj["Size"], + "LastModified": obj.get("LastModified"), + "ETag": obj.get("ETag"), + } + ) + + if response.get("IsTruncated"): + continuation_token = response["NextContinuationToken"] + else: + break + + if not results: + raise FileNotFoundError(path) + + return results + + entries = await asyncio.to_thread(_do_ls) + + if detail: + return entries + return [entry["name"] for entry in entries] + + async def _mkdir( + self, path: str, create_parents: bool = True, **kwargs: Any + ) -> None: + """No-op: S3 does not have real directories.""" + pass + + async def _rm_file(self, path: str, **kwargs: Any) -> None: + """Delete a single object from S3.""" + bucket, key = self._split_path(path) + + if not key: + # Cannot delete a bare bucket via rm_file + return + + def _do_rm(): + # Check existence first + try: + self._s3_client.head_object(Bucket=bucket, Key=key) + except ClientError as e: + code = e.response["Error"]["Code"] + if code in ("404", "NoSuchKey"): + raise FileNotFoundError(path) from e + if code in ("403", "AccessDenied"): + raise PermissionError(path) from e + raise OSError( + f"S3 error {code}: {e.response['Error'].get('Message', '')}" + ) from e + + self._s3_client.delete_object(Bucket=bucket, Key=key) + + await asyncio.to_thread(_do_rm) + + async def _rm(self, path, recursive=False, batch_size=None, **kwargs): + """Delete files, skipping directory placeholders that don't exist as objects.""" + from fsspec.asyn import _run_coros_in_chunks + + batch_size = batch_size or self.batch_size + paths = await self._expand_path(path, recursive=recursive) + + async def _safe_rm(p): + try: + await self._rm_file(p, **kwargs) + except FileNotFoundError: + # During recursive delete, expanded paths include directory + # prefixes that don't exist as objects — skip them silently. + pass + + return await _run_coros_in_chunks( + [_safe_rm(p) for p in reversed(paths)], + batch_size=batch_size, + nofiles=True, + ) + + async def _cp_file(self, path1: str, path2: str, **kwargs: Any) -> None: + """Copy an S3 object to another location.""" + bucket1, key1 = self._split_path(path1) + bucket2, key2 = self._split_path(path2) + + def _do_copy(): + try: + self._s3_client.copy_object( + Bucket=bucket2, + Key=key2, + CopySource={"Bucket": bucket1, "Key": key1}, + ) + except ClientError as e: + code = e.response["Error"]["Code"] + if code in ("404", "NoSuchKey"): + raise FileNotFoundError(path1) from e + raise OSError( + f"S3 copy error {code}: {e.response['Error'].get('Message', '')}" + ) from e + + await asyncio.to_thread(_do_copy) + + async def _put_file(self, lpath: str, rpath: str, **kwargs: Any) -> None: + """Upload a local file to S3.""" + bucket, key = self._split_path(rpath) + + def _do_upload(): + try: + self._s3_client.upload_file(lpath, bucket, key) + except ClientError as e: + code = e.response["Error"]["Code"] + raise OSError( + f"S3 upload error {code}: {e.response['Error'].get('Message', '')}" + ) from e + + await asyncio.to_thread(_do_upload) + + async def _get_file(self, rpath: str, lpath: str, **kwargs: Any) -> None: + """Download an S3 object to a local file.""" + bucket, key = self._split_path(rpath) + + def _do_download(): + try: + self._s3_client.download_file(bucket, key, lpath) + except ClientError as e: + code = e.response["Error"]["Code"] + if code in ("404", "NoSuchKey"): + raise FileNotFoundError(rpath) from e + raise OSError( + f"S3 download error {code}: {e.response['Error'].get('Message', '')}" + ) from e + + await asyncio.to_thread(_do_download) + + def _open( + self, + path: str, + mode: str = "rb", + block_size: int | None = None, + autocommit: bool = True, + cache_options: dict | None = None, + **kwargs: Any, + ) -> S3File: + """Return a file-like object for S3.""" + path = self._strip_protocol(path) + + if "a" in mode: + # Append mode: pre-fetch existing content (non-atomic read-modify-write) + try: + existing = self.cat_file(path) + except FileNotFoundError: + existing = b"" + f = S3File( + self, + path, + mode="wb", + block_size=block_size, + autocommit=autocommit, + cache_options=cache_options, + **kwargs, + ) + f.write(existing) + return f + + return S3File( + self, + path, + mode=mode, + block_size=block_size, + autocommit=autocommit, + cache_options=cache_options, + **kwargs, + ) diff --git a/pyproject.toml b/pyproject.toml index 886929b..ac73735 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,7 @@ dependencies = [ "jmespath>=0.7.1,<2.0.0", "boto3>=1.34.129", "universal-pathlib>=0.2.1", - "fsspec[http,s3]>=2023.6.0", + "fsspec[http]>=2023.6.0", ] [project.optional-dependencies] @@ -36,6 +36,10 @@ Repository = "https://github.com/awslabs/llmeter" requires = ["hatchling"] build-backend = "hatchling.build" +[project.entry-points."fsspec.specs"] +s3 = "llmeter.s3fs:Boto3S3FileSystem" +s3a = "llmeter.s3fs:Boto3S3FileSystem" + [dependency-groups] dev = [ "boto3-stubs[bedrock,bedrock-runtime,essential,sagemaker,sagemaker-runtime]>=1.35.24", diff --git a/tests/integ/conftest.py b/tests/integ/conftest.py index 07b4a5b..223841d 100644 --- a/tests/integ/conftest.py +++ b/tests/integ/conftest.py @@ -4,6 +4,14 @@ """ Shared fixtures and configuration for integration tests. +S3 Filesystem Integration Test Fixtures: +- s3_test_bucket: Reads S3FS_TEST_BUCKET env var (skips if not set) +- s3_test_prefix: Generates a unique prefix per test run for isolation +- s3_cleanup: Deletes all objects under the test prefix after tests complete + +To run S3 integration tests: + S3FS_TEST_BUCKET=my-bucket uv run pytest -m integ tests/integ/test_s3fs_integ.py + This module provides session-scoped fixtures for AWS credentials, configuration, and test utilities used across all integration tests. @@ -306,3 +314,62 @@ def test_payload_with_image(): ], "inferenceConfig": {"maxTokens": 100}, } + + +# --- S3 Filesystem Integration Test Fixtures --- + + +@pytest.fixture(scope="session") +def s3_test_bucket(): + """ + Read S3 test bucket name from S3FS_TEST_BUCKET environment variable. + + Skips all S3 integration tests if the environment variable is not set. + + Returns: + str: The S3 bucket name for integration testing. + """ + bucket = os.environ.get("S3FS_TEST_BUCKET") + if not bucket: + pytest.skip( + "S3FS_TEST_BUCKET environment variable not set. " + "Set it to a writable S3 bucket to run S3 integration tests." + ) + return bucket + + +@pytest.fixture(scope="session") +def s3_test_prefix(): + """ + Generate a unique prefix per test run for isolation. + + Uses a UUID to ensure different test runs don't interfere with each other. + + Returns: + str: A unique prefix like 'llmeter-test/{uuid}/' + """ + import uuid + + return f"llmeter-test/{uuid.uuid4()}/" + + +@pytest.fixture(autouse=True, scope="session") +def s3_cleanup(s3_test_bucket, s3_test_prefix, aws_credentials): + """ + Delete all objects under the test prefix after tests complete. + + This fixture runs automatically for the session and cleans up all + test objects after the integration tests finish. + """ + yield + + # Cleanup: delete all objects under the test prefix + s3 = aws_credentials.client("s3") + paginator = s3.get_paginator("list_objects_v2") + pages = paginator.paginate(Bucket=s3_test_bucket, Prefix=s3_test_prefix) + + for page in pages: + objects = page.get("Contents", []) + if objects: + delete_keys = [{"Key": obj["Key"]} for obj in objects] + s3.delete_objects(Bucket=s3_test_bucket, Delete={"Objects": delete_keys}) diff --git a/tests/integ/test_s3fs_integ.py b/tests/integ/test_s3fs_integ.py new file mode 100644 index 0000000..cf72c92 --- /dev/null +++ b/tests/integ/test_s3fs_integ.py @@ -0,0 +1,458 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Integration tests for Boto3S3FileSystem against real S3. + +These tests require: +- Valid AWS credentials with S3 read/write permissions +- S3FS_TEST_BUCKET environment variable set to a writable bucket + +Run with: uv run pytest -m integ tests/integ/test_s3fs_integ.py + +Requirements validated: +- R1-AC3: UPath integration without s3fs installed +- R2-AC1-4: File read operations +- R3-AC1-4: File write operations +- R4-AC2,3,5,6: Directory operations (_ls, _info) +- R5-AC1-3: Existence checks +- R11-AC1,2: Round-trip data integrity +- R12-AC1,2: Async execution model (to_thread, concurrent gather) +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from llmeter.s3fs import Boto3S3FileSystem + +pytestmark = pytest.mark.integ + + +# --- Helper to build S3 paths --- + + +def s3_path(bucket: str, prefix: str, key: str) -> str: + """Build a full S3 path (without protocol) from bucket, prefix, and key.""" + return f"{bucket}/{prefix}{key}" + + +# --- Round-trip write/read tests --- + + +class TestRoundTrip: + """Test round-trip write/read with real S3 (binary and text, various sizes).""" + + @pytest.mark.asyncio + async def test_roundtrip_binary_small( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """Write and read back small binary content (100 bytes).""" + fs = Boto3S3FileSystem(session=aws_credentials) + path = s3_path(s3_test_bucket, s3_test_prefix, "roundtrip/small.bin") + data = b"x" * 100 + + await fs._pipe_file(path, data) + result = await fs._cat_file(path) + + assert result == data + + @pytest.mark.asyncio + async def test_roundtrip_binary_empty( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """Write and read back empty content (0 bytes).""" + fs = Boto3S3FileSystem(session=aws_credentials) + path = s3_path(s3_test_bucket, s3_test_prefix, "roundtrip/empty.bin") + data = b"" + + await fs._pipe_file(path, data) + result = await fs._cat_file(path) + + assert result == data + + @pytest.mark.asyncio + async def test_roundtrip_binary_medium( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """Write and read back medium binary content (1MB).""" + fs = Boto3S3FileSystem(session=aws_credentials) + path = s3_path(s3_test_bucket, s3_test_prefix, "roundtrip/medium.bin") + data = b"\xab\xcd" * (1024 * 512) # 1MB + + await fs._pipe_file(path, data) + result = await fs._cat_file(path) + + assert result == data + assert len(result) == 1024 * 1024 + + @pytest.mark.asyncio + async def test_roundtrip_text( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """Write and read back text content via open() modes.""" + fs = Boto3S3FileSystem(session=aws_credentials) + path = s3_path(s3_test_bucket, s3_test_prefix, "roundtrip/text.txt") + text = "Hello, world!\nLine 2\nUnicode: café résumé naïve" + + # Write text + with fs.open(path, "w") as f: + f.write(text) + + # Read text + with fs.open(path, "r") as f: + result = f.read() + + assert result == text + + @pytest.mark.asyncio + async def test_roundtrip_binary_via_open( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """Write and read back binary content via open() modes.""" + fs = Boto3S3FileSystem(session=aws_credentials) + path = s3_path(s3_test_bucket, s3_test_prefix, "roundtrip/binary_open.bin") + data = bytes(range(256)) * 4 + + with fs.open(path, "wb") as f: + f.write(data) + + with fs.open(path, "rb") as f: + result = f.read() + + assert result == data + + +# --- Directory listing tests --- + + +class TestListing: + """Test _ls listing with real S3 objects.""" + + @pytest.mark.asyncio + async def test_ls_detail_false( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """_ls with detail=False returns list of path strings.""" + fs = Boto3S3FileSystem(session=aws_credentials) + prefix = f"{s3_test_prefix}ls_test/" + + # Create test objects + await fs._pipe_file(f"{s3_test_bucket}/{prefix}file1.txt", b"a") + await fs._pipe_file(f"{s3_test_bucket}/{prefix}file2.txt", b"b") + await fs._pipe_file(f"{s3_test_bucket}/{prefix}subdir/file3.txt", b"c") + + # List + result = await fs._ls(f"{s3_test_bucket}/{prefix.rstrip('/')}", detail=False) + + # Should contain file1, file2, and subdir/ as immediate children + names = [r.split("/")[-1] for r in result] + assert "file1.txt" in names + assert "file2.txt" in names + assert "subdir" in names + + @pytest.mark.asyncio + async def test_ls_detail_true( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """_ls with detail=True returns list of metadata dicts.""" + fs = Boto3S3FileSystem(session=aws_credentials) + prefix = f"{s3_test_prefix}ls_detail/" + + await fs._pipe_file(f"{s3_test_bucket}/{prefix}data.json", b'{"key": "value"}') + + result = await fs._ls(f"{s3_test_bucket}/{prefix.rstrip('/')}", detail=True) + + assert len(result) >= 1 + entry = next(e for e in result if e["name"].endswith("data.json")) + assert entry["type"] == "file" + assert entry["size"] == len(b'{"key": "value"}') + assert "name" in entry + + @pytest.mark.asyncio + async def test_ls_bare_bucket( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """_ls on a prefix that has objects returns results.""" + fs = Boto3S3FileSystem(session=aws_credentials) + + # Ensure at least one object exists + await fs._pipe_file( + f"{s3_test_bucket}/{s3_test_prefix}ls_bare/marker.txt", b"exists" + ) + + # List the test prefix (simulates listing under a "directory") + result = await fs._ls( + f"{s3_test_bucket}/{s3_test_prefix.rstrip('/')}", detail=False + ) + assert len(result) > 0 + + +# --- Info and exists tests --- + + +class TestInfoExists: + """Test _info and _exists with real S3.""" + + @pytest.mark.asyncio + async def test_info_file(self, s3_test_bucket, s3_test_prefix, aws_credentials): + """_info on an existing object returns file metadata.""" + fs = Boto3S3FileSystem(session=aws_credentials) + path = s3_path(s3_test_bucket, s3_test_prefix, "info/file.txt") + content = b"hello info" + + await fs._pipe_file(path, content) + info = await fs._info(path) + + assert info["type"] == "file" + assert info["size"] == len(content) + assert info["name"] == path + + @pytest.mark.asyncio + async def test_info_directory( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """_info on a prefix with children returns directory metadata.""" + fs = Boto3S3FileSystem(session=aws_credentials) + prefix = f"{s3_test_prefix}info_dir" + + # Create a child object + await fs._pipe_file(f"{s3_test_bucket}/{prefix}/child.txt", b"child") + + info = await fs._info(f"{s3_test_bucket}/{prefix}") + assert info["type"] == "directory" + + @pytest.mark.asyncio + async def test_info_not_found( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """_info on a non-existent path raises FileNotFoundError.""" + fs = Boto3S3FileSystem(session=aws_credentials) + path = s3_path(s3_test_bucket, s3_test_prefix, "info/nonexistent.txt") + + with pytest.raises(FileNotFoundError): + await fs._info(path) + + @pytest.mark.asyncio + async def test_exists_true(self, s3_test_bucket, s3_test_prefix, aws_credentials): + """_exists returns True for existing objects.""" + fs = Boto3S3FileSystem(session=aws_credentials) + path = s3_path(s3_test_bucket, s3_test_prefix, "exists/file.txt") + + await fs._pipe_file(path, b"exists") + assert await fs._exists(path) is True + + @pytest.mark.asyncio + async def test_exists_false(self, s3_test_bucket, s3_test_prefix, aws_credentials): + """_exists returns False for non-existent paths.""" + fs = Boto3S3FileSystem(session=aws_credentials) + path = s3_path(s3_test_bucket, s3_test_prefix, "exists/nope.txt") + + assert await fs._exists(path) is False + + @pytest.mark.asyncio + async def test_exists_prefix(self, s3_test_bucket, s3_test_prefix, aws_credentials): + """_exists returns True for non-empty prefixes.""" + fs = Boto3S3FileSystem(session=aws_credentials) + prefix = f"{s3_test_prefix}exists_prefix" + + await fs._pipe_file(f"{s3_test_bucket}/{prefix}/child.txt", b"data") + assert await fs._exists(f"{s3_test_bucket}/{prefix}") is True + + +# --- Removal tests --- + + +class TestRemoval: + """Test _rm and recursive deletion on real S3.""" + + @pytest.mark.asyncio + async def test_rm_single_file( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """_rm_file deletes a single object.""" + fs = Boto3S3FileSystem(session=aws_credentials) + path = s3_path(s3_test_bucket, s3_test_prefix, "rm/single.txt") + + await fs._pipe_file(path, b"to delete") + assert await fs._exists(path) is True + + await fs._rm_file(path) + assert await fs._exists(path) is False + + @pytest.mark.asyncio + async def test_rm_nonexistent_raises( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """_rm_file on non-existent path raises FileNotFoundError.""" + fs = Boto3S3FileSystem(session=aws_credentials) + path = s3_path(s3_test_bucket, s3_test_prefix, "rm/ghost.txt") + + with pytest.raises(FileNotFoundError): + await fs._rm_file(path) + + @pytest.mark.asyncio + async def test_rm_recursive(self, s3_test_bucket, s3_test_prefix, aws_credentials): + """_rm with recursive=True deletes all objects under a prefix.""" + fs = Boto3S3FileSystem(session=aws_credentials) + prefix = f"{s3_test_prefix}rm_recursive" + base = f"{s3_test_bucket}/{prefix}" + + # Create multiple objects + await fs._pipe_file(f"{base}/a.txt", b"a") + await fs._pipe_file(f"{base}/b.txt", b"b") + await fs._pipe_file(f"{base}/sub/c.txt", b"c") + + # Recursive delete + fs.rm(f"{base}", recursive=True) + + # Verify all are gone + assert await fs._exists(f"{base}/a.txt") is False + assert await fs._exists(f"{base}/b.txt") is False + assert await fs._exists(f"{base}/sub/c.txt") is False + + +# --- UPath integration tests --- + + +class TestUPathIntegration: + """Test UPath integration with real S3 bucket.""" + + def test_upath_write_text_read_text( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """UPath read_text/write_text round-trip.""" + from upath import UPath + + path = UPath( + f"s3://{s3_test_bucket}/{s3_test_prefix}upath/text.txt", + session=aws_credentials, + ) + text = "Hello from UPath!\nLine 2" + + path.write_text(text) + result = path.read_text() + + assert result == text + + def test_upath_iterdir(self, s3_test_bucket, s3_test_prefix, aws_credentials): + """UPath iterdir lists directory contents.""" + from upath import UPath + + base = UPath( + f"s3://{s3_test_bucket}/{s3_test_prefix}upath_iterdir", + session=aws_credentials, + ) + + # Create files + (base / "one.txt").write_text("one") + (base / "two.txt").write_text("two") + + # Iterdir + names = sorted(p.name for p in base.iterdir()) + assert "one.txt" in names + assert "two.txt" in names + + def test_upath_glob(self, s3_test_bucket, s3_test_prefix, aws_credentials): + """UPath glob matches patterns.""" + from upath import UPath + + base = UPath( + f"s3://{s3_test_bucket}/{s3_test_prefix}upath_glob", + session=aws_credentials, + ) + + # Create files with different extensions + (base / "data.json").write_text('{"a": 1}') + (base / "data.csv").write_text("a,b,c") + (base / "sub" / "nested.json").write_text('{"b": 2}') + + # Glob for json files (non-recursive) + json_files = sorted(p.name for p in base.glob("*.json")) + assert "data.json" in json_files + assert "data.csv" not in json_files + + # Recursive glob + all_json = sorted(p.name for p in base.glob("**/*.json")) + assert "data.json" in all_json + assert "nested.json" in all_json + + +# --- Concurrent async operations --- + + +class TestConcurrentAsync: + """Test concurrent async operations via asyncio.gather on real S3.""" + + @pytest.mark.asyncio + async def test_concurrent_writes( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """Multiple concurrent pipe_file operations complete successfully.""" + fs = Boto3S3FileSystem(session=aws_credentials) + prefix = f"{s3_test_prefix}concurrent/writes" + + tasks = [ + fs._pipe_file( + f"{s3_test_bucket}/{prefix}/file_{i}.txt", + f"content_{i}".encode(), + ) + for i in range(10) + ] + + await asyncio.gather(*tasks) + + # Verify all files exist with correct content + for i in range(10): + path = f"{s3_test_bucket}/{prefix}/file_{i}.txt" + data = await fs._cat_file(path) + assert data == f"content_{i}".encode() + + @pytest.mark.asyncio + async def test_concurrent_reads( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """Multiple concurrent cat_file operations complete successfully.""" + fs = Boto3S3FileSystem(session=aws_credentials) + prefix = f"{s3_test_prefix}concurrent/reads" + + # Write files first + for i in range(10): + await fs._pipe_file( + f"{s3_test_bucket}/{prefix}/file_{i}.txt", + f"read_content_{i}".encode(), + ) + + # Read all concurrently + tasks = [ + fs._cat_file(f"{s3_test_bucket}/{prefix}/file_{i}.txt") for i in range(10) + ] + results = await asyncio.gather(*tasks) + + for i, result in enumerate(results): + assert result == f"read_content_{i}".encode() + + @pytest.mark.asyncio + async def test_concurrent_mixed_operations( + self, s3_test_bucket, s3_test_prefix, aws_credentials + ): + """Mixed concurrent operations (write, read, exists) work correctly.""" + fs = Boto3S3FileSystem(session=aws_credentials) + prefix = f"{s3_test_prefix}concurrent/mixed" + + # Write a file first + path = f"{s3_test_bucket}/{prefix}/target.txt" + await fs._pipe_file(path, b"target data") + + # Run mixed operations concurrently + results = await asyncio.gather( + fs._cat_file(path), + fs._exists(path), + fs._info(path), + fs._pipe_file(f"{s3_test_bucket}/{prefix}/new.txt", b"new"), + ) + + assert results[0] == b"target data" + assert results[1] is True + assert results[2]["type"] == "file" diff --git a/tests/test_s3fs.py b/tests/test_s3fs.py new file mode 100644 index 0000000..721f806 --- /dev/null +++ b/tests/test_s3fs.py @@ -0,0 +1,618 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for Boto3S3FileSystem using moto mock_aws.""" + +from __future__ import annotations + +import asyncio +import sys +import types +from unittest.mock import patch + +import boto3 +import pytest +from moto import mock_aws + +from llmeter.s3fs import Boto3S3FileSystem + +BUCKET = "test-bucket" + + +# --- Test _cat_file / _pipe_file round-trip --- + + +class TestCatFilePipeFile: + """Tests for _cat_file and _pipe_file round-trip operations.""" + + @mock_aws + def test_roundtrip_binary(self): + """Binary data round-trips correctly.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + data = b"\x00\x01\x02\xff" * 100 + fs.pipe_file(f"{BUCKET}/binary.dat", data) + result = fs.cat_file(f"{BUCKET}/binary.dat") + assert result == data + + @mock_aws + def test_roundtrip_text(self): + """UTF-8 text round-trips correctly via pipe/cat.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + text = "Hello, world! \U0001f30d\nLine 2\n" + data = text.encode("utf-8") + fs.pipe_file(f"{BUCKET}/text.txt", data) + result = fs.cat_file(f"{BUCKET}/text.txt") + assert result == data + assert result.decode("utf-8") == text + + @mock_aws + def test_roundtrip_empty_file(self): + """Empty file round-trips correctly.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/empty.dat", b"") + result = fs.cat_file(f"{BUCKET}/empty.dat") + assert result == b"" + + +# --- Test _info --- + + +class TestInfo: + """Tests for _info metadata retrieval.""" + + @mock_aws + def test_info_file(self): + """_info returns correct metadata for a file.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/dir/file.txt", b"hello") + info = fs.info(f"{BUCKET}/dir/file.txt") + assert info["type"] == "file" + assert info["size"] == 5 + assert "name" in info + + @mock_aws + def test_info_directory(self): + """_info returns directory type for a prefix with children.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/dir/file.txt", b"hello") + info = fs.info(f"{BUCKET}/dir") + assert info["type"] == "directory" + assert info["size"] == 0 + + @mock_aws + def test_info_nonexistent_raises(self): + """_info raises FileNotFoundError for nonexistent paths.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + with pytest.raises(FileNotFoundError): + fs.info(f"{BUCKET}/nonexistent/path") + + +# --- Test _exists --- + + +class TestExists: + """Tests for _exists path existence checks.""" + + @mock_aws + def test_exists_file(self): + """_exists returns True for an existing file.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/exists.txt", b"data") + assert fs.exists(f"{BUCKET}/exists.txt") is True + + @mock_aws + def test_exists_directory(self): + """_exists returns True for a non-empty prefix (directory).""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/dir/file.txt", b"data") + assert fs.exists(f"{BUCKET}/dir") is True + + @mock_aws + def test_exists_nonexistent(self): + """_exists returns False for non-existent paths.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + assert fs.exists(f"{BUCKET}/nope") is False + + +# --- Test _ls --- + + +class TestLs: + """Tests for _ls directory listing operations.""" + + @mock_aws + def test_ls_detail_false(self): + """_ls with detail=False returns list of path strings.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/dir/a.txt", b"a") + fs.pipe_file(f"{BUCKET}/dir/b.txt", b"b") + result = fs.ls(f"{BUCKET}/dir", detail=False) + assert sorted(result) == sorted([f"{BUCKET}/dir/a.txt", f"{BUCKET}/dir/b.txt"]) + + @mock_aws + def test_ls_detail_true(self): + """_ls with detail=True returns list of info dicts.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/dir/file.txt", b"hello") + result = fs.ls(f"{BUCKET}/dir", detail=True) + assert len(result) == 1 + assert result[0]["name"] == f"{BUCKET}/dir/file.txt" + assert result[0]["type"] == "file" + assert result[0]["size"] == 5 + + @mock_aws + def test_ls_with_subdirectories(self): + """_ls shows both files and subdirectory prefixes.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/dir/file.txt", b"data") + fs.pipe_file(f"{BUCKET}/dir/subdir/nested.txt", b"nested") + result = fs.ls(f"{BUCKET}/dir", detail=False) + assert f"{BUCKET}/dir/file.txt" in result + assert f"{BUCKET}/dir/subdir" in result + + @mock_aws + def test_ls_empty_prefix_raises(self): + """_ls raises FileNotFoundError for empty prefix.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + with pytest.raises(FileNotFoundError): + fs.ls(f"{BUCKET}/nonexistent") + + @mock_aws + def test_ls_bare_bucket(self): + """_ls with bare bucket name lists top-level objects and prefixes.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/top.txt", b"top") + fs.pipe_file(f"{BUCKET}/subdir/nested.txt", b"nested") + result = fs.ls(BUCKET, detail=False) + assert f"{BUCKET}/top.txt" in result + assert f"{BUCKET}/subdir" in result + + @mock_aws + def test_ls_pagination(self): + """_ls handles pagination for >1000 objects.""" + client = boto3.client("s3", region_name="us-east-1") + client.create_bucket(Bucket=BUCKET) + # Create 1005 objects to trigger pagination + for i in range(1005): + client.put_object(Bucket=BUCKET, Key=f"many/{i:04d}.txt", Body=b"x") + fs = Boto3S3FileSystem() + result = fs.ls(f"{BUCKET}/many", detail=False) + assert len(result) == 1005 + + +# --- Test _rm_file and recursive _rm --- + + +class TestRm: + """Tests for _rm_file and recursive _rm operations.""" + + @mock_aws + def test_rm_file(self): + """_rm_file deletes a single object.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/to_delete.txt", b"delete me") + assert fs.exists(f"{BUCKET}/to_delete.txt") is True + fs.rm_file(f"{BUCKET}/to_delete.txt") + assert fs.exists(f"{BUCKET}/to_delete.txt") is False + + @mock_aws + def test_rm_file_nonexistent_raises(self): + """_rm_file raises FileNotFoundError for missing objects.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + with pytest.raises(FileNotFoundError): + fs.rm_file(f"{BUCKET}/nope.txt") + + @mock_aws + def test_rm_recursive(self): + """_rm with recursive=True deletes all objects under a prefix.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/dir/a.txt", b"a") + fs.pipe_file(f"{BUCKET}/dir/b.txt", b"b") + fs.pipe_file(f"{BUCKET}/dir/sub/c.txt", b"c") + fs.rm(f"{BUCKET}/dir", recursive=True) + assert fs.exists(f"{BUCKET}/dir") is False + + +# --- Test _put_file and _get_file --- + + +class TestPutGetFile: + """Tests for _put_file and _get_file (local to S3 transfers).""" + + @mock_aws + def test_put_file(self, tmp_path): + """_put_file uploads a local file to S3.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + local_file = tmp_path / "upload.txt" + local_file.write_bytes(b"local content") + fs.put_file(str(local_file), f"{BUCKET}/uploaded.txt") + result = fs.cat_file(f"{BUCKET}/uploaded.txt") + assert result == b"local content" + + @mock_aws + def test_get_file(self, tmp_path): + """_get_file downloads an S3 object to a local file.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/remote.txt", b"remote content") + local_file = tmp_path / "download.txt" + fs.get_file(f"{BUCKET}/remote.txt", str(local_file)) + assert local_file.read_bytes() == b"remote content" + + +# --- Test _open with various modes --- + + +class TestOpen: + """Tests for _open with various file modes.""" + + @mock_aws + def test_open_rb(self): + """open(mode='rb') reads binary content.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/file.bin", b"\x00\x01\x02") + with fs.open(f"{BUCKET}/file.bin", "rb") as f: + assert f.read() == b"\x00\x01\x02" + + @mock_aws + def test_open_wb(self): + """open(mode='wb') writes binary content.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + with fs.open(f"{BUCKET}/file.bin", "wb") as f: + f.write(b"\x03\x04\x05") + assert fs.cat_file(f"{BUCKET}/file.bin") == b"\x03\x04\x05" + + @mock_aws + def test_open_r_text(self): + """open(mode='r') reads text content with UTF-8 encoding.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + text = "Hello, world!\n" + fs.pipe_file(f"{BUCKET}/text.txt", text.encode("utf-8")) + with fs.open(f"{BUCKET}/text.txt", "r") as f: + assert f.read() == text + + @mock_aws + def test_open_w_text(self): + """open(mode='w') writes text content with UTF-8 encoding.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + text = "Hello, world!\n" + with fs.open(f"{BUCKET}/text.txt", "w") as f: + f.write(text) + result = fs.cat_file(f"{BUCKET}/text.txt") + assert result.decode("utf-8") == text + + @mock_aws + def test_open_append_existing(self): + """open(mode='ab') appends to existing file content.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/append.txt", b"first") + with fs.open(f"{BUCKET}/append.txt", "ab") as f: + f.write(b"second") + result = fs.cat_file(f"{BUCKET}/append.txt") + assert result == b"firstsecond" + + @mock_aws + def test_open_append_new_file(self): + """open(mode='ab') creates a new file if it doesn't exist.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + with fs.open(f"{BUCKET}/new_append.txt", "ab") as f: + f.write(b"new content") + result = fs.cat_file(f"{BUCKET}/new_append.txt") + assert result == b"new content" + + +# --- Test glob patterns --- + + +class TestGlob: + """Tests for glob pattern matching.""" + + @mock_aws + def test_glob_star(self): + """glob with * matches files in a directory.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/data/a.json", b"{}") + fs.pipe_file(f"{BUCKET}/data/b.json", b"{}") + fs.pipe_file(f"{BUCKET}/data/c.txt", b"text") + result = fs.glob(f"{BUCKET}/data/*.json") + assert sorted(result) == sorted( + [f"{BUCKET}/data/a.json", f"{BUCKET}/data/b.json"] + ) + + @mock_aws + def test_glob_recursive(self): + """glob with ** matches files recursively.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/root/a.json", b"{}") + fs.pipe_file(f"{BUCKET}/root/sub/b.json", b"{}") + fs.pipe_file(f"{BUCKET}/root/sub/deep/c.json", b"{}") + result = fs.glob(f"{BUCKET}/root/**/*.json") + assert len(result) == 3 + assert f"{BUCKET}/root/a.json" in result + assert f"{BUCKET}/root/sub/b.json" in result + assert f"{BUCKET}/root/sub/deep/c.json" in result + + @mock_aws + def test_glob_no_match(self): + """glob returns empty list when no matches.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + fs.pipe_file(f"{BUCKET}/data/a.txt", b"text") + result = fs.glob(f"{BUCKET}/data/*.csv") + assert result == [] + + +# --- Test UPath integration --- + + +class TestUPathIntegration: + """Tests for universal-pathlib UPath integration.""" + + @mock_aws + def test_upath_write_text_read_text(self): + """UPath read_text/write_text work with Boto3S3FileSystem.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + from upath import UPath + + p = UPath(f"s3://{BUCKET}/upath_test.txt") + p.write_text("hello from upath") + assert p.read_text() == "hello from upath" + + @mock_aws + def test_upath_exists(self): + """UPath.exists() works with Boto3S3FileSystem.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + from upath import UPath + + p = UPath(f"s3://{BUCKET}/exists_test.txt") + assert not p.exists() + p.write_bytes(b"data") + assert p.exists() + + @mock_aws + def test_upath_iterdir(self): + """UPath.iterdir() lists directory contents.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + from upath import UPath + + base = UPath(f"s3://{BUCKET}/iterdir_test") + (base / "a.txt").write_bytes(b"a") + (base / "b.txt").write_bytes(b"b") + names = sorted(p.name for p in base.iterdir()) + assert names == ["a.txt", "b.txt"] + + +# --- Test async operations --- + + +class TestAsync: + """Tests for async operations via pytest-asyncio.""" + + @pytest.mark.asyncio + async def test_async_cat_pipe_roundtrip(self): + """Async _cat_file/_pipe_file round-trip works.""" + with mock_aws(): + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem(asynchronous=True) + await fs._pipe_file(f"{BUCKET}/async.txt", b"async data") + result = await fs._cat_file(f"{BUCKET}/async.txt") + assert result == b"async data" + + @pytest.mark.asyncio + async def test_async_concurrent_operations(self): + """Multiple async operations can run concurrently.""" + with mock_aws(): + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem(asynchronous=True) + # Write multiple files concurrently + paths = [f"{BUCKET}/concurrent/{i}.txt" for i in range(5)] + await asyncio.gather( + *[fs._pipe_file(p, f"data-{i}".encode()) for i, p in enumerate(paths)] + ) + # Read them back concurrently + results = await asyncio.gather(*[fs._cat_file(p) for p in paths]) + for i, result in enumerate(results): + assert result == f"data-{i}".encode() + + @pytest.mark.asyncio + async def test_async_ls(self): + """Async _ls works correctly.""" + with mock_aws(): + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem(asynchronous=True) + await fs._pipe_file(f"{BUCKET}/async_dir/file1.txt", b"1") + await fs._pipe_file(f"{BUCKET}/async_dir/file2.txt", b"2") + result = await fs._ls(f"{BUCKET}/async_dir", detail=False) + assert sorted(result) == sorted( + [f"{BUCKET}/async_dir/file1.txt", f"{BUCKET}/async_dir/file2.txt"] + ) + + @pytest.mark.asyncio + async def test_async_exists(self): + """Async _exists works correctly.""" + with mock_aws(): + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem(asynchronous=True) + assert await fs._exists(f"{BUCKET}/nope") is False + await fs._pipe_file(f"{BUCKET}/yes.txt", b"yes") + assert await fs._exists(f"{BUCKET}/yes.txt") is True + + @pytest.mark.asyncio + async def test_async_info(self): + """Async _info works correctly.""" + with mock_aws(): + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem(asynchronous=True) + await fs._pipe_file(f"{BUCKET}/info.txt", b"12345") + info = await fs._info(f"{BUCKET}/info.txt") + assert info["type"] == "file" + assert info["size"] == 5 + + @pytest.mark.asyncio + async def test_async_rm(self): + """Async _rm_file works correctly.""" + with mock_aws(): + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem(asynchronous=True) + await fs._pipe_file(f"{BUCKET}/rm_async.txt", b"del") + assert await fs._exists(f"{BUCKET}/rm_async.txt") is True + await fs._rm_file(f"{BUCKET}/rm_async.txt") + assert await fs._exists(f"{BUCKET}/rm_async.txt") is False + + +# --- Test error handling --- + + +class TestErrorHandling: + """Tests for FileNotFoundError and PermissionError exceptions.""" + + @mock_aws + def test_cat_file_not_found(self): + """_cat_file raises FileNotFoundError for missing objects.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + with pytest.raises(FileNotFoundError): + fs.cat_file(f"{BUCKET}/missing.txt") + + @mock_aws + def test_info_not_found(self): + """_info raises FileNotFoundError for missing paths.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + with pytest.raises(FileNotFoundError): + fs.info(f"{BUCKET}/no/such/path") + + @mock_aws + def test_rm_file_not_found(self): + """_rm_file raises FileNotFoundError for missing objects.""" + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + with pytest.raises(FileNotFoundError): + fs.rm_file(f"{BUCKET}/nope.txt") + + @mock_aws + def test_permission_error_on_cat(self): + """_cat_file raises PermissionError on 403.""" + from botocore.exceptions import ClientError + + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + + def mock_get(**kwargs): + raise ClientError( + {"Error": {"Code": "403", "Message": "Access Denied"}}, + "GetObject", + ) + + fs._s3_client.get_object = mock_get + with pytest.raises(PermissionError): + fs.cat_file(f"{BUCKET}/forbidden.txt") + + @mock_aws + def test_permission_error_on_info(self): + """_info raises PermissionError on AccessDenied.""" + from botocore.exceptions import ClientError + + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + fs = Boto3S3FileSystem() + + def mock_head(**kwargs): + raise ClientError( + {"Error": {"Code": "403", "Message": "Access Denied"}}, + "HeadObject", + ) + + fs._s3_client.head_object = mock_head + with pytest.raises(PermissionError): + fs.info(f"{BUCKET}/forbidden.txt") + + +# --- Test registration fallback behavior --- + + +class TestRegistration: + """Tests for registration fallback behavior.""" + + def test_registration_skips_when_s3fs_available(self): + """Registration is skipped when s3fs is importable.""" + fake_s3fs = types.ModuleType("s3fs") + with patch.dict(sys.modules, {"s3fs": fake_s3fs}): + import llmeter + + # The function should return early without registering + llmeter._register_s3_filesystem() + + def test_registration_skips_when_obstore_available(self): + """Registration is skipped when obstore.fsspec is importable.""" + fake_obstore = types.ModuleType("obstore") + fake_obstore_fsspec = types.ModuleType("obstore.fsspec") + fake_obstore_fsspec.FsspecStore = object + with patch.dict( + sys.modules, + { + "obstore": fake_obstore, + "obstore.fsspec": fake_obstore_fsspec, + "s3fs": None, # Simulate s3fs not available (raises ImportError) + }, + ): + import llmeter + + llmeter._register_s3_filesystem() + + @mock_aws + def test_registration_registers_when_no_alternatives(self): + """Registration registers Boto3S3FileSystem when no alternatives exist.""" + # Ensure s3fs and obstore are NOT importable + with patch.dict( + sys.modules, + {"s3fs": None, "obstore": None, "obstore.fsspec": None}, + ): + import llmeter + + llmeter._register_s3_filesystem() + # Verify registration worked by using the filesystem + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + import fsspec + + fs = fsspec.filesystem("s3") + assert isinstance(fs, Boto3S3FileSystem) + + def test_registration_handles_error_gracefully(self): + """Registration does not raise if fsspec.register_implementation fails.""" + with patch.dict( + sys.modules, + {"s3fs": None, "obstore": None, "obstore.fsspec": None}, + ): + with patch( + "fsspec.register_implementation", side_effect=RuntimeError("oops") + ): + import llmeter + + # Should not raise - just logs a warning + llmeter._register_s3_filesystem() From efc8aee652e7df3bd22d771b09bd62794e40a832 Mon Sep 17 00:00:00 2001 From: Alessandro Cere Date: Wed, 3 Jun 2026 19:53:25 +0800 Subject: [PATCH 2/3] docs: add S3 storage user guide and API reference - Add docs/user_guide/s3_storage.md covering usage, credentials, custom config, fallback behavior, limitations, and async support - Add docs/reference/s3fs.md for auto-generated API docs - Update installation guide with S3 storage section - Add both pages to mkdocs.yml navigation --- docs/reference/s3fs.md | 3 + docs/user_guide/installation.md | 16 ++++++ docs/user_guide/s3_storage.md | 97 +++++++++++++++++++++++++++++++++ mkdocs.yml | 2 + 4 files changed, 118 insertions(+) create mode 100644 docs/reference/s3fs.md create mode 100644 docs/user_guide/s3_storage.md diff --git a/docs/reference/s3fs.md b/docs/reference/s3fs.md new file mode 100644 index 0000000..05dcd56 --- /dev/null +++ b/docs/reference/s3fs.md @@ -0,0 +1,3 @@ +# s3fs + +::: llmeter.s3fs diff --git a/docs/user_guide/installation.md b/docs/user_guide/installation.md index 56f99f5..996d824 100644 --- a/docs/user_guide/installation.md +++ b/docs/user_guide/installation.md @@ -52,3 +52,19 @@ LLMeter measures **end-to-end latencies** and uses pure-Python ([asyncio](https: > If hosting LLMeter on [burstable](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/burstable-performance-instances.html) Cloud instance types (e.g. AWS `t3`, `t4g`, etc), be aware that sustained network and compute limits are lower than short-term burst resources. Because generative AI inference is a compute-intensive task, load testing LLMs has not typically required very high-powered (compute, network) resources from the client side. This led us to a deliberate decision to keep LLMeter simple to install and use by avoiding more complex distributed computing frameworks, in favour of plain Python+asyncio. If you have a use-case with such a fast LLM or large request volume that this approach is limiting to you - please share your feedback! + +## S3 Storage Support + +LLMeter includes built-in support for Amazon S3 paths (`s3://`) with no extra dependencies. The `boto3` package (already a core dependency) provides the S3 client, and LLMeter's built-in filesystem backend handles the rest. + +No additional installation is needed — just ensure your AWS credentials are configured (via environment variables, `~/.aws/credentials`, or an IAM role). + +```python +from upath import UPath as Path + +# S3 paths work out of the box +results = Path("s3://my-bucket/experiments/run-001/metrics.json") +data = results.read_text() +``` + +See the [S3 Storage guide](s3_storage.md) for details on configuration, async usage, and fallback behavior. diff --git a/docs/user_guide/s3_storage.md b/docs/user_guide/s3_storage.md new file mode 100644 index 0000000..17947ec --- /dev/null +++ b/docs/user_guide/s3_storage.md @@ -0,0 +1,97 @@ +# S3 Storage + +LLMeter supports reading and writing experiment results to Amazon S3 using standard path operations. This works transparently through the [UPath](https://github.com/fsspec/universal_pathlib) integration — you can use `s3://` paths anywhere you'd use local file paths. + +## How It Works + +LLMeter includes a built-in S3 filesystem backend (`Boto3S3FileSystem`) that uses `boto3` directly. This replaces the previous `s3fs`/`aiobotocore` dependency chain, eliminating the version conflicts those packages caused with `boto3`. + +The backend registers itself automatically when you import `llmeter`. No extra installation or configuration is needed beyond having valid AWS credentials. + +```python +from upath import UPath as Path + +# These just work — no extra packages needed +results_path = Path("s3://my-bucket/experiments/run-001/") +results_path.mkdir(parents=True, exist_ok=True) + +# Write results +(results_path / "metrics.json").write_text('{"latency_p50": 0.42}') + +# Read results +data = (results_path / "metrics.json").read_text() + +# List files +for p in results_path.iterdir(): + print(p.name) + +# Glob patterns +json_files = list(results_path.glob("**/*.json")) +``` + +## Credentials + +The backend uses standard AWS credential resolution — the same mechanisms `boto3` uses: + +1. Environment variables (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`) +2. AWS config files (`~/.aws/credentials`, `~/.aws/config`) +3. IAM instance profiles (on EC2/ECS/Lambda) +4. SSO/credential process configurations + +If `boto3` works in your environment, S3 paths in LLMeter will too. + +## Custom Configuration + +For advanced use cases, you can instantiate the filesystem directly: + +```python +import boto3 +from llmeter.s3fs import Boto3S3FileSystem + +# Custom session (e.g., specific profile) +session = boto3.Session(profile_name="my-profile") +fs = Boto3S3FileSystem(session=session) + +# Custom region +fs = Boto3S3FileSystem(region_name="eu-west-1") + +# S3-compatible services (MinIO, LocalStack) +fs = Boto3S3FileSystem(endpoint_url="http://localhost:9000") +``` + +## Fallback Behavior + +If you prefer a different S3 backend, the built-in one steps aside: + +- If `s3fs` is installed, LLMeter defers to it +- If `obstore` is installed, LLMeter defers to it +- Otherwise, LLMeter's built-in `Boto3S3FileSystem` handles S3 paths + +This means existing setups with `s3fs` continue to work unchanged. + +## Limitations + +- **File size**: The backend uses `put_object` for writes, which has a 5 GiB limit. This is fine for LLMeter's typical use case (JSON/JSONL result files in the KB–MB range). +- **Append mode**: `open(mode="a")` uses read-modify-write semantics. Concurrent writers to the same key may cause data loss. +- **No multipart upload**: Files larger than 5 GiB require multipart upload, which is not implemented. + +## Async Support + +The backend is async-native, wrapping synchronous boto3 calls in `asyncio.to_thread()` for non-blocking execution. This means it works well in async code: + +```python +import asyncio +from llmeter.s3fs import Boto3S3FileSystem + +async def main(): + fs = Boto3S3FileSystem() + + # Concurrent reads + results = await asyncio.gather( + fs._cat_file("my-bucket/file1.json"), + fs._cat_file("my-bucket/file2.json"), + fs._cat_file("my-bucket/file3.json"), + ) + +asyncio.run(main()) +``` diff --git a/mkdocs.yml b/mkdocs.yml index 37ec0a4..81cf08f 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -37,6 +37,7 @@ nav: - Metrics and Statistics: user_guide/metrics.md - Connect to Endpoints: user_guide/connect_endpoints.md - Run Experiments: user_guide/run_experiments.md + - S3 Storage: user_guide/s3_storage.md - API Reference: - reference/index.md - callbacks: @@ -69,6 +70,7 @@ nav: - prompt_utils: reference/prompt_utils.md - results: reference/results.md - runner: reference/runner.md + - s3fs: reference/s3fs.md - tokenizers: reference/tokenizers.md - utils: reference/utils.md repo_url: https://github.com/awslabs/llmeter From d589ab383765906cf2ae20ae85fa069ed4b227ac Mon Sep 17 00:00:00 2001 From: Alessandro Cere Date: Wed, 3 Jun 2026 21:14:02 +0800 Subject: [PATCH 3/3] refactor(s3fs): simplify error handling and remove defensive patterns - Extract _translate_error() helper to replace repeated error code checks - Remove unnecessary boto3 ImportError guard (it's a core dependency) - Remove head_object pre-check in _rm_file (S3 delete is idempotent) - Let _put_file propagate boto3 errors directly (no wrapping needed) - Remove unused `import fsspec` (only fsspec.spec is needed) - Raise ValueError on bare bucket delete instead of silent return - Update tests to reflect idempotent delete semantics --- llmeter/s3fs.py | 114 ++++++++------------------------- tests/integ/test_s3fs_integ.py | 8 +-- tests/test_s3fs.py | 16 ++--- 3 files changed, 39 insertions(+), 99 deletions(-) diff --git a/llmeter/s3fs.py b/llmeter/s3fs.py index 65958f7..24ced1b 100644 --- a/llmeter/s3fs.py +++ b/llmeter/s3fs.py @@ -15,22 +15,24 @@ import logging from typing import Any -try: - import boto3 - from botocore.exceptions import ClientError -except ImportError as e: - raise ImportError( - "boto3 is required for the llmeter S3 filesystem backend. " - "Install it with: pip install boto3" - ) from e - -import fsspec +import boto3 import fsspec.spec +from botocore.exceptions import ClientError from fsspec.asyn import AsyncFileSystem logger = logging.getLogger(__name__) +def _translate_error(e: ClientError, path: str) -> OSError: + """Map a boto3 ClientError to the appropriate Python exception.""" + code = e.response["Error"]["Code"] + if code in ("NoSuchKey", "404", "NoSuchBucket"): + return FileNotFoundError(path) + if code in ("403", "AccessDenied"): + return PermissionError(path) + return OSError(f"S3 error {code}: {e.response['Error'].get('Message', '')}") + + class S3File(fsspec.spec.AbstractBufferedFile): """File-like object for S3 backed by in-memory buffer.""" @@ -39,22 +41,11 @@ def _fetch_range(self, start: int, end: int) -> bytes: bucket, key = self.fs._split_path(self.path) try: response = self.fs._s3_client.get_object( - Bucket=bucket, - Key=key, - Range=f"bytes={start}-{end - 1}", + Bucket=bucket, Key=key, Range=f"bytes={start}-{end - 1}" ) return response["Body"].read() - except self.fs._s3_client.exceptions.NoSuchKey: - raise FileNotFoundError(self.path) except ClientError as e: - code = e.response["Error"]["Code"] - if code in ("403", "AccessDenied"): - raise PermissionError(self.path) from e - if code in ("404", "NoSuchBucket"): - raise FileNotFoundError(self.path) from e - raise OSError( - f"S3 error {code}: {e.response['Error'].get('Message', '')}" - ) from e + raise _translate_error(e, self.path) from e def _initiate_upload(self) -> None: """Initialize upload state. Buffer is already set by AbstractBufferedFile.""" @@ -69,10 +60,7 @@ def _upload_chunk(self, final: bool = False) -> None: try: self.fs._s3_client.put_object(Bucket=bucket, Key=key, Body=data) except ClientError as e: - code = e.response["Error"]["Code"] - raise OSError( - f"S3 upload error {code}: {e.response['Error'].get('Message', '')}" - ) from e + raise _translate_error(e, self.path) from e class Boto3S3FileSystem(AsyncFileSystem): @@ -161,17 +149,8 @@ def _do_get(): try: response = self._s3_client.get_object(**get_kwargs) return response["Body"].read() - except self._s3_client.exceptions.NoSuchKey: - raise FileNotFoundError(path) except ClientError as e: - code = e.response["Error"]["Code"] - if code in ("403", "AccessDenied"): - raise PermissionError(path) from e - if code in ("404", "NoSuchBucket"): - raise FileNotFoundError(path) from e - raise OSError( - f"S3 error {code}: {e.response['Error'].get('Message', '')}" - ) from e + raise _translate_error(e, path) from e return await asyncio.to_thread(_do_get) @@ -183,10 +162,7 @@ def _do_put(): try: self._s3_client.put_object(Bucket=bucket, Key=key, Body=value) except ClientError as e: - code = e.response["Error"]["Code"] - raise OSError( - f"S3 upload error {code}: {e.response['Error'].get('Message', '')}" - ) from e + raise _translate_error(e, path) from e await asyncio.to_thread(_do_put) @@ -207,13 +183,9 @@ def _do_info(): "ETag": response.get("ETag"), } except ClientError as e: - code = e.response["Error"]["Code"] - if code in ("403", "AccessDenied"): - raise PermissionError(path) from e - if code not in ("404", "NoSuchKey"): - raise OSError( - f"S3 error {code}: {e.response['Error'].get('Message', '')}" - ) from e + err = _translate_error(e, path) + if not isinstance(err, FileNotFoundError): + raise err from e # Try as directory (prefix check) prefix = f"{key}/" if key else "" @@ -267,7 +239,6 @@ def _do_ls(): # Objects (files) for obj in response.get("Contents", []): obj_key = obj["Key"] - # Skip the prefix itself if it appears as an object if obj_key == prefix: continue name = f"{bucket}/{obj_key}" @@ -306,25 +277,11 @@ async def _mkdir( async def _rm_file(self, path: str, **kwargs: Any) -> None: """Delete a single object from S3.""" bucket, key = self._split_path(path) - if not key: - # Cannot delete a bare bucket via rm_file - return + raise ValueError(f"Cannot delete a bare bucket: {path}") def _do_rm(): - # Check existence first - try: - self._s3_client.head_object(Bucket=bucket, Key=key) - except ClientError as e: - code = e.response["Error"]["Code"] - if code in ("404", "NoSuchKey"): - raise FileNotFoundError(path) from e - if code in ("403", "AccessDenied"): - raise PermissionError(path) from e - raise OSError( - f"S3 error {code}: {e.response['Error'].get('Message', '')}" - ) from e - + # delete_object is idempotent on S3 — no existence check needed self._s3_client.delete_object(Bucket=bucket, Key=key) await asyncio.to_thread(_do_rm) @@ -339,9 +296,8 @@ async def _rm(self, path, recursive=False, batch_size=None, **kwargs): async def _safe_rm(p): try: await self._rm_file(p, **kwargs) - except FileNotFoundError: - # During recursive delete, expanded paths include directory - # prefixes that don't exist as objects — skip them silently. + except (FileNotFoundError, ValueError): + # Expanded paths may include directory prefixes or bare buckets pass return await _run_coros_in_chunks( @@ -363,12 +319,7 @@ def _do_copy(): CopySource={"Bucket": bucket1, "Key": key1}, ) except ClientError as e: - code = e.response["Error"]["Code"] - if code in ("404", "NoSuchKey"): - raise FileNotFoundError(path1) from e - raise OSError( - f"S3 copy error {code}: {e.response['Error'].get('Message', '')}" - ) from e + raise _translate_error(e, path1) from e await asyncio.to_thread(_do_copy) @@ -377,13 +328,7 @@ async def _put_file(self, lpath: str, rpath: str, **kwargs: Any) -> None: bucket, key = self._split_path(rpath) def _do_upload(): - try: - self._s3_client.upload_file(lpath, bucket, key) - except ClientError as e: - code = e.response["Error"]["Code"] - raise OSError( - f"S3 upload error {code}: {e.response['Error'].get('Message', '')}" - ) from e + self._s3_client.upload_file(lpath, bucket, key) await asyncio.to_thread(_do_upload) @@ -395,12 +340,7 @@ def _do_download(): try: self._s3_client.download_file(bucket, key, lpath) except ClientError as e: - code = e.response["Error"]["Code"] - if code in ("404", "NoSuchKey"): - raise FileNotFoundError(rpath) from e - raise OSError( - f"S3 download error {code}: {e.response['Error'].get('Message', '')}" - ) from e + raise _translate_error(e, rpath) from e await asyncio.to_thread(_do_download) diff --git a/tests/integ/test_s3fs_integ.py b/tests/integ/test_s3fs_integ.py index cf72c92..1caca9e 100644 --- a/tests/integ/test_s3fs_integ.py +++ b/tests/integ/test_s3fs_integ.py @@ -282,15 +282,15 @@ async def test_rm_single_file( assert await fs._exists(path) is False @pytest.mark.asyncio - async def test_rm_nonexistent_raises( + async def test_rm_nonexistent_is_noop( self, s3_test_bucket, s3_test_prefix, aws_credentials ): - """_rm_file on non-existent path raises FileNotFoundError.""" + """_rm_file on non-existent path is a no-op (S3 delete is idempotent).""" fs = Boto3S3FileSystem(session=aws_credentials) path = s3_path(s3_test_bucket, s3_test_prefix, "rm/ghost.txt") - with pytest.raises(FileNotFoundError): - await fs._rm_file(path) + # Should not raise + await fs._rm_file(path) @pytest.mark.asyncio async def test_rm_recursive(self, s3_test_bucket, s3_test_prefix, aws_credentials): diff --git a/tests/test_s3fs.py b/tests/test_s3fs.py index 721f806..8af5007 100644 --- a/tests/test_s3fs.py +++ b/tests/test_s3fs.py @@ -211,12 +211,12 @@ def test_rm_file(self): assert fs.exists(f"{BUCKET}/to_delete.txt") is False @mock_aws - def test_rm_file_nonexistent_raises(self): - """_rm_file raises FileNotFoundError for missing objects.""" + def test_rm_file_nonexistent_is_noop(self): + """_rm_file on non-existent key is a no-op (S3 delete is idempotent).""" boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) fs = Boto3S3FileSystem() - with pytest.raises(FileNotFoundError): - fs.rm_file(f"{BUCKET}/nope.txt") + # Should not raise — S3 delete_object is idempotent + fs.rm_file(f"{BUCKET}/nope.txt") @mock_aws def test_rm_recursive(self): @@ -509,12 +509,12 @@ def test_info_not_found(self): fs.info(f"{BUCKET}/no/such/path") @mock_aws - def test_rm_file_not_found(self): - """_rm_file raises FileNotFoundError for missing objects.""" + def test_rm_file_not_found_is_noop(self): + """_rm_file is idempotent — no error on missing objects.""" boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) fs = Boto3S3FileSystem() - with pytest.raises(FileNotFoundError): - fs.rm_file(f"{BUCKET}/nope.txt") + # Should not raise + fs.rm_file(f"{BUCKET}/nope.txt") @mock_aws def test_permission_error_on_cat(self):