diff --git a/app/core/associations/endpoints_associations.py b/app/core/associations/endpoints_associations.py index c88f9d7e32..17901b0b17 100644 --- a/app/core/associations/endpoints_associations.py +++ b/app/core/associations/endpoints_associations.py @@ -19,7 +19,10 @@ ) from app.types.content_type import ContentType from app.types.module import CoreModule -from app.utils.tools import get_file_from_data, save_file_as_data +from app.utils.tools import ( + compress_and_save_image_file, + get_file_from_data, +) router = APIRouter(tags=["Associations"]) @@ -173,16 +176,19 @@ async def create_association_logo( if not association: raise HTTPException(status_code=404, detail="Association not found") - await save_file_as_data( + await compress_and_save_image_file( upload_file=image, directory="associations/logos", filename=association_id, - max_file_size=4 * 1024 * 1024, accepted_content_types=[ ContentType.jpg, ContentType.png, ContentType.webp, ], + max_file_size=1024 * 1024 * 5, # 5 MB + height=300, + width=300, + quality=85, ) diff --git a/app/core/groups/endpoints_groups.py b/app/core/groups/endpoints_groups.py index 8949fd240d..570b109488 100644 --- a/app/core/groups/endpoints_groups.py +++ b/app/core/groups/endpoints_groups.py @@ -27,9 +27,9 @@ from app.types.module import CoreModule from app.utils.communication.notifications import NotificationManager from app.utils.tools import ( + compress_and_save_image_file, get_file_from_data, is_user_member_of_any_group, - save_file_as_data, ) router = APIRouter(tags=["Groups"]) @@ -374,16 +374,19 @@ async def create_group_logo( if not group: raise HTTPException(status_code=404, detail="Group not found") - await save_file_as_data( + await compress_and_save_image_file( upload_file=image, directory="groups/logos", filename=group_id, - max_file_size=4 * 1024 * 1024, accepted_content_types=[ ContentType.jpg, ContentType.png, ContentType.webp, ], + max_file_size=1024 * 1024 * 5, # 5 MB + height=300, + width=300, + quality=85, ) diff --git a/app/core/users/endpoints_users.py b/app/core/users/endpoints_users.py index 898aadae35..6c38f1ca53 100644 --- a/app/core/users/endpoints_users.py +++ b/app/core/users/endpoints_users.py @@ -49,9 +49,9 @@ from app.utils.communication.notifications import NotificationManager from app.utils.mail.mailworker import send_email from app.utils.tools import ( + compress_and_save_image_file, create_and_send_email_migration, get_file_from_data, - save_file_as_data, sort_user, ) @@ -1114,16 +1114,19 @@ async def create_current_user_profile_picture( **The user must be authenticated to use this endpoint** """ - await save_file_as_data( + await compress_and_save_image_file( upload_file=image, directory="profile-pictures", filename=user.id, - max_file_size=4 * 1024 * 1024, accepted_content_types=[ ContentType.jpg, ContentType.png, ContentType.webp, ], + max_file_size=1024 * 1024 * 5, # 5 MB + height=300, + width=300, + quality=85, ) return standard_responses.Result(success=True) diff --git a/app/modules/advert/endpoints_advert.py b/app/modules/advert/endpoints_advert.py index 5b001c4917..b0440fb5d8 100644 --- a/app/modules/advert/endpoints_advert.py +++ b/app/modules/advert/endpoints_advert.py @@ -28,10 +28,10 @@ from app.types.module import Module from app.utils.communication.notifications import NotificationManager, NotificationTool from app.utils.tools import ( + compress_and_save_image_file, get_file_from_data, is_user_member_of_an_association, is_user_member_of_an_association_id, - save_file_as_data, ) root = "advert" @@ -323,14 +323,18 @@ async def create_advert_image( detail=f"Unauthorized to manage {advert.advertiser_id} adverts", ) - await save_file_as_data( + await compress_and_save_image_file( upload_file=image, directory="adverts", filename=advert_id, - max_file_size=4 * 1024 * 1024, accepted_content_types=[ ContentType.jpg, ContentType.png, ContentType.webp, ], + max_file_size=1024 * 1024 * 5, # 5 MB + height=315, + width=851, + quality=85, + fit=True, ) diff --git a/app/modules/calendar/endpoints_calendar.py b/app/modules/calendar/endpoints_calendar.py index 45d7526376..9a58d231a4 100644 --- a/app/modules/calendar/endpoints_calendar.py +++ b/app/modules/calendar/endpoints_calendar.py @@ -37,11 +37,11 @@ from app.types.module import Module from app.utils.communication.notifications import NotificationManager, NotificationTool from app.utils.tools import ( + compress_and_save_image_file, delete_file_from_data, get_file_from_data, is_user_member_of_an_association, is_user_member_of_any_group, - save_file_as_data, ) module = Module( @@ -229,16 +229,20 @@ async def create_event_image( detail="You are not allowed to access this event", ) - await save_file_as_data( + await compress_and_save_image_file( upload_file=image, directory="event", filename=event_id, - max_file_size=4 * 1024 * 1024, accepted_content_types=[ ContentType.jpg, ContentType.png, ContentType.webp, ], + max_file_size=1024 * 1024 * 5, # 5 MB + height=315, + width=851, + quality=85, + fit=True, ) diff --git a/app/modules/campaign/endpoints_campaign.py b/app/modules/campaign/endpoints_campaign.py index 8f632898ee..4b9e2500bd 100644 --- a/app/modules/campaign/endpoints_campaign.py +++ b/app/modules/campaign/endpoints_campaign.py @@ -28,9 +28,9 @@ from app.types.content_type import ContentType from app.types.module import Module from app.utils.tools import ( + compress_and_save_image_file, get_file_from_data, is_user_member_of_any_group, - save_file_as_data, ) module = Module( @@ -816,16 +816,19 @@ async def create_campaigns_logo( detail="The list does not exist.", ) - await save_file_as_data( + await compress_and_save_image_file( upload_file=image, directory="campaigns", filename=str(list_id), - max_file_size=4 * 1024 * 1024, accepted_content_types=[ ContentType.jpg, ContentType.png, ContentType.webp, ], + max_file_size=1024 * 1024 * 5, # 5 MB + height=300, + width=300, + quality=85, ) return standard_responses.Result(success=True) diff --git a/app/modules/cinema/endpoints_cinema.py b/app/modules/cinema/endpoints_cinema.py index 96d15e49a1..c781275c16 100644 --- a/app/modules/cinema/endpoints_cinema.py +++ b/app/modules/cinema/endpoints_cinema.py @@ -31,7 +31,10 @@ get_previous_sunday, ) from app.utils.communication.notifications import NotificationTool -from app.utils.tools import get_file_from_data, save_file_as_data +from app.utils.tools import ( + compress_and_save_image_file, + get_file_from_data, +) root = "cinema" cinema_topic = Topic( @@ -214,16 +217,20 @@ async def create_campaigns_logo( detail="The session does not exist.", ) - await save_file_as_data( + await compress_and_save_image_file( upload_file=image, directory="cinemasessions", filename=str(session_id), - max_file_size=4 * 1024 * 1024, accepted_content_types=[ ContentType.jpg, ContentType.png, ContentType.webp, ], + max_file_size=1024 * 1024 * 5, # 5 MB + height=750, + width=500, + quality=85, + fit=True, ) return standard_responses.Result(success=True) diff --git a/app/modules/phonebook/endpoints_phonebook.py b/app/modules/phonebook/endpoints_phonebook.py index ee62ca0a0b..b2a4838bd4 100644 --- a/app/modules/phonebook/endpoints_phonebook.py +++ b/app/modules/phonebook/endpoints_phonebook.py @@ -20,9 +20,9 @@ from app.types.content_type import ContentType from app.types.module import Module from app.utils.tools import ( + compress_and_save_image_file, get_file_from_data, is_user_member_of_any_group, - save_file_as_data, ) module = Module( @@ -690,17 +690,18 @@ async def create_association_logo( detail="The Association does not exist.", ) - await save_file_as_data( + await compress_and_save_image_file( upload_file=image, directory="associations", filename=association_id, - max_file_size=4 * 1024 * 1024, accepted_content_types=[ ContentType.jpg, ContentType.png, ContentType.webp, ], + max_file_size=1024 * 1024 * 5, # 5 MB ) + return standard_responses.Result(success=True) diff --git a/app/modules/raffle/endpoints_raffle.py b/app/modules/raffle/endpoints_raffle.py index 500f106ae1..ba8337ab57 100644 --- a/app/modules/raffle/endpoints_raffle.py +++ b/app/modules/raffle/endpoints_raffle.py @@ -24,6 +24,7 @@ from app.types.module import Module from app.utils.redis import locker_get, locker_set from app.utils.tools import ( + compress_and_save_image_file, get_file_from_data, is_user_member_of_any_group, save_file_as_data, @@ -228,16 +229,17 @@ async def create_current_raffle_logo( detail=f"Raffle {raffle_id} is not in Creation Mode", ) - await save_file_as_data( + await compress_and_save_image_file( upload_file=image, directory="raffle-pictures", filename=str(raffle_id), - max_file_size=4 * 1024 * 1024, accepted_content_types=[ ContentType.jpg, ContentType.png, ContentType.webp, ], + max_file_size=1024 * 1024 * 5, # 5 MB + fit=True, ) return standard_responses.Result(success=True) diff --git a/app/modules/recommendation/endpoints_recommendation.py b/app/modules/recommendation/endpoints_recommendation.py index c51dfdd0fd..cac35e4324 100644 --- a/app/modules/recommendation/endpoints_recommendation.py +++ b/app/modules/recommendation/endpoints_recommendation.py @@ -21,7 +21,10 @@ from app.types import standard_responses from app.types.content_type import ContentType from app.types.module import Module -from app.utils.tools import get_file_from_data, save_file_as_data +from app.utils.tools import ( + compress_and_save_image_file, + get_file_from_data, +) router = APIRouter() @@ -184,16 +187,19 @@ async def create_recommendation_image( if not recommendation: raise HTTPException(status_code=404, detail="The recommendation does not exist") - await save_file_as_data( + await compress_and_save_image_file( upload_file=image, directory="recommendations", filename=str(recommendation_id), - max_file_size=4 * 1024 * 1024, accepted_content_types=[ ContentType.jpg, ContentType.png, ContentType.webp, ], + max_file_size=1024 * 1024 * 5, # 5 MB + height=300, + width=300, + quality=85, ) return standard_responses.Result(success=True) diff --git a/app/types/content_type.py b/app/types/content_type.py index e82c6356dd..18c5230b45 100644 --- a/app/types/content_type.py +++ b/app/types/content_type.py @@ -1,5 +1,7 @@ from enum import Enum +from app.types.exceptions import UnknownContentTypeExtensionError + class ContentType(str, Enum): """ @@ -10,3 +12,31 @@ class ContentType(str, Enum): png = "image/png" webp = "image/webp" pdf = "application/pdf" + + @property + def extension(self) -> str: + """ + Get the file extension corresponding to the content type + """ + if self == ContentType.jpg: + return "jpg" + if self == ContentType.png: + return "png" + if self == ContentType.webp: + return "webp" + if self == ContentType.pdf: + return "pdf" + raise UnknownContentTypeExtensionError(content_type=self.value) + + def __str__(self): + return self.extension + + +class PillowImageFormat(str, Enum): + """ + Accepted image formats for Pillow + """ + + jpg = "JPEG" + png = "PNG" + webp = "WEBP" diff --git a/app/types/exceptions.py b/app/types/exceptions.py index 0d5dee2ea8..625975fcbd 100644 --- a/app/types/exceptions.py +++ b/app/types/exceptions.py @@ -217,3 +217,10 @@ def __init__(self, object_name: str): super().__init__( f"Newly added object {object_name} not found in the database", ) + + +class UnknownContentTypeExtensionError(Exception): + def __init__(self, content_type: str): + super().__init__( + f"Unknown content type extension for content type: {content_type}", + ) diff --git a/app/utils/tools.py b/app/utils/tools.py index 4cdb1613e8..f396241add 100644 --- a/app/utils/tools.py +++ b/app/utils/tools.py @@ -7,6 +7,7 @@ import unicodedata from collections.abc import Callable, Sequence from inspect import iscoroutinefunction +from io import BytesIO from pathlib import Path from typing import TYPE_CHECKING, Any, TypeVar from uuid import UUID @@ -19,6 +20,7 @@ from fastapi.templating import Jinja2Templates from jellyfish import jaro_winkler_similarity from jinja2 import Environment, FileSystemLoader, select_autoescape +from PIL import Image from pydantic import ValidationError from sqlalchemy.ext.asyncio import AsyncSession from weasyprint import CSS, HTML @@ -31,7 +33,7 @@ from app.core.users.models_users import CoreUser from app.core.utils import security from app.types import core_data -from app.types.content_type import ContentType +from app.types.content_type import ContentType, PillowImageFormat from app.types.exceptions import ( CoreDataNotFoundError, FileDoesNotExistError, @@ -162,35 +164,21 @@ async def is_user_id_valid(user_id: str, db: AsyncSession) -> bool: return await cruds_users.get_user_by_id(db=db, user_id=user_id) is not None -async def save_file_as_data( +async def ensure_file_properties( upload_file: UploadFile, - directory: str, - filename: str | UUID, - max_file_size: int = 1024 * 1024 * 2, # 2 MB accepted_content_types: list[ContentType] | None = None, -): + max_file_size: int = 1024 * 1024 * 5, # 5 MB +) -> None: """ - Save an image or pdf file to the data folder. - - - The file will be saved in the `data` folder: "data/{directory}/{filename}.ext" - - Maximum size is 2MB by default, it can be changed using `max_file_size` (in bytes) parameter. + Ensure that the provided file respects the properties: + - Maximum size is 5 MB by default, it can be changed using `max_file_size` (in bytes) parameter. - `accepted_content_types` is a list of accepted content types. By default, all format are accepted. Use: `["image/jpeg", "image/png", "image/webp"]` to accept only images. - - Filename should be an uuid. - The file extension will be inferred from the provided content file. - There should only be one file with the same filename, thus, saving a new file will remove the existing even if its extension was different. - Currently, compatible extensions are defined in the enum `ContentType` + An HTTP Exception will be raised if an error occurs. - An HTTP Exception will be raised if an error occurres. - - The filename should be a uuid. - - WARNING: **NEVER** trust user input when calling this function. Always check that parameters are valid. + The file will not be saved nor modified. """ - if isinstance(filename, UUID): - filename = str(filename) - if accepted_content_types is None: # Accept only images by default accepted_content_types = [ @@ -200,12 +188,6 @@ async def save_file_as_data( ContentType.pdf, ] - if not uuid_regex.match(filename): - hyperion_error_logger.error( - f"save_file_as_data: security issue, the filename is not a valid UUID: {filename}.", - ) - raise FileNameIsNotAnUUIDError() - if upload_file.content_type not in accepted_content_types: raise HTTPException( status_code=400, @@ -224,6 +206,48 @@ async def save_file_as_data( # We go back to the beginning of the file to save it on the disk await upload_file.seek(0) + +async def save_file_as_data( + upload_file: UploadFile, + directory: str, + filename: str | UUID, + max_file_size: int = 1024 * 1024 * 2, # 2 MB + accepted_content_types: list[ContentType] | None = None, +): + """ + Save an image or pdf file to the data folder. + + - The file will be saved in the `data` folder: "data/{directory}/{filename}.ext" + - Maximum size is 2MB by default, it can be changed using `max_file_size` (in bytes) parameter. + - `accepted_content_types` is a list of accepted content types. By default, all format are accepted. + Use: `["image/jpeg", "image/png", "image/webp"]` to accept only images. + - Filename should be an uuid. + + The file extension will be inferred from the provided content file. + There should only be one file with the same filename, thus, saving a new file will remove the existing even if its extension was different. + Currently, compatible extensions are defined in the enum `ContentType` + + An HTTP Exception will be raised if an error occurs. + + The filename should be a uuid. + + WARNING: **NEVER** trust user input when calling this function. Always check that parameters are valid. + """ + if isinstance(filename, UUID): + filename = str(filename) + + if not uuid_regex.match(filename): + hyperion_error_logger.error( + f"save_file_as_data: security issue, the filename is not a valid UUID: {filename}.", + ) + raise FileNameIsNotAnUUIDError() + + await ensure_file_properties( + upload_file=upload_file, + accepted_content_types=accepted_content_types, + max_file_size=max_file_size, + ) + extension = ContentType(upload_file.content_type).name # Remove the existing file if any and create the new one @@ -485,6 +509,136 @@ async def save_pdf_first_page_as_image( ) +def compress_image( + file_bytes: bytes, + height: int | None = None, + width: int | None = None, + quality: int = 85, + output_format: PillowImageFormat = PillowImageFormat.webp, + fit: bool = False, +) -> bytes: + """ + Resize, crop and compress an image using Pillow. + + - If `height` or `width` is None, the original image dimension will be used. + - The image aspect ratio will be preserved. + - The resulting image will be centered if cropping is needed. + + If `fit` is True, the image will be enlarged to fit the whole area defined by `height` and `width`. + Otherwise, the image will be reduced to fit inside the area defined by `height` and `width` without cropping. + + Don't forget to take into account the output format when saving the image. + """ + # We want to add an Alpha layer so that cropping does not produce black borders + image = Image.open(BytesIO(file_bytes)).convert("RGBA") + + if height is None: + height = image.height + if width is None: + width = image.width + + # Preserve aspect ratio + if fit: + ratio = max(width / image.width, height / image.height) + else: + ratio = min(width / image.width, height / image.height) + new_size = (int(image.width * ratio), int(image.height * ratio)) + resized_image = image.resize(new_size) + + # We may want to crop the image, the resulting image will be centered + left = (resized_image.width - width) // 2 + top = (resized_image.height - height) // 2 + right = (resized_image.width + width) // 2 + bottom = (resized_image.height + height) // 2 + + cropped_image = resized_image.crop( + ( + left, + top, + right, + bottom, + ), + ) + + output = BytesIO() + cropped_image.save(output, format=output_format, quality=quality) + return output.getvalue() + + +async def compress_and_save_image_file( + upload_file: UploadFile, + directory: str, + filename: str | UUID, + accepted_content_types: list[ContentType] | None = None, + max_file_size: int = 1024 * 1024 * 5, # 5 MB + height: int | None = None, + width: int | None = None, + quality: int = 85, + fit: bool = False, +): + """ + Save a compressed webp version of an input image in the data folder. + + The filename should be a uuid. + No verifications will be made about the content of the file, it is up to the caller to ensure the content is valid and safe. + + - The file will be saved in the `data` folder: "data/{directory}/{filename}.webp" + - An original copy of the file will be saved in "data/{directory}/original/{filename}.{upload_file extension}" + + Ensure that the provided file respects the properties: + - Maximum size is 5 MB by default, it can be changed using `max_file_size` (in bytes) parameter. + - `accepted_content_types` is a list of accepted content types. By default, all format are accepted. + Use: `["image/jpeg", "image/png", "image/webp"]` to accept only images. + + The image will be resized, cropped and compressed using Pillow. + + - If `height` or `width` is None, the original image dimension will be used. + - The image aspect ratio will be preserved. + - The resulting image will be centered if cropping is needed. + + If `fit` is True, the image will be enlarged to fit the whole area defined by `height` and `width`. + Otherwise, the image will be reduced to fit inside the area defined by `height` and `width` without cropping. + + An HTTP Exception will be raised if an error occurs. + + WARNING: **NEVER** trust user input when calling this function. Always check that parameters are valid. + """ + await ensure_file_properties( + upload_file=upload_file, + accepted_content_types=accepted_content_types, + max_file_size=max_file_size, + ) + + original_file_bytes = await upload_file.read() + + original_directory = ( + f"{directory}{'/' if not directory.endswith('/') else ''}original" + ) + + await save_bytes_as_data( + file_bytes=original_file_bytes, + directory=original_directory, + filename=filename, + extension=ContentType(upload_file.content_type).extension, + ) + + file_bytes = compress_image( + file_bytes=original_file_bytes, + height=height, + width=width, + quality=quality, + output_format=PillowImageFormat.webp, + fit=fit, + ) + + await save_bytes_as_data( + file_bytes=file_bytes, + directory=directory, + filename=filename, + extension=ContentType.webp, + ) + + def get_random_string(length: int = 5) -> str: return "".join( secrets.choice("abcdefghijklmnopqrstuvwxyz0123456789") for _ in range(length) diff --git a/migrations/versions/49-compress_images.py b/migrations/versions/49-compress_images.py new file mode 100644 index 0000000000..755e61c3e2 --- /dev/null +++ b/migrations/versions/49-compress_images.py @@ -0,0 +1,131 @@ +"""empty message + +Create Date: 2025-12-08 11:28:24.682540 +""" + +from collections.abc import Sequence +from pathlib import Path +from typing import TYPE_CHECKING + +from app.types.content_type import PillowImageFormat +from app.utils.tools import compress_image + +if TYPE_CHECKING: + from pytest_alembic import MigrationContext + +import sqlalchemy as sa + +# revision identifiers, used by Alembic. +revision: str = "1ec573d854a1" +down_revision: str | None = "ecd89212ca0" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +data_sources: dict[str, dict[str, int]] = { + "associations/logos": { + "height": 300, + "width": 300, + "quality": 85, + }, + "groups/logos": { + "height": 300, + "width": 300, + "quality": 85, + }, + "profile-pictures": { + "height": 300, + "width": 300, + "quality": 85, + }, + "adverts": { + "height": 315, + "width": 851, + "quality": 85, + "fit": 1, + }, + "event": { + "height": 315, + "width": 851, + "quality": 85, + "fit": 1, + }, + "campaigns": { + "height": 300, + "width": 300, + "quality": 85, + }, + "cinemasessions": { + "height": 750, + "width": 500, + "quality": 85, + "fit": 1, + }, + "recommendations": { + "height": 300, + "width": 300, + "quality": 85, + }, +} + + +def upgrade() -> None: + for data_folder, params in data_sources.items(): + print("__________________________________________") # noqa: T201 + print(f"Processing folder: {data_folder}") # noqa: T201 + height = params.get("height") + width = params.get("width") + quality = params.get("quality", 85) + fit = bool(params.get("fit", 0)) + if Path("data/" + data_folder).exists(): + for file_path in Path("data/" + data_folder).iterdir(): + print(" - ", file_path) # noqa: T201 + if file_path.suffix in (".png", ".jpg", ".webp"): + with Path(file_path).open("rb") as file: + file_bytes = file.read() + + Path(f"data/{data_folder}/original/").mkdir( + parents=True, + exist_ok=True, + ) + + # Save the original file + with Path(f"data/{data_folder}/original/{file_path.name}").open( + "wb", + ) as out_file: + out_file.write(file_bytes) + + # Compress and save the image + res = compress_image( + file_bytes, + height=height, + width=width, + quality=quality, + output_format=PillowImageFormat.webp, + fit=fit, + ) + + # Delete the original file + Path(f"data/{data_folder}/{file_path.name}").unlink() + + with Path(f"data/{data_folder}/{file_path.stem}.webp").open( + "wb", + ) as out_file: + out_file.write(res) + + +def downgrade() -> None: + pass + + +def pre_test_upgrade( + alembic_runner: "MigrationContext", + alembic_connection: sa.Connection, +) -> None: + pass + + +def test_upgrade( + alembic_runner: "MigrationContext", + alembic_connection: sa.Connection, +) -> None: + pass diff --git a/tests/test_utils.py b/tests/test_utils.py index 0474e994e9..1495f8d364 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,16 +1,21 @@ import shutil import uuid +from io import BytesIO from pathlib import Path import pytest import pytest_asyncio from fastapi import HTTPException, UploadFile +from PIL import Image from starlette.datastructures import Headers from app.core.core_endpoints import models_core +from app.types.content_type import ContentType, PillowImageFormat from app.types.core_data import BaseCoreData from app.types.exceptions import CoreDataNotFoundError, FileNameIsNotAnUUIDError from app.utils.tools import ( + compress_and_save_image_file, + compress_image, delete_file_from_data, get_core_data, get_file_from_data, @@ -237,6 +242,58 @@ def test_delete_file_raise_a_value_error_if_filename_isnt_an_uuid() -> None: ) +@pytest.mark.parametrize( + ("height", "width"), + [ + (100, 100), + (300, 300), + (50, 100), + (100, 50), + (100, None), + (None, 100), + ], +) +async def test_compress(height: int, width: int) -> None: + with Path("assets/images/default_profile_picture.png").open("rb") as file: + file_bytes = file.read() + res = compress_image( + file_bytes, + height=height, + width=width, + quality=70, + output_format=PillowImageFormat.webp, + ) + + res_image = Image.open(BytesIO(res)) + if height is not None: + assert res_image.height == height + if width is not None: + assert res_image.width == width + assert res_image.format == "WEBP" + + +async def test_compress_and_save_image_file() -> None: + valid_uuid = str(uuid.uuid4()) + with Path("assets/images/default_profile_picture.png").open("rb") as file: + await compress_and_save_image_file( + upload_file=UploadFile( + file, + headers=Headers({"content-type": "image/png"}), + ), + directory="test/compressed", + filename=valid_uuid, + accepted_content_types=[ + ContentType.png, + ], + max_file_size=1024 * 1024 * 5, # 5 MB + height=300, + width=300, + quality=85, + ) + assert Path(f"data/test/compressed/{valid_uuid}.webp").exists() + assert Path(f"data/test/compressed/original/{valid_uuid}.png").exists() + + async def test_save_pdf_first_page_as_image() -> None: valid_uuid = str(uuid.uuid4())