from datetime import date, datetime, timedelta, timezone import pytest from app_new.extensions import db from app_new.core.models import ( Alert, Company, Consumable, ConsumableOrder, ConsumableUsage, Equipment, EquipmentConsumable, Meter, MeterReading, PhotocopierContract, PhotocopierContractPeriod, User, ) from app_new.core.services.c4_service import ( C4DomainError, assign_equipment_to_contract, calculate_contract_usage, cancel_consumable_order, close_photocopier_contract, create_consumable_order, create_contract, estimate_consumable_stock_coverage, receive_consumable_order, visible_open_c4_alerts, ) def _company(name): row = Company(name=name) db.session.add(row) db.session.flush() return row def _equipment(name, device_type="photocopier"): row = Equipment(name=name, device_type=device_type, serial_number=f"TEST_UI_C4_COMPLETE_{name}") db.session.add(row) db.session.flush() return row def _meter(equipment, name, role="COPIER_BW"): row = Meter(equipment=equipment, name=name, meter_type="photocopieur", unit="pages", contract_role=role) db.session.add(row) db.session.flush() return row 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_complete_estimates_missing_contract_start_and_quality(app): with app.app_context(): supplier = _company("TEST_UI_C4_COMPLETE_ESTIMATE_SUPPLIER") contract = create_contract(supplier_id=supplier.id, name="TEST_UI_C4_COMPLETE_ESTIMATE", start_date=date(2026, 10, 1), end_date=date(2027, 9, 30), anniversary_month=10, anniversary_day=1) period = contract.periods[0] machine = _equipment("TEST_UI_C4_COMPLETE_ESTIMATE_MACHINE") assign_equipment_to_contract(contract=contract, equipment=machine, entry_date=period.start_date) meter = _meter(machine, "N&B", "COPIER_BW") _reading(meter, 9000, date(2026, 9, 1)) _reading(meter, 9140, date(2026, 9, 15)) # historique fiable : 10/jour _reading(meter, 9440, date(2026, 10, 15)) _reading(meter, 9610, date(2026, 11, 1)) usage = calculate_contract_usage(period) # 10 jours scolaires environ avant le premier relevé + delta mesuré. assert usage["bw"] > 170 assert usage["quality"] == "ESTIMATED" def test_c4_complete_missing_contract_start_without_history_is_partial(app): with app.app_context(): supplier = _company("TEST_UI_C4_COMPLETE_PARTIAL_SUPPLIER") contract = create_contract(supplier_id=supplier.id, name="TEST_UI_C4_COMPLETE_PARTIAL", start_date=date(2026, 10, 1), end_date=date(2027, 9, 30), anniversary_month=10, anniversary_day=1) period = contract.periods[0] machine = _equipment("TEST_UI_C4_COMPLETE_PARTIAL_MACHINE") assign_equipment_to_contract(contract=contract, equipment=machine, entry_date=period.start_date) meter = _meter(machine, "N&B", "COPIER_BW") _reading(meter, 100, date(2026, 10, 15)) _reading(meter, 200, date(2026, 11, 1)) usage = calculate_contract_usage(period) assert usage["bw"] == 100 assert usage["quality"] == "PARTIAL" def test_c4_complete_unknown_contract_role_is_ignored(app): with app.app_context(): supplier = _company("TEST_UI_C4_COMPLETE_ROLE_SUPPLIER") contract = create_contract(supplier_id=supplier.id, name="TEST_UI_C4_COMPLETE_ROLE", start_date=date(2026, 1, 1), end_date=date(2026, 12, 31), anniversary_month=1, anniversary_day=1) period = contract.periods[0] machine = _equipment("TEST_UI_C4_COMPLETE_ROLE_MACHINE") assign_equipment_to_contract(contract=contract, equipment=machine, entry_date=period.start_date) _meter(machine, "Heures", "NONE") usage = calculate_contract_usage(period) assert usage["bw"] == 0 and usage["color"] == 0 def test_c4_complete_close_contract_truncates_current_and_future_history(app): with app.app_context(): supplier = _company("TEST_UI_C4_COMPLETE_CLOSE_SUPPLIER") contract = create_contract(supplier_id=supplier.id, name="TEST_UI_C4_COMPLETE_CLOSE", start_date=date(2026, 10, 1), end_date=date(2031, 9, 30), anniversary_month=10, anniversary_day=1) machine = _equipment("TEST_UI_C4_COMPLETE_CLOSE_MACHINE") assignment = assign_equipment_to_contract(contract=contract, equipment=machine, entry_date=date(2026, 10, 1)) close_photocopier_contract(contract=contract, closed_at=date(2028, 3, 15)) assert contract.status == "CLOSED" and contract.end_date == date(2028, 3, 15) assert assignment.exit_date == date(2028, 3, 15) assert contract.periods[1].status == "CLOSED" assert contract.periods[2].status == "CANCELLED" assert contract.periods[2].start_date == date(2028, 10, 1) def test_c4_complete_new_contract_after_close_preserves_meter_identity(app): with app.app_context(): supplier = _company("TEST_UI_C4_COMPLETE_RESET_SUPPLIER") old = create_contract(supplier_id=supplier.id, name="TEST_UI_C4_COMPLETE_OLD", start_date=date(2026, 1, 1), end_date=date(2026, 12, 31), anniversary_month=1, anniversary_day=1) machine = _equipment("TEST_UI_C4_COMPLETE_RESET_MACHINE") assignment = assign_equipment_to_contract(contract=old, equipment=machine, entry_date=date(2026, 1, 1)) meter = _meter(machine, "N&B", "COPIER_BW") _reading(meter, 100, date(2026, 1, 1)) close_photocopier_contract(contract=old, closed_at=date(2026, 9, 30)) new = create_contract(supplier_id=supplier.id, name="TEST_UI_C4_COMPLETE_NEW", start_date=date(2026, 10, 1), end_date=date(2027, 9, 30), anniversary_month=10, anniversary_day=1) new_assignment = assign_equipment_to_contract(contract=new, equipment=machine, entry_date=date(2026, 10, 1)) assert assignment.exit_date == date(2026, 9, 30) assert new_assignment.entry_date == date(2026, 10, 1) assert meter.id and meter.current_value == 0 def test_c4_complete_daily_view_alerts_are_open_and_permission_filtered(app, admin_user): with app.app_context(): quota = Alert(alert_type="QUOTA_OVERAGE_PROJECTED", title="Quota", message="Quota à examiner", is_resolved=False) stock = Alert(alert_type="CONSUMABLE_STOCK_LOW", title="Stock", message="Stock à examiner", is_resolved=False) closed = Alert(alert_type="CONSUMABLE_STOCK_LOW", title="Ancienne", message="Ne pas afficher", is_resolved=True) db.session.add_all([quota, stock, closed]) db.session.commit() rows = visible_open_c4_alerts(db.session.get(User, admin_user["id"])) assert {row.alert_type for row in rows} >= {"QUOTA_OVERAGE_PROJECTED", "CONSUMABLE_STOCK_LOW"} assert all(row.is_resolved is False for row in rows) def test_c4_complete_cancel_partial_order_keeps_stock_and_aggregates_open_orders(app, admin_user): with app.app_context(): consumable = Consumable(name="TEST_UI_C4_COMPLETE_ORDER", quantity=1, min_quantity=2, target_quantity=8) db.session.add(consumable) db.session.commit() first = create_consumable_order(consumable=consumable, quantity=5, user_id=admin_user["id"]) second = create_consumable_order(consumable=consumable, quantity=4, user_id=admin_user["id"]) receive_consumable_order(order=first, quantity=3, user_id=admin_user["id"]) assert first.remaining_quantity == 2 and second.remaining_quantity == 4 cancel_consumable_order(order=first, user_id=admin_user["id"], reason="TEST_UI_C4_COMPLETE_CANCEL") assert first.status == "CANCELLED" assert consumable.quantity == 4 # stock reçu conservé assert second.remaining_quantity == 4 with pytest.raises(C4DomainError): cancel_consumable_order(order=first, user_id=admin_user["id"], reason="second appel") def test_c4_complete_stock_coverage_uses_active_compatible_machines(app): with app.app_context(): consumable = Consumable(name="TEST_UI_C4_COMPLETE_COVERAGE", quantity=3) a, b, retired = (_equipment("TEST_UI_C4_COMPLETE_COVERAGE_A", "printer"), _equipment("TEST_UI_C4_COMPLETE_COVERAGE_B", "printer"), _equipment("TEST_UI_C4_COMPLETE_COVERAGE_RETIRED", "printer")) retired.lifecycle_status = "retired" db.session.add(consumable) db.session.flush() db.session.add_all([EquipmentConsumable(equipment=item, consumable=consumable) for item in (a, b, retired)]) base = datetime(2026, 1, 1, tzinfo=timezone.utc) for equipment, offsets in ((a, (0, 30, 60)), (b, (0, 60, 120))): for offset in offsets: db.session.add(ConsumableUsage(consumable=consumable, equipment=equipment, quantity_used=1, used_at=base + timedelta(days=offset))) db.session.commit() coverage = estimate_consumable_stock_coverage(consumable=consumable) assert coverage["machines_total"] == 2 assert coverage["machines_with_history"] == 2 assert coverage["quality"] == "ESTIMATED" assert round(coverage["coverage_days"]) == 60 ConsumableUsage.query.update({ConsumableUsage.used_by_id: None}, synchronize_session=False) db.session.commit()