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..720ea5bf20 --- /dev/null +++ b/app/modules/sdec_facturation/cruds_sdec_facturation.py @@ -0,0 +1,1160 @@ +"""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 + ) + + +# ---------------------------------------------------------------------------- # +# Mandate # +# ---------------------------------------------------------------------------- # +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 + ) + + +# ---------------------------------------------------------------------------- # +# Association # +# ---------------------------------------------------------------------------- # + + +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, + modified_date=association_db.modified_date, + visible=association_db.visible, + ) + + +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.ProductAndPriceBase, + db: AsyncSession, +) -> 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, + category=product.category, + for_sale=product.for_sale, + creation_date=datetime.now(tz=UTC), + ) + db.add(product_db) + await db.flush() + + price_db = models_sdec_facturation.ProductPrice( + id=uuid.uuid4(), + 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(price_db) + await db.flush() + return schemas_sdec_facturation.ProductAndPriceComplete( + id=product_db.id, + code=product_db.code, + name=product_db.name, + 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 update_product( + product_id: uuid.UUID, + product_edit: schemas_sdec_facturation.ProductUpdate, + db: AsyncSession, +): + """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.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_id: uuid.UUID, + db: AsyncSession, +): + """Delete a product from the database""" + + await db.execute( + update(models_sdec_facturation.Product) + .where(models_sdec_facturation.Product.id == product_id) + .values( + for_sale=False, + ), + ) + await db.flush() + + +async def get_all_products_and_price( + db: AsyncSession, +) -> Sequence[schemas_sdec_facturation.ProductAndPriceComplete]: + """Get all products from the database""" + + 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.ProductAndPriceComplete( + 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, + effective_date=product_price.effective_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, + 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, + 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, + category=result.category, + for_sale=result.for_sale, + creation_date=result.creation_date, + ) + if result + else None + ) + + +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 # +# ---------------------------------------------------------------------------- # + + +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(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]: + """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, + 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, + 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, + 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, + paid=facture_association_db.paid, + payment_date=facture_association_db.payment_date, + ) + + +async def update_facture_association( + facture_association_id: uuid.UUID, + facture_association_edit: schemas_sdec_facturation.FactureUpdate, + 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( + paid=facture_association_edit.paid, + 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]: + """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, + 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, + 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, + 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, + paid=result.paid, + payment_date=result.payment_date, + ) + if result + else None + ) + + +# ---------------------------------------------------------------------------- # +# Facture Individual # +# ---------------------------------------------------------------------------- # +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_individual_db = models_sdec_facturation.FactureIndividual( + id=uuid.uuid4(), + 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_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_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.FactureUpdate, + 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_individual_id) + .values( + paid=facture_individual_edit.paid, + payment_date=current_date, + ), + ) + await db.flush() + + +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_individual = result.scalars().all() + return [ + schemas_sdec_facturation.FactureIndividualComplete( + 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_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""" + result = ( + ( + await db.execute( + select(models_sdec_facturation.FactureIndividual).where( + models_sdec_facturation.FactureIndividual.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, + ), + ) + ) + .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..99ba652fe1 --- /dev/null +++ b/app/modules/sdec_facturation/endpoints_sdec_facturation.py @@ -0,0 +1,960 @@ +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 + +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/member/", + 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/member/", + 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", + ) + + 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/member/{member_id}", + status_code=204, +) +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", + ) + + 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, + db, + ) + + +@module.router.delete( + "/sdec_facturation/member/{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/mandate/", + 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/mandate/", + 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/mandate/{mandate_year}", + status_code=204, +) +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/mandate/{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=204, +) +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.ProductAndPriceComplete], + status_code=200, +) +async def get_all_products( + db: AsyncSession = Depends(get_db), + _=Depends(is_user()), +) -> Sequence[schemas_sdec_facturation.ProductAndPriceComplete]: + """Get all product items from the database""" + return await cruds_sdec_facturation.get_all_products_and_price(db) + + +@module.router.post( + "/sdec_facturation/product/", + response_model=schemas_sdec_facturation.ProductComplete, + status_code=201, +) +async def create_product( + product: schemas_sdec_facturation.ProductAndPriceBase, + db: AsyncSession = Depends(get_db), + _=Depends(is_user_in([GroupType.sdec_facturation_admin])), +) -> schemas_sdec_facturation.ProductAndPriceComplete: + """ + 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=204, +) +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** + """ + + 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", + ) + + 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.id, + product_edit, + db, + ) + + +@module.router.patch( + "/sdec_facturation/product/price/{product_id}", + status_code=204, +) +async def update_price( + product_id: uuid.UUID, + price_edit: schemas_sdec_facturation.ProductPriceUpdate, + 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** + """ + + 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, + ) + if product_db is None: + raise HTTPException( + status_code=404, + detail="Product item not found", + ) + + 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, + ) + + +@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_id, db) + + +# ---------------------------------------------------------------------------- # +# Order # +# ---------------------------------------------------------------------------- # + + +@module.router.get( + "/sdec_facturation/order/", + 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) + + +@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), + _=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) + + +@module.router.patch( + "/sdec_facturation/order/{order_id}", + status_code=204, +) +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=204, +) +async def update_facture_association( + facture_association_id: uuid.UUID, + facture_association_edit: schemas_sdec_facturation.FactureUpdate, + 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 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 facture_individual.country.strip() == "": + raise HTTPException( + status_code=400, + detail="Country 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=204, +) +async def update_facture_individual( + facture_individual_id: uuid.UUID, + facture_individual_edit: schemas_sdec_facturation.FactureUpdate, + 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 new file mode 100644 index 0000000000..188fbf9dcb --- /dev/null +++ b/app/modules/sdec_facturation/models_sdec_facturation.py @@ -0,0 +1,111 @@ +"""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, + ProductCategoryType, + 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] + 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] + effective_date: Mapped[date] + + +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"), + ) + start_date: Mapped[date] + end_date: Mapped[date] + price: Mapped[float] + facture_date: Mapped[date] + paid: Mapped[bool] = mapped_column(default=False) + valid: Mapped[bool] = mapped_column(default=True) + 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] + 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 new file mode 100644 index 0000000000..516cb99770 --- /dev/null +++ b/app/modules/sdec_facturation/schemas_sdec_facturation.py @@ -0,0 +1,154 @@ +"""Schemas file for endpoint /sdec_facturation""" + +from datetime import date +from uuid import UUID + +from pydantic import BaseModel + +from app.modules.sdec_facturation.types_sdec_facturation import ( + AssociationStructureType, + AssociationType, + IndividualCategoryType, + ProductCategoryType, + RoleType, +) + + +class MemberBase(BaseModel): + name: str + mandate: int + role: RoleType + visible: bool = True + + +class MemberComplete(MemberBase): + id: 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 + modified_date: date + + +class ProductBase(BaseModel): + code: str + name: str + category: ProductCategoryType + for_sale: bool = True + + +class ProductComplete(ProductBase): + id: UUID + creation_date: date + + +class ProductUpdate(BaseModel): + name: str | None = None + category: ProductCategoryType | None = None + for_sale: bool | None = None + + +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 + member_id: UUID + order: str + valid: bool = True + + +class OrderComplete(OrderBase): + id: UUID + creation_date: date + + +class OrderUpdate(BaseModel): + order: str | None = None + + +class FactureAssociationBase(BaseModel): + facture_number: str + member_id: UUID + association_id: UUID + start_date: date + end_date: date + price: float + valid: bool = True + paid: bool = False + payment_date: date | None = None + + +class FactureAssociationComplete(FactureAssociationBase): + facture_date: date + id: UUID + + +class FactureIndividualBase(BaseModel): + facture_number: str + member_id: UUID + individual_order: str + individual_category: IndividualCategoryType + price: float + firstname: str + lastname: str + adresse: str + postal_code: str + city: str + country: str + valid: bool = True + paid: bool = False + payment_date: date | None = None + + +class FactureIndividualComplete(FactureIndividualBase): + facture_date: date + id: UUID + + +class FactureUpdate(BaseModel): + paid: bool | None = None 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..8575bdc751 --- /dev/null +++ b/app/modules/sdec_facturation/types_sdec_facturation.py @@ -0,0 +1,66 @@ +from enum import Enum + + +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}" + + +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}" 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