Method test_persist

tests/test_pipeline_files.py:599–636

(self)

class TestS3FilesStore:
    @inline_callbacks_test
    def test_persist(self):
        bucket = "mybucket"
        key = "export.csv"
        uri = f"s3://{bucket}/{key}"
        buffer = mock.MagicMock()
        meta = {"foo": "bar"}
        path = ""
        content_type = "image/png"

        store = S3FilesStore(uri)
        from botocore.stub import Stubber  # noqa: PLC0415

        with Stubber(store.s3_client) as stub:
            stub.add_response(
                "put_object",
                expected_params={
                    "ACL": S3FilesStore.POLICY,
                    "Body": buffer,
                    "Bucket": bucket,
                    "CacheControl": S3FilesStore.HEADERS["Cache-Control"],
                    "ContentType": content_type,
                    "Key": key,
                    "Metadata": meta,
                },
                service_response={},
            )

            yield store.persist_file(
                path,
                buffer,
                info=None,
                meta=meta,
                headers={"Content-Type": content_type},
            )

            stub.assert_no_pending_responses()
            # The call to read does not happen with Stubber
            assert buffer.method_calls == [mock.call.seek(0)]

    @inline_callbacks_test
    def test_stat(self):