Storage check

[PetrDlouhy]

I have a check for Django storages. It goes through all storages, tries to write a file, read and then delete a file on them.
For S3 storages it also makes the file public and tries to access it's URL.

Here is the code:

import random
import string

import requests
from botocore.exceptions import ClientError
from django.conf import settings
from django.core.files.base import ContentFile
from django.core.files.storage import storages
from django_alive import HealthcheckFailure
from storages.backends.s3boto3 import S3Boto3Storage


def generate_random_content(size=20):
    """Generate a random string of fixed size"""
    return "".join(random.choices(string.ascii_letters + string.digits, k=size))


def make_s3_file_public(storage, name):
    """Make an S3 file public after uploading"""
    try:
        storage.bucket.Object(name).Acl().put(ACL="public-read")
    except ClientError as e:
        raise HealthcheckFailure(f"Failed to set ACL for S3 file '{name}': {e}") from e


def check_url(storage, storage_name, name, test_content):
    # Check URL only for S3Boto3Storage
    if isinstance(storage, S3Boto3Storage) and hasattr(storage, "url"):
        print(storage_name)
        file_url = storage.url(name)
        print(file_url)
        response = requests.get(file_url, timeout=5)
        print(response.content)
        print(test_content)
        if response.content != test_content:
            raise HealthcheckFailure(
                f"HTTP downloaded content does not match for S3 storage '{storage_name}'"
            )


def check_storages():
    errors = []

    for storage_name in settings.STORAGES.keys():
        test_file_name = f"storage_test_file_{storage_name}_{generate_random_content(size=5)}.txt"
        storage = storages[storage_name]
        test_content = f"{storage_name} {generate_random_content()}".encode("utf-8")
        try:
            try:
                storage.delete(test_file_name)
            except FileNotFoundError:
                pass

            # Write operation
            name = storage.save(test_file_name, ContentFile(test_content))

            # For S3 storage, make the file public
            if isinstance(storage, S3Boto3Storage):
                make_s3_file_public(
                    storage,
                    storage.location + "/" + name if storage.location else name,
                )

            # Read operation
            with storage.open(name, "rb") as file:
                content = file.read()
                if content != test_content:
                    raise HealthcheckFailure(
                        f"Read content does not match written content for storage '{storage_name}'"
                    )

            check_url(storage, storage_name, name, test_content)

            # Clean up: Delete the test file
            storage.delete(name)

        except Exception as e:  # noqa
            errors.append(f"Storage '{storage_name}' failed: {e}")

    if errors:
        raise HealthcheckFailure("; ".join(errors))

The basic check for the storages (write, read, delete) could be performed on any Django storage using just pure Django (although I am not sure how much it is useful for filebased storages).

The second part of the testing is probably storage dependent.

@ipmb What do you think about this test? Do you think that the basic test fits django-alive (after some work on the code), or should I place the whole test to https://github.com/PetrDlouhy/django-alive-checks?

[ipmb]

Yeah, I think it makes sense in your extension. I'll accept a PR that links to it from the README. 👍