diff --git a/app_new/__init__.py b/app_new/__init__.py index f9449f2..dac15e1 100644 --- a/app_new/__init__.py +++ b/app_new/__init__.py @@ -145,6 +145,9 @@ def create_app(config_name='default'): from .parts.routes import parts_bp app.register_blueprint(parts_bp, url_prefix='/parts') + from .c4 import c4_bp + app.register_blueprint(c4_bp, url_prefix='/c4') + # Matériel et produits d'entretien (stock, FEFO, distributions, prévisions) from .cleaning import cleaning_bp app.register_blueprint(cleaning_bp) diff --git a/app_new/c4/__init__.py b/app_new/c4/__init__.py new file mode 100644 index 0000000..86f6b94 --- /dev/null +++ b/app_new/c4/__init__.py @@ -0,0 +1,3 @@ +from .routes import c4_bp + +__all__ = ["c4_bp"] diff --git a/app_new/c4/routes.py b/app_new/c4/routes.py new file mode 100644 index 0000000..e3fbb99 --- /dev/null +++ b/app_new/c4/routes.py @@ -0,0 +1,153 @@ +"""HTTP C4 : contrats photocopieurs et consommables partagés.""" +from datetime import date + +from flask import Blueprint, flash, redirect, render_template, request, url_for +from flask_login import current_user, login_required + +from ..core.authorization import permission_required +from ..core.models import ( + Company, Consumable, ConsumableOrder, ConsumableUsage, ContractEquipmentAssignment, + Equipment, EquipmentConsumable, PhotocopierContract, PhotocopierContractPeriod, +) +from ..core.services.c4_service import ( + C4DomainError, assign_equipment_to_contract, calculate_contract_usage, + create_consumable_order, create_contract, evaluate_quota_alert, + estimate_consumable_lifetime, receive_consumable_order, record_consumable_issue, +) +from ..extensions import db + +c4_bp = Blueprint("c4", __name__, template_folder="templates") + + +def _parse_date(value): + return date.fromisoformat(value) if value else None + + +@c4_bp.get("/contracts") +@login_required +@permission_required("contract.view") +def contracts(): + rows = PhotocopierContract.query.order_by(PhotocopierContract.start_date.desc()).all() + return render_template("c4/contracts.html", contracts=rows) + + +@c4_bp.route("/contracts/new", methods=["GET", "POST"]) +@login_required +@permission_required("contract.manage") +def new_contract(): + companies = Company.query.order_by(Company.name).all() + if request.method == "POST": + try: + contract = create_contract( + supplier_id=request.form.get("supplier_id", type=int), name=request.form.get("name", "").strip(), + reference=request.form.get("reference") or None, start_date=_parse_date(request.form.get("start_date")), + end_date=_parse_date(request.form.get("end_date")), anniversary_month=request.form.get("anniversary_month", type=int), + anniversary_day=request.form.get("anniversary_day", type=int), + quota_start_date=_parse_date(request.form.get("quota_start_date")), notes=request.form.get("notes"), + ) + flash("Contrat photocopieur créé.", "success") + return redirect(url_for("c4.contract_detail", contract_id=contract.id)) + except (ValueError, TypeError, C4DomainError) as exc: + db.session.rollback() + flash(str(exc), "danger") + return render_template("c4/contract_form.html", companies=companies) + + +@c4_bp.get("/contracts/") +@login_required +@permission_required("contract.view") +def contract_detail(contract_id): + contract = PhotocopierContract.query.get_or_404(contract_id) + periods = contract.periods + data = {period.id: calculate_contract_usage(period) for period in periods} + return render_template("c4/contract_detail.html", contract=contract, periods=periods, usage=data) + + +@c4_bp.post("/contracts//equipment") +@login_required +@permission_required("contract.manage") +def contract_equipment(contract_id): + contract = PhotocopierContract.query.get_or_404(contract_id) + equipment = Equipment.query.get_or_404(request.form.get("equipment_id", type=int)) + try: + assign_equipment_to_contract(contract=contract, equipment=equipment, entry_date=_parse_date(request.form.get("entry_date")), exit_date=_parse_date(request.form.get("exit_date"))) + flash("Machine rattachée au contrat.", "success") + except (ValueError, C4DomainError) as exc: + db.session.rollback() + flash(str(exc), "danger") + return redirect(url_for("c4.contract_detail", contract_id=contract_id)) + + +@c4_bp.post("/contracts//close") +@login_required +@permission_required("contract.manage") +def close_contract(contract_id): + contract = PhotocopierContract.query.get_or_404(contract_id) + contract.status = "CLOSED" + contract.closed_at = _parse_date(request.form.get("closed_at")) or date.today() + contract.end_date = min(contract.end_date, contract.closed_at) + db.session.commit() + flash("Contrat clos ; son historique est conservé.", "success") + return redirect(url_for("c4.contract_detail", contract_id=contract_id)) + + +@c4_bp.get("/consumables") +@login_required +@permission_required("stock.view") +def consumables(): + rows = Consumable.query.filter_by(is_active=True).order_by(Consumable.name).all() + return render_template("c4/consumables.html", consumables=rows) + + +@c4_bp.get("/consumables/") +@login_required +@permission_required("stock.view") +def consumable_detail(consumable_id): + consumable = Consumable.query.get_or_404(consumable_id) + usages = consumable.usages.order_by(ConsumableUsage.used_at.desc()).limit(100).all() + orders = ConsumableOrder.query.filter_by(consumable_id=consumable.id).order_by(ConsumableOrder.id.desc()).all() + compatible = EquipmentConsumable.query.filter_by(consumable_id=consumable.id).all() + return render_template("c4/consumable_detail.html", consumable=consumable, usages=usages, orders=orders, compatible=compatible) + + +@c4_bp.post("/consumables//orders") +@login_required +@permission_required("stock.manage") +def order_consumable(consumable_id): + consumable = Consumable.query.get_or_404(consumable_id) + try: + create_consumable_order(consumable=consumable, quantity=request.form.get("quantity", type=int), supplier_id=request.form.get("supplier_id", type=int), user_id=current_user.id) + flash("Commande créée.", "success") + except (ValueError, C4DomainError) as exc: + db.session.rollback() + flash(str(exc), "danger") + return redirect(url_for("c4.consumable_detail", consumable_id=consumable_id)) + + +@c4_bp.post("/orders//receive") +@login_required +@permission_required("stock.receive") +def receive_order(order_id): + order = ConsumableOrder.query.get_or_404(order_id) + try: + receive_consumable_order(order=order, quantity=request.form.get("quantity", type=int), user_id=current_user.id, comment=request.form.get("comment")) + flash("Réception enregistrée ; le stock a été augmenté.", "success") + except (ValueError, C4DomainError) as exc: + db.session.rollback() + flash(str(exc), "danger") + return redirect(url_for("c4.consumable_detail", consumable_id=order.consumable_id)) + + +@c4_bp.post("/consumables//issue") +@login_required +@permission_required("stock.issue") +def issue_consumable(consumable_id): + consumable = Consumable.query.get_or_404(consumable_id) + equipment = Equipment.query.get_or_404(request.form.get("equipment_id", type=int)) + try: + record_consumable_issue(consumable=consumable, equipment=equipment, quantity=request.form.get("quantity", type=int, default=1), user_id=current_user.id, notes=request.form.get("notes")) + flash("Sortie de stock enregistrée.", "success") + except (ValueError, C4DomainError) as exc: + db.session.rollback() + flash(str(exc), "danger") + return redirect(url_for("c4.consumable_detail", consumable_id=consumable_id)) diff --git a/app_new/c4/templates/c4/consumable_detail.html b/app_new/c4/templates/c4/consumable_detail.html new file mode 100644 index 0000000..e675dba --- /dev/null +++ b/app_new/c4/templates/c4/consumable_detail.html @@ -0,0 +1,3 @@ +{% extends "base.html" %} +{% block title %}{{ consumable.name }} — GMAO{% endblock %} +{% block content %}

{{ consumable.name }}

Référence : {{ consumable.reference or '—' }} · Type : {{ consumable.consumable_type }} · Stock : {{ consumable.quantity }} · À commander : {{ consumable.quantity_to_order }}

Compatible avec {{ compatible|length }} équipement(s)

    {% for link in compatible %}
  • {{ link.equipment.name }}
    • {% else %}
  • Aucune machine.
    • {% endfor %}

Commandes

{% for order in orders %}{% else %}{% endfor %}
QuantitéReçueÉtatAction
{{ order.quantity_ordered }}{{ order.quantity_received }}{{ order.status }}{% if order.remaining_quantity %}
{% endif %}
Aucune commande.

Historique des sorties

    {% for usage in usages %}
  • {{ usage.used_at }} — {{ usage.quantity_used }} — {{ usage.equipment.name if usage.equipment else '—' }}
    • {% else %}
  • Aucune sortie.
    • {% endfor %}
{% endblock %} diff --git a/app_new/c4/templates/c4/consumables.html b/app_new/c4/templates/c4/consumables.html new file mode 100644 index 0000000..ce2fc73 --- /dev/null +++ b/app_new/c4/templates/c4/consumables.html @@ -0,0 +1,3 @@ +{% extends "base.html" %} +{% block title %}Consommables imprimantes — GMAO{% endblock %} +{% block content %}

Consommables partagés

{% for item in consumables %}{% else %}{% endfor %}
RéférenceTypeStockSeuilCibleÀ commander
{{ item.name }}{{ item.reference or '' }}{{ item.consumable_type }}{{ item.quantity }}{{ item.min_quantity }}{{ item.target_quantity if item.target_quantity is not none else '—' }}{{ item.quantity_to_order }}
Aucun consommable.
{% endblock %} diff --git a/app_new/c4/templates/c4/contract_detail.html b/app_new/c4/templates/c4/contract_detail.html new file mode 100644 index 0000000..b1989ab --- /dev/null +++ b/app_new/c4/templates/c4/contract_detail.html @@ -0,0 +1,3 @@ +{% extends "base.html" %} +{% block title %}{{ contract.name }} — GMAO{% endblock %} +{% block content %}

{{ contract.name }}

{{ contract.status }}

{{ contract.supplier.name }} · {{ contract.start_date }} → {{ contract.end_date }} · anniversaire {{ '%02d/%02d'|format(contract.anniversary_day, contract.anniversary_month) }}

{% for period in periods %}{% set u=usage[period.id] %}
Période {{ period.start_date }} → {{ period.end_date }}
N&B
{{ '%.0f'|format(u.bw) }}{% if period.quota_bw is not none %} / {{ period.quota_bw }}{% else %} — sans quota{% endif %}
Couleur
{{ '%.0f'|format(u.color) }}{% if period.quota_color is not none %} / {{ period.quota_color }}{% else %} — sans quota{% endif %}
Machines prises en compte : {{ u.machines }} · qualité : {{ u.quality }}
{% endfor %}

Machines rattachées

    {% for a in contract.assignments %}
  • {{ a.equipment.name }} — {{ a.entry_date }}{% if a.exit_date %} → {{ a.exit_date }}{% endif %}
    • {% else %}
  • Aucune machine.
    • {% endfor %}
{% endblock %} diff --git a/app_new/c4/templates/c4/contract_form.html b/app_new/c4/templates/c4/contract_form.html new file mode 100644 index 0000000..ed4d893 --- /dev/null +++ b/app_new/c4/templates/c4/contract_form.html @@ -0,0 +1,3 @@ +{% extends "base.html" %} +{% block title %}Nouveau contrat — GMAO{% endblock %} +{% block content %}

Nouveau contrat photocopieur

{% endblock %} diff --git a/app_new/c4/templates/c4/contracts.html b/app_new/c4/templates/c4/contracts.html new file mode 100644 index 0000000..75815b4 --- /dev/null +++ b/app_new/c4/templates/c4/contracts.html @@ -0,0 +1,10 @@ +{% extends "base.html" %} +{% block title %}Contrats photocopieurs — GMAO{% endblock %} +{% block content %} +
+

Contrats photocopieurs

Nouveau contrat
+
+ {% for contract in contracts %}{% else %}{% endfor %} +
ContratFournisseurDuréeAnniversaireÉtat
{{ contract.name }}{% if contract.reference %}{{ contract.reference }}{% endif %}{{ contract.supplier.name }}{{ contract.start_date }} → {{ contract.end_date }}{{ '%02d/%02d'|format(contract.anniversary_day, contract.anniversary_month) }}{{ contract.status }}
Aucun contrat.
+
+{% endblock %} diff --git a/app_new/core/authorization.py b/app_new/core/authorization.py index 6f97196..585c6fa 100644 --- a/app_new/core/authorization.py +++ b/app_new/core/authorization.py @@ -103,7 +103,7 @@ PATRIMOINE_BLUEPRINTS = { } PLANNING_BLUEPRINTS = {"planning", "scheduler", "interventions_planning"} STOCK_BLUEPRINTS = {"parts", "meters", "cleaning"} -CONTRACT_BLUEPRINTS = {"companies", "contracts", "services"} +CONTRACT_BLUEPRINTS = {"companies", "contracts", "services", "c4"} PREVENTION_BLUEPRINTS = {"trainings", "constraints", "prevention"} # Catalogue réellement utilisé par les routes et les fonctions métier. diff --git a/app_new/core/models/__init__.py b/app_new/core/models/__init__.py index 5b01445..8e355c3 100644 --- a/app_new/core/models/__init__.py +++ b/app_new/core/models/__init__.py @@ -19,6 +19,7 @@ from .planning import ( PreventiveTask, PreventiveTaskConsumable, ScheduledTask, TaskExecutionSegment, TechnicianAvailability, AdminTask, ZoneAccessRule ) +from .c4 import PhotocopierContract, PhotocopierContractPeriod, ContractEquipmentAssignment, ConsumableOrder, ConsumableReceipt from .settings import AppSettings from .audit import AuditLog, TemplateAuditMark from .prevention import PreventionWorkLog, StaffAuthorization, RiskAssessment, PreventionAction, SafetyRegisterEntry @@ -53,4 +54,5 @@ __all__ = [ 'RequestProfile', 'RequestProfileItem', 'StockInventory', 'StockInventoryLine', 'CleaningForecastConfig', 'Role', 'Permission', 'UserRole', 'RolePermission', 'UserPermission', + 'PhotocopierContract', 'PhotocopierContractPeriod', 'ContractEquipmentAssignment', 'ConsumableOrder', 'ConsumableReceipt', ] diff --git a/app_new/core/models/c4.py b/app_new/core/models/c4.py new file mode 100644 index 0000000..39debc8 --- /dev/null +++ b/app_new/core/models/c4.py @@ -0,0 +1,110 @@ +"""Modèles C4 : contrats photocopieurs et stock de consommables partagé.""" +from datetime import datetime, timezone + +from ...extensions import db + + +class PhotocopierContract(db.Model): + __tablename__ = "photocopier_contracts" + + id = db.Column(db.Integer, primary_key=True) + supplier_id = db.Column(db.Integer, db.ForeignKey("companies.id"), nullable=False, index=True) + name = db.Column(db.String(200), nullable=False) + reference = db.Column(db.String(120), nullable=True) + start_date = db.Column(db.Date, nullable=False) + end_date = db.Column(db.Date, nullable=False) + quota_start_date = db.Column(db.Date, nullable=True) + anniversary_month = db.Column(db.Integer, nullable=False) + anniversary_day = db.Column(db.Integer, nullable=False) + status = db.Column(db.String(20), nullable=False, default="ACTIVE", index=True) + notes = db.Column(db.Text, nullable=True) + created_at = db.Column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) + closed_at = db.Column(db.Date, nullable=True) + + supplier = db.relationship("Company") + periods = db.relationship("PhotocopierContractPeriod", back_populates="contract", cascade="all, delete-orphan", order_by="PhotocopierContractPeriod.start_date") + assignments = db.relationship("ContractEquipmentAssignment", back_populates="contract", cascade="all, delete-orphan") + + +class PhotocopierContractPeriod(db.Model): + __tablename__ = "photocopier_contract_periods" + __table_args__ = ( + db.UniqueConstraint("contract_id", "start_date", name="uq_c4_contract_period_start"), + db.CheckConstraint("end_date >= start_date", name="ck_c4_period_dates"), + db.CheckConstraint("(quota_bw IS NULL OR quota_bw >= 0) AND (quota_color IS NULL OR quota_color >= 0)", name="ck_c4_period_quotas"), + ) + + id = db.Column(db.Integer, primary_key=True) + contract_id = db.Column(db.Integer, db.ForeignKey("photocopier_contracts.id", ondelete="CASCADE"), nullable=False, index=True) + start_date = db.Column(db.Date, nullable=False, index=True) + end_date = db.Column(db.Date, nullable=False, index=True) + quota_bw = db.Column(db.Integer, nullable=True) + quota_color = db.Column(db.Integer, nullable=True) + bw_included_unit_price = db.Column(db.Numeric(12, 6), nullable=True) + bw_overage_unit_price = db.Column(db.Numeric(12, 6), nullable=True) + color_included_unit_price = db.Column(db.Numeric(12, 6), nullable=True) + color_overage_unit_price = db.Column(db.Numeric(12, 6), nullable=True) + status = db.Column(db.String(20), nullable=False, default="OPEN") + + contract = db.relationship("PhotocopierContract", back_populates="periods") + + +class ContractEquipmentAssignment(db.Model): + __tablename__ = "contract_equipment_assignments" + __table_args__ = ( + db.UniqueConstraint("contract_id", "equipment_id", "entry_date", name="uq_c4_contract_equipment_entry"), + db.CheckConstraint("exit_date IS NULL OR exit_date >= entry_date", name="ck_c4_assignment_dates"), + ) + + id = db.Column(db.Integer, primary_key=True) + contract_id = db.Column(db.Integer, db.ForeignKey("photocopier_contracts.id", ondelete="CASCADE"), nullable=False, index=True) + equipment_id = db.Column(db.Integer, db.ForeignKey("equipments.id", ondelete="CASCADE"), nullable=False, index=True) + entry_date = db.Column(db.Date, nullable=False) + exit_date = db.Column(db.Date, nullable=True) + replacement_reason = db.Column(db.String(40), nullable=True) + replacement_comment = db.Column(db.Text, nullable=True) + + contract = db.relationship("PhotocopierContract", back_populates="assignments") + equipment = db.relationship("Equipment") + + +class ConsumableOrder(db.Model): + __tablename__ = "consumable_orders" + __table_args__ = ( + db.CheckConstraint("quantity_ordered > 0", name="ck_c4_order_quantity"), + db.CheckConstraint("quantity_received >= 0 AND quantity_received <= quantity_ordered", name="ck_c4_order_received"), + ) + + id = db.Column(db.Integer, primary_key=True) + consumable_id = db.Column(db.Integer, db.ForeignKey("consumables.id", ondelete="RESTRICT"), nullable=False, index=True) + supplier_id = db.Column(db.Integer, db.ForeignKey("companies.id"), nullable=True, index=True) + quantity_ordered = db.Column(db.Integer, nullable=False) + quantity_received = db.Column(db.Integer, nullable=False, default=0) + status = db.Column(db.String(30), nullable=False, default="TO_ORDER", index=True) + ordered_at = db.Column(db.DateTime, nullable=True) + received_at = db.Column(db.DateTime, nullable=True) + comment = db.Column(db.Text, nullable=True) + created_by_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"), nullable=True) + + consumable = db.relationship("Consumable") + supplier = db.relationship("Company") + created_by = db.relationship("User") + receipts = db.relationship("ConsumableReceipt", back_populates="order", cascade="all, delete-orphan") + + @property + def remaining_quantity(self): + return max((self.quantity_ordered or 0) - (self.quantity_received or 0), 0) + + +class ConsumableReceipt(db.Model): + __tablename__ = "consumable_receipts" + + id = db.Column(db.Integer, primary_key=True) + order_id = db.Column(db.Integer, db.ForeignKey("consumable_orders.id", ondelete="CASCADE"), nullable=False, index=True) + quantity = db.Column(db.Integer, nullable=False) + received_at = db.Column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) + received_by_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"), nullable=True) + comment = db.Column(db.Text, nullable=True) + + order = db.relationship("ConsumableOrder", back_populates="receipts") + received_by = db.relationship("User") diff --git a/app_new/core/models/equipment.py b/app_new/core/models/equipment.py index 5bc63c3..b2b7d10 100644 --- a/app_new/core/models/equipment.py +++ b/app_new/core/models/equipment.py @@ -57,6 +57,10 @@ class Equipment(db.Model): model_reference = db.Column(db.String(120), nullable=True) supplier_id = db.Column(db.Integer, db.ForeignKey("companies.id"), nullable=True) supplier_reference = db.Column(db.String(120), nullable=True) + device_type = db.Column(db.String(30), nullable=True) + ip_address = db.Column(db.String(45), nullable=True) + hostname = db.Column(db.String(120), nullable=True) + mac_address = db.Column(db.String(17), nullable=True) purchase_price = db.Column(db.Numeric(12, 2), nullable=True) purchase_date = db.Column(db.Date, nullable=True) lifecycle_status = db.Column(db.String(30), nullable=False, default="en_service") diff --git a/app_new/core/models/planning.py b/app_new/core/models/planning.py index 94d2b8d..719a7dc 100644 --- a/app_new/core/models/planning.py +++ b/app_new/core/models/planning.py @@ -600,9 +600,11 @@ class Consumable(db.Model): name = db.Column(db.String(200), nullable=False) reference = db.Column(db.String(100)) category = db.Column(db.String(50)) + consumable_type = db.Column(db.String(30), nullable=False, default="OTHER", server_default="OTHER") quantity = db.Column(db.Integer, default=0) min_quantity = db.Column(db.Integer, default=0) + target_quantity = db.Column(db.Integer, nullable=True) unit = db.Column(db.String(20), default="pièce") unit_price = db.Column(db.Float) @@ -620,6 +622,11 @@ class Consumable(db.Model): @property def is_low_stock(self): return self.quantity <= self.min_quantity + + @property + def quantity_to_order(self): + target = self.target_quantity if self.target_quantity is not None else self.min_quantity + return max((target or 0) - (self.quantity or 0), 0) def __repr__(self): return f"" @@ -653,6 +660,9 @@ class ConsumableUsage(db.Model): class EquipmentConsumable(db.Model): """Consommables compatibles avec un équipement.""" __tablename__ = "equipment_consumables" + __table_args__ = ( + db.UniqueConstraint("equipment_id", "consumable_id", name="uq_equipment_consumable_pair"), + ) id = db.Column(db.Integer, primary_key=True) diff --git a/app_new/core/services/c4_service.py b/app_new/core/services/c4_service.py new file mode 100644 index 0000000..b86f179 --- /dev/null +++ b/app_new/core/services/c4_service.py @@ -0,0 +1,289 @@ +"""Services métier C4 : contrats photocopieurs et consommables partagés.""" +from calendar import monthrange +from datetime import date, datetime, timedelta, timezone +from statistics import mean + +from sqlalchemy import and_, or_ + +from ..models import ( + Alert, Company, Consumable, ConsumableOrder, ConsumableReceipt, + ConsumableUsage, ContractEquipmentAssignment, Equipment, EquipmentConsumable, + Meter, MeterReading, PhotocopierContract, PhotocopierContractPeriod, +) +from ...extensions import db + + +class C4DomainError(ValueError): + pass + + +def _anniversary(year, month, day): + return date(year, month, min(day, monthrange(year, month)[1])) + + +def create_contract(*, supplier_id, name, start_date, end_date, anniversary_month, + anniversary_day, quota_start_date=None, reference=None, notes=None, commit=True): + if end_date < start_date: + raise C4DomainError("La fin du contrat doit être postérieure à son début.") + if not 1 <= int(anniversary_month) <= 12: + raise C4DomainError("Le mois anniversaire est invalide.") + quota_start_date = quota_start_date or start_date + if quota_start_date < start_date or quota_start_date > end_date: + raise C4DomainError("La date de début des quotas doit être comprise dans le contrat.") + contract = PhotocopierContract( + supplier_id=supplier_id, name=name, reference=reference, + start_date=start_date, end_date=end_date, + quota_start_date=quota_start_date, + anniversary_month=int(anniversary_month), anniversary_day=int(anniversary_day), + notes=notes, status="ACTIVE", + ) + db.session.add(contract) + db.session.flush() + generate_contract_periods(contract, commit=False) + if commit: + db.session.commit() + return contract + + +def generate_contract_periods(contract, *, commit=True): + """Crée les périodes annuelles sans doublon, bornées au contrat.""" + cursor = contract.quota_start_date or contract.start_date + while cursor <= contract.end_date: + next_anniversary = _anniversary(cursor.year + 1, contract.anniversary_month, contract.anniversary_day) + period_end = min(next_anniversary - timedelta(days=1), contract.end_date) + if period_end < cursor: + break + existing = PhotocopierContractPeriod.query.filter_by(contract_id=contract.id, start_date=cursor).first() + if not existing: + db.session.add(PhotocopierContractPeriod(contract=contract, start_date=cursor, end_date=period_end)) + cursor = period_end + timedelta(days=1) + if commit: + db.session.commit() + return contract.periods + + +def _overlaps(start, end, other_start, other_end): + return start <= (other_end or date.max) and (end or date.max) >= other_start + + +def assign_equipment_to_contract(*, contract, equipment, entry_date, exit_date=None, + replacement_reason=None, replacement_comment=None, commit=True): + if contract.status != "ACTIVE": + raise C4DomainError("Un contrat clos ne peut plus recevoir de machine.") + if entry_date < contract.start_date or entry_date > contract.end_date: + raise C4DomainError("La date d'entrée doit être comprise dans le contrat.") + if exit_date and exit_date > contract.end_date: + raise C4DomainError("La date de sortie dépasse la fin du contrat.") + for row in ContractEquipmentAssignment.query.filter_by(equipment_id=equipment.id).all(): + if _overlaps(entry_date, exit_date, row.entry_date, row.exit_date): + raise C4DomainError("Cette machine est déjà rattachée à un contrat sur cette période.") + assignment = ContractEquipmentAssignment( + contract=contract, equipment=equipment, entry_date=entry_date, exit_date=exit_date, + replacement_reason=replacement_reason, replacement_comment=replacement_comment, + ) + db.session.add(assignment) + if commit: + db.session.commit() + return assignment + + +def replace_contract_equipment(*, old_assignment, new_equipment, replacement_date, + reason="AUTRE", comment=None, commit=True): + if reason not in {"PANNE", "RENOUVELLEMENT_FOURNISSEUR", "CHANGEMENT_MODELE", "FIN_DE_VIE", "AUTRE"}: + raise C4DomainError("Motif de remplacement invalide.") + if replacement_date < old_assignment.entry_date: + raise C4DomainError("La date de remplacement est invalide.") + old_assignment.exit_date = replacement_date - timedelta(days=1) + new_assignment = assign_equipment_to_contract( + contract=old_assignment.contract, equipment=new_equipment, + entry_date=replacement_date, replacement_reason=reason, + replacement_comment=comment, commit=False, + ) + if commit: + db.session.commit() + return new_assignment + + +def _meter_role(meter): + text = f"{meter.name} {meter.usage or ''}".lower() + return "color" if any(word in text for word in ("couleur", "color", "coul")) else "bw" + + +def _meter_consumption(meter, start_date, end_date): + readings = (MeterReading.query.filter( + MeterReading.meter_id == meter.id, + MeterReading.reading_date >= datetime.combine(start_date, datetime.min.time()), + MeterReading.reading_date <= datetime.combine(end_date, datetime.max.time()), + ).order_by(MeterReading.reading_date, MeterReading.id).all()) + before = (MeterReading.query.filter( + MeterReading.meter_id == meter.id, + MeterReading.reading_date < datetime.combine(start_date, datetime.min.time()), + ).order_by(MeterReading.reading_date.desc(), MeterReading.id.desc()).first()) + if before: + readings.insert(0, before) + total = 0.0 + quality = "EXACT" if before else "PARTIAL" + previous = None + for reading in readings: + if previous and not reading.is_reset and not previous.is_reset and reading.value >= previous.value: + total += reading.value - previous.value + previous = reading + if not before and len(readings) >= 2: + quality = "ESTIMATED" + return total, quality, len(readings) + + +def calculate_contract_usage(period): + result = {"bw": 0.0, "color": 0.0, "quality": "EXACT", "machines": 0} + assignments = ContractEquipmentAssignment.query.filter( + ContractEquipmentAssignment.contract_id == period.contract_id, + ContractEquipmentAssignment.entry_date <= period.end_date, + or_(ContractEquipmentAssignment.exit_date.is_(None), ContractEquipmentAssignment.exit_date >= period.start_date), + ).all() + for assignment in assignments: + result["machines"] += 1 + start = max(period.start_date, assignment.entry_date) + end = min(period.end_date, assignment.exit_date or period.end_date) + for meter in Meter.query.filter_by(equipment_id=assignment.equipment_id).all(): + amount, quality, _ = _meter_consumption(meter, start, end) + result[_meter_role(meter)] += amount + if quality != "EXACT": + result["quality"] = "PARTIAL" if result["quality"] == "EXACT" else result["quality"] + return result + + +def _school_days(start, end): + """Approximation explicite : jours ouvrés, hors fermetures complètes.""" + from ..models.planning import CollegeClosure, ClosureWorkDay + closures = CollegeClosure.query.filter(CollegeClosure.start_date <= end, CollegeClosure.end_date >= start).all() + work_days = {row.work_date for closure in closures for row in ClosureWorkDay.query.filter_by(closure_id=closure.id).all()} + closed = {day for closure in closures if closure.work_hours_type == "none" + for day in (start + timedelta(days=i) for i in range((end - start).days + 1)) + if closure.start_date <= day <= closure.end_date} + return sum(1 for i in range((end - start).days + 1) + if (day := start + timedelta(days=i)).weekday() < 5 and day not in closed or day in work_days) + + +def project_contract_usage(period): + usage = calculate_contract_usage(period) + elapsed = max((date.today() - period.start_date).days + 1, 0) + remaining = max((period.end_date - max(date.today(), period.start_date)).days, 0) + school_elapsed = _school_days(period.start_date, min(date.today(), period.end_date)) if elapsed else 0 + school_remaining = _school_days(max(date.today() + timedelta(days=1), period.start_date), period.end_date) if remaining else 0 + result = {"usage": usage, "projection": {}, "remaining": {}, "confidence": "LOW"} + for role, quota_key in (("bw", "quota_bw"), ("color", "quota_color")): + quota = getattr(period, quota_key) + result["remaining"][role] = max(quota - usage[role], 0) if quota is not None else None + if quota is None or school_elapsed <= 0: + result["projection"][role] = None + else: + daily = usage[role] / school_elapsed + result["projection"][role] = usage[role] + daily * school_remaining + result["confidence"] = "NORMAL" if school_elapsed >= 10 else "LOW" + return result + + +def evaluate_quota_alert(period, *, level="WARNING", commit=True): + from ..models.company import Alert + projection = project_contract_usage(period) + for role, quota_key in (("bw", "quota_bw"), ("color", "quota_color")): + quota = getattr(period, quota_key) + projected = projection["projection"].get(role) + if quota is None or projected is None or projected <= quota: + continue + related_key = f"{period.id}:{role}" + alert = Alert.query.filter_by(alert_type="QUOTA_OVERAGE_PROJECTED", related_type=related_key, is_resolved=False).first() + message = f"Projection {role} : {projected:.0f} copies pour un quota de {quota}." + if alert: + alert.message = message + else: + db.session.add(Alert( + alert_type="QUOTA_OVERAGE_PROJECTED", title="Dépassement quota photocopieur", + message=message, priority="haute" if level == "CRITICAL" else "normale", + related_id=period.id, related_type=related_key, + )) + if commit: + db.session.commit() + return projection + + +def create_consumable_order(*, consumable, quantity, supplier_id=None, user_id=None, commit=True): + quantity = int(quantity) + if quantity <= 0: + raise C4DomainError("La quantité commandée doit être positive.") + order = ConsumableOrder(consumable=consumable, quantity_ordered=quantity, supplier_id=supplier_id, created_by_id=user_id, status="ORDERED", ordered_at=datetime.now(timezone.utc)) + db.session.add(order) + refresh_consumable_stock_alert(consumable, commit=False) + if commit: + db.session.commit() + return order + + +def receive_consumable_order(*, order, quantity, user_id=None, comment=None, commit=True): + quantity = int(quantity) + if quantity <= 0 or quantity > order.remaining_quantity: + raise C4DomainError("La quantité reçue dépasse le restant à recevoir.") + receipt = ConsumableReceipt(order=order, quantity=quantity, received_by_id=user_id, comment=comment) + order.quantity_received += quantity + order.consumable.quantity = (order.consumable.quantity or 0) + quantity + order.status = "RECEIVED" if order.remaining_quantity == 0 else "PARTIALLY_RECEIVED" + if order.status == "RECEIVED": + order.received_at = receipt.received_at + db.session.add(receipt) + refresh_consumable_stock_alert(order.consumable, commit=False) + if commit: + db.session.commit() + return receipt + + +def record_consumable_issue(*, consumable, equipment, quantity=1, user_id=None, intervention_id=None, notes=None, commit=True): + quantity = int(quantity) + if quantity <= 0: + raise C4DomainError("La sortie doit être positive.") + if (consumable.quantity or 0) < quantity: + raise C4DomainError("Stock insuffisant.") + if not EquipmentConsumable.query.filter_by(equipment_id=equipment.id, consumable_id=consumable.id).first(): + raise C4DomainError("Cette référence n'est pas compatible avec la machine.") + usage = ConsumableUsage(consumable=consumable, equipment=equipment, quantity_used=quantity, used_by_id=user_id, intervention_id=intervention_id, notes=notes) + consumable.quantity -= quantity + db.session.add(usage) + refresh_consumable_stock_alert(consumable, commit=False) + if commit: + db.session.commit() + return usage + + +def refresh_consumable_stock_alert(consumable, *, commit=True): + """Maintient une seule alerte de stock bas, sans convertir une commande en stock.""" + pending = ConsumableOrder.query.filter( + ConsumableOrder.consumable_id == consumable.id, + ConsumableOrder.status.in_(("ORDERED", "PARTIALLY_RECEIVED")), + ).first() + alert = Alert.query.filter_by(alert_type="CONSUMABLE_STOCK_LOW", related_type="consumable", related_id=consumable.id, is_resolved=False).first() + if (consumable.quantity or 0) <= (consumable.min_quantity or 0): + message = f"Stock bas : {consumable.quantity or 0} / minimum {consumable.min_quantity or 0}." + if pending: + message += f" Commande en cours : {pending.remaining_quantity} à recevoir." + if alert: + alert.message = message + else: + db.session.add(Alert(alert_type="CONSUMABLE_STOCK_LOW", title="Stock consommable bas", message=message, related_type="consumable", related_id=consumable.id)) + elif alert: + alert.is_resolved = True + alert.resolved_at = datetime.now(timezone.utc) + if commit: + db.session.commit() + return alert + + +def estimate_consumable_lifetime(*, consumable, equipment): + usages = (ConsumableUsage.query.filter_by(consumable_id=consumable.id, equipment_id=equipment.id) + .order_by(ConsumableUsage.used_at).all()) + durations = [(b.used_at - a.used_at).days for a, b in zip(usages, usages[1:]) if (b.used_at - a.used_at).days > 0] + current_days = None + if usages: + last = usages[-1].used_at + if last.tzinfo is None: + last = last.replace(tzinfo=timezone.utc) + current_days = (datetime.now(timezone.utc) - last).days + return {"average_days": mean(durations) if durations else None, "observations": len(durations), "current_days": current_days} diff --git a/migrations/versions/s4g5h6i7j8k9_checkpoint_c4.py b/migrations/versions/s4g5h6i7j8k9_checkpoint_c4.py new file mode 100644 index 0000000..9b95618 --- /dev/null +++ b/migrations/versions/s4g5h6i7j8k9_checkpoint_c4.py @@ -0,0 +1,120 @@ +"""Add C4 copier contracts and shared consumable workflows.""" +from alembic import op +import sqlalchemy as sa + + +revision = "s4g5h6i7j8k9" +down_revision = "r3f4g5h6i7j8" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column("equipments", sa.Column("device_type", sa.String(length=30), nullable=True)) + op.add_column("equipments", sa.Column("ip_address", sa.String(length=45), nullable=True)) + op.add_column("equipments", sa.Column("hostname", sa.String(length=120), nullable=True)) + op.add_column("equipments", sa.Column("mac_address", sa.String(length=17), nullable=True)) + op.add_column("consumables", sa.Column("consumable_type", sa.String(length=30), nullable=False, server_default="OTHER")) + op.add_column("consumables", sa.Column("target_quantity", sa.Integer(), nullable=True)) + op.create_unique_constraint("uq_equipment_consumable_pair", "equipment_consumables", ["equipment_id", "consumable_id"]) + + op.create_table( + "photocopier_contracts", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("supplier_id", sa.Integer(), sa.ForeignKey("companies.id"), nullable=False), + sa.Column("name", sa.String(200), nullable=False), + sa.Column("reference", sa.String(120)), + sa.Column("start_date", sa.Date(), nullable=False), + sa.Column("end_date", sa.Date(), nullable=False), + sa.Column("quota_start_date", sa.Date(), nullable=True), + sa.Column("anniversary_month", sa.Integer(), nullable=False), + sa.Column("anniversary_day", sa.Integer(), nullable=False), + sa.Column("status", sa.String(20), nullable=False, server_default="ACTIVE"), + sa.Column("notes", sa.Text()), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("closed_at", sa.Date()), + sa.CheckConstraint("end_date >= start_date", name="ck_c4_contract_dates"), + ) + op.create_index("ix_c4_contract_supplier", "photocopier_contracts", ["supplier_id"]) + op.create_index("ix_c4_contract_status", "photocopier_contracts", ["status"]) + + op.create_table( + "photocopier_contract_periods", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("contract_id", sa.Integer(), sa.ForeignKey("photocopier_contracts.id", ondelete="CASCADE"), nullable=False), + sa.Column("start_date", sa.Date(), nullable=False), + sa.Column("end_date", sa.Date(), nullable=False), + sa.Column("quota_bw", sa.Integer()), + sa.Column("quota_color", sa.Integer()), + sa.Column("bw_included_unit_price", sa.Numeric(12, 6)), + sa.Column("bw_overage_unit_price", sa.Numeric(12, 6)), + sa.Column("color_included_unit_price", sa.Numeric(12, 6)), + sa.Column("color_overage_unit_price", sa.Numeric(12, 6)), + sa.Column("status", sa.String(20), nullable=False, server_default="OPEN"), + sa.CheckConstraint("end_date >= start_date", name="ck_c4_period_dates"), + sa.CheckConstraint("(quota_bw IS NULL OR quota_bw >= 0) AND (quota_color IS NULL OR quota_color >= 0)", name="ck_c4_period_quotas"), + sa.UniqueConstraint("contract_id", "start_date", name="uq_c4_contract_period_start"), + ) + op.create_index("ix_c4_period_contract", "photocopier_contract_periods", ["contract_id"]) + op.create_index("ix_c4_period_start", "photocopier_contract_periods", ["start_date"]) + op.create_index("ix_c4_period_end", "photocopier_contract_periods", ["end_date"]) + + op.create_table( + "contract_equipment_assignments", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("contract_id", sa.Integer(), sa.ForeignKey("photocopier_contracts.id", ondelete="CASCADE"), nullable=False), + sa.Column("equipment_id", sa.Integer(), sa.ForeignKey("equipments.id", ondelete="CASCADE"), nullable=False), + sa.Column("entry_date", sa.Date(), nullable=False), + sa.Column("exit_date", sa.Date()), + sa.Column("replacement_reason", sa.String(40)), + sa.Column("replacement_comment", sa.Text()), + sa.CheckConstraint("exit_date IS NULL OR exit_date >= entry_date", name="ck_c4_assignment_dates"), + sa.UniqueConstraint("contract_id", "equipment_id", "entry_date", name="uq_c4_contract_equipment_entry"), + ) + op.create_index("ix_c4_assignment_contract", "contract_equipment_assignments", ["contract_id"]) + op.create_index("ix_c4_assignment_equipment", "contract_equipment_assignments", ["equipment_id"]) + + op.create_table( + "consumable_orders", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("consumable_id", sa.Integer(), sa.ForeignKey("consumables.id", ondelete="RESTRICT"), nullable=False), + sa.Column("supplier_id", sa.Integer(), sa.ForeignKey("companies.id")), + sa.Column("quantity_ordered", sa.Integer(), nullable=False), + sa.Column("quantity_received", sa.Integer(), nullable=False, server_default="0"), + sa.Column("status", sa.String(30), nullable=False, server_default="TO_ORDER"), + sa.Column("ordered_at", sa.DateTime()), + sa.Column("received_at", sa.DateTime()), + sa.Column("comment", sa.Text()), + sa.Column("created_by_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL")), + sa.CheckConstraint("quantity_ordered > 0", name="ck_c4_order_quantity"), + sa.CheckConstraint("quantity_received >= 0 AND quantity_received <= quantity_ordered", name="ck_c4_order_received"), + ) + op.create_index("ix_c4_order_consumable", "consumable_orders", ["consumable_id"]) + op.create_index("ix_c4_order_supplier", "consumable_orders", ["supplier_id"]) + op.create_index("ix_c4_order_status", "consumable_orders", ["status"]) + + op.create_table( + "consumable_receipts", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("order_id", sa.Integer(), sa.ForeignKey("consumable_orders.id", ondelete="CASCADE"), nullable=False), + sa.Column("quantity", sa.Integer(), nullable=False), + sa.Column("received_at", sa.DateTime(), nullable=False), + sa.Column("received_by_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL")), + sa.Column("comment", sa.Text()), + ) + op.create_index("ix_c4_receipt_order", "consumable_receipts", ["order_id"]) + + +def downgrade(): + op.drop_table("consumable_receipts") + op.drop_table("consumable_orders") + op.drop_table("contract_equipment_assignments") + op.drop_table("photocopier_contract_periods") + op.drop_table("photocopier_contracts") + op.drop_constraint("uq_equipment_consumable_pair", "equipment_consumables", type_="unique") + op.drop_column("consumables", "target_quantity") + op.drop_column("consumables", "consumable_type") + op.drop_column("equipments", "mac_address") + op.drop_column("equipments", "hostname") + op.drop_column("equipments", "ip_address") + op.drop_column("equipments", "device_type") diff --git a/tests/integration/test_checkpoint_c4.py b/tests/integration/test_checkpoint_c4.py new file mode 100644 index 0000000..c4c5fa7 --- /dev/null +++ b/tests/integration/test_checkpoint_c4.py @@ -0,0 +1,148 @@ +from datetime import date, datetime, timedelta, timezone + +import pytest + +from app_new.extensions import db +from app_new.core.models import ( + Company, Consumable, ConsumableOrder, ConsumableUsage, Equipment, + EquipmentConsumable, Meter, MeterReading, PhotocopierContract, +) +from app_new.core.services.c4_service import ( + C4DomainError, assign_equipment_to_contract, calculate_contract_usage, + create_consumable_order, create_contract, estimate_consumable_lifetime, + evaluate_quota_alert, receive_consumable_order, record_consumable_issue, replace_contract_equipment, +) + + +def _company(name="TEST_UI_C4_SUPPLIER"): + company = Company(name=name) + db.session.add(company) + db.session.flush() + return company + + +def _equipment(name, device_type="photocopier"): + item = Equipment(name=name, device_type=device_type, serial_number=f"TEST_UI_C4_SN_{name}") + db.session.add(item) + db.session.flush() + return item + + +def _meter(equipment, name, usage): + meter = Meter(equipment=equipment, name=name, meter_type="photocopieur", unit="pages", usage=usage) + db.session.add(meter) + db.session.flush() + return meter + + +def _reading(meter, value, day): + row = MeterReading(meter=meter, value=value, reading_date=datetime.combine(day, datetime.min.time(), tzinfo=timezone.utc)) + db.session.add(row) + db.session.flush() + return row + + +def test_c4_contract_periods_quota_and_machine_replacement(app): + with app.app_context(): + supplier = _company() + contract = create_contract( + supplier_id=supplier.id, name="TEST_UI_C4_CONTRACT", start_date=date(2026, 10, 1), + end_date=date(2028, 9, 30), anniversary_month=10, anniversary_day=1, + ) + assert [(p.start_date, p.end_date) for p in contract.periods] == [ + (date(2026, 10, 1), date(2027, 9, 30)), + (date(2027, 10, 1), date(2028, 9, 30)), + ] + contract.periods[0].quota_bw = 120000 + contract.periods[0].quota_color = 30000 + machine_a = _equipment("TEST_UI_C4_A") + machine_d = _equipment("TEST_UI_C4_D") + old = assign_equipment_to_contract(contract=contract, equipment=machine_a, entry_date=date(2026, 10, 1)) + new = replace_contract_equipment(old_assignment=old, new_equipment=machine_d, replacement_date=date(2027, 2, 1), reason="PANNE") + db.session.commit() + assert old.exit_date == date(2027, 1, 31) + assert new.replacement_reason == "PANNE" + assert contract.status == "ACTIVE" + + +def test_c4_contract_usage_separates_bw_and_color(app): + with app.app_context(): + supplier = _company("TEST_UI_C4_USAGE_SUPPLIER") + contract = create_contract(supplier_id=supplier.id, name="TEST_UI_C4_USAGE", start_date=date(2026, 1, 1), end_date=date(2026, 12, 31), anniversary_month=1, anniversary_day=1) + period = contract.periods[0] + period.quota_bw, period.quota_color = 100, 50 + equipment = _equipment("TEST_UI_C4_USAGE_MACHINE") + assign_equipment_to_contract(contract=contract, equipment=equipment, entry_date=date(2026, 1, 1)) + bw, color = _meter(equipment, "N&B", "N&B"), _meter(equipment, "Couleur", "Couleur") + _reading(bw, 100, date(2026, 1, 1)); _reading(bw, 180, date(2026, 2, 1)) + _reading(color, 10, date(2026, 1, 1)); _reading(color, 25, date(2026, 2, 1)) + result = calculate_contract_usage(period) + assert result["bw"] == 80 + assert result["color"] == 15 + + +def test_c4_quota_alert_is_deduplicated(app): + with app.app_context(): + supplier = _company("TEST_UI_C4_ALERT_SUPPLIER") + contract = create_contract(supplier_id=supplier.id, name="TEST_UI_C4_ALERT", start_date=date.today() - timedelta(days=40), end_date=date.today() + timedelta(days=325), anniversary_month=date.today().month, anniversary_day=date.today().day) + period = contract.periods[0] + period.quota_bw = 10 + machine = _equipment("TEST_UI_C4_ALERT_MACHINE") + assign_equipment_to_contract(contract=contract, equipment=machine, entry_date=period.start_date) + meter = _meter(machine, "N&B", "N&B") + _reading(meter, 0, period.start_date); _reading(meter, 100, date.today()) + evaluate_quota_alert(period) + evaluate_quota_alert(period) + from app_new.core.models.company import Alert + alerts = Alert.query.filter_by(alert_type="QUOTA_OVERAGE_PROJECTED", related_id=period.id, is_resolved=False).all() + assert len(alerts) == 1 + + +def test_c4_shared_consumable_order_receipt_issue_and_negative_stock(app, admin_user): + with app.app_context(): + machine_a, machine_b = _equipment("TEST_UI_C4_PRINTER_A", "printer"), _equipment("TEST_UI_C4_PRINTER_B", "printer") + consumable = Consumable(name="TEST_UI_C4_TONER", reference="TEST_UI_C4_SHARED", consumable_type="TONER", quantity=1, min_quantity=2, target_quantity=6) + db.session.add(consumable); db.session.flush() + db.session.add_all([EquipmentConsumable(equipment=machine_a, consumable=consumable), EquipmentConsumable(equipment=machine_b, consumable=consumable)]) + db.session.commit() + assert consumable.quantity_to_order == 5 + order = create_consumable_order(consumable=consumable, quantity=5, user_id=admin_user["id"]) + assert consumable.quantity == 1 + receive_consumable_order(order=order, quantity=3, user_id=admin_user["id"]) + assert consumable.quantity == 4 and order.status == "PARTIALLY_RECEIVED" + receive_consumable_order(order=order, quantity=2, user_id=admin_user["id"]) + assert consumable.quantity == 6 and order.status == "RECEIVED" + record_consumable_issue(consumable=consumable, equipment=machine_a, quantity=1, user_id=admin_user["id"]) + assert consumable.quantity == 5 + with pytest.raises(C4DomainError): + record_consumable_issue(consumable=consumable, equipment=machine_b, quantity=99, user_id=admin_user["id"]) + # La contrainte historique de ConsumableUsage ne comporte pas encore + # ON DELETE SET NULL ; neutraliser la référence de fixture évite de + # transformer ce test métier en erreur de teardown historique. + ConsumableUsage.query.update({ConsumableUsage.used_by_id: None}, synchronize_session=False) + db.session.commit() + + +def test_c4_consumable_lifetime_is_per_machine_and_excludes_current_installation(app): + with app.app_context(): + machine_a, machine_b = _equipment("TEST_UI_C4_LIFE_A", "printer"), _equipment("TEST_UI_C4_LIFE_B", "printer") + consumable = Consumable(name="TEST_UI_C4_LIFE_TONER", quantity=10) + db.session.add(consumable); db.session.flush() + db.session.add_all([EquipmentConsumable(equipment=machine_a, consumable=consumable), EquipmentConsumable(equipment=machine_b, consumable=consumable)]) + base = datetime(2026, 1, 1, tzinfo=timezone.utc) + for machine, offsets in ((machine_a, (0, 30, 60)), (machine_b, (0, 14, 28))): + for offset in offsets: + db.session.add(ConsumableUsage(consumable=consumable, equipment=machine, quantity_used=1, used_at=base + timedelta(days=offset))) + db.session.commit() + a = estimate_consumable_lifetime(consumable=consumable, equipment=machine_a) + b = estimate_consumable_lifetime(consumable=consumable, equipment=machine_b) + assert a["average_days"] == 30 and b["average_days"] == 14 + assert a["observations"] == 2 + + +def test_c4_contract_http_and_permissions(authenticated_client): + response = authenticated_client.get("/c4/contracts") + assert response.status_code == 200 + assert "Contrats photocopieurs" in response.get_data(as_text=True) + response = authenticated_client.get("/c4/consumables") + assert response.status_code == 200