from datetime import date, datetime, timedelta, timezone from time import perf_counter 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, project_contract_usage, 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 def test_c4_performance_paths(authenticated_client, app): """Mesure les chemins C4 sans confondre le coût d'un GET et d'un calcul métier.""" with app.app_context(): supplier = _company("TEST_UI_C4_PERF_SUPPLIER") contract = create_contract(supplier_id=supplier.id, name="TEST_UI_C4_PERF_CONTRACT", start_date=date(2026, 1, 1), end_date=date(2027, 12, 31), anniversary_month=1, anniversary_day=1) period = contract.periods[0] period.quota_bw = 10000; period.quota_color = 5000 machines = [_equipment(f"TEST_UI_C4_PERF_MACHINE_{i}") for i in range(5)] for machine in machines: assign_equipment_to_contract(contract=contract, equipment=machine, entry_date=period.start_date, commit=False) bw = _meter(machine, f"N&B {machine.name}", "N&B") color = _meter(machine, f"Couleur {machine.name}", "Couleur") for index in range(1, 8): _reading(bw, index * 100, date(2026, index, 1)) _reading(color, index * 40, date(2026, index, 1)) consumable = Consumable(name="TEST_UI_C4_PERF_TONER", quantity=10, min_quantity=2) db.session.add(consumable); db.session.flush() for machine in machines: db.session.add(EquipmentConsumable(equipment=machine, consumable=consumable)) db.session.commit() from sqlalchemy import event counts = {"n": 0} def count(_conn, _cursor, _statement, _params, _ctx, _executemany): counts["n"] += 1 event.listen(db.engine, "before_cursor_execute", count) started = perf_counter(); usage = calculate_contract_usage(period); projection = project_contract_usage(period); service_seconds = perf_counter() - started service_sql = counts["n"] counts["n"] = 0 started = perf_counter(); response = authenticated_client.get(f"/c4/contracts/{contract.id}"); contract_seconds = perf_counter() - started; contract_sql = counts["n"] counts["n"] = 0 started = perf_counter(); response_consumables = authenticated_client.get(f"/c4/consumables/{consumable.id}"); consumable_seconds = perf_counter() - started; consumable_sql = counts["n"] event.remove(db.engine, "before_cursor_execute", count) assert response.status_code == 200 and response_consumables.status_code == 200 assert usage["machines"] == 5 and projection["usage"]["bw"] > 0 print(f"C4_PERF service_seconds={service_seconds:.4f} service_queries={service_sql} contract_seconds={contract_seconds:.4f} contract_queries={contract_sql} consumable_seconds={consumable_seconds:.4f} consumable_queries={consumable_sql}")