feat(c4): add copier contracts and shared consumable stock

This commit is contained in:
root 2026-08-24 15:56:13 +00:00
parent 78d0090275
commit a2374d8157
16 changed files with 865 additions and 1 deletions

View file

@ -145,6 +145,9 @@ def create_app(config_name='default'):
from .parts.routes import parts_bp from .parts.routes import parts_bp
app.register_blueprint(parts_bp, url_prefix='/parts') 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) # Matériel et produits d'entretien (stock, FEFO, distributions, prévisions)
from .cleaning import cleaning_bp from .cleaning import cleaning_bp
app.register_blueprint(cleaning_bp) app.register_blueprint(cleaning_bp)

3
app_new/c4/__init__.py Normal file
View file

@ -0,0 +1,3 @@
from .routes import c4_bp
__all__ = ["c4_bp"]

153
app_new/c4/routes.py Normal file
View file

@ -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/<int:contract_id>")
@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/<int:contract_id>/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/<int:contract_id>/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/<int:consumable_id>")
@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/<int:consumable_id>/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/<int:order_id>/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/<int:consumable_id>/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))

View file

@ -0,0 +1,3 @@
{% extends "base.html" %}
{% block title %}{{ consumable.name }} — GMAO{% endblock %}
{% block content %}<div class="container-fluid"><h1 class="h4">{{ consumable.name }}</h1><p>Référence : {{ consumable.reference or '—' }} · Type : {{ consumable.consumable_type }} · Stock : {{ consumable.quantity }} · À commander : {{ consumable.quantity_to_order }}</p><h2 class="h5">Compatible avec {{ compatible|length }} équipement(s)</h2><ul>{% for link in compatible %}<li>{{ link.equipment.name }}</li>{% else %}<li>Aucune machine.</li>{% endfor %}</ul><h2 class="h5">Commandes</h2><div class="table-responsive"><table class="table table-sm"><tr><th>Quantité</th><th>Reçue</th><th>État</th><th>Action</th></tr>{% for order in orders %}<tr><td>{{ order.quantity_ordered }}</td><td>{{ order.quantity_received }}</td><td>{{ order.status }}</td><td>{% if order.remaining_quantity %}<form method="post" action="{{ url_for('c4.receive_order', order_id=order.id) }}" class="d-flex gap-2"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><input class="form-control form-control-sm" type="number" name="quantity" min="1" max="{{ order.remaining_quantity }}" value="{{ order.remaining_quantity }}"><button class="btn btn-sm btn-success">Réceptionner</button></form>{% endif %}</td></tr>{% else %}<tr><td colspan="4">Aucune commande.</td></tr>{% endfor %}</table></div><form method="post" action="{{ url_for('c4.order_consumable', consumable_id=consumable.id) }}" class="row g-2 mt-3"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><div class="col-auto"><label class="form-label">Nouvelle commande</label><input class="form-control" type="number" name="quantity" min="1" value="{{ consumable.quantity_to_order or 1 }}"></div><div class="col-auto align-self-end"><button class="btn btn-primary">Commander</button></div></form><h2 class="h5 mt-4">Historique des sorties</h2><ul>{% for usage in usages %}<li>{{ usage.used_at }} — {{ usage.quantity_used }} — {{ usage.equipment.name if usage.equipment else '—' }}</li>{% else %}<li>Aucune sortie.</li>{% endfor %}</ul></div>{% endblock %}

View file

@ -0,0 +1,3 @@
{% extends "base.html" %}
{% block title %}Consommables imprimantes — GMAO{% endblock %}
{% block content %}<div class="container-fluid"><h1 class="h4 mb-3">Consommables partagés</h1><div class="table-responsive table-responsive-stack"><table class="table table-striped"><thead><tr><th>Référence</th><th>Type</th><th>Stock</th><th>Seuil</th><th>Cible</th><th>À commander</th></tr></thead><tbody>{% for item in consumables %}<tr><td data-label="Référence"><a href="{{ url_for('c4.consumable_detail', consumable_id=item.id) }}">{{ item.name }}</a><small class="d-block text-muted">{{ item.reference or '' }}</small></td><td data-label="Type">{{ item.consumable_type }}</td><td data-label="Stock">{{ item.quantity }}</td><td data-label="Seuil">{{ item.min_quantity }}</td><td data-label="Cible">{{ item.target_quantity if item.target_quantity is not none else '—' }}</td><td data-label="À commander">{{ item.quantity_to_order }}</td></tr>{% else %}<tr><td colspan="6">Aucun consommable.</td></tr>{% endfor %}</tbody></table></div></div>{% endblock %}

View file

@ -0,0 +1,3 @@
{% extends "base.html" %}
{% block title %}{{ contract.name }} — GMAO{% endblock %}
{% block content %}<div class="container-fluid"><div class="d-flex justify-content-between"><h1 class="h4">{{ contract.name }}</h1><span class="badge text-bg-secondary">{{ contract.status }}</span></div><p>{{ contract.supplier.name }} · {{ contract.start_date }} → {{ contract.end_date }} · anniversaire {{ '%02d/%02d'|format(contract.anniversary_day, contract.anniversary_month) }}</p><form method="post" action="{{ url_for('c4.close_contract', contract_id=contract.id) }}" class="mb-3"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><button class="btn btn-outline-danger btn-sm">Clore le contrat</button></form>{% for period in periods %}{% set u=usage[period.id] %}<section class="card mb-3"><div class="card-header">Période {{ period.start_date }} → {{ period.end_date }}</div><div class="card-body"><div class="row"><div class="col-md-6"><strong>N&B</strong><br>{{ '%.0f'|format(u.bw) }}{% if period.quota_bw is not none %} / {{ period.quota_bw }}{% else %} — sans quota{% endif %}</div><div class="col-md-6"><strong>Couleur</strong><br>{{ '%.0f'|format(u.color) }}{% if period.quota_color is not none %} / {{ period.quota_color }}{% else %} — sans quota{% endif %}</div></div><small>Machines prises en compte : {{ u.machines }} · qualité : {{ u.quality }}</small></div></section>{% endfor %}<h2 class="h5">Machines rattachées</h2><ul>{% for a in contract.assignments %}<li>{{ a.equipment.name }} — {{ a.entry_date }}{% if a.exit_date %} → {{ a.exit_date }}{% endif %}</li>{% else %}<li>Aucune machine.</li>{% endfor %}</ul></div>{% endblock %}

View file

@ -0,0 +1,3 @@
{% extends "base.html" %}
{% block title %}Nouveau contrat — GMAO{% endblock %}
{% block content %}<div class="container"><h1 class="h4 mb-3">Nouveau contrat photocopieur</h1><form method="post"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><div class="row g-3"><div class="col-md-6"><label class="form-label">Nom *</label><input class="form-control" name="name" required></div><div class="col-md-6"><label class="form-label">Fournisseur *</label><select class="form-select" name="supplier_id" required>{% for company in companies %}<option value="{{ company.id }}">{{ company.name }}</option>{% endfor %}</select></div><div class="col-md-4"><label class="form-label">Référence</label><input class="form-control" name="reference"></div><div class="col-md-4"><label class="form-label">Début global *</label><input class="form-control" type="date" name="start_date" required></div><div class="col-md-4"><label class="form-label">Fin globale *</label><input class="form-control" type="date" name="end_date" required></div><div class="col-md-3"><label class="form-label">Début quota</label><input class="form-control" type="date" name="quota_start_date"></div><div class="col-md-3"><label class="form-label">Mois anniversaire *</label><input class="form-control" type="number" min="1" max="12" name="anniversary_month" required></div><div class="col-md-3"><label class="form-label">Jour anniversaire *</label><input class="form-control" type="number" min="1" max="31" name="anniversary_day" required></div><div class="col-12"><label class="form-label">Notes</label><textarea class="form-control" name="notes"></textarea></div></div><button class="btn btn-primary mt-3">Créer</button></form></div>{% endblock %}

View file

@ -0,0 +1,10 @@
{% extends "base.html" %}
{% block title %}Contrats photocopieurs — GMAO{% endblock %}
{% block content %}
<div class="container-fluid">
<div class="d-flex justify-content-between align-items-center mb-3"><h1 class="h4">Contrats photocopieurs</h1><a class="btn btn-primary" href="{{ url_for('c4.new_contract') }}">Nouveau contrat</a></div>
<div class="table-responsive table-responsive-stack"><table class="table table-sm table-striped"><thead><tr><th>Contrat</th><th>Fournisseur</th><th>Durée</th><th>Anniversaire</th><th>État</th><th></th></tr></thead><tbody>
{% for contract in contracts %}<tr><td data-label="Contrat"><a href="{{ url_for('c4.contract_detail', contract_id=contract.id) }}">{{ contract.name }}</a>{% if contract.reference %}<small class="text-muted d-block">{{ contract.reference }}</small>{% endif %}</td><td data-label="Fournisseur">{{ contract.supplier.name }}</td><td data-label="Durée">{{ contract.start_date }} → {{ contract.end_date }}</td><td data-label="Anniversaire">{{ '%02d/%02d'|format(contract.anniversary_day, contract.anniversary_month) }}</td><td data-label="État">{{ contract.status }}</td><td></td></tr>{% else %}<tr><td colspan="6">Aucun contrat.</td></tr>{% endfor %}
</tbody></table></div>
</div>
{% endblock %}

View file

@ -103,7 +103,7 @@ PATRIMOINE_BLUEPRINTS = {
} }
PLANNING_BLUEPRINTS = {"planning", "scheduler", "interventions_planning"} PLANNING_BLUEPRINTS = {"planning", "scheduler", "interventions_planning"}
STOCK_BLUEPRINTS = {"parts", "meters", "cleaning"} STOCK_BLUEPRINTS = {"parts", "meters", "cleaning"}
CONTRACT_BLUEPRINTS = {"companies", "contracts", "services"} CONTRACT_BLUEPRINTS = {"companies", "contracts", "services", "c4"}
PREVENTION_BLUEPRINTS = {"trainings", "constraints", "prevention"} PREVENTION_BLUEPRINTS = {"trainings", "constraints", "prevention"}
# Catalogue réellement utilisé par les routes et les fonctions métier. # Catalogue réellement utilisé par les routes et les fonctions métier.

View file

@ -19,6 +19,7 @@ from .planning import (
PreventiveTask, PreventiveTaskConsumable, ScheduledTask, TaskExecutionSegment, PreventiveTask, PreventiveTaskConsumable, ScheduledTask, TaskExecutionSegment,
TechnicianAvailability, AdminTask, ZoneAccessRule TechnicianAvailability, AdminTask, ZoneAccessRule
) )
from .c4 import PhotocopierContract, PhotocopierContractPeriod, ContractEquipmentAssignment, ConsumableOrder, ConsumableReceipt
from .settings import AppSettings from .settings import AppSettings
from .audit import AuditLog, TemplateAuditMark from .audit import AuditLog, TemplateAuditMark
from .prevention import PreventionWorkLog, StaffAuthorization, RiskAssessment, PreventionAction, SafetyRegisterEntry from .prevention import PreventionWorkLog, StaffAuthorization, RiskAssessment, PreventionAction, SafetyRegisterEntry
@ -53,4 +54,5 @@ __all__ = [
'RequestProfile', 'RequestProfileItem', 'StockInventory', 'StockInventoryLine', 'RequestProfile', 'RequestProfileItem', 'StockInventory', 'StockInventoryLine',
'CleaningForecastConfig', 'CleaningForecastConfig',
'Role', 'Permission', 'UserRole', 'RolePermission', 'UserPermission', 'Role', 'Permission', 'UserRole', 'RolePermission', 'UserPermission',
'PhotocopierContract', 'PhotocopierContractPeriod', 'ContractEquipmentAssignment', 'ConsumableOrder', 'ConsumableReceipt',
] ]

110
app_new/core/models/c4.py Normal file
View file

@ -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")

View file

@ -57,6 +57,10 @@ class Equipment(db.Model):
model_reference = db.Column(db.String(120), nullable=True) model_reference = db.Column(db.String(120), nullable=True)
supplier_id = db.Column(db.Integer, db.ForeignKey("companies.id"), nullable=True) supplier_id = db.Column(db.Integer, db.ForeignKey("companies.id"), nullable=True)
supplier_reference = db.Column(db.String(120), 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_price = db.Column(db.Numeric(12, 2), nullable=True)
purchase_date = db.Column(db.Date, nullable=True) purchase_date = db.Column(db.Date, nullable=True)
lifecycle_status = db.Column(db.String(30), nullable=False, default="en_service") lifecycle_status = db.Column(db.String(30), nullable=False, default="en_service")

View file

@ -600,9 +600,11 @@ class Consumable(db.Model):
name = db.Column(db.String(200), nullable=False) name = db.Column(db.String(200), nullable=False)
reference = db.Column(db.String(100)) reference = db.Column(db.String(100))
category = db.Column(db.String(50)) 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) quantity = db.Column(db.Integer, default=0)
min_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 = db.Column(db.String(20), default="pièce")
unit_price = db.Column(db.Float) unit_price = db.Column(db.Float)
@ -621,6 +623,11 @@ class Consumable(db.Model):
def is_low_stock(self): def is_low_stock(self):
return self.quantity <= self.min_quantity 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): def __repr__(self):
return f"<Consumable {self.name}>" return f"<Consumable {self.name}>"
@ -653,6 +660,9 @@ class ConsumableUsage(db.Model):
class EquipmentConsumable(db.Model): class EquipmentConsumable(db.Model):
"""Consommables compatibles avec un équipement.""" """Consommables compatibles avec un équipement."""
__tablename__ = "equipment_consumables" __tablename__ = "equipment_consumables"
__table_args__ = (
db.UniqueConstraint("equipment_id", "consumable_id", name="uq_equipment_consumable_pair"),
)
id = db.Column(db.Integer, primary_key=True) id = db.Column(db.Integer, primary_key=True)

View file

@ -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}

View file

@ -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")

View file

@ -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