Skip to content

Shortcuts

upload_file(filepath, s3_params)

Upload file to s3.

Source code in saritasa_s3_tools/testing/shortcuts.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def upload_file(
    filepath: str,
    s3_params: client.S3UploadParams,
) -> httpx.Response:
    """Upload file to s3."""
    url = s3_params.url
    params = s3_params.params
    # Test file upload itself
    with (
        httpx.Client() as client,
        pathlib.Path(filepath).open("rb") as upload_file,
    ):
        upload_response = client.post(
            url=url,
            data={
                key: value
                for key, value in params.items()
                if value is not None
            },
            files={"file": upload_file.read()},
        )
    return upload_response

upload_file_and_verify(filepath, s3_params, is_minio=False)

Upload and verify that file is uploaded.

Source code in saritasa_s3_tools/testing/shortcuts.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def upload_file_and_verify(
    filepath: str,
    s3_params: client.S3UploadParams,
    is_minio: bool = False,
) -> tuple[str, str]:
    """Upload and verify that file is uploaded."""
    upload_response = upload_file(
        filepath=filepath,
        s3_params=s3_params,
    )
    assert upload_response.is_success, upload_response.content  # noqa: S101
    parsed_response = xml.etree.ElementTree.fromstring(  # noqa: S314
        upload_response.content.decode(),
    )
    if is_minio:
        file_key = parsed_response[1].text
        file_url = parsed_response[3].text
    else:
        file_key = parsed_response[2].text
        file_url = parsed_response[0].text
    assert file_url, upload_response.content  # noqa: S101
    assert file_key, upload_response.content  # noqa: S101
    return file_url, file_key