### Installing and Running Tests with Poetry

Source: https://github.com/phillipuniverse/pytest-aioboto3/blob/main/README.md

Use Poetry to install dependencies and execute Pytest for the plugin. Requires Poetry 1.6+ and Python 3.10+. This sets up the environment and runs the example test suite.

```shell
poetry install
poetry run pytest
```

--------------------------------

### Reusable Fixtures for S3 Bucket Setup and Data Population

Source: https://context7.com/phillipuniverse/pytest-aioboto3/llms.txt

Provides a pattern for creating reusable pytest fixtures to manage S3 buckets. It includes fixtures for defining a bucket name, creating a bucket, and populating it with test data. The example also demonstrates testing bucket contents and prefix filtering using these fixtures. Requires `pytest`, `aioboto3`, and `types_aiobotocore_s3`.

```python
import pytest
from types_aiobotocore_s3 import S3Client

@pytest.fixture
def bucket_name() -> str:
    """Define the bucket name for tests."""
    return "test-app-bucket"

@pytest.fixture
async def test_bucket(bucket_name: str, aioboto3_s3_client: S3Client) -> str:
    """Create a bucket and verify it's ready."""
    resp = await aioboto3_s3_client.create_bucket(Bucket=bucket_name)
    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200

    # Verify bucket exists
    head = await aioboto3_s3_client.head_bucket(Bucket=bucket_name)
    assert head["ResponseMetadata"]["HTTPStatusCode"] == 200

    return bucket_name

@pytest.fixture
async def bucket_with_data(test_bucket: str, aioboto3_s3_client: S3Client) -> str:
    """Create a bucket with pre-populated test data."""
    # Upload sample files
    files = {
        "config.json": b'{"version": "1.0"}',
        "data/records.csv": b"id,name,value\n1,test,100\n2,demo,200",
        "logs/app.log": b"[INFO] Application started\n[INFO] Processing complete"
    }

    for key, body in files.items():
        await aioboto3_s3_client.put_object(
            Bucket=test_bucket,
            Key=key,
            Body=body
        )

    return test_bucket

async def test_list_bucket_contents(bucket_with_data: str, aioboto3_s3_client: S3Client) -> None:
    """Test using pre-populated bucket."""
    response = await aioboto3_s3_client.list_objects_v2(Bucket=bucket_with_data)

    keys = sorted([obj["Key"] for obj in response["Contents"]])
    assert keys == ["config.json", "data/records.csv", "logs/app.log"]

    # Test prefix filtering
    response = await aioboto3_s3_client.list_objects_v2(
        Bucket=bucket_with_data,
        Prefix="data/"
    )
    assert len(response["Contents"]) == 1
    assert response["Contents"][0]["Key"] == "data/records.csv"
```

--------------------------------

### Test S3 Presigned URL Generation with Aiobotocore

Source: https://context7.com/phillipuniverse/pytest-aioboto3/llms.txt

This test case validates the generation of presigned URLs for S3 objects using aiobotocore. It covers generating URLs for both GET and PUT operations with specified expiration times. Requires an aiobotocore S3 client.

```python
from types_aiobotocore_s3 import S3Client

async def test_presigned_url_generation(aioboto3_s3_client: S3Client) -> None:
    """Test generating presigned URLs for S3 objects."""
    bucket = "presigned-test-bucket"
    key = "secure-document.pdf"

    # Setup
    await aioboto3_s3_client.create_bucket(Bucket=bucket)
    await aioboto3_s3_client.put_object(
        Bucket=bucket,
        Key=key,
        Body=b"Secret document content"
    )

    # Generate presigned URL for GET
    presigned_url = await aioboto3_s3_client.generate_presigned_url(
        "get_object",
        Params={"Bucket": bucket, "Key": key},
        ExpiresIn=3600  # 1 hour
    )

    # Verify URL was generated
    assert presigned_url is not None
    assert bucket in presigned_url
    assert key in presigned_url
    assert "X-Amz-Algorithm" in presigned_url
    assert "X-Amz-Expires" in presigned_url

    # Generate presigned URL for PUT
    put_url = await aioboto3_s3_client.generate_presigned_url(
        "put_object",
        Params={"Bucket": bucket, "Key": "upload.txt"},
        ExpiresIn=300  # 5 minutes
    )

    assert put_url is not None
    assert "upload.txt" in put_url
```

--------------------------------

### Use moto_services Fixture for Direct AWS Service Mocking

Source: https://context7.com/phillipuniverse/pytest-aioboto3/llms.txt

Shows how to use the `moto_services` fixture to obtain service endpoint URLs for direct mocking of AWS services. This example configures an `aioboto3` client to use a specific S3 endpoint provided by Moto, creates a bucket, and verifies its existence via a HEAD request. It depends on `pytest`, `aioboto3`, and `typing.Mapping`.

```python
import pytest
from typing import Mapping

@pytest.fixture
async def custom_s3_client(moto_services: Mapping[str, str]) -> None:
    """Example showing how to use moto_services directly."""
    import aioboto3

    # moto_services provides: {"s3": "http://localhost:5000"}
    s3_endpoint = moto_services["s3"]

    session = aioboto3.Session(
        aws_access_key_id="testing",
        aws_secret_access_key="testing",
        region_name="us-east-1"
    )

    async with session.client("s3", endpoint_url=s3_endpoint) as client:
        # Create bucket directly using endpoint
        await client.create_bucket(Bucket="custom-bucket")

        # Verify it exists
        response = await client.head_bucket(Bucket="custom-bucket")
        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
```

--------------------------------

### Testing S3 Bucket Access with aioboto3_s3_client Fixture

Source: https://github.com/phillipuniverse/pytest-aioboto3/blob/main/README.md

Inject the aioboto3_s3_client fixture into async tests to get a mocked S3 client. It handles session patching automatically. Outputs are mocked responses from Moto for S3 operations like list_buckets.

```python
async def test_aio_aws_bucket_access(aioboto3_s3_client: S3Client) -> None:
    resp = await aioboto3_s3_client.list_buckets()
    ...
```

--------------------------------

### Create Custom aioboto3 Sessions with Region Mocking

Source: https://context7.com/phillipuniverse/pytest-aioboto3/llms.txt

Demonstrates creating custom aioboto3 sessions for different AWS regions (e.g., EU and US) using the `moto_patch_session` fixture. It includes creating buckets in specific regions and verifying their locations and the visibility of buckets across different mocked regions. This requires the `aioboto3` and `pytest` libraries.

```python
import aioboto3
import pytest

async def test_custom_session_regions(moto_patch_session: None) -> None:
    # Create session for EU region
    eu_session = aioboto3.Session(region_name="eu-west-1")
    async with eu_session.client("s3", region_name="eu-west-1") as eu_client:
        await eu_client.create_bucket(
            Bucket="eu-bucket",
            CreateBucketConfiguration={"LocationConstraint": "eu-west-1"}
        )

        # Verify bucket location
        location = await eu_client.get_bucket_location(Bucket="eu-bucket")
        assert location["LocationConstraint"] == "eu-west-1"

    # Create session for US region
    us_session = aioboto3.Session(region_name="us-east-1")
    async with us_session.client("s3", region_name="us-east-1") as us_client:
        await us_client.create_bucket(Bucket="us-bucket")

        # List all buckets - should see both
        response = await us_client.list_buckets()
        bucket_names = sorted([b["Name"] for b in response["Buckets"]])
        assert bucket_names == ["eu-bucket", "us-bucket"]
```

--------------------------------

### Python: aioboto3_s3_client Fixture

Source: https://context7.com/phillipuniverse/pytest-aioboto3/llms.txt

Provides a pre-configured async S3 client fixture for testing S3 operations with automatic cleanup and endpoint mocking. This fixture simplifies interaction with S3 services within tests, leveraging Moto's mocking capabilities.

```python
import pytest
from types_aiobotocore_s3 import S3Client

async def test_s3_bucket_operations(aioboto3_s3_client: S3Client) -> None:
    # Create a bucket
    create_resp = await aioboto3_s3_client.create_bucket(Bucket="test-bucket")
    assert create_resp["ResponseMetadata"]["HTTPStatusCode"] == 200

    # Upload an object
    await aioboto3_s3_client.put_object(
        Bucket="test-bucket",
        Key="data.json",
        Body=b'{"status": "success"}',
        ContentType="application/json"
    )

    # List objects in the bucket
    list_resp = await aioboto3_s3_client.list_objects_v2(Bucket="test-bucket")
    assert list_resp["KeyCount"] == 1
    assert list_resp["Contents"][0]["Key"] == "data.json"

    # Retrieve the object
    get_resp = await aioboto3_s3_client.get_object(Bucket="test-bucket", Key="data.json")
    body = await get_resp["Body"].read()
    assert body == b'{"status": "success"}'

    # Delete the object
    await aioboto3_s3_client.delete_object(Bucket="test-bucket", Key="data.json")

    # Verify deletion
    list_resp = await aioboto3_s3_client.list_objects_v2(Bucket="test-bucket")
    assert list_resp["KeyCount"] == 0
```

--------------------------------

### Python: moto_patch_session Fixture

Source: https://context7.com/phillipuniverse/pytest-aioboto3/llms.txt

Patches aioboto3 and boto3 session creation globally, enabling testing of production code without modifications. This core fixture manages Moto mock servers in the background.

```python
import aioboto3
from types_aiobotocore_s3 import S3Client

# Production code - no test awareness
async def upload_user_data(user_id: str, data: dict) -> str:
    """Production function that uploads data to S3."""
    import json
    session = aioboto3.Session()
    async with session.client("s3", region_name="us-west-2") as s3:
        key = f"users/{user_id}/data.json"
        await s3.put_object(
            Bucket="production-bucket",
            Key=key,
            Body=json.dumps(data).encode(),
            ContentType="application/json"
        )
        return key

# Test code
async def test_upload_user_data(moto_patch_session: None, aioboto3_s3_client: S3Client) -> None:
    # Setup: create the bucket that production code expects
    await aioboto3_s3_client.create_bucket(Bucket="production-bucket")

    # Execute: call production code - it automatically uses mocked S3
    user_data = {"name": "Alice", "email": "alice@example.com"}
    key = await upload_user_data("user123", user_data)

    # Verify: check the uploaded data
    assert key == "users/user123/data.json"
    response = await aioboto3_s3_client.get_object(Bucket="production-bucket", Key=key)
    body = await response["Body"].read()

    import json
    uploaded_data = json.loads(body)
    assert uploaded_data == user_data
```

--------------------------------

### Creating Custom aioboto3 Session with moto_patch_session Fixture

Source: https://github.com/phillipuniverse/pytest-aioboto3/blob/main/README.md

Use the moto_patch_session fixture to patch aioboto3 sessions in tests. Create a new Session and client for S3 with region specification. Yields the client for use in async contexts; limited to mocked services like S3.

```python
async def test_some_s3_thing(moto_patch_session: None) -> None:
    session = aioboto3.Session(region_name="us-east-1")
    async with session.client("s3", region_name="us-east-1") as client:  # type: S3Client
        yield client
```

--------------------------------

### Python: aioboto3_s3_resource Fixture

Source: https://context7.com/phillipuniverse/pytest-aioboto3/llms.txt

Provides a pre-configured async S3 resource fixture for higher-level object-oriented interaction with S3, offering automatic cleanup and mocking. This allows for more intuitive testing of S3 resource operations.

```python
import pytest
from types_aiobotocore_s3.service_resource import S3ServiceResource

async def test_s3_resource_api(aioboto3_s3_resource: S3ServiceResource) -> None:
    # Create a bucket using resource API
    bucket = await aioboto3_s3_resource.Bucket("resource-test-bucket")
    await bucket.create()

    # Upload file using bucket object
    obj = await bucket.Object("file.txt")
    await obj.put(Body=b"Hello from resource API")

    # Read the file back
    response = await obj.get()
    content = await response["Body"].read()
    assert content == b"Hello from resource API"

    # List all objects
    objects = []
    async for obj in bucket.objects.all():
        objects.append(obj.key)
    assert objects == ["file.txt"]

    # Delete the object
    await obj.delete()

    # Verify bucket is empty
    objects = []
    async for obj in bucket.objects.all():
        objects.append(obj.key)
    assert objects == []
```

--------------------------------

### Test S3 Error Handling with Pytest and Aiobotocore

Source: https://context7.com/phillipuniverse/pytest-aioboto3/llms.txt

This snippet demonstrates how to test S3 error conditions such as NoSuchBucket, NoSuchKey, and BucketAlreadyOwnedByYou using pytest's exception handling. It requires an aiobotocore S3 client and the botocore.exceptions.ClientError.

```python
import pytest
from botocore.exceptions import ClientError
from types_aiobotocore_s3 import S3Client

async def test_bucket_not_found_error(aioboto3_s3_client: S3Client) -> None:
    """Test handling of NoSuchBucket error."""
    with pytest.raises(ClientError) as exc_info:
        await aioboto3_s3_client.head_bucket(Bucket="nonexistent-bucket")

    error = exc_info.value
    assert error.response["Error"]["Code"] == "404"
    assert "NoSuchBucket" in str(error) or "Not Found" in str(error)

async def test_object_not_found_error(aioboto3_s3_client: S3Client) -> None:
    """Test handling of NoSuchKey error."""
    # Create bucket first
    await aioboto3_s3_client.create_bucket(Bucket="test-bucket")

    # Try to get non-existent object
    with pytest.raises(ClientError) as exc_info:
        await aioboto3_s3_client.get_object(Bucket="test-bucket", Key="missing.txt")

    error = exc_info.value
    assert error.response["Error"]["Code"] == "NoSuchKey"

async def test_duplicate_bucket_error(aioboto3_s3_client: S3Client) -> None:
    """Test handling of BucketAlreadyOwnedByYou error."""
    bucket_name = "duplicate-test-bucket"

    # Create bucket
    await aioboto3_s3_client.create_bucket(Bucket=bucket_name)

    # Try to create the same bucket again
    with pytest.raises(ClientError) as exc_info:
        await aioboto3_s3_client.create_bucket(Bucket=bucket_name)

    error = exc_info.value
    assert error.response["Error"]["Code"] in ["BucketAlreadyOwnedByYou", "BucketAlreadyExists"]
```

--------------------------------

### Test S3 Multipart Upload with Aiobotocore

Source: https://context7.com/phillipuniverse/pytest-aioboto3/llms.txt

This function tests the S3 multipart upload process, including initiating the upload, uploading multiple parts, completing the upload, and verifying the object's existence and content length. It requires an aiobotocore S3 client.

```python
from types_aiobotocore_s3 import S3Client

async def test_multipart_upload(aioboto3_s3_client: S3Client) -> None:
    """Test S3 multipart upload functionality."""
    bucket = "multipart-test-bucket"
    key = "large-file.bin"

    # Create bucket
    await aioboto3_s3_client.create_bucket(Bucket=bucket)

    # Initiate multipart upload
    response = await aioboto3_s3_client.create_multipart_upload(
        Bucket=bucket,
        Key=key
    )
    upload_id = response["UploadId"]

    # Upload parts
    parts = []
    for i in range(1, 4):
        part_data = f"Part {i} data: " + ("x" * 1024 * 5)
        part_data = part_data.encode()

        part_response = await aioboto3_s3_client.upload_part(
            Bucket=bucket,
            Key=key,
            PartNumber=i,
            UploadId=upload_id,
            Body=part_data
        )

        parts.append({
            "PartNumber": i,
            "ETag": part_response["ETag"]
        })

    # Complete multipart upload
    await aioboto3_s3_client.complete_multipart_upload(
        Bucket=bucket,
        Key=key,
        UploadId=upload_id,
        MultipartUpload={"Parts": parts}
    )

    # Verify file exists
    response = await aioboto3_s3_client.head_object(Bucket=bucket, Key=key)
    assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
    assert response["ContentLength"] > 0
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.