From 1658edf11fc3bdc501d6723a3b2ff388f2b6abfe Mon Sep 17 00:00:00 2001 From: Azarta2Zygoto Date: Sun, 9 Nov 2025 00:17:01 +0100 Subject: [PATCH 1/3] Mise en place de la facturation du SDeC --- app/modules/sdec_facturation/__init__.py | 0 .../cruds_sdec_facturation.py | 977 ++++++++++++++++++ .../endpoints_sdec_facturation.py | 572 ++++++++++ .../models_sdec_facturation.py | 102 ++ .../schemas_sdec_facturation.py | 133 +++ .../types_sdec_facturation.py | 48 + 6 files changed, 1832 insertions(+) create mode 100644 app/modules/sdec_facturation/__init__.py create mode 100644 app/modules/sdec_facturation/cruds_sdec_facturation.py create mode 100644 app/modules/sdec_facturation/endpoints_sdec_facturation.py create mode 100644 app/modules/sdec_facturation/models_sdec_facturation.py create mode 100644 app/modules/sdec_facturation/schemas_sdec_facturation.py create mode 100644 app/modules/sdec_facturation/types_sdec_facturation.py diff --git a/app/modules/sdec_facturation/__init__.py b/app/modules/sdec_facturation/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/app/modules/sdec_facturation/cruds_sdec_facturation.py b/app/modules/sdec_facturation/cruds_sdec_facturation.py new file mode 100644 index 0000000000..d7ad8c4b60 --- /dev/null +++ b/app/modules/sdec_facturation/cruds_sdec_facturation.py @@ -0,0 +1,977 @@ +"""File defining the functions called by the endpoints, making queries to the table using the models""" + +import uuid +from collections.abc import Sequence +from datetime import UTC, datetime + +from sqlalchemy import delete, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.modules.sdec_facturation import ( + models_sdec_facturation, + schemas_sdec_facturation, +) + +# ---------------------------------------------------------------------------- # +# Member # +# ---------------------------------------------------------------------------- # + + +async def create_member( + member: schemas_sdec_facturation.MemberBase, + db: AsyncSession, +) -> schemas_sdec_facturation.MemberComplete: + """Create a new member in the database""" + + member_db = models_sdec_facturation.Member( + id=uuid.uuid4(), + name=member.name, + mandate=member.mandate, + role=member.role, + visible=member.visible, + modified_date=datetime.now(tz=UTC), + ) + db.add(member_db) + await db.flush() + return schemas_sdec_facturation.MemberComplete( + id=member_db.id, + name=member_db.name, + mandate=member.mandate, + role=member.role, + visible=member.visible, + modified_date=member_db.modified_date, + ) + + +async def update_member( + member_id: uuid.UUID, + member_edit: schemas_sdec_facturation.MemberBase, + db: AsyncSession, +): + """Update a member in the database""" + + await db.execute( + update(models_sdec_facturation.Member) + .where(models_sdec_facturation.Member.id == member_id) + .values(**member_edit.model_dump(), modified_date=datetime.now(tz=UTC)), + ) + await db.flush() + + +async def delete_member( + member_id: uuid.UUID, + db: AsyncSession, +): + """Delete a member from the database""" + + await db.execute( + update(models_sdec_facturation.Member) + .where(models_sdec_facturation.Member.id == member_id) + .values( + visible=False, + ), + ) + await db.flush() + + +async def get_all_members( + db: AsyncSession, +) -> Sequence[schemas_sdec_facturation.MemberComplete]: + """Get all members from the database""" + result = await db.execute(select(models_sdec_facturation.Member)) + members = result.scalars().all() + return [ + schemas_sdec_facturation.MemberComplete( + id=member.id, + name=member.name, + mandate=member.mandate, + role=member.role, + visible=member.visible, + modified_date=member.modified_date, + ) + for member in members + ] + + +async def get_member_by_id( + member_id: uuid.UUID, + db: AsyncSession, +) -> schemas_sdec_facturation.MemberComplete | None: + """Get a specific member by its ID from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.Member).where( + models_sdec_facturation.Member.id == member_id, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.MemberComplete( + id=result.id, + name=result.name, + mandate=result.mandate, + role=result.role, + visible=result.visible, + modified_date=result.modified_date, + ) + if result + else None + ) + + +async def get_member_by_name( + member_name: str, + db: AsyncSession, +) -> schemas_sdec_facturation.MemberComplete | None: + """Get a specific member by its name from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.Member).where( + models_sdec_facturation.Member.name == member_name, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.MemberComplete( + id=result.id, + name=result.name, + mandate=result.mandate, + role=result.role, + visible=result.visible, + modified_date=result.modified_date, + ) + if result + else None + ) + + +# ---------------------------------------------------------------------------- # +# Mandat # +# ---------------------------------------------------------------------------- # +async def create_mandate( + mandate: schemas_sdec_facturation.MandateComplete, + db: AsyncSession, +) -> schemas_sdec_facturation.MandateComplete: + """Create a new mandate in the database""" + + mandate_db = models_sdec_facturation.Mandate( + year=mandate.year, + name=mandate.name, + ) + db.add(mandate_db) + await db.flush() + return schemas_sdec_facturation.MandateComplete( + year=mandate_db.year, + name=mandate_db.name, + ) + + +async def update_mandate( + year: int, + mandate_edit: schemas_sdec_facturation.MandateUpdate, + db: AsyncSession, +): + """Update a mandate in the database""" + + await db.execute( + update(models_sdec_facturation.Mandate) + .where(models_sdec_facturation.Mandate.year == year) + .values( + name=mandate_edit.name, + ), + ) + await db.flush() + + +async def delete_mandate( + year: int, + db: AsyncSession, +): + """Delete a mandate from the database""" + + await db.execute( + delete(models_sdec_facturation.Mandate).where( + models_sdec_facturation.Mandate.year == year, + ), + ) + await db.flush() + + +async def get_all_mandates( + db: AsyncSession, +) -> Sequence[schemas_sdec_facturation.MandateComplete]: + """Get all mandates from the database""" + result = await db.execute(select(models_sdec_facturation.Mandate)) + mandats = result.scalars().all() + return [ + schemas_sdec_facturation.MandateComplete( + year=mandate.year, + name=mandate.name, + ) + for mandate in mandats + ] + + +async def get_mandate_by_year( + year: int, + db: AsyncSession, +) -> schemas_sdec_facturation.MandateComplete | None: + """Get a specific mandate by its year from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.Mandate).where( + models_sdec_facturation.Mandate.year == year, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.MandateComplete( + year=result.year, + name=result.name, + ) + if result + else None + ) + + +# ---------------------------------------------------------------------------- # +# Associationciation # +# ---------------------------------------------------------------------------- # + + +async def create_association( + association: schemas_sdec_facturation.AssociationBase, + db: AsyncSession, +) -> schemas_sdec_facturation.AssociationComplete: + """Create a new associationciation in the database""" + + association_db = models_sdec_facturation.Association( + id=uuid.uuid4(), + name=association.name, + type=association.type, + structure=association.structure, + modified_date=datetime.now(tz=UTC), + visible=association.visible, + ) + db.add(association_db) + await db.flush() + return schemas_sdec_facturation.AssociationComplete( + id=association_db.id, + name=association_db.name, + type=association_db.type, + structure=association_db.structure, + visible=association_db.visible, + modified_date=association_db.modified_date, + ) + + +async def delete_association( + association_id: uuid.UUID, + db: AsyncSession, +): + """Delete an associationciation from the database""" + + await db.execute( + update(models_sdec_facturation.Association) + .where(models_sdec_facturation.Association.id == association_id) + .values( + visible=False, + ), + ) + await db.flush() + + +async def update_association( + association_id: uuid.UUID, + association_edit: schemas_sdec_facturation.AssociationBase, + db: AsyncSession, +): + """Update an associationciation in the database""" + + await db.execute( + update(models_sdec_facturation.Association) + .where(models_sdec_facturation.Association.id == association_id) + .values(**association_edit.model_dump(), modified_date=datetime.now(tz=UTC)), + ) + await db.flush() + + +async def get_all_associations( + db: AsyncSession, +) -> Sequence[schemas_sdec_facturation.AssociationComplete]: + """Get all associationciations from the database""" + result = await db.execute(select(models_sdec_facturation.Association)) + association = result.scalars().all() + return [ + schemas_sdec_facturation.AssociationComplete( + id=association.id, + name=association.name, + type=association.type, + structure=association.structure, + visible=association.visible, + modified_date=association.modified_date, + ) + for association in association + ] + + +async def get_association_by_id( + association_id: uuid.UUID, + db: AsyncSession, +) -> schemas_sdec_facturation.AssociationComplete | None: + """Get a specific associationciation by its ID from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.Association).where( + models_sdec_facturation.Association.id == association_id, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.AssociationComplete( + id=result.id, + name=result.name, + type=result.type, + structure=result.structure, + visible=result.visible, + modified_date=result.modified_date, + ) + if result + else None + ) + + +async def get_association_by_name( + association_name: str, + db: AsyncSession, +) -> schemas_sdec_facturation.AssociationComplete | None: + """Get a specific associationciation by its name from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.Association).where( + models_sdec_facturation.Association.name == association_name, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.AssociationComplete( + id=result.id, + name=result.name, + type=result.type, + structure=result.structure, + visible=result.visible, + modified_date=result.modified_date, + ) + if result + else None + ) + + +# ---------------------------------------------------------------------------- # +# Product # +# ---------------------------------------------------------------------------- # + + +async def create_product( + product: schemas_sdec_facturation.ProductBase, + db: AsyncSession, +) -> schemas_sdec_facturation.ProductComplete: + """Create a new product in the database""" + + product_db = models_sdec_facturation.Product( + id=uuid.uuid4(), + code=product.code, + name=product.name, + individual_price=product.individual_price, + association_price=product.association_price, + ae_price=product.ae_price, + category=product.category, + for_sale=product.for_sale, + creation_date=datetime.now(tz=UTC), + ) + db.add(product_db) + await db.flush() + return schemas_sdec_facturation.ProductComplete( + id=product_db.id, + code=product_db.code, + name=product_db.name, + individual_price=product_db.individual_price, + association_price=product_db.association_price, + ae_price=product_db.ae_price, + category=product_db.category, + for_sale=product_db.for_sale, + creation_date=product_db.creation_date, + ) + + +async def update_product( + product_code: str, + product_edit: schemas_sdec_facturation.ProductBase, + db: AsyncSession, +): + """Update a product in the database""" + + product_db = models_sdec_facturation.Product( + id=uuid.uuid4(), + code=product_code, + name=product_edit.name, + individual_price=product_edit.individual_price, + association_price=product_edit.association_price, + ae_price=product_edit.ae_price, + category=product_edit.category, + for_sale=product_edit.for_sale, + creation_date=datetime.now(tz=UTC), + ) + db.add(product_db) + await db.flush() + return schemas_sdec_facturation.ProductComplete( + id=product_db.id, + code=product_db.code, + name=product_db.name, + individual_price=product_db.individual_price, + association_price=product_db.association_price, + ae_price=product_db.ae_price, + category=product_db.category, + for_sale=product_db.for_sale, + creation_date=product_db.creation_date, + ) + + +async def minor_update_product( + product_code: str, + product_edit: schemas_sdec_facturation.ProductMinorUpdate, + db: AsyncSession, +): + """Minor update of a product in the database""" + + update_values = { + key: value + for key, value in product_edit.model_dump().items() + if value is not None + } + + await db.execute( + update(models_sdec_facturation.Product) + .where(models_sdec_facturation.Product.code == product_code) + .values(**update_values), + ) + await db.flush() + + +async def delete_product( + product_code: str, + db: AsyncSession, +): + """Delete a product from the database""" + + await db.execute( + update(models_sdec_facturation.Product) + .where(models_sdec_facturation.Product.code == product_code) + .values( + for_sale=False, + ), + ) + await db.flush() + + +async def get_all_products( + db: AsyncSession, +) -> Sequence[schemas_sdec_facturation.ProductComplete]: + """Get all products from the database""" + result = await db.execute(select(models_sdec_facturation.Product)) + products = result.scalars().all() + return [ + schemas_sdec_facturation.ProductComplete( + id=product.id, + code=product.code, + name=product.name, + individual_price=product.individual_price, + association_price=product.association_price, + ae_price=product.ae_price, + category=product.category, + for_sale=product.for_sale, + creation_date=product.creation_date, + ) + for product in products + ] + + +async def get_product_by_id( + product_id: uuid.UUID, + db: AsyncSession, +) -> schemas_sdec_facturation.ProductComplete | None: + """Get a specific product by its ID from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.Product).where( + models_sdec_facturation.Product.id == product_id, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.ProductComplete( + id=result.id, + code=result.code, + name=result.name, + individual_price=result.individual_price, + association_price=result.association_price, + ae_price=result.ae_price, + category=result.category, + for_sale=result.for_sale, + creation_date=result.creation_date, + ) + if result + else None + ) + + +async def get_product_by_code( + product_code: str, + db: AsyncSession, +) -> schemas_sdec_facturation.ProductComplete | None: + """Get a specific product by its code from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.Product).where( + models_sdec_facturation.Product.code == product_code, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.ProductComplete( + id=result.id, + code=result.code, + name=result.name, + individual_price=result.individual_price, + association_price=result.association_price, + ae_price=result.ae_price, + category=result.category, + for_sale=result.for_sale, + creation_date=result.creation_date, + ) + if result + else None + ) + + +async def get_product_by_name( + product_name: str, + db: AsyncSession, +) -> schemas_sdec_facturation.ProductComplete | None: + """Get a specific product by its name from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.Product).where( + models_sdec_facturation.Product.name == product_name, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.ProductComplete( + id=result.id, + code=result.code, + name=result.name, + individual_price=result.individual_price, + association_price=result.association_price, + ae_price=result.ae_price, + category=result.category, + for_sale=result.for_sale, + creation_date=result.creation_date, + ) + if result + else None + ) + + +# ---------------------------------------------------------------------------- # +# Order # +# ---------------------------------------------------------------------------- # + + +async def create_order( + order: schemas_sdec_facturation.OrderBase, + db: AsyncSession, +) -> schemas_sdec_facturation.OrderComplete: + """Create a new order in the database""" + + order_db = models_sdec_facturation.Order( + id=uuid.uuid4(), + association_id=order.association_id, + member_id=order.member_id, + order=order.order, + creation_date=datetime.now(tz=UTC), + valid=order.valid, + ) + db.add(order_db) + await db.flush() + return schemas_sdec_facturation.OrderComplete( + id=order_db.id, + association_id=order_db.association_id, + member_id=order_db.member_id, + order=order_db.order, + creation_date=order_db.creation_date, + valid=order_db.valid, + ) + + +async def update_order( + order_id: uuid.UUID, + order_edit: schemas_sdec_facturation.OrderUpdate, + db: AsyncSession, +): + """Update an order in the database""" + + await db.execute( + update(models_sdec_facturation.Order) + .where(models_sdec_facturation.Order.id == order_id) + .values(valid=order_edit.valid, order=order_edit.order), + ) + await db.flush() + + +async def get_all_orders( + db: AsyncSession, +) -> Sequence[schemas_sdec_facturation.OrderComplete]: + """Get all orders from the database""" + result = await db.execute(select(models_sdec_facturation.Order)) + orders = result.scalars().all() + return [ + schemas_sdec_facturation.OrderComplete( + id=order.id, + association_id=order.association_id, + member_id=order.member_id, + order=order.order, + creation_date=order.creation_date, + valid=order.valid, + ) + for order in orders + ] + + +async def get_order_by_id( + order_id: uuid.UUID, + db: AsyncSession, +) -> schemas_sdec_facturation.OrderComplete | None: + """Get a specific order by its ID from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.Order).where( + models_sdec_facturation.Order.id == order_id, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.OrderComplete( + id=result.id, + association_id=result.association_id, + member_id=result.member_id, + order=result.order, + creation_date=result.creation_date, + valid=result.valid, + ) + if result + else None + ) + + +# ---------------------------------------------------------------------------- # +# Facture Association # +# ---------------------------------------------------------------------------- # +async def create_facture_association( + facture_association: schemas_sdec_facturation.FactureAssociationBase, + db: AsyncSession, +) -> schemas_sdec_facturation.FactureAssociationComplete: + """Create a new associationciation invoice in the database""" + + facture_association_db = models_sdec_facturation.FactureAssociation( + id=uuid.uuid4(), + facture_number=facture_association.facture_number, + member_id=facture_association.member_id, + association_id=facture_association.association_id, + association_order=",".join( + [str(cmd_id) for cmd_id in facture_association.association_order], + ), + price=facture_association.price, + facture_date=datetime.now(tz=UTC), + valid=facture_association.valid, + paid=facture_association.paid, + payment_date=facture_association.payment_date, + ) + db.add(facture_association_db) + await db.flush() + return schemas_sdec_facturation.FactureAssociationComplete( + id=facture_association_db.id, + facture_number=facture_association_db.facture_number, + member_id=facture_association_db.member_id, + association_id=facture_association_db.association_id, + association_order=[ + int(cmd_id) + for cmd_id in facture_association_db.association_order.split(",") + ], + price=facture_association_db.price, + facture_date=facture_association_db.facture_date, + valid=facture_association_db.valid, + paid=facture_association_db.paid, + payment_date=facture_association_db.payment_date, + ) + + +async def upfacture_date_association( + facture_association_id: uuid.UUID, + facture_association_edit: schemas_sdec_facturation.FactureAssociationBase, + db: AsyncSession, +): + """Update an associationciation invoice in the database""" + + await db.execute( + update(models_sdec_facturation.FactureAssociation) + .where(models_sdec_facturation.FactureAssociation.id == facture_association_id) + .values( + valid=facture_association_edit.valid, + paid=facture_association_edit.paid, + payment_date=facture_association_edit.payment_date, + ), + ) + await db.flush() + + +async def get_all_factures_association( + db: AsyncSession, +) -> Sequence[schemas_sdec_facturation.FactureAssociationComplete]: + """Get all associationciation invoices from the database""" + result = await db.execute(select(models_sdec_facturation.FactureAssociation)) + factures_association = result.scalars().all() + return [ + schemas_sdec_facturation.FactureAssociationComplete( + id=facture_association.id, + facture_number=facture_association.facture_number, + member_id=facture_association.member_id, + association_id=facture_association.association_id, + association_order=[ + int(cmd_id) + for cmd_id in facture_association.association_order.split(",") + ], + price=facture_association.price, + facture_date=facture_association.facture_date, + valid=facture_association.valid, + paid=facture_association.paid, + payment_date=facture_association.payment_date, + ) + for facture_association in factures_association + ] + + +async def get_facture_association_by_id( + facture_association_id: uuid.UUID, + db: AsyncSession, +) -> schemas_sdec_facturation.FactureAssociationComplete | None: + """Get a specific associationciation invoice by its ID from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.FactureAssociation).where( + models_sdec_facturation.FactureAssociation.id + == facture_association_id, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.FactureAssociationComplete( + id=result.id, + facture_number=result.facture_number, + member_id=result.member_id, + association_id=result.association_id, + association_order=[ + int(cmd_id) for cmd_id in result.association_order.split(",") + ], + price=result.price, + facture_date=result.facture_date, + valid=result.valid, + paid=result.paid, + payment_date=result.payment_date, + ) + if result + else None + ) + + +# ---------------------------------------------------------------------------- # +# Facture Individual # +# ---------------------------------------------------------------------------- # +async def create_facture_particulier( + facture_particulier: schemas_sdec_facturation.FactureIndividualBase, + db: AsyncSession, +) -> schemas_sdec_facturation.FactureIndividualComplete: + """Create a new individual invoice in the database""" + + facture_particulier_db = models_sdec_facturation.FactureIndividual( + id=uuid.uuid4(), + facture_number=facture_particulier.facture_number, + member_id=facture_particulier.member_id, + individual_order=facture_particulier.individual_order, + individual_category=facture_particulier.individual_category, + price=facture_particulier.price, + facture_date=datetime.now(tz=UTC), + firstname=facture_particulier.firstname, + lastname=facture_particulier.lastname, + adresse=facture_particulier.adresse, + postal_code=facture_particulier.postal_code, + city=facture_particulier.city, + country=facture_particulier.country, + valid=facture_particulier.valid, + paid=facture_particulier.paid, + payment_date=facture_particulier.payment_date, + ) + db.add(facture_particulier_db) + await db.flush() + return schemas_sdec_facturation.FactureIndividualComplete( + id=facture_particulier_db.id, + facture_number=facture_particulier_db.facture_number, + member_id=facture_particulier_db.member_id, + individual_order=facture_particulier_db.individual_order, + individual_category=facture_particulier_db.individual_category, + price=facture_particulier_db.price, + facture_date=facture_particulier_db.facture_date, + firstname=facture_particulier_db.firstname, + lastname=facture_particulier_db.lastname, + adresse=facture_particulier_db.adresse, + postal_code=facture_particulier_db.postal_code, + city=facture_particulier_db.city, + country=facture_particulier_db.country, + valid=facture_particulier_db.valid, + paid=facture_particulier_db.paid, + payment_date=facture_particulier_db.payment_date, + ) + + +async def up_date_facture_particulier( + facture_particulier_id: uuid.UUID, + facture_particulier_edit: schemas_sdec_facturation.FactureIndividualBase, + db: AsyncSession, +): + """Update an individual invoice in the database""" + + await db.execute( + update(models_sdec_facturation.FactureIndividual) + .where(models_sdec_facturation.FactureIndividual.id == facture_particulier_id) + .values( + firstname=facture_particulier_edit.firstname, + lastname=facture_particulier_edit.lastname, + adresse=facture_particulier_edit.adresse, + postal_code=facture_particulier_edit.postal_code, + city=facture_particulier_edit.city, + valid=facture_particulier_edit.valid, + paid=facture_particulier_edit.paid, + payment_date=facture_particulier_edit.payment_date, + ), + ) + await db.flush() + + +async def get_all_factures_particulier( + db: AsyncSession, +) -> Sequence[schemas_sdec_facturation.FactureIndividualComplete]: + """Get all individual invoices from the database""" + result = await db.execute(select(models_sdec_facturation.FactureIndividual)) + factures_particulier = result.scalars().all() + return [ + schemas_sdec_facturation.FactureIndividualComplete( + id=facture_particulier.id, + facture_number=facture_particulier.facture_number, + member_id=facture_particulier.member_id, + individual_order=facture_particulier.individual_order, + individual_category=facture_particulier.individual_category, + price=facture_particulier.price, + facture_date=facture_particulier.facture_date, + firstname=facture_particulier.firstname, + lastname=facture_particulier.lastname, + adresse=facture_particulier.adresse, + postal_code=facture_particulier.postal_code, + city=facture_particulier.city, + country=facture_particulier.country, + valid=facture_particulier.valid, + paid=facture_particulier.paid, + payment_date=facture_particulier.payment_date, + ) + for facture_particulier in factures_particulier + ] + + +async def get_facture_particulier_by_id( + facture_particulier_id: uuid.UUID, + db: AsyncSession, +) -> schemas_sdec_facturation.FactureIndividualComplete | None: + """Get a specific individual invoice by its ID from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.FactureIndividual).where( + models_sdec_facturation.FactureIndividual.id + == facture_particulier_id, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.FactureIndividualComplete( + id=result.id, + facture_number=result.facture_number, + member_id=result.member_id, + individual_order=result.individual_order, + individual_category=result.individual_category, + price=result.price, + facture_date=result.facture_date, + firstname=result.firstname, + lastname=result.lastname, + adresse=result.adresse, + postal_code=result.postal_code, + city=result.city, + country=result.country, + valid=result.valid, + paid=result.paid, + payment_date=result.payment_date, + ) + if result + else None + ) diff --git a/app/modules/sdec_facturation/endpoints_sdec_facturation.py b/app/modules/sdec_facturation/endpoints_sdec_facturation.py new file mode 100644 index 0000000000..61e187f136 --- /dev/null +++ b/app/modules/sdec_facturation/endpoints_sdec_facturation.py @@ -0,0 +1,572 @@ +import logging +import uuid +from collections.abc import Sequence + +from fastapi import Depends, HTTPException +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.groups.groups_type import GroupType +from app.dependencies import ( + get_db, + is_user, + is_user_in, +) +from app.modules.sdec_facturation import ( + cruds_sdec_facturation, + schemas_sdec_facturation, +) +from app.types.module import Module + +module = Module( + root="sdec_facturation", + tag="sdec_facturation", + factory=None, +) + + +hyperion_error_logger = logging.getLogger("hyperion.error") + +# ---------------------------------------------------------------------------- # +# Member # +# ---------------------------------------------------------------------------- # + + +@module.router.get( + "/sdec_facturation/members/", + response_model=list[schemas_sdec_facturation.MemberComplete], + status_code=200, +) +async def get_all_members( + db: AsyncSession = Depends(get_db), + _=Depends(is_user()), +) -> Sequence[schemas_sdec_facturation.MemberComplete]: + """Get all members from the database""" + return await cruds_sdec_facturation.get_all_members(db) + + +@module.router.post( + "/sdec_facturation/members/", + response_model=schemas_sdec_facturation.MemberComplete, + status_code=201, +) +async def create_member( + member: schemas_sdec_facturation.MemberBase, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> schemas_sdec_facturation.MemberComplete: + """ + Create a new member in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + if ( + await cruds_sdec_facturation.get_member_by_name( + member.name, + db, + ) + ) is not None: + raise HTTPException( + status_code=400, + detail="User is already a member", + ) + + return await cruds_sdec_facturation.create_member(member, db) + + +@module.router.patch( + "/sdec_facturation/members/{member_id}", + status_code=200, +) +async def update_member( + member_id: uuid.UUID, + member_edit: schemas_sdec_facturation.MemberBase, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Update a member in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + + member_db = await cruds_sdec_facturation.get_member_by_id( + member_id, + db, + ) + if member_db is None: + raise HTTPException( + status_code=404, + detail="Member not found", + ) + + if ( + await cruds_sdec_facturation.get_member_by_name( + member_edit.name, + db, + ) + ) is not None and member_edit.name != member_db.name: + raise HTTPException( + status_code=400, + detail="User is already a member", + ) + + await cruds_sdec_facturation.update_member( + member_id, + member_edit, + db, + ) + + +@module.router.delete( + "/sdec_facturation/members/{member_id}", + status_code=204, +) +async def delete_member( + member_id: uuid.UUID, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Delete a member from the database + **This endpoint is only usable by SDEC Facturation admins** + """ + member_db = await cruds_sdec_facturation.get_member_by_id( + member_id, + db, + ) + if member_db is None: + raise HTTPException( + status_code=404, + detail="Member not found", + ) + + await cruds_sdec_facturation.delete_member(member_id, db) + + +# ---------------------------------------------------------------------------- # +# Mandate # +# ---------------------------------------------------------------------------- # +@module.router.get( + "/sdec_facturation/mandates/", + response_model=list[schemas_sdec_facturation.MandateComplete], + status_code=200, +) +async def get_all_mandates( + db: AsyncSession = Depends(get_db), + _=Depends(is_user()), +) -> Sequence[schemas_sdec_facturation.MandateComplete]: + """Get all mandates from the database""" + return await cruds_sdec_facturation.get_all_mandates(db) + + +@module.router.post( + "/sdec_facturation/mandates/", + response_model=schemas_sdec_facturation.MandateComplete, + status_code=201, +) +async def create_mandate( + mandate: schemas_sdec_facturation.MandateComplete, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> schemas_sdec_facturation.MandateComplete: + """ + Create a new mandate in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + if ( + await cruds_sdec_facturation.get_mandate_by_year( + mandate.year, + db, + ) + ) is not None: + raise HTTPException( + status_code=400, + detail="Mandate year already exists", + ) + + return await cruds_sdec_facturation.create_mandate(mandate, db) + + +@module.router.patch( + "/sdec_facturation/mandates/{mandate_year}", + status_code=200, +) +async def update_mandate( + mandate_year: int, + mandate_edit: schemas_sdec_facturation.MandateUpdate, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Update a mandate in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + + mandate_db = await cruds_sdec_facturation.get_mandate_by_year( + mandate_year, + db, + ) + if mandate_db is None: + raise HTTPException( + status_code=404, + detail="Mandate not found", + ) + + await cruds_sdec_facturation.update_mandate( + mandate_year, + mandate_edit, + db, + ) + + +@module.router.delete( + "/sdec_facturation/mandates/{mandate_year}", + status_code=204, +) +async def delete_mandate( + mandate_year: int, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Delete a mandate from the database + **This endpoint is only usable by SDEC Facturation admins** + """ + mandate_db = await cruds_sdec_facturation.get_mandate_by_year( + mandate_year, + db, + ) + if mandate_db is None: + raise HTTPException( + status_code=404, + detail="Mandate not found", + ) + + await cruds_sdec_facturation.delete_mandate(mandate_year, db) + + +# ---------------------------------------------------------------------------- # +# Association # +# ---------------------------------------------------------------------------- # + + +@module.router.get( + "/sdec_facturation/association/", + response_model=list[schemas_sdec_facturation.AssociationComplete], + status_code=200, +) +async def get_all_associations( + db: AsyncSession = Depends(get_db), + _=Depends(is_user()), +) -> Sequence[schemas_sdec_facturation.AssociationComplete]: + """Get all associations from the database""" + return await cruds_sdec_facturation.get_all_associations(db) + + +@module.router.post( + "/sdec_facturation/association/", + response_model=schemas_sdec_facturation.AssociationComplete, + status_code=201, +) +async def create_association( + association: schemas_sdec_facturation.AssociationBase, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> schemas_sdec_facturation.AssociationComplete: + """ + Create a new association in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + if ( + await cruds_sdec_facturation.get_association_by_name(association.name, db) + ) is not None: + raise HTTPException( + status_code=400, + detail="Association name already used", + ) + + return await cruds_sdec_facturation.create_association(association, db) + + +@module.router.patch( + "/sdec_facturation/association/{association_id}", + status_code=200, +) +async def update_association( + association_id: uuid.UUID, + association_edit: schemas_sdec_facturation.AssociationBase, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Create a new association in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + association_db = await cruds_sdec_facturation.get_association_by_id( + association_id, + db, + ) + if association_db is None: + raise HTTPException( + status_code=404, + detail="Association not found", + ) + + if ( + await cruds_sdec_facturation.get_association_by_name(association_edit.name, db) + ) is not None and association_edit.name != association_db.name: + raise HTTPException( + status_code=400, + detail="Association name already used", + ) + + await cruds_sdec_facturation.update_association( + association_id, + association_edit, + db, + ) + + +@module.router.delete( + "/sdec_facturation/association/{association_id}", + status_code=204, +) +async def delete_association( + association_id: uuid.UUID, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Delete an association from the database + **This endpoint is only usable by SDEC Facturation admins** + """ + association_db = await cruds_sdec_facturation.get_association_by_id( + association_id, + db, + ) + if association_db is None: + raise HTTPException( + status_code=404, + detail="Association not found", + ) + + await cruds_sdec_facturation.delete_association(association_id, db) + + +# ---------------------------------------------------------------------------- # +# Product # +# ---------------------------------------------------------------------------- # + + +@module.router.get( + "/sdec_facturation/product/", + response_model=list[schemas_sdec_facturation.ProductComplete], + status_code=200, +) +async def get_all_products( + db: AsyncSession = Depends(get_db), + _=Depends(is_user()), +) -> Sequence[schemas_sdec_facturation.ProductComplete]: + """Get all product items from the database""" + return await cruds_sdec_facturation.get_all_products(db) + + +@module.router.post( + "/sdec_facturation/product/", + response_model=schemas_sdec_facturation.ProductComplete, + status_code=201, +) +async def create_product( + product: schemas_sdec_facturation.ProductBase, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> schemas_sdec_facturation.ProductComplete: + """ + Create a new product item in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + if ( + product.individual_price < 0 + or product.association_price < 0 + or product.ae_price < 0 + ): + raise HTTPException( + status_code=400, + detail="Product item prices must be positive", + ) + if (await cruds_sdec_facturation.get_product_by_code(product.code, db)) is not None: + raise HTTPException( + status_code=400, + detail="Product item code already used", + ) + if (await cruds_sdec_facturation.get_product_by_name(product.name, db)) is not None: + raise HTTPException( + status_code=400, + detail="Product item name already used", + ) + + return await cruds_sdec_facturation.create_product(product, db) + + +@module.router.patch( + "/sdec_facturation/product/{product_id}", + status_code=200, +) +async def update_product( + product_id: uuid.UUID, + product_edit: schemas_sdec_facturation.ProductUpdate, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Update a product item in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + + if ( + product_edit.individual_price < 0 + or product_edit.association_price < 0 + or product_edit.ae_price < 0 + ): + raise HTTPException( + status_code=400, + detail="Product item prices must be positive", + ) + + product_db = await cruds_sdec_facturation.get_product_by_id( + product_id, + db, + ) + if product_db is None: + raise HTTPException( + status_code=404, + detail="Product item not found", + ) + + product_base = schemas_sdec_facturation.ProductBase( + code=product_db.code, + name=product_db.name, + individual_price=product_edit.individual_price, + association_price=product_edit.association_price, + ae_price=product_edit.ae_price, + category=product_db.category, + for_sale=product_db.for_sale, + ) + + await cruds_sdec_facturation.update_product( + product_db.code, + product_base, + db, + ) + + +@module.router.patch( + "/sdec_facturation/product/minor/{product_id}", + status_code=200, +) +async def minor_update_product( + product_id: uuid.UUID, + product_edit: schemas_sdec_facturation.ProductMinorUpdate, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Minor update a product item in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + + product_db = await cruds_sdec_facturation.get_product_by_id( + product_id, + db, + ) + if product_db is None: + raise HTTPException( + status_code=404, + detail="Product item not found", + ) + + await cruds_sdec_facturation.minor_update_product( + product_db.code, + product_edit, + db, + ) + + +@module.router.delete( + "/sdec_facturation/product/{product_id}", + status_code=204, +) +async def delete_product( + product_id: uuid.UUID, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Delete a product item from the database + **This endpoint is only usable by SDEC Facturation admins** + """ + product_db = await cruds_sdec_facturation.get_product_by_id( + product_id, + db, + ) + if product_db is None: + raise HTTPException( + status_code=404, + detail="Product item not found", + ) + + await cruds_sdec_facturation.delete_product(product_db.code, db) + + +# ---------------------------------------------------------------------------- # +# Order # +# ---------------------------------------------------------------------------- # + + +@module.router.get( + "/sdec_facturation/orders/", + response_model=list[schemas_sdec_facturation.OrderComplete], + status_code=200, +) +async def get_all_orders( + db: AsyncSession = Depends(get_db), + _=Depends(is_user()), +) -> Sequence[schemas_sdec_facturation.OrderComplete]: + """Get all orders from the database""" + return await cruds_sdec_facturation.get_all_orders(db) + + +async def create_order( + order: schemas_sdec_facturation.OrderBase, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> schemas_sdec_facturation.OrderComplete: + """ + Create a new order in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + if ( + await cruds_sdec_facturation.get_association_by_id( + order.association_id, + db, + ) + ) is None: + raise HTTPException( + status_code=400, + detail="Association does not exist", + ) + + if ( + await cruds_sdec_facturation.get_member_by_id( + order.member_id, + db, + ) + ) is None: + raise HTTPException( + status_code=400, + detail="Member does not exist", + ) + return await cruds_sdec_facturation.create_order(order, db) + + +# ---------------------------------------------------------------------------- # +# Facture Association # +# ---------------------------------------------------------------------------- # diff --git a/app/modules/sdec_facturation/models_sdec_facturation.py b/app/modules/sdec_facturation/models_sdec_facturation.py new file mode 100644 index 0000000000..7b62c48e0e --- /dev/null +++ b/app/modules/sdec_facturation/models_sdec_facturation.py @@ -0,0 +1,102 @@ +"""Models file for SDeC facturation website""" + +from datetime import date +from uuid import UUID + +from sqlalchemy import ForeignKey +from sqlalchemy.orm import Mapped, mapped_column + +from app.modules.sdec_facturation.types_sdec_facturation import ( + AssociationStructureType, + AssociationType, + IndividualCategoryType, + RoleType, +) +from app.types.sqlalchemy import Base, PrimaryKey + + +class Member(Base): + __tablename__ = "sdec_facturation_member" + id: Mapped[PrimaryKey] + name: Mapped[str] = mapped_column(index=True) + mandate: Mapped[int] = mapped_column(ForeignKey("sdec_facturation_mandate.year")) + role: Mapped[RoleType] + modified_date: Mapped[date] + visible: Mapped[bool] = mapped_column(default=True) + + +class Mandate(Base): + __tablename__ = "sdec_facturation_mandate" + year: Mapped[int] = mapped_column(primary_key=True, index=True) + name: Mapped[str] = mapped_column(index=True) + + +class Association(Base): + __tablename__ = "sdec_facturation_association" + id: Mapped[PrimaryKey] + name: Mapped[str] = mapped_column(index=True) + type: Mapped[AssociationType] + structure: Mapped[AssociationStructureType] + modified_date: Mapped[date] + visible: Mapped[bool] = mapped_column(default=True) + + +class Product(Base): + __tablename__ = "sdec_facturation_product" + id: Mapped[PrimaryKey] + code: Mapped[str] + name: Mapped[str] + individual_price: Mapped[float] + association_price: Mapped[float] + ae_price: Mapped[float] + category: Mapped[str] + creation_date: Mapped[date] + for_sale: Mapped[bool] = mapped_column(default=True) + + +class Order(Base): + __tablename__ = "sdec_facturation_order" + id: Mapped[PrimaryKey] + association_id: Mapped[UUID] = mapped_column( + ForeignKey("sdec_facturation_association.id"), + ) + member_id: Mapped[UUID] = mapped_column(ForeignKey("sdec_facturation_member.id")) + order: Mapped[str] + creation_date: Mapped[date] + valid: Mapped[bool] = mapped_column(default=True) + + +class FactureAssociation(Base): + __tablename__ = "sdec_facturation_facture_association" + id: Mapped[PrimaryKey] + facture_number: Mapped[str] = mapped_column(unique=True) + member_id: Mapped[UUID] = mapped_column(ForeignKey("sdec_facturation_member.id")) + association_id: Mapped[UUID] = mapped_column( + ForeignKey("sdec_facturation_association.id"), + ) + association_order: Mapped[str] + price: Mapped[float] + facture_date: Mapped[date] + valid: Mapped[bool] + paid: Mapped[bool] + payment_date: Mapped[date | None] = mapped_column(default=None) + + +class FactureIndividual(Base): + __tablename__ = "sdec_facturation_facture_individual" + id: Mapped[PrimaryKey] + facture_number: Mapped[str] = mapped_column(unique=True) + member_id: Mapped[UUID] = mapped_column(ForeignKey("sdec_facturation_member.id")) + individual_order: Mapped[str] + individual_category: Mapped[IndividualCategoryType] + price: Mapped[float] + facture_date: Mapped[date] + firstname: Mapped[str] + lastname: Mapped[str] + adresse: Mapped[str] + postal_code: Mapped[str] + city: Mapped[str] + country: Mapped[str] + valid: Mapped[bool] + paid: Mapped[bool] + payment_date: Mapped[date | None] = mapped_column(default=None) diff --git a/app/modules/sdec_facturation/schemas_sdec_facturation.py b/app/modules/sdec_facturation/schemas_sdec_facturation.py new file mode 100644 index 0000000000..285e0f32dd --- /dev/null +++ b/app/modules/sdec_facturation/schemas_sdec_facturation.py @@ -0,0 +1,133 @@ +"""Schemas file for endpoint /sdec-facturation""" + +import uuid +from datetime import date + +from pydantic import BaseModel + +from app.modules.sdec_facturation.types_sdec_facturation import ( + AssociationStructureType, + AssociationType, + IndividualCategoryType, + RoleType, +) + + +class MemberBase(BaseModel): + name: str + mandate: int + role: RoleType + visible: bool = True + + +class MemberComplete(MemberBase): + id: uuid.UUID + modified_date: date + + +class MandateComplete(BaseModel): + year: int + name: str + + +class MandateUpdate(BaseModel): + name: str + + +class AssociationBase(BaseModel): + name: str + type: AssociationType + structure: AssociationStructureType + visible: bool = True + + +class AssociationComplete(AssociationBase): + id: uuid.UUID + modified_date: date + + +class ProductBase(BaseModel): + code: str + name: str + individual_price: float + association_price: float + ae_price: float + category: str + for_sale: bool = True + + +class ProductComplete(ProductBase): + id: uuid.UUID + creation_date: date + + +class ProductMinorUpdate(BaseModel): + name: str | None = None + category: str | None = None + + +class ProductUpdate(BaseModel): + individual_price: float + association_price: float + ae_price: float + + +class OrderBase(BaseModel): + association_id: uuid.UUID + member_id: uuid.UUID + order: str + valid: bool = True + + +class OrderComplete(OrderBase): + id: uuid.UUID + creation_date: date + + +class OrderUpdate(BaseModel): + valid: bool + order: str + + +class FactureAssociationBase(BaseModel): + facture_number: str + member_id: uuid.UUID + association_id: uuid.UUID + association_order: list[int] + price: float + valid: bool + paid: bool + payment_date: date | None = None + + +class FactureAssociationComplete(FactureAssociationBase): + facture_date: date + id: uuid.UUID + + +class FactureUpdate(BaseModel): + valid: bool | None = None + paid: bool | None = None + payment_date: date | None = None + + +class FactureIndividualBase(BaseModel): + facture_number: str + member_id: uuid.UUID + individual_order: str + individual_category: IndividualCategoryType + price: float + firstname: str + lastname: str + adresse: str + postal_code: str + city: str + country: str + valid: bool + paid: bool + payment_date: date | None = None + + +class FactureIndividualComplete(FactureIndividualBase): + facture_date: date + id: uuid.UUID diff --git a/app/modules/sdec_facturation/types_sdec_facturation.py b/app/modules/sdec_facturation/types_sdec_facturation.py new file mode 100644 index 0000000000..96cd9eeb60 --- /dev/null +++ b/app/modules/sdec_facturation/types_sdec_facturation.py @@ -0,0 +1,48 @@ +from enum import Enum + + +class IndividualCategoryType(str, Enum): + pe = "pe" + pa = "pa" + autre = "autre" + tfe = "tfe" + + def __str__(self) -> str: + return f"{self.name}<{self.value}" + + +class RoleType(str, Enum): + prez = "prez" + trez = "trez" + trez_int = "trez_int" + trez_ext = "trez_ext" + sg = "sg" + com = "com" + profs = "profs" + matos = "matos" + appro = "appro" + te = "te" + projets = "projets" + boutique = "boutique" + perms = "perms" + + def __str__(self) -> str: + return f"{self.name}<{self.value}" + + +class AssociationStructureType(str, Enum): + asso = "asso" + club = "club" + section = "section" + + def __str__(self) -> str: + return f"{self.name}<{self.value}" + + +class AssociationType(str, Enum): + aeecl = "aeecl" + useecl = "useecl" + independant = "independant" + + def __str__(self) -> str: + return f"{self.name}<{self.value}" From 343393c852b43c9d75b5e77e1e9cd8997031cdeb Mon Sep 17 00:00:00 2001 From: Azarta2Zygoto Date: Sun, 9 Nov 2025 23:24:36 +0100 Subject: [PATCH 2/3] Multiples corrections et finalisation des endpoints --- .../cruds_sdec_facturation.py | 498 ++++++++++++------ .../endpoints_sdec_facturation.py | 460 ++++++++++++++-- .../models_sdec_facturation.py | 25 +- .../schemas_sdec_facturation.py | 91 ++-- .../types_sdec_facturation.py | 38 +- 5 files changed, 866 insertions(+), 246 deletions(-) diff --git a/app/modules/sdec_facturation/cruds_sdec_facturation.py b/app/modules/sdec_facturation/cruds_sdec_facturation.py index d7ad8c4b60..61641a5f06 100644 --- a/app/modules/sdec_facturation/cruds_sdec_facturation.py +++ b/app/modules/sdec_facturation/cruds_sdec_facturation.py @@ -154,7 +154,7 @@ async def get_member_by_name( # ---------------------------------------------------------------------------- # -# Mandat # +# Mandate # # ---------------------------------------------------------------------------- # async def create_mandate( mandate: schemas_sdec_facturation.MandateComplete, @@ -247,7 +247,7 @@ async def get_mandate_by_year( # ---------------------------------------------------------------------------- # -# Associationciation # +# Association # # ---------------------------------------------------------------------------- # @@ -272,8 +272,8 @@ async def create_association( name=association_db.name, type=association_db.type, structure=association_db.structure, - visible=association_db.visible, modified_date=association_db.modified_date, + visible=association_db.visible, ) @@ -393,100 +393,120 @@ async def get_association_by_name( async def create_product( - product: schemas_sdec_facturation.ProductBase, + product: schemas_sdec_facturation.ProductAndPriceBase, db: AsyncSession, -) -> schemas_sdec_facturation.ProductComplete: +) -> schemas_sdec_facturation.ProductAndPriceComplete: """Create a new product in the database""" product_db = models_sdec_facturation.Product( id=uuid.uuid4(), code=product.code, name=product.name, - individual_price=product.individual_price, - association_price=product.association_price, - ae_price=product.ae_price, category=product.category, for_sale=product.for_sale, creation_date=datetime.now(tz=UTC), ) db.add(product_db) await db.flush() - return schemas_sdec_facturation.ProductComplete( - id=product_db.id, - code=product_db.code, - name=product_db.name, - individual_price=product_db.individual_price, - association_price=product_db.association_price, - ae_price=product_db.ae_price, - category=product_db.category, - for_sale=product_db.for_sale, - creation_date=product_db.creation_date, - ) - -async def update_product( - product_code: str, - product_edit: schemas_sdec_facturation.ProductBase, - db: AsyncSession, -): - """Update a product in the database""" - - product_db = models_sdec_facturation.Product( + price_db = models_sdec_facturation.ProductPrice( id=uuid.uuid4(), - code=product_code, - name=product_edit.name, - individual_price=product_edit.individual_price, - association_price=product_edit.association_price, - ae_price=product_edit.ae_price, - category=product_edit.category, - for_sale=product_edit.for_sale, - creation_date=datetime.now(tz=UTC), + product_id=product_db.id, + individual_price=product.individual_price, + association_price=product.association_price, + ae_price=product.ae_price, + effective_date=datetime.now(tz=UTC), ) - db.add(product_db) + db.add(price_db) await db.flush() - return schemas_sdec_facturation.ProductComplete( + return schemas_sdec_facturation.ProductAndPriceComplete( id=product_db.id, code=product_db.code, name=product_db.name, - individual_price=product_db.individual_price, - association_price=product_db.association_price, - ae_price=product_db.ae_price, + individual_price=price_db.individual_price, + association_price=price_db.association_price, + ae_price=price_db.ae_price, category=product_db.category, for_sale=product_db.for_sale, creation_date=product_db.creation_date, + effective_date=price_db.effective_date, ) -async def minor_update_product( - product_code: str, - product_edit: schemas_sdec_facturation.ProductMinorUpdate, +async def update_product( + product_id: uuid.UUID, + product_edit: schemas_sdec_facturation.ProductUpdate, db: AsyncSession, ): - """Minor update of a product in the database""" + """Update a product in the database""" update_values = { key: value for key, value in product_edit.model_dump().items() if value is not None } - await db.execute( update(models_sdec_facturation.Product) - .where(models_sdec_facturation.Product.code == product_code) + .where(models_sdec_facturation.Product.id == product_id) .values(**update_values), ) await db.flush() +async def create_price( + product_id: uuid.UUID, + price_edit: schemas_sdec_facturation.ProductPriceUpdate, + db: AsyncSession, +) -> schemas_sdec_facturation.ProductPriceComplete: + """Minor update of a product in the database""" + + price_db = models_sdec_facturation.ProductPrice( + id=uuid.uuid4(), + product_id=product_id, + individual_price=price_edit.individual_price, + association_price=price_edit.association_price, + ae_price=price_edit.ae_price, + effective_date=datetime.now(tz=UTC), + ) + + db.add(price_db) + await db.flush() + + return schemas_sdec_facturation.ProductPriceComplete( + id=price_db.id, + product_id=price_db.product_id, + individual_price=price_db.individual_price, + association_price=price_db.association_price, + ae_price=price_db.ae_price, + effective_date=price_db.effective_date, + ) + + +async def update_price( + product_id: uuid.UUID, + price_edit: schemas_sdec_facturation.ProductPriceUpdate, + db: AsyncSession, +): + """Update the price of a product in the database""" + current_date = datetime.now(tz=UTC) + await db.execute( + update(models_sdec_facturation.ProductPrice) + .where(models_sdec_facturation.ProductPrice.id == product_id) + .where(models_sdec_facturation.ProductPrice.effective_date == current_date) + .values(**price_edit.model_dump()), + ) + await db.flush() + + async def delete_product( - product_code: str, + product_id: uuid.UUID, db: AsyncSession, ): """Delete a product from the database""" await db.execute( update(models_sdec_facturation.Product) - .where(models_sdec_facturation.Product.code == product_code) + .where(models_sdec_facturation.Product.id == product_id) .values( for_sale=False, ), @@ -494,14 +514,45 @@ async def delete_product( await db.flush() -async def get_all_products( +async def get_all_products_and_price( db: AsyncSession, -) -> Sequence[schemas_sdec_facturation.ProductComplete]: +) -> Sequence[schemas_sdec_facturation.ProductAndPriceComplete]: """Get all products from the database""" - result = await db.execute(select(models_sdec_facturation.Product)) - products = result.scalars().all() + + query = select( + models_sdec_facturation.Product, + models_sdec_facturation.ProductPrice, + ).outerjoin( + models_sdec_facturation.ProductPrice, + models_sdec_facturation.Product.id + == models_sdec_facturation.ProductPrice.product_id, + ) + result = await db.execute(query) + rows = result.all() # list of (Product, ProductPrice|None) + + products = [] + for product, product_price in rows: + individual_price = product_price.individual_price if product_price else 0.0 + association_price = product_price.association_price if product_price else 0.0 + ae_price = product_price.ae_price if product_price else 0.0 + + products.append( + schemas_sdec_facturation.ProductAndPriceComplete( + id=product.id, + code=product.code, + name=product.name, + individual_price=individual_price, + association_price=association_price, + ae_price=ae_price, + category=product.category, + for_sale=product.for_sale, + creation_date=product.creation_date, + effective_date=product_price.effective_date if product_price else None, + ), + ) + return [ - schemas_sdec_facturation.ProductComplete( + schemas_sdec_facturation.ProductAndPriceComplete( id=product.id, code=product.code, name=product.name, @@ -511,6 +562,7 @@ async def get_all_products( category=product.category, for_sale=product.for_sale, creation_date=product.creation_date, + effective_date=product_price.effective_date, ) for product in products ] @@ -521,6 +573,7 @@ async def get_product_by_id( db: AsyncSession, ) -> schemas_sdec_facturation.ProductComplete | None: """Get a specific product by its ID from the database""" + result = ( ( await db.execute( @@ -537,9 +590,6 @@ async def get_product_by_id( id=result.id, code=result.code, name=result.name, - individual_price=result.individual_price, - association_price=result.association_price, - ae_price=result.ae_price, category=result.category, for_sale=result.for_sale, creation_date=result.creation_date, @@ -570,9 +620,6 @@ async def get_product_by_code( id=result.id, code=result.code, name=result.name, - individual_price=result.individual_price, - association_price=result.association_price, - ae_price=result.ae_price, category=result.category, for_sale=result.for_sale, creation_date=result.creation_date, @@ -603,9 +650,6 @@ async def get_product_by_name( id=result.id, code=result.code, name=result.name, - individual_price=result.individual_price, - association_price=result.association_price, - ae_price=result.ae_price, category=result.category, for_sale=result.for_sale, creation_date=result.creation_date, @@ -615,6 +659,37 @@ async def get_product_by_name( ) +async def get_prices_by_product_id_and_date( + product_id: uuid.UUID, + db: AsyncSession, +) -> schemas_sdec_facturation.ProductPriceComplete | None: + """Get the price of a product by its ID and a specific date from the database""" + date = datetime.now(tz=UTC) + result = ( + ( + await db.execute( + select(models_sdec_facturation.ProductPrice) + .where(models_sdec_facturation.ProductPrice.product_id == product_id) + .where(models_sdec_facturation.ProductPrice.effective_date == date), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.ProductPriceComplete( + id=result.id, + product_id=result.product_id, + individual_price=result.individual_price, + association_price=result.association_price, + ae_price=result.ae_price, + effective_date=result.effective_date, + ) + if result + else None + ) + + # ---------------------------------------------------------------------------- # # Order # # ---------------------------------------------------------------------------- # @@ -656,11 +731,24 @@ async def update_order( await db.execute( update(models_sdec_facturation.Order) .where(models_sdec_facturation.Order.id == order_id) - .values(valid=order_edit.valid, order=order_edit.order), + .values(order=order_edit.order), ) await db.flush() +async def delete_order( + order_id: uuid.UUID, + db: AsyncSession, +): + """Delete an order from the database""" + + await db.execute( + update(models_sdec_facturation.Order) + .where(models_sdec_facturation.Order.id == order_id) + .values(valid=False), + ) + + async def get_all_orders( db: AsyncSession, ) -> Sequence[schemas_sdec_facturation.OrderComplete]: @@ -724,9 +812,8 @@ async def create_facture_association( facture_number=facture_association.facture_number, member_id=facture_association.member_id, association_id=facture_association.association_id, - association_order=",".join( - [str(cmd_id) for cmd_id in facture_association.association_order], - ), + start_date=facture_association.start_date, + end_date=facture_association.end_date, price=facture_association.price, facture_date=datetime.now(tz=UTC), valid=facture_association.valid, @@ -740,10 +827,8 @@ async def create_facture_association( facture_number=facture_association_db.facture_number, member_id=facture_association_db.member_id, association_id=facture_association_db.association_id, - association_order=[ - int(cmd_id) - for cmd_id in facture_association_db.association_order.split(",") - ], + start_date=facture_association.start_date, + end_date=facture_association.end_date, price=facture_association_db.price, facture_date=facture_association_db.facture_date, valid=facture_association_db.valid, @@ -752,25 +837,41 @@ async def create_facture_association( ) -async def upfacture_date_association( +async def update_facture_association( facture_association_id: uuid.UUID, - facture_association_edit: schemas_sdec_facturation.FactureAssociationBase, + facture_association_edit: schemas_sdec_facturation.FactureAssociationUpdate, db: AsyncSession, ): """Update an associationciation invoice in the database""" + current_date: datetime | None = datetime.now(tz=UTC) + if not facture_association_edit.paid: + current_date = None await db.execute( update(models_sdec_facturation.FactureAssociation) .where(models_sdec_facturation.FactureAssociation.id == facture_association_id) .values( - valid=facture_association_edit.valid, paid=facture_association_edit.paid, - payment_date=facture_association_edit.payment_date, + payment_date=current_date, ), ) await db.flush() +async def delete_facture_association( + facture_association_id: uuid.UUID, + db: AsyncSession, +): + """Delete an associationciation invoice from the database""" + + await db.execute( + update(models_sdec_facturation.FactureAssociation) + .where(models_sdec_facturation.FactureAssociation.id == facture_association_id) + .values(valid=False), + ) + await db.flush() + + async def get_all_factures_association( db: AsyncSession, ) -> Sequence[schemas_sdec_facturation.FactureAssociationComplete]: @@ -783,10 +884,8 @@ async def get_all_factures_association( facture_number=facture_association.facture_number, member_id=facture_association.member_id, association_id=facture_association.association_id, - association_order=[ - int(cmd_id) - for cmd_id in facture_association.association_order.split(",") - ], + start_date=facture_association.start_date, + end_date=facture_association.end_date, price=facture_association.price, facture_date=facture_association.facture_date, valid=facture_association.valid, @@ -820,9 +919,44 @@ async def get_facture_association_by_id( facture_number=result.facture_number, member_id=result.member_id, association_id=result.association_id, - association_order=[ - int(cmd_id) for cmd_id in result.association_order.split(",") - ], + start_date=result.start_date, + end_date=result.end_date, + price=result.price, + facture_date=result.facture_date, + valid=result.valid, + paid=result.paid, + payment_date=result.payment_date, + ) + if result + else None + ) + + +async def get_facture_association_by_number( + facture_number: str, + db: AsyncSession, +) -> schemas_sdec_facturation.FactureAssociationComplete | None: + """Get specific associationciation invoices by their facture number from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.FactureAssociation).where( + models_sdec_facturation.FactureAssociation.facture_number + == facture_number, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.FactureAssociationComplete( + id=result.id, + facture_number=result.facture_number, + member_id=result.member_id, + association_id=result.association_id, + start_date=result.start_date, + end_date=result.end_date, price=result.price, facture_date=result.facture_date, valid=result.valid, @@ -837,107 +971,120 @@ async def get_facture_association_by_id( # ---------------------------------------------------------------------------- # # Facture Individual # # ---------------------------------------------------------------------------- # -async def create_facture_particulier( - facture_particulier: schemas_sdec_facturation.FactureIndividualBase, +async def create_facture_individual( + facture_individual: schemas_sdec_facturation.FactureIndividualBase, db: AsyncSession, ) -> schemas_sdec_facturation.FactureIndividualComplete: """Create a new individual invoice in the database""" - facture_particulier_db = models_sdec_facturation.FactureIndividual( + facture_individual_db = models_sdec_facturation.FactureIndividual( id=uuid.uuid4(), - facture_number=facture_particulier.facture_number, - member_id=facture_particulier.member_id, - individual_order=facture_particulier.individual_order, - individual_category=facture_particulier.individual_category, - price=facture_particulier.price, + facture_number=facture_individual.facture_number, + member_id=facture_individual.member_id, + individual_order=facture_individual.individual_order, + individual_category=facture_individual.individual_category, + price=facture_individual.price, facture_date=datetime.now(tz=UTC), - firstname=facture_particulier.firstname, - lastname=facture_particulier.lastname, - adresse=facture_particulier.adresse, - postal_code=facture_particulier.postal_code, - city=facture_particulier.city, - country=facture_particulier.country, - valid=facture_particulier.valid, - paid=facture_particulier.paid, - payment_date=facture_particulier.payment_date, - ) - db.add(facture_particulier_db) + firstname=facture_individual.firstname, + lastname=facture_individual.lastname, + adresse=facture_individual.adresse, + postal_code=facture_individual.postal_code, + city=facture_individual.city, + country=facture_individual.country, + ) + db.add(facture_individual_db) await db.flush() return schemas_sdec_facturation.FactureIndividualComplete( - id=facture_particulier_db.id, - facture_number=facture_particulier_db.facture_number, - member_id=facture_particulier_db.member_id, - individual_order=facture_particulier_db.individual_order, - individual_category=facture_particulier_db.individual_category, - price=facture_particulier_db.price, - facture_date=facture_particulier_db.facture_date, - firstname=facture_particulier_db.firstname, - lastname=facture_particulier_db.lastname, - adresse=facture_particulier_db.adresse, - postal_code=facture_particulier_db.postal_code, - city=facture_particulier_db.city, - country=facture_particulier_db.country, - valid=facture_particulier_db.valid, - paid=facture_particulier_db.paid, - payment_date=facture_particulier_db.payment_date, - ) - - -async def up_date_facture_particulier( - facture_particulier_id: uuid.UUID, - facture_particulier_edit: schemas_sdec_facturation.FactureIndividualBase, + id=facture_individual_db.id, + facture_number=facture_individual_db.facture_number, + member_id=facture_individual_db.member_id, + individual_order=facture_individual_db.individual_order, + individual_category=facture_individual_db.individual_category, + price=facture_individual_db.price, + facture_date=facture_individual_db.facture_date, + firstname=facture_individual_db.firstname, + lastname=facture_individual_db.lastname, + adresse=facture_individual_db.adresse, + postal_code=facture_individual_db.postal_code, + city=facture_individual_db.city, + country=facture_individual_db.country, + valid=facture_individual_db.valid, + paid=facture_individual_db.paid, + payment_date=facture_individual_db.payment_date, + ) + + +async def update_facture_individual( + facture_individual_id: uuid.UUID, + facture_individual_edit: schemas_sdec_facturation.FactureIndividualUpdate, db: AsyncSession, ): """Update an individual invoice in the database""" + current_date: datetime | None = datetime.now(tz=UTC) + if not facture_individual_edit.paid: + current_date = None await db.execute( update(models_sdec_facturation.FactureIndividual) - .where(models_sdec_facturation.FactureIndividual.id == facture_particulier_id) + .where(models_sdec_facturation.FactureIndividual.id == facture_individual_id) .values( - firstname=facture_particulier_edit.firstname, - lastname=facture_particulier_edit.lastname, - adresse=facture_particulier_edit.adresse, - postal_code=facture_particulier_edit.postal_code, - city=facture_particulier_edit.city, - valid=facture_particulier_edit.valid, - paid=facture_particulier_edit.paid, - payment_date=facture_particulier_edit.payment_date, + firstname=facture_individual_edit.firstname, + lastname=facture_individual_edit.lastname, + adresse=facture_individual_edit.adresse, + postal_code=facture_individual_edit.postal_code, + city=facture_individual_edit.city, + paid=facture_individual_edit.paid, + payment_date=current_date, ), ) await db.flush() -async def get_all_factures_particulier( +async def delete_facture_individual( + facture_individual_id: uuid.UUID, + db: AsyncSession, +): + """Delete an individual invoice from the database""" + + await db.execute( + update(models_sdec_facturation.FactureIndividual) + .where(models_sdec_facturation.FactureIndividual.id == facture_individual_id) + .values(valid=False), + ) + await db.flush() + + +async def get_all_factures_individual( db: AsyncSession, ) -> Sequence[schemas_sdec_facturation.FactureIndividualComplete]: """Get all individual invoices from the database""" result = await db.execute(select(models_sdec_facturation.FactureIndividual)) - factures_particulier = result.scalars().all() + factures_individual = result.scalars().all() return [ schemas_sdec_facturation.FactureIndividualComplete( - id=facture_particulier.id, - facture_number=facture_particulier.facture_number, - member_id=facture_particulier.member_id, - individual_order=facture_particulier.individual_order, - individual_category=facture_particulier.individual_category, - price=facture_particulier.price, - facture_date=facture_particulier.facture_date, - firstname=facture_particulier.firstname, - lastname=facture_particulier.lastname, - adresse=facture_particulier.adresse, - postal_code=facture_particulier.postal_code, - city=facture_particulier.city, - country=facture_particulier.country, - valid=facture_particulier.valid, - paid=facture_particulier.paid, - payment_date=facture_particulier.payment_date, - ) - for facture_particulier in factures_particulier + id=facture_individual.id, + facture_number=facture_individual.facture_number, + member_id=facture_individual.member_id, + individual_order=facture_individual.individual_order, + individual_category=facture_individual.individual_category, + price=facture_individual.price, + facture_date=facture_individual.facture_date, + firstname=facture_individual.firstname, + lastname=facture_individual.lastname, + adresse=facture_individual.adresse, + postal_code=facture_individual.postal_code, + city=facture_individual.city, + country=facture_individual.country, + valid=facture_individual.valid, + paid=facture_individual.paid, + payment_date=facture_individual.payment_date, + ) + for facture_individual in factures_individual ] -async def get_facture_particulier_by_id( - facture_particulier_id: uuid.UUID, +async def get_facture_individual_by_id( + facture_individual_id: uuid.UUID, db: AsyncSession, ) -> schemas_sdec_facturation.FactureIndividualComplete | None: """Get a specific individual invoice by its ID from the database""" @@ -946,7 +1093,48 @@ async def get_facture_particulier_by_id( await db.execute( select(models_sdec_facturation.FactureIndividual).where( models_sdec_facturation.FactureIndividual.id - == facture_particulier_id, + == facture_individual_id, + ), + ) + ) + .scalars() + .first() + ) + return ( + schemas_sdec_facturation.FactureIndividualComplete( + id=result.id, + facture_number=result.facture_number, + member_id=result.member_id, + individual_order=result.individual_order, + individual_category=result.individual_category, + price=result.price, + facture_date=result.facture_date, + firstname=result.firstname, + lastname=result.lastname, + adresse=result.adresse, + postal_code=result.postal_code, + city=result.city, + country=result.country, + valid=result.valid, + paid=result.paid, + payment_date=result.payment_date, + ) + if result + else None + ) + + +async def get_facture_individual_by_number( + facture_number: str, + db: AsyncSession, +) -> schemas_sdec_facturation.FactureIndividualComplete | None: + """Get specific individual invoices by their facture number from the database""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.FactureIndividual).where( + models_sdec_facturation.FactureIndividual.facture_number + == facture_number, ), ) ) diff --git a/app/modules/sdec_facturation/endpoints_sdec_facturation.py b/app/modules/sdec_facturation/endpoints_sdec_facturation.py index 61e187f136..6041e1c9cf 100644 --- a/app/modules/sdec_facturation/endpoints_sdec_facturation.py +++ b/app/modules/sdec_facturation/endpoints_sdec_facturation.py @@ -1,6 +1,7 @@ import logging import uuid from collections.abc import Sequence +from datetime import UTC, datetime from fastapi import Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession @@ -32,7 +33,7 @@ @module.router.get( - "/sdec_facturation/members/", + "/sdec_facturation/member/", response_model=list[schemas_sdec_facturation.MemberComplete], status_code=200, ) @@ -45,7 +46,7 @@ async def get_all_members( @module.router.post( - "/sdec_facturation/members/", + "/sdec_facturation/member/", response_model=schemas_sdec_facturation.MemberComplete, status_code=201, ) @@ -69,11 +70,17 @@ async def create_member( detail="User is already a member", ) + if (member.mandate < 2000) or (member.mandate > datetime.now(tz=UTC).year + 1): + raise HTTPException( + status_code=400, + detail="Mandate year is not valid", + ) + return await cruds_sdec_facturation.create_member(member, db) @module.router.patch( - "/sdec_facturation/members/{member_id}", + "/sdec_facturation/member/{member_id}", status_code=200, ) async def update_member( @@ -108,6 +115,14 @@ async def update_member( detail="User is already a member", ) + if (member_edit.mandate < 2000) or ( + member_edit.mandate > datetime.now(tz=UTC).year + 1 + ): + raise HTTPException( + status_code=400, + detail="Mandate year is not valid", + ) + await cruds_sdec_facturation.update_member( member_id, member_edit, @@ -116,7 +131,7 @@ async def update_member( @module.router.delete( - "/sdec_facturation/members/{member_id}", + "/sdec_facturation/member/{member_id}", status_code=204, ) async def delete_member( @@ -145,7 +160,7 @@ async def delete_member( # Mandate # # ---------------------------------------------------------------------------- # @module.router.get( - "/sdec_facturation/mandates/", + "/sdec_facturation/mandate/", response_model=list[schemas_sdec_facturation.MandateComplete], status_code=200, ) @@ -158,7 +173,7 @@ async def get_all_mandates( @module.router.post( - "/sdec_facturation/mandates/", + "/sdec_facturation/mandate/", response_model=schemas_sdec_facturation.MandateComplete, status_code=201, ) @@ -186,7 +201,7 @@ async def create_mandate( @module.router.patch( - "/sdec_facturation/mandates/{mandate_year}", + "/sdec_facturation/mandate/{mandate_year}", status_code=200, ) async def update_mandate( @@ -218,7 +233,7 @@ async def update_mandate( @module.router.delete( - "/sdec_facturation/mandates/{mandate_year}", + "/sdec_facturation/mandate/{mandate_year}", status_code=204, ) async def delete_mandate( @@ -358,15 +373,15 @@ async def delete_association( @module.router.get( "/sdec_facturation/product/", - response_model=list[schemas_sdec_facturation.ProductComplete], + response_model=list[schemas_sdec_facturation.ProductAndPriceComplete], status_code=200, ) async def get_all_products( db: AsyncSession = Depends(get_db), _=Depends(is_user()), -) -> Sequence[schemas_sdec_facturation.ProductComplete]: +) -> Sequence[schemas_sdec_facturation.ProductAndPriceComplete]: """Get all product items from the database""" - return await cruds_sdec_facturation.get_all_products(db) + return await cruds_sdec_facturation.get_all_products_and_price(db) @module.router.post( @@ -375,10 +390,10 @@ async def get_all_products( status_code=201, ) async def create_product( - product: schemas_sdec_facturation.ProductBase, + product: schemas_sdec_facturation.ProductAndPriceBase, db: AsyncSession = Depends(get_db), _=Depends(is_user_in([GroupType.sdec_facturation_admin])), -) -> schemas_sdec_facturation.ProductComplete: +) -> schemas_sdec_facturation.ProductAndPriceComplete: """ Create a new product item in the database **This endpoint is only usable by SDEC Facturation admins** @@ -421,16 +436,6 @@ async def update_product( **This endpoint is only usable by SDEC Facturation admins** """ - if ( - product_edit.individual_price < 0 - or product_edit.association_price < 0 - or product_edit.ae_price < 0 - ): - raise HTTPException( - status_code=400, - detail="Product item prices must be positive", - ) - product_db = await cruds_sdec_facturation.get_product_by_id( product_id, db, @@ -441,30 +446,31 @@ async def update_product( detail="Product item not found", ) - product_base = schemas_sdec_facturation.ProductBase( - code=product_db.code, - name=product_db.name, - individual_price=product_edit.individual_price, - association_price=product_edit.association_price, - ae_price=product_edit.ae_price, - category=product_db.category, - for_sale=product_db.for_sale, - ) + if ( + product_edit.name is not None + and (await cruds_sdec_facturation.get_product_by_name(product_edit.name, db)) + is not None + and product_edit.name != product_db.name + ): + raise HTTPException( + status_code=400, + detail="Product item name already used", + ) await cruds_sdec_facturation.update_product( - product_db.code, - product_base, + product_db.id, + product_edit, db, ) @module.router.patch( - "/sdec_facturation/product/minor/{product_id}", + "/sdec_facturation/product/price/{product_id}", status_code=200, ) -async def minor_update_product( +async def update_price( product_id: uuid.UUID, - product_edit: schemas_sdec_facturation.ProductMinorUpdate, + price_edit: schemas_sdec_facturation.ProductPriceUpdate, db: AsyncSession = Depends(get_db), _=Depends(is_user_in([GroupType.sdec_facturation_admin])), ) -> None: @@ -483,9 +489,26 @@ async def minor_update_product( detail="Product item not found", ) - await cruds_sdec_facturation.minor_update_product( - product_db.code, - product_edit, + price_db = await cruds_sdec_facturation.get_prices_by_product_id_and_date( + product_id, + db, + ) + current_date = datetime.now(tz=UTC) + if price_db is not None and current_date <= price_db.effective_date: + raise HTTPException( + status_code=400, + detail="New price effective date must be after the current one", + ) + if price_db is not None and current_date == price_db.effective_date: + await cruds_sdec_facturation.update_price( + product_db.id, + price_edit, + db, + ) + + await cruds_sdec_facturation.create_price( + product_db.id, + price_edit, db, ) @@ -513,7 +536,7 @@ async def delete_product( detail="Product item not found", ) - await cruds_sdec_facturation.delete_product(product_db.code, db) + await cruds_sdec_facturation.delete_product(product_id, db) # ---------------------------------------------------------------------------- # @@ -522,7 +545,7 @@ async def delete_product( @module.router.get( - "/sdec_facturation/orders/", + "/sdec_facturation/order/", response_model=list[schemas_sdec_facturation.OrderComplete], status_code=200, ) @@ -534,6 +557,11 @@ async def get_all_orders( return await cruds_sdec_facturation.get_all_orders(db) +@module.router.post( + "/sdec_facturation/order/", + response_model=schemas_sdec_facturation.OrderComplete, + status_code=201, +) async def create_order( order: schemas_sdec_facturation.OrderBase, db: AsyncSession = Depends(get_db), @@ -567,6 +595,352 @@ async def create_order( return await cruds_sdec_facturation.create_order(order, db) +@module.router.patch( + "/sdec_facturation/order/{order_id}", + status_code=200, +) +async def update_order( + order_id: uuid.UUID, + order_edit: schemas_sdec_facturation.OrderUpdate, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Update an order in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + + order_db = await cruds_sdec_facturation.get_order_by_id( + order_id, + db, + ) + if order_db is None: + raise HTTPException( + status_code=404, + detail="Order not found", + ) + + await cruds_sdec_facturation.update_order( + order_id, + order_edit, + db, + ) + + +@module.router.delete( + "/sdec_facturation/order/{order_id}", + status_code=204, +) +async def delete_order( + order_id: uuid.UUID, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Delete an order from the database + **This endpoint is only usable by SDEC Facturation admins** + """ + order_db = await cruds_sdec_facturation.get_order_by_id( + order_id, + db, + ) + if order_db is None: + raise HTTPException( + status_code=404, + detail="Order not found", + ) + + await cruds_sdec_facturation.delete_order(order_id, db) + + +# ---------------------------------------------------------------------------- # +# Facture Association # +# ---------------------------------------------------------------------------- # +@module.router.get( + "/sdec_facturation/facture_association/", + response_model=list[schemas_sdec_facturation.FactureAssociationComplete], + status_code=200, +) +async def get_all_facture_associations( + db: AsyncSession = Depends(get_db), + _=Depends(is_user()), +) -> Sequence[schemas_sdec_facturation.FactureAssociationComplete]: + """Get all facture associations from the database""" + return await cruds_sdec_facturation.get_all_factures_association(db) + + +@module.router.post( + "/sdec_facturation/facture_association/", + response_model=schemas_sdec_facturation.FactureAssociationComplete, + status_code=201, +) +async def create_facture_association( + facture_association: schemas_sdec_facturation.FactureAssociationBase, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> schemas_sdec_facturation.FactureAssociationComplete: + """ + Create a new facture association in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + if ( + await cruds_sdec_facturation.get_facture_association_by_number( + facture_association.facture_number, + db, + ) + ) is not None: + raise HTTPException( + status_code=400, + detail="Facture number already used", + ) + + if ( + await cruds_sdec_facturation.get_member_by_id( + facture_association.member_id, + db, + ) + ) is None: + raise HTTPException( + status_code=400, + detail="Member does not exist", + ) + + if ( + await cruds_sdec_facturation.get_association_by_id( + facture_association.association_id, + db, + ) + ) is None: + raise HTTPException( + status_code=400, + detail="Association does not exist", + ) + + if facture_association.price < 0: + raise HTTPException( + status_code=400, + detail="Facture price must be positive", + ) + + if facture_association.start_date >= facture_association.end_date: + raise HTTPException( + status_code=400, + detail="Facture start date must be before end date", + ) + + return await cruds_sdec_facturation.create_facture_association( + facture_association, + db, + ) + + +@module.router.patch( + "/sdec_facturation/facture_association/{facture_association_id}", + status_code=200, +) +async def update_facture_association( + facture_association_id: uuid.UUID, + facture_association_edit: schemas_sdec_facturation.FactureAssociationUpdate, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Update a facture association in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + + facture_association_db = await cruds_sdec_facturation.get_facture_association_by_id( + facture_association_id, + db, + ) + if facture_association_db is None: + raise HTTPException( + status_code=404, + detail="Facture association not found", + ) + + await cruds_sdec_facturation.update_facture_association( + facture_association_id, + facture_association_edit, + db, + ) + + +@module.router.delete( + "/sdec_facturation/facture_association/{facture_association_id}", + status_code=204, +) +async def delete_facture_association( + facture_association_id: uuid.UUID, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Delete a facture association from the database + **This endpoint is only usable by SDEC Facturation admins** + """ + facture_association_db = await cruds_sdec_facturation.get_facture_association_by_id( + facture_association_id, + db, + ) + if facture_association_db is None: + raise HTTPException( + status_code=404, + detail="Facture association not found", + ) + + await cruds_sdec_facturation.delete_facture_association( + facture_association_id, + db, + ) + + # ---------------------------------------------------------------------------- # -# Facture Association # +# Facture Individual # # ---------------------------------------------------------------------------- # +@module.router.get( + "/sdec_facturation/facture_individual/", + response_model=list[schemas_sdec_facturation.FactureIndividualComplete], + status_code=200, +) +async def get_all_facture_individuals( + db: AsyncSession = Depends(get_db), + _=Depends(is_user()), +) -> Sequence[schemas_sdec_facturation.FactureIndividualComplete]: + """Get all facture individuals from the database""" + return await cruds_sdec_facturation.get_all_factures_individual(db) + + +@module.router.post( + "/sdec_facturation/facture_individual/", + response_model=schemas_sdec_facturation.FactureIndividualComplete, + status_code=201, +) +async def create_facture_individual( + facture_individual: schemas_sdec_facturation.FactureIndividualBase, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> schemas_sdec_facturation.FactureIndividualComplete: + """ + Create a new facture individual in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + if facture_individual.firstname.strip() == "": + raise HTTPException( + status_code=400, + detail="Firstname cannot be empty", + ) + if facture_individual.lastname.strip() == "": + raise HTTPException( + status_code=400, + detail="Lastname cannot be empty", + ) + if facture_individual.adresse.strip() == "": + raise HTTPException( + status_code=400, + detail="Adresse cannot be empty", + ) + if facture_individual.postal_code.strip() == "": + raise HTTPException( + status_code=400, + detail="Postal code cannot be empty", + ) + if facture_individual.city.strip() == "": + raise HTTPException( + status_code=400, + detail="City cannot be empty", + ) + + if ( + await cruds_sdec_facturation.get_facture_individual_by_number( + facture_individual.facture_number, + db, + ) + ) is not None: + raise HTTPException( + status_code=400, + detail="Facture number already used", + ) + + if ( + await cruds_sdec_facturation.get_member_by_id( + facture_individual.member_id, + db, + ) + ) is None: + raise HTTPException( + status_code=400, + detail="Member does not exist", + ) + + if facture_individual.price < 0: + raise HTTPException( + status_code=400, + detail="Facture price must be positive", + ) + + return await cruds_sdec_facturation.create_facture_individual( + facture_individual, + db, + ) + + +@module.router.patch( + "/sdec_facturation/facture_individual/{facture_individual_id}", + status_code=200, +) +async def update_facture_individual( + facture_individual_id: uuid.UUID, + facture_individual_edit: schemas_sdec_facturation.FactureIndividualUpdate, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Update a facture individual in the database + **This endpoint is only usable by SDEC Facturation admins** + """ + + facture_individual_db = await cruds_sdec_facturation.get_facture_individual_by_id( + facture_individual_id, + db, + ) + if facture_individual_db is None: + raise HTTPException( + status_code=404, + detail="Facture individual not found", + ) + + await cruds_sdec_facturation.update_facture_individual( + facture_individual_id, + facture_individual_edit, + db, + ) + + +@module.router.delete( + "/sdec_facturation/facture_individual/{facture_individual_id}", + status_code=204, +) +async def delete_facture_individual( + facture_individual_id: uuid.UUID, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> None: + """ + Delete a facture individual from the database + **This endpoint is only usable by SDEC Facturation admins** + """ + facture_individual_db = await cruds_sdec_facturation.get_facture_individual_by_id( + facture_individual_id, + db, + ) + if facture_individual_db is None: + raise HTTPException( + status_code=404, + detail="Facture individual not found", + ) + + await cruds_sdec_facturation.delete_facture_individual( + facture_individual_id, + db, + ) diff --git a/app/modules/sdec_facturation/models_sdec_facturation.py b/app/modules/sdec_facturation/models_sdec_facturation.py index 7b62c48e0e..188fbf9dcb 100644 --- a/app/modules/sdec_facturation/models_sdec_facturation.py +++ b/app/modules/sdec_facturation/models_sdec_facturation.py @@ -10,6 +10,7 @@ AssociationStructureType, AssociationType, IndividualCategoryType, + ProductCategoryType, RoleType, ) from app.types.sqlalchemy import Base, PrimaryKey @@ -46,12 +47,19 @@ class Product(Base): id: Mapped[PrimaryKey] code: Mapped[str] name: Mapped[str] + category: Mapped[ProductCategoryType] + creation_date: Mapped[date] + for_sale: Mapped[bool] = mapped_column(default=True) + + +class ProductPrice(Base): + __tablename__ = "sdec_facturation_product_price" + id: Mapped[PrimaryKey] + product_id: Mapped[UUID] = mapped_column(ForeignKey("sdec_facturation_product.id")) individual_price: Mapped[float] association_price: Mapped[float] ae_price: Mapped[float] - category: Mapped[str] - creation_date: Mapped[date] - for_sale: Mapped[bool] = mapped_column(default=True) + effective_date: Mapped[date] class Order(Base): @@ -74,11 +82,12 @@ class FactureAssociation(Base): association_id: Mapped[UUID] = mapped_column( ForeignKey("sdec_facturation_association.id"), ) - association_order: Mapped[str] + start_date: Mapped[date] + end_date: Mapped[date] price: Mapped[float] facture_date: Mapped[date] - valid: Mapped[bool] - paid: Mapped[bool] + paid: Mapped[bool] = mapped_column(default=False) + valid: Mapped[bool] = mapped_column(default=True) payment_date: Mapped[date | None] = mapped_column(default=None) @@ -97,6 +106,6 @@ class FactureIndividual(Base): postal_code: Mapped[str] city: Mapped[str] country: Mapped[str] - valid: Mapped[bool] - paid: Mapped[bool] + paid: Mapped[bool] = mapped_column(default=False) + valid: Mapped[bool] = mapped_column(default=True) payment_date: Mapped[date | None] = mapped_column(default=None) diff --git a/app/modules/sdec_facturation/schemas_sdec_facturation.py b/app/modules/sdec_facturation/schemas_sdec_facturation.py index 285e0f32dd..bbe47cd2dc 100644 --- a/app/modules/sdec_facturation/schemas_sdec_facturation.py +++ b/app/modules/sdec_facturation/schemas_sdec_facturation.py @@ -1,7 +1,7 @@ -"""Schemas file for endpoint /sdec-facturation""" +"""Schemas file for endpoint /sdec_facturation""" -import uuid from datetime import date +from uuid import UUID from pydantic import BaseModel @@ -9,6 +9,7 @@ AssociationStructureType, AssociationType, IndividualCategoryType, + ProductCategoryType, RoleType, ) @@ -21,7 +22,7 @@ class MemberBase(BaseModel): class MemberComplete(MemberBase): - id: uuid.UUID + id: UUID modified_date: date @@ -42,78 +43,98 @@ class AssociationBase(BaseModel): class AssociationComplete(AssociationBase): - id: uuid.UUID + id: UUID modified_date: date class ProductBase(BaseModel): code: str name: str - individual_price: float - association_price: float - ae_price: float - category: str + category: ProductCategoryType for_sale: bool = True class ProductComplete(ProductBase): - id: uuid.UUID + id: UUID creation_date: date -class ProductMinorUpdate(BaseModel): +class ProductUpdate(BaseModel): name: str | None = None - category: str | None = None + category: ProductCategoryType | None = None + for_sale: bool | None = None -class ProductUpdate(BaseModel): +class ProductPriceBase(BaseModel): + product_id: UUID + individual_price: float + association_price: float + ae_price: float + + +class ProductPriceComplete(ProductPriceBase): + id: UUID + effective_date: date + + +class ProductPriceUpdate(BaseModel): individual_price: float association_price: float ae_price: float +class ProductAndPriceBase(ProductBase): + individual_price: float + association_price: float + ae_price: float + + +class ProductAndPriceComplete(ProductAndPriceBase): + id: UUID + creation_date: date + effective_date: date + + class OrderBase(BaseModel): - association_id: uuid.UUID - member_id: uuid.UUID + association_id: UUID + member_id: UUID order: str valid: bool = True class OrderComplete(OrderBase): - id: uuid.UUID + id: UUID creation_date: date class OrderUpdate(BaseModel): - valid: bool - order: str + order: str | None = None class FactureAssociationBase(BaseModel): facture_number: str - member_id: uuid.UUID - association_id: uuid.UUID - association_order: list[int] + member_id: UUID + association_id: UUID + start_date: date + end_date: date price: float - valid: bool - paid: bool + valid: bool = True + paid: bool = False payment_date: date | None = None class FactureAssociationComplete(FactureAssociationBase): facture_date: date - id: uuid.UUID + id: UUID -class FactureUpdate(BaseModel): - valid: bool | None = None +class FactureAssociationUpdate(BaseModel): paid: bool | None = None - payment_date: date | None = None class FactureIndividualBase(BaseModel): facture_number: str - member_id: uuid.UUID + member_id: UUID individual_order: str individual_category: IndividualCategoryType price: float @@ -123,11 +144,21 @@ class FactureIndividualBase(BaseModel): postal_code: str city: str country: str - valid: bool - paid: bool + valid: bool = True + paid: bool = False payment_date: date | None = None class FactureIndividualComplete(FactureIndividualBase): facture_date: date - id: uuid.UUID + id: UUID + + +class FactureIndividualUpdate(BaseModel): + firstname: str | None = None + lastname: str | None = None + adresse: str | None = None + postal_code: str | None = None + city: str | None = None + country: str | None = None + paid: bool | None = None diff --git a/app/modules/sdec_facturation/types_sdec_facturation.py b/app/modules/sdec_facturation/types_sdec_facturation.py index 96cd9eeb60..8575bdc751 100644 --- a/app/modules/sdec_facturation/types_sdec_facturation.py +++ b/app/modules/sdec_facturation/types_sdec_facturation.py @@ -1,16 +1,6 @@ from enum import Enum -class IndividualCategoryType(str, Enum): - pe = "pe" - pa = "pa" - autre = "autre" - tfe = "tfe" - - def __str__(self) -> str: - return f"{self.name}<{self.value}" - - class RoleType(str, Enum): prez = "prez" trez = "trez" @@ -46,3 +36,31 @@ class AssociationType(str, Enum): def __str__(self) -> str: return f"{self.name}<{self.value}" + + +class ProductCategoryType(str, Enum): + impression = "impression" + papier_a4 = "papier_a4" + papier_a3 = "papier_a3" + enveloppe = "enveloppe" + ticket = "ticket" + reliure_plastification = "reliure_plastification" + petite_fourniture = "petite_fourniture" + grosse_fourniture = "grosse_fourniture" + poly = "poly" + papier_tasoeur = "papier_tasoeur" + tshirt_flocage = "tshirt_flocage" + divers = "divers" + + def __str__(self) -> str: + return f"{self.name}<{self.value}" + + +class IndividualCategoryType(str, Enum): + pe = "pe" + pa = "pa" + autre = "autre" + tfe = "tfe" + + def __str__(self) -> str: + return f"{self.name}<{self.value}" From 94550e3c70ed85c15fe63e0dc1943b726599da49 Mon Sep 17 00:00:00 2001 From: Azarta2Zygoto Date: Mon, 10 Nov 2025 23:18:18 +0100 Subject: [PATCH 3/3] =?UTF-8?q?Cr=C3=A9ation=20tests=20+=20ajustement=20de?= =?UTF-8?q?=20la=20base=20de=20donn=C3=A9es?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cruds_sdec_facturation.py | 9 +- .../endpoints_sdec_facturation.py | 40 +- .../schemas_sdec_facturation.py | 12 +- tests/test_sdec_facturation.py | 1312 +++++++++++++++++ 4 files changed, 1342 insertions(+), 31 deletions(-) create mode 100644 tests/test_sdec_facturation.py diff --git a/app/modules/sdec_facturation/cruds_sdec_facturation.py b/app/modules/sdec_facturation/cruds_sdec_facturation.py index 61641a5f06..720ea5bf20 100644 --- a/app/modules/sdec_facturation/cruds_sdec_facturation.py +++ b/app/modules/sdec_facturation/cruds_sdec_facturation.py @@ -839,7 +839,7 @@ async def create_facture_association( async def update_facture_association( facture_association_id: uuid.UUID, - facture_association_edit: schemas_sdec_facturation.FactureAssociationUpdate, + facture_association_edit: schemas_sdec_facturation.FactureUpdate, db: AsyncSession, ): """Update an associationciation invoice in the database""" @@ -1016,7 +1016,7 @@ async def create_facture_individual( async def update_facture_individual( facture_individual_id: uuid.UUID, - facture_individual_edit: schemas_sdec_facturation.FactureIndividualUpdate, + facture_individual_edit: schemas_sdec_facturation.FactureUpdate, db: AsyncSession, ): """Update an individual invoice in the database""" @@ -1028,11 +1028,6 @@ async def update_facture_individual( update(models_sdec_facturation.FactureIndividual) .where(models_sdec_facturation.FactureIndividual.id == facture_individual_id) .values( - firstname=facture_individual_edit.firstname, - lastname=facture_individual_edit.lastname, - adresse=facture_individual_edit.adresse, - postal_code=facture_individual_edit.postal_code, - city=facture_individual_edit.city, paid=facture_individual_edit.paid, payment_date=current_date, ), diff --git a/app/modules/sdec_facturation/endpoints_sdec_facturation.py b/app/modules/sdec_facturation/endpoints_sdec_facturation.py index 6041e1c9cf..99ba652fe1 100644 --- a/app/modules/sdec_facturation/endpoints_sdec_facturation.py +++ b/app/modules/sdec_facturation/endpoints_sdec_facturation.py @@ -81,7 +81,7 @@ async def create_member( @module.router.patch( "/sdec_facturation/member/{member_id}", - status_code=200, + status_code=204, ) async def update_member( member_id: uuid.UUID, @@ -202,7 +202,7 @@ async def create_mandate( @module.router.patch( "/sdec_facturation/mandate/{mandate_year}", - status_code=200, + status_code=204, ) async def update_mandate( mandate_year: int, @@ -303,7 +303,7 @@ async def create_association( @module.router.patch( "/sdec_facturation/association/{association_id}", - status_code=200, + status_code=204, ) async def update_association( association_id: uuid.UUID, @@ -423,7 +423,7 @@ async def create_product( @module.router.patch( "/sdec_facturation/product/{product_id}", - status_code=200, + status_code=204, ) async def update_product( product_id: uuid.UUID, @@ -466,7 +466,7 @@ async def update_product( @module.router.patch( "/sdec_facturation/product/price/{product_id}", - status_code=200, + status_code=204, ) async def update_price( product_id: uuid.UUID, @@ -479,6 +479,16 @@ async def update_price( **This endpoint is only usable by SDEC Facturation admins** """ + if ( + price_edit.individual_price < 0 + or price_edit.association_price < 0 + or price_edit.ae_price < 0 + ): + raise HTTPException( + status_code=400, + detail="Product item prices must be positive", + ) + product_db = await cruds_sdec_facturation.get_product_by_id( product_id, db, @@ -494,7 +504,7 @@ async def update_price( db, ) current_date = datetime.now(tz=UTC) - if price_db is not None and current_date <= price_db.effective_date: + if price_db is not None and current_date < price_db.effective_date: raise HTTPException( status_code=400, detail="New price effective date must be after the current one", @@ -540,7 +550,7 @@ async def delete_product( # ---------------------------------------------------------------------------- # -# Order # +# Order # # ---------------------------------------------------------------------------- # @@ -597,7 +607,7 @@ async def create_order( @module.router.patch( "/sdec_facturation/order/{order_id}", - status_code=200, + status_code=204, ) async def update_order( order_id: uuid.UUID, @@ -736,11 +746,11 @@ async def create_facture_association( @module.router.patch( "/sdec_facturation/facture_association/{facture_association_id}", - status_code=200, + status_code=204, ) async def update_facture_association( facture_association_id: uuid.UUID, - facture_association_edit: schemas_sdec_facturation.FactureAssociationUpdate, + facture_association_edit: schemas_sdec_facturation.FactureUpdate, db: AsyncSession = Depends(get_db), _=Depends(is_user_in([GroupType.sdec_facturation_admin])), ) -> None: @@ -850,6 +860,11 @@ async def create_facture_individual( status_code=400, detail="City cannot be empty", ) + if facture_individual.country.strip() == "": + raise HTTPException( + status_code=400, + detail="Country cannot be empty", + ) if ( await cruds_sdec_facturation.get_facture_individual_by_number( @@ -887,11 +902,11 @@ async def create_facture_individual( @module.router.patch( "/sdec_facturation/facture_individual/{facture_individual_id}", - status_code=200, + status_code=204, ) async def update_facture_individual( facture_individual_id: uuid.UUID, - facture_individual_edit: schemas_sdec_facturation.FactureIndividualUpdate, + facture_individual_edit: schemas_sdec_facturation.FactureUpdate, db: AsyncSession = Depends(get_db), _=Depends(is_user_in([GroupType.sdec_facturation_admin])), ) -> None: @@ -899,7 +914,6 @@ async def update_facture_individual( Update a facture individual in the database **This endpoint is only usable by SDEC Facturation admins** """ - facture_individual_db = await cruds_sdec_facturation.get_facture_individual_by_id( facture_individual_id, db, diff --git a/app/modules/sdec_facturation/schemas_sdec_facturation.py b/app/modules/sdec_facturation/schemas_sdec_facturation.py index bbe47cd2dc..516cb99770 100644 --- a/app/modules/sdec_facturation/schemas_sdec_facturation.py +++ b/app/modules/sdec_facturation/schemas_sdec_facturation.py @@ -128,10 +128,6 @@ class FactureAssociationComplete(FactureAssociationBase): id: UUID -class FactureAssociationUpdate(BaseModel): - paid: bool | None = None - - class FactureIndividualBase(BaseModel): facture_number: str member_id: UUID @@ -154,11 +150,5 @@ class FactureIndividualComplete(FactureIndividualBase): id: UUID -class FactureIndividualUpdate(BaseModel): - firstname: str | None = None - lastname: str | None = None - adresse: str | None = None - postal_code: str | None = None - city: str | None = None - country: str | None = None +class FactureUpdate(BaseModel): paid: bool | None = None diff --git a/tests/test_sdec_facturation.py b/tests/test_sdec_facturation.py new file mode 100644 index 0000000000..fdcde25ff4 --- /dev/null +++ b/tests/test_sdec_facturation.py @@ -0,0 +1,1312 @@ +import uuid +from datetime import UTC, datetime + +import pytest_asyncio +from fastapi.testclient import TestClient + +from app.core.groups import models_groups +from app.core.groups.groups_type import GroupType +from app.core.users import models_users +from app.modules.sdec_facturation import ( + cruds_sdec_facturation, + models_sdec_facturation, + schemas_sdec_facturation, +) +from app.modules.sdec_facturation.types_sdec_facturation import ( + AssociationStructureType, + AssociationType, + IndividualCategoryType, + ProductCategoryType, + RoleType, +) +from tests.commons import ( + add_object_to_db, + create_api_access_token, + create_user_with_groups, + get_TestingSessionLocal, +) + +sdec_facturation_admin: models_users.CoreUser +sdec_facturation_user: models_users.CoreUser +token_admin: str +token_user: str + +member1: models_sdec_facturation.Member +member2: models_sdec_facturation.Member + +mandate: models_sdec_facturation.Mandate + +association1: models_sdec_facturation.Association +association2: models_sdec_facturation.Association +association3: models_sdec_facturation.Association + +product1: models_sdec_facturation.Product +product2: models_sdec_facturation.Product +product3: models_sdec_facturation.Product +product4: models_sdec_facturation.Product + +productPrice1: models_sdec_facturation.ProductPrice +productPrice2: models_sdec_facturation.ProductPrice +productPrice3: models_sdec_facturation.ProductPrice +productPrice4: models_sdec_facturation.ProductPrice +productPrice5: models_sdec_facturation.ProductPrice + +order1: models_sdec_facturation.Order +order2: models_sdec_facturation.Order + +factureAssociation1: models_sdec_facturation.FactureAssociation + +FactureIndividual1: models_sdec_facturation.FactureIndividual + + +@pytest_asyncio.fixture(scope="module", autouse=True) +async def init_objects(): + global sdec_facturation_admin, token_admin + sdec_facturation_admin = await create_user_with_groups( + [GroupType.sdec_facturation_admin], + ) + token_admin = create_api_access_token(sdec_facturation_admin) + + global sdec_facturation_user, token_user + sdec_facturation_user = await create_user_with_groups( + [], + ) + token_user = create_api_access_token(sdec_facturation_user) + + global member1, member2 + member1 = models_sdec_facturation.Member( + id=uuid.uuid4(), + name="Member 1", + mandate=2023, + role=RoleType.prez, + modified_date=datetime(2023, 1, 1, tzinfo=UTC), + visible=True, + ) + await add_object_to_db(member1) + member2 = models_sdec_facturation.Member( + id=uuid.uuid4(), + name="Member 2", + mandate=2023, + role=RoleType.trez, + modified_date=datetime(2023, 2, 1, tzinfo=UTC), + visible=True, + ) + await add_object_to_db(member2) + + global mandate + mandate = models_sdec_facturation.Mandate( + year=2023, + name="Mandate 2023", + ) + await add_object_to_db(mandate) + + global association1, association2, association3 + association1 = models_sdec_facturation.Association( + id=uuid.uuid4(), + name="Association 1", + type=AssociationType.aeecl, + structure=AssociationStructureType.asso, + modified_date=datetime(2023, 1, 15, tzinfo=UTC), + visible=True, + ) + await add_object_to_db(association1) + association2 = models_sdec_facturation.Association( + id=uuid.uuid4(), + name="Association 2", + type=AssociationType.useecl, + structure=AssociationStructureType.club, + modified_date=datetime(2023, 2, 15, tzinfo=UTC), + visible=True, + ) + await add_object_to_db(association2) + association3 = models_sdec_facturation.Association( + id=uuid.uuid4(), + name="Association 3", + type=AssociationType.independant, + structure=AssociationStructureType.section, + modified_date=datetime(2023, 3, 15, tzinfo=UTC), + visible=True, + ) + await add_object_to_db(association3) + + global product1, product2, product3, product4 + product1 = models_sdec_facturation.Product( + id=uuid.uuid4(), + code="P001", + name="Product 1", + category=ProductCategoryType.impression, + creation_date=datetime(2023, 1, 10, tzinfo=UTC), + for_sale=True, + ) + await add_object_to_db(product1) + product2 = models_sdec_facturation.Product( + id=uuid.uuid4(), + code="P002", + name="Product 2", + category=ProductCategoryType.papier_a4, + creation_date=datetime(2023, 1, 10, tzinfo=UTC), + for_sale=True, + ) + await add_object_to_db(product2) + product3 = models_sdec_facturation.Product( + id=uuid.uuid4(), + code="P003", + name="Product 3", + category=ProductCategoryType.enveloppe, + creation_date=datetime(2023, 1, 10, tzinfo=UTC), + for_sale=True, + ) + await add_object_to_db(product3) + product4 = models_sdec_facturation.Product( + id=uuid.uuid4(), + code="P004", + name="Product 4", + category=ProductCategoryType.ticket, + creation_date=datetime(2023, 1, 10, tzinfo=UTC), + for_sale=True, + ) + await add_object_to_db(product4) + + global productPrice1, productPrice2, productPrice3, productPrice4, productPrice5 + productPrice1 = models_sdec_facturation.ProductPrice( + id=uuid.uuid4(), + product_id=product1.id, + individual_price=1.0, + association_price=0.8, + ae_price=0.5, + effective_date=datetime(2023, 1, 15, tzinfo=UTC), + ) + await add_object_to_db(productPrice1) + productPrice2 = models_sdec_facturation.ProductPrice( + id=uuid.uuid4(), + product_id=product2.id, + individual_price=2.0, + association_price=1.5, + ae_price=1.0, + effective_date=datetime(2023, 1, 15, tzinfo=UTC), + ) + await add_object_to_db(productPrice2) + productPrice3 = models_sdec_facturation.ProductPrice( + id=uuid.uuid4(), + product_id=product3.id, + individual_price=3.0, + association_price=2.5, + ae_price=2.0, + effective_date=datetime(2023, 1, 15, tzinfo=UTC), + ) + await add_object_to_db(productPrice3) + productPrice4 = models_sdec_facturation.ProductPrice( + id=uuid.uuid4(), + product_id=product4.id, + individual_price=4.0, + association_price=3.5, + ae_price=3.0, + effective_date=datetime.now(tz=UTC), + ) + await add_object_to_db(productPrice4) + productPrice5 = models_sdec_facturation.ProductPrice( + id=uuid.uuid4(), + product_id=product1.id, + individual_price=1.2, + association_price=0.9, + ae_price=0.6, + effective_date=datetime(2023, 2, 15, tzinfo=UTC), + ) + await add_object_to_db(productPrice5) + + global order1, order2 + order1 = models_sdec_facturation.Order( + id=uuid.uuid4(), + association_id=association1.id, + member_id=member1.id, + order="Product 1:10,Product 2:5", + creation_date=datetime(2023, 3, 1, tzinfo=UTC), + valid=True, + ) + await add_object_to_db(order1) + order2 = models_sdec_facturation.Order( + id=uuid.uuid4(), + association_id=association2.id, + member_id=member2.id, + order="Product 3:7,Product 4:3", + creation_date=datetime(2023, 3, 5, tzinfo=UTC), + valid=True, + ) + await add_object_to_db(order2) + + global factureAssociation1 + factureAssociation1 = models_sdec_facturation.FactureAssociation( + id=uuid.uuid4(), + facture_number="FA2023001", + member_id=member1.id, + association_id=association1.id, + start_date=datetime(2023, 1, 1, tzinfo=UTC), + end_date=datetime(2023, 12, 31, tzinfo=UTC), + price=150.0, + facture_date=datetime(2023, 3, 10, tzinfo=UTC), + paid=False, + valid=True, + payment_date=None, + ) + await add_object_to_db(factureAssociation1) + + global FactureIndividual1 + FactureIndividual1 = models_sdec_facturation.FactureIndividual( + id=uuid.uuid4(), + facture_number="FI2023001", + member_id=member2.id, + individual_order="Product 1:2,Product 4:1", + individual_category=IndividualCategoryType.pe, + price=6.4, + facture_date=datetime(2023, 3, 12, tzinfo=UTC), + firstname="John", + lastname="Doe", + adresse="123 Main St", + postal_code="69000", + city="Lyon", + country="France", + paid=False, + valid=True, + payment_date=None, + ) + await add_object_to_db(FactureIndividual1) + + +# ---------------------------------------------------------------------------- # +# Get tests # +# ---------------------------------------------------------------------------- # + +# ---------------------------------------------------------------------------- # +# Member # +# ---------------------------------------------------------------------------- # + + +def test_get_all_members(client: TestClient): + """Test retrieving all members.""" + response = client.get( + "/sdec_facturation/member/", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 200 + assert len(response.json()) == 2 + + +def test_create_member(client: TestClient): + """Test creating a new member.""" + new_member_data = { + "name": "Member 3", + "mandate": 2023, + "role": "com", + "visible": True, + } + response = client.post( + "/sdec_facturation/member/", + json=new_member_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 201 + member = response.json() + assert member["name"] == "Member 3" + assert member["mandate"] == 2023 + assert member["role"] == "com" + assert member["visible"] is True + modified_date = datetime.fromisoformat(member["modified_date"]).date() + current_date = datetime.now(tz=UTC).date() + assert modified_date == current_date + assert isinstance(member["id"], str) + + repeated_member_data = { + "name": "Member 2", + "mandate": 2023, + "role": "com", + "visible": True, + } + response = client.post( + "/sdec_facturation/member/", + json=repeated_member_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to duplicate member name + + +def test_create_member_as_lambda(client: TestClient): + """Test that a non-admin user cannot create a new member.""" + new_member_data = { + "name": "Member 4", + "mandate": 2023, + "role": "profs", + "visible": True, + } + response = client.post( + "/sdec_facturation/member/", + json=new_member_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_update_member(client: TestClient): + """Test updating an existing member.""" + update_data = { + "name": "Updated Member 1", + "role": "trez_ext", + "visible": False, + } + response = client.put( + f"/sdec_facturation/member/{member1.id}", + json=update_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + repeated_name_data = { + "name": "Member 2", + "role": "sg", + "visible": True, + } + response = client.put( + f"/sdec_facturation/member/{member1.id}", + json=repeated_name_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + + +def test_update_member_as_lambda(client: TestClient): + """Test that a non-admin user cannot update a member.""" + update_data = { + "name": "Malicious Update", + "role": "sg", + "visible": True, + } + response = client.put( + f"/sdec_facturation/member/{member1.id}", + json=update_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_delete_member(client: TestClient): + """Test deleting a member.""" + response = client.delete( + f"/sdec_facturation/member/{member2.id}", + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + # Verify deletion + get_response = client.get( + f"/sdec_facturation/member/{member2.id}", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert get_response.status_code == 404 + + +def test_delete_member_as_lambda(client: TestClient): + """Test that a non-admin user cannot delete a member.""" + response = client.delete( + f"/sdec_facturation/member/{member1.id}", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +# ---------------------------------------------------------------------------- # +# Mandate # +# ---------------------------------------------------------------------------- # +def test_get_all_mandates(client: TestClient): + """Test retrieving all mandates.""" + response = client.get( + "/sdec_facturation/mandate/", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 200 + assert len(response.json()) == 1 + + +def test_create_mandate(client: TestClient): + """Test creating a new mandate.""" + new_mandate_data = { + "year": 2024, + "name": "Mandate 2024", + } + response = client.post( + "/sdec_facturation/mandate/", + json=new_mandate_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 201 + + mandate = response.json() + assert mandate["year"] == 2024 + assert mandate["name"] == "Mandate 2024" + + repeated_mandate_data = { + "year": 2023, + "name": "Duplicate Mandate 2023", + } + response = client.post( + "/sdec_facturation/mandate/", + json=repeated_mandate_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to duplicate mandate year + + +def test_create_mandate_as_lambda(client: TestClient): + """Test that a non-admin user cannot create a new mandate.""" + new_mandate_data = { + "year": 2025, + "name": "Mandate 2025", + } + response = client.post( + "/sdec_facturation/mandate/", + json=new_mandate_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_update_mandate(client: TestClient): + """Test updating an existing mandate.""" + update_data = { + "name": "Updated Mandate 2023", + } + response = client.put( + f"/sdec_facturation/mandate/{mandate.year}", + json=update_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + +def test_update_mandate_as_lambda(client: TestClient): + """Test that a non-admin user cannot update a mandate.""" + update_data = { + "name": "Malicious Mandate Update", + } + response = client.put( + f"/sdec_facturation/mandate/{mandate.year}", + json=update_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_delete_mandate(client: TestClient): + """Test deleting a mandate.""" + response = client.delete( + f"/sdec_facturation/mandate/{mandate.year}", + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + # Verify deletion + get_response = client.get( + f"/sdec_facturation/mandate/{mandate.year}", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert get_response.status_code == 404 + + +def test_delete_mandate_as_lambda(client: TestClient): + """Test that a non-admin user cannot delete a mandate.""" + response = client.delete( + f"/sdec_facturation/mandate/{mandate.year}", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +# ---------------------------------------------------------------------------- # +# Association # +# ---------------------------------------------------------------------------- # +def test_get_all_associations(client: TestClient): + """Test retrieving all associations.""" + response = client.get( + "/sdec_facturation/association/", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 200 + assert len(response.json()) == 3 + + +def test_create_association(client: TestClient): + """Test creating a new association.""" + new_association_data = { + "name": "Association 4", + "type": "aeecl", + "structure": "club", + "visible": True, + } + response = client.post( + "/sdec_facturation/association/", + json=new_association_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 201 + + association = response.json() + assert association["name"] == "Association 4" + assert association["type"] == "aeecl" + assert association["structure"] == "club" + assert association["visible"] is True + modified_date = datetime.fromisoformat(association["modified_date"]).date() + current_date = datetime.now(tz=UTC).date() + assert modified_date == current_date + assert isinstance(association["id"], str) + + repeated_association_data = { + "name": "Association 1", + "type": "useecl", + "structure": "asso", + "visible": True, + } + response = client.post( + "/sdec_facturation/association/", + json=repeated_association_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to duplicate association name + + +def test_create_association_as_lambda(client: TestClient): + """Test that a non-admin user cannot create a new association.""" + new_association_data = { + "name": "Association 5", + "type": "useecl", + "structure": "section", + "visible": True, + } + response = client.post( + "/sdec_facturation/association/", + json=new_association_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_update_association(client: TestClient): + """Test updating an existing association.""" + update_data = { + "name": "Updated Association 1", + "structure": "section", + "visible": False, + } + response = client.put( + f"/sdec_facturation/association/{association1.id}", + json=update_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + repeated_name_data = { + "name": "Association 2", + "structure": "club", + "visible": True, + } + response = client.put( + f"/sdec_facturation/association/{association1.id}", + json=repeated_name_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to duplicate association name + + +def test_update_association_as_lambda(client: TestClient): + """Test that a non-admin user cannot update an association.""" + update_data = { + "name": "Malicious Association Update", + "structure": "asso", + "visible": True, + } + response = client.put( + f"/sdec_facturation/association/{association1.id}", + json=update_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_delete_association(client: TestClient): + """Test deleting an association.""" + response = client.delete( + f"/sdec_facturation/association/{association2.id}", + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + # Verify deletion + get_response = client.get( + f"/sdec_facturation/association/{association2.id}", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert get_response.status_code == 404 + + +def test_delete_association_as_lambda(client: TestClient): + """Test that a non-admin user cannot delete an association.""" + response = client.delete( + f"/sdec_facturation/association/{association1.id}", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +# ---------------------------------------------------------------------------- # +# Product # +# ---------------------------------------------------------------------------- # +def test_get_all_products(client: TestClient): + """Test retrieving all products.""" + response = client.get( + "/sdec_facturation/product/", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 200 + assert len(response.json()) == 4 + + +def test_create_product(client: TestClient): + """Test creating a new product.""" + new_product_data = { + "code": "P005", + "name": "Product 5", + "category": "divers", + "for_sale": True, + "individual_price": 5.0, + "association_price": 4.0, + "ae_price": 3.0, + } + response = client.post( + "/sdec_facturation/product/", + json=new_product_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 201 + product = response.json() + assert product["code"] == "P005" + assert product["name"] == "Product 5" + assert product["category"] == "divers" + assert product["for_sale"] is True + assert product["individual_price"] == 5.0 + assert product["association_price"] == 4.0 + assert product["ae_price"] == 3.0 + creation_date = datetime.fromisoformat(product["creation_date"]).date() + current_date = datetime.now(tz=UTC).date() + assert creation_date == current_date + effective_date = datetime.fromisoformat(product["effective_date"]).date() + current_date = datetime.now(tz=UTC).date() + assert effective_date == current_date + assert isinstance(product["id"], str) + + repeated_product_data = { + "code": "P001", + "name": "Duplicate Product 1", + "category": "impression", + "for_sale": True, + "individual_price": 1.0, + "association_price": 0.8, + "ae_price": 0.5, + } + response = client.post( + "/sdec_facturation/product/", + json=repeated_product_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to duplicate product code + + +def test_create_product_lack_price(client: TestClient): + """Test creating a new product without price fields.""" + lack_price_data = { + "code": "P006", + "name": "Product 6", + "category": "divers", + "for_sale": True, + } + response = client.post( + "/sdec_facturation/product/", + json=lack_price_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to missing price fields + + +def test_create_product_invalid_price(client: TestClient): + """Test creating a new product with invalid price values.""" + invalid_price_data = { + "code": "P007", + "name": "Product 7", + "category": "divers", + "for_sale": True, + "individual_price": -1.0, # Invalid negative price + "association_price": 2.0, + "ae_price": 1.0, + } + response = client.post( + "/sdec_facturation/product/", + json=invalid_price_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to invalid price value + + +def test_create_product_as_lambda(client: TestClient): + """Test that a non-admin user cannot create a new product.""" + new_product_data = { + "code": "P008", + "name": "Product 8", + "category": "divers", + "for_sale": True, + "individual_price": 8.0, + "association_price": 6.0, + "ae_price": 4.0, + } + response = client.post( + "/sdec_facturation/product/", + json=new_product_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_update_product(client: TestClient): + """Test updating an existing product.""" + update_data = { + "name": "Updated Product 1", + "category": "tshirt_flocage", + "for_sale": False, + } + response = client.put( + f"/sdec_facturation/product/{product1.id}", + json=update_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + repeated_name_data = { + "name": "Product 2", + "category": "impression", + "for_sale": True, + } + response = client.put( + f"/sdec_facturation/product/{product1.id}", + json=repeated_name_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to duplicate product name + + +def test_update_product_as_lambda(client: TestClient): + """Test that a non-admin user cannot update a product.""" + update_data = { + "name": "Malicious Product Update", + "category": "divers", + "for_sale": True, + } + response = client.put( + f"/sdec_facturation/product/{product1.id}", + json=update_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_update_price_not_today(client: TestClient): + """Test updating the price of an existing product.""" + update_price_data = { + "individual_price": 2.15, + "association_price": 1.51, + "ae_price": 1.12, + } + response = client.put( + f"/sdec_facturation/product/price/{product2.id}", + json=update_price_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + +def test_update_price_today(client: TestClient): + """Test updating the price of an existing product with today's date.""" + update_price_data = { + "individual_price": 2.50, + "association_price": 1.75, + "ae_price": 1.25, + } + response = client.put( + f"/sdec_facturation/product/price/{product4.id}", + json=update_price_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + +def test_update_price_invalid(client: TestClient): + """Test updating a product's price with invalid values.""" + invalid_price_data = { + "individual_price": -3.0, # Invalid negative price + "association_price": 2.0, + "ae_price": 1.0, + } + response = client.put( + f"/sdec_facturation/product/price/{product2.id}", + json=invalid_price_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to invalid price value + + +def test_update_price_as_lambda(client: TestClient): + """Test that a non-admin user cannot update a product's price.""" + update_price_data = { + "individual_price": 3.0, + "association_price": 2.0, + "ae_price": 1.0, + } + response = client.put( + f"/sdec_facturation/product/price/{product2.id}", + json=update_price_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_delete_product(client: TestClient): + """Test deleting a product.""" + response = client.delete( + f"/sdec_facturation/product/{product3.id}", + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + +def test_delete_product_as_lambda(client: TestClient): + """Test that a non-admin user cannot delete a product.""" + response = client.delete( + f"/sdec_facturation/product/{product1.id}", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +# ---------------------------------------------------------------------------- # +# Order # +# ---------------------------------------------------------------------------- # + + +def test_get_all_orders(client: TestClient): + """Test retrieving all orders.""" + response = client.get( + "/sdec_facturation/order/", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 200 + assert len(response.json()) == 2 + + +def test_create_order(client: TestClient): + """Test creating a new order.""" + new_order_data = { + "association_id": str(association3.id), + "member_id": str(member1.id), + "order": "Product 2:4,Product 4:2", + "valid": True, + } + response = client.post( + "/sdec_facturation/order/", + json=new_order_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 201 + + order = response.json() + assert order["association_id"] == str(association3.id) + assert order["member_id"] == str(member1.id) + assert order["order"] == "Product 2:4,Product 4:2" + assert order["valid"] is True + creation_date = datetime.fromisoformat(order["creation_date"]).date() + current_date = datetime.now(tz=UTC).date() + assert creation_date == current_date + assert isinstance(order["id"], str) + + +def test_create_order_as_lambda(client: TestClient): + """Test that a non-admin user cannot create a new order.""" + new_order_data = { + "association_id": str(association3.id), + "member_id": str(member1.id), + "order": "Product 2:4,Product 4:2", + "valid": True, + } + response = client.post( + "/sdec_facturation/order/", + json=new_order_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_update_order(client: TestClient): + """Test updating an existing order.""" + update_data = { + "order": "Product 1:5,Product 3:3", + } + response = client.put( + f"/sdec_facturation/order/{order1.id}", + json=update_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + +def test_update_order_as_lambda(client: TestClient): + """Test that a non-admin user cannot update an order.""" + update_data = { + "order": "Product 2:6,Product 4:4", + } + response = client.put( + f"/sdec_facturation/order/{order1.id}", + json=update_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_delete_order(client: TestClient): + """Test deleting an order.""" + response = client.delete( + f"/sdec_facturation/order/{order2.id}", + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + +# ---------------------------------------------------------------------------- # +# Facture Association # +# ---------------------------------------------------------------------------- # + + +def test_get_all_facture_associations(client: TestClient): + """Test retrieving all association invoices.""" + response = client.get( + "/sdec_facturation/facture_association/", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 200 + assert len(response.json()) == 1 + + +def test_create_facture_association(client: TestClient): + """Test creating a new association invoice.""" + new_facture_data = { + "facture_number": "FA2023002", + "member_id": str(member2.id), + "association_id": str(association2.id), + "start_date": "2023-01-01", + "end_date": "2023-12-31", + "price": 200.0, + "paid": False, + "valid": True, + "payment_date": None, + } + response = client.post( + "/sdec_facturation/facture_association/", + json=new_facture_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 201 + + facture_association = response.json() + assert facture_association["facture_number"] == "FA2023002" + assert facture_association["member_id"] == str(member2.id) + assert facture_association["association_id"] == str(association2.id) + assert facture_association["start_date"] == "2023-01-01T00:00:00+00:00" + assert facture_association["end_date"] == "2023-12-31T00:00:00+00:00" + assert facture_association["price"] == 200.0 + assert facture_association["paid"] is False + assert facture_association["valid"] is True + facture_date = datetime.fromisoformat(facture_association["facture_date"]).date() + assert facture_date == datetime.now(tz=UTC).date() + assert isinstance(facture_association["id"], str) + + +def test_create_facture_association_invalid_price(client: TestClient): + """Test creating a new association invoice with invalid price.""" + new_facture_data = { + "facture_number": "FA2023003", + "member_id": str(member2.id), + "association_id": str(association2.id), + "start_date": "2023-01-01", + "end_date": "2023-12-31", + "price": -100.0, # Invalid negative price + "paid": False, + "valid": True, + "payment_date": None, + } + response = client.post( + "/sdec_facturation/facture_association/", + json=new_facture_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to invalid price value + + +def test_create_facture_association_invalid_dates(client: TestClient): + """Test creating a new association invoice with invalid dates.""" + new_facture_data = { + "facture_number": "FA2023004", + "member_id": str(member2.id), + "association_id": str(association2.id), + "start_date": "2023-12-31", + "end_date": "2023-01-01", # End date before start date + "price": 150.0, + "paid": False, + "valid": True, + "payment_date": None, + } + response = client.post( + "/sdec_facturation/facture_association/", + json=new_facture_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to invalid date range + + +def test_create_facture_association_as_lambda(client: TestClient): + """Test that a non-admin user cannot create a new association invoice.""" + new_facture_data = { + "facture_number": "FA2023005", + "member_id": str(member2.id), + "association_id": str(association2.id), + "start_date": "2023-01-01", + "end_date": "2023-12-31", + "price": 180.0, + "paid": False, + "valid": True, + "payment_date": None, + } + response = client.post( + "/sdec_facturation/facture_association/", + json=new_facture_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_update_facture_association(client: TestClient): + """Test updating an existing association invoice.""" + update_data = { + "paid": True, + } + response = client.put( + f"/sdec_facturation/facture_association/{factureAssociation1.id}", + json=update_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + +def test_update_facture_association_as_lambda(client: TestClient): + """Test that a non-admin user cannot update an association invoice.""" + update_data = { + "paid": True, + } + response = client.put( + f"/sdec_facturation/facture_association/{factureAssociation1.id}", + json=update_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_delete_facture_association(client: TestClient): + """Test deleting an association invoice.""" + response = client.delete( + f"/sdec_facturation/facture_association/{factureAssociation1.id}", + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + +def test_delete_facture_association_as_lambda(client: TestClient): + """Test that a non-admin user cannot delete an association invoice.""" + response = client.delete( + f"/sdec_facturation/facture_association/{factureAssociation1.id}", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +# ---------------------------------------------------------------------------- # +# Facture Individual # +# ---------------------------------------------------------------------------- # +def test_get_all_facture_individuals(client: TestClient): + """Test retrieving all individual invoices.""" + response = client.get( + "/sdec_facturation/facture_individual/", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 200 + assert len(response.json()) == 1 + + +def test_create_facture_individual(client: TestClient): + """Test creating a new individual invoice.""" + new_facture_data = { + "facture_number": "FI2023002", + "member_id": str(member1.id), + "individual_order": "Product 2:3,Product 3:2", + "individual_category": "profs", + "price": 10.5, + "firstname": "Alice", + "lastname": "Smith", + "adresse": "456 Elm St", + "postal_code": "75001", + "city": "Paris", + "country": "France", + "paid": False, + "valid": True, + "payment_date": None, + } + response = client.post( + "/sdec_facturation/facture_individual/", + json=new_facture_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 201 + + facture_individual = response.json() + assert facture_individual["facture_number"] == "FI2023002" + assert facture_individual["member_id"] == str(member1.id) + assert facture_individual["individual_order"] == "Product 2:3,Product 3:2" + assert facture_individual["individual_category"] == "profs" + assert facture_individual["price"] == 10.5 + assert facture_individual["firstname"] == "Alice" + assert facture_individual["lastname"] == "Smith" + assert facture_individual["adresse"] == "456 Elm St" + assert facture_individual["postal_code"] == "75001" + assert facture_individual["city"] == "Paris" + assert facture_individual["country"] == "France" + assert facture_individual["paid"] is False + assert facture_individual["valid"] is True + facture_date = datetime.fromisoformat(facture_individual["facture_date"]).date() + assert facture_date == datetime.now(tz=UTC).date() + assert isinstance(facture_individual["id"], str) + + +def test_create_facture_individual_invalid_price(client: TestClient): + """Test creating a new individual invoice with invalid price.""" + new_facture_data = { + "facture_number": "FI2023003", + "member_id": str(member1.id), + "individual_order": "Product 2:3,Product 3:2", + "individual_category": "profs", + "price": -5.0, # Invalid negative price + "firstname": "Bob", + "lastname": "Brown", + "adresse": "789 Oak St", + "postal_code": "13001", + "city": "Marseille", + "country": "France", + "paid": False, + "valid": True, + "payment_date": None, + } + response = client.post( + "/sdec_facturation/facture_individual/", + json=new_facture_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to invalid price value + + +def test_create_facture_individual_invalid_name(client: TestClient): + """Test creating a new individual invoice with invalid name.""" + new_facture_data = { + "facture_number": "FI2023004", + "member_id": str(member1.id), + "individual_order": "Product 2:3,Product 3:2", + "individual_category": "profs", + "price": 15.0, + "firstname": "", # Invalid empty firstname + "lastname": "Green", + "adresse": "101 Pine St", + "postal_code": "31000", + "city": "Toulouse", + "country": "France", + "paid": False, + "valid": True, + "payment_date": None, + } + response = client.post( + "/sdec_facturation/facture_individual/", + json=new_facture_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 400 # Bad Request due to invalid firstname + + +def test_create_facture_individual_as_lambda(client: TestClient): + """Test that a non-admin user cannot create a new individual invoice.""" + new_facture_data = { + "facture_number": "FI2023005", + "member_id": str(member1.id), + "individual_order": "Product 2:3,Product 3:2", + "individual_category": "profs", + "price": 12.0, + "firstname": "Charlie", + "lastname": "Davis", + "adresse": "202 Birch St", + "postal_code": "44000", + "city": "Nantes", + "country": "France", + "paid": False, + } + response = client.post( + "/sdec_facturation/facture_individual/", + json=new_facture_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_update_facture_individual(client: TestClient): + """Test updating an existing individual invoice.""" + update_data = { + "paid": True, + } + response = client.put( + f"/sdec_facturation/facture_individual/{FactureIndividual1.id}", + json=update_data, + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + +def test_update_facture_individual_as_lambda(client: TestClient): + """Test that a non-admin user cannot update an individual invoice.""" + update_data = { + "paid": True, + } + response = client.put( + f"/sdec_facturation/facture_individual/{FactureIndividual1.id}", + json=update_data, + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403 + + +def test_delete_facture_individual(client: TestClient): + """Test deleting an individual invoice.""" + response = client.delete( + f"/sdec_facturation/facture_individual/{FactureIndividual1.id}", + headers={"Authorization": f"Bearer {token_admin}"}, + ) + assert response.status_code == 204 + + +def test_delete_facture_individual_as_lambda(client: TestClient): + """Test that a non-admin user cannot delete an individual invoice.""" + response = client.delete( + f"/sdec_facturation/facture_individual/{FactureIndividual1.id}", + headers={"Authorization": f"Bearer {token_user}"}, + ) + assert response.status_code == 403