from datetime import date, datetime, time, timezone import re from app_new.extensions import db from app_new.core.models import ( CollegeClosure, Company, ContractEquipmentAssignment, Equipment, Meter, MeterReading, MeterReadingOccurrence, User, WorkSchedule, ) from app_new.core.models.planning import MeterReadingSchedule from app_new.core.services.c4_service import ( _school_days, calculate_contract_usage, classify_legacy_copier_meter, create_contract, ) from app_new.core.services.meter_service import create_meter from app_new.core.services.meter_reading_planning import ( create_or_update_schedule, generate_occurrences, record_occurrence_reading, ) def _supplier(name): row = Company(name=name) db.session.add(row) db.session.flush() return row def _machine(name): row = Equipment(name=name, device_type="photocopier", serial_number=f"TEST_UI_C4_FINAL_{name}") db.session.add(row) db.session.flush() return row def _meter(machine, name, role="COPIER_BW", current=0): row = Meter(equipment=machine, name=name, meter_type="photocopieur", unit="pages", contract_role=role, current_value=current) 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, time.min, tzinfo=timezone.utc), ) db.session.add(row) db.session.flush() return row def test_c4_final_c2_anticipated_reading_is_exact_contract_reference(app, admin_user): with app.app_context(): supplier = _supplier("TEST_UI_C4_FINAL_C2_SUPPLIER") contract = create_contract( supplier_id=supplier.id, name="TEST_UI_C4_FINAL_C2_CONTRACT", start_date=date(2028, 10, 1), end_date=date(2029, 9, 30), anniversary_month=10, anniversary_day=1, ) period = contract.periods[0] machine = _machine("TEST_UI_C4_FINAL_C2_MACHINE") db.session.add(ContractEquipmentAssignment( contract=contract, equipment=machine, entry_date=period.start_date, )) meter = _meter(machine, "N&B", current=1000) _reading(meter, 900, date(2028, 9, 1)) _reading(meter, 1000, date(2028, 9, 15)) schedule = create_or_update_schedule( meter=meter, frequency="FIXED_ANNUAL_DATE", reference_date=date(2028, 10, 1), assigned_to_id=admin_user["id"], fixed_month=10, fixed_day=1, commit=False, ) # Le calendrier C2 connaît le vendredi comme journée travaillée. db.session.add(WorkSchedule( user_id=admin_user["id"], day_of_week=4, start_time=time(8), end_time=time(17), is_active=True, )) db.session.flush() generate_occurrences( start_date=date(2028, 10, 1), end_date=date(2028, 10, 1), schedule_ids=[schedule.id], ) occurrence = MeterReadingOccurrence.query.filter_by( schedule_id=schedule.id, target_date=date(2028, 10, 1), ).one() assert occurrence.operational_date == date(2028, 9, 29) processed = record_occurrence_reading( occurrence_id=occurrence.id, value=1100, user_id=admin_user["id"], reading_date=datetime(2028, 9, 29, tzinfo=timezone.utc), ) db.session.expire_all() assert processed.meter_reading.reading_date.date() == date(2028, 9, 29) assert processed.target_date == date(2028, 10, 1) _reading(meter, 1200, date(2028, 10, 15)) usage = calculate_contract_usage(period) assert usage["quality"] == "EXACT" assert usage["bw"] == 100 def test_c4_final_early_spontaneous_reading_is_not_exact(app): with app.app_context(): supplier = _supplier("TEST_UI_C4_FINAL_SPONTANEOUS_SUPPLIER") contract = create_contract( supplier_id=supplier.id, name="TEST_UI_C4_FINAL_SPONTANEOUS", start_date=date(2028, 10, 1), end_date=date(2029, 9, 30), anniversary_month=10, anniversary_day=1, ) period = contract.periods[0] machine = _machine("TEST_UI_C4_FINAL_SPONTANEOUS_MACHINE") db.session.add(ContractEquipmentAssignment( contract=contract, equipment=machine, entry_date=period.start_date, )) meter = _meter(machine, "N&B", current=1000) _reading(meter, 900, date(2028, 9, 1)) _reading(meter, 1000, date(2028, 9, 15)) _reading(meter, 1100, date(2028, 9, 29)) _reading(meter, 1200, date(2028, 10, 15)) usage = calculate_contract_usage(period) assert usage["quality"] in {"ESTIMATED", "PARTIAL"} def test_c4_final_legacy_roles_are_safe_and_explicit(app): with app.app_context(): machine = _machine("TEST_UI_C4_FINAL_LEGACY_MACHINE") bw = _meter(machine, "N&B", role="NONE") color = _meter(machine, "Couleur", role="NONE") ambiguous = _meter(machine, "Heures fonctionnement", role="NONE") assert classify_legacy_copier_meter(bw, commit=False) == "COPIER_BW" assert classify_legacy_copier_meter(color, commit=False) == "COPIER_COLOR" assert classify_legacy_copier_meter(ambiguous, commit=False) == "NONE" assert bw.contract_role == "COPIER_BW" assert color.contract_role == "COPIER_COLOR" assert ambiguous.contract_role == "NONE" def test_c4_final_meter_contract_role_can_be_edited_http(app, admin_user, authenticated_client): with app.app_context(): machine = _machine("TEST_UI_C4_FINAL_HTTP_MACHINE") meter = create_meter(name="Compteur historique", meter_type="photocopieur", unit="pages", contract_role="NONE", equipment_id=machine.id) meter_id = meter.id machine_id = machine.id db.session.commit() assert db.session.get(Meter, meter_id) is not None with app.app_context(): token_page = authenticated_client.get(f'/meters/{meter_id}/edit') assert token_page.status_code == 200 token = re.search(r'name="csrf_token" value="([^"]+)"', token_page.get_data(as_text=True)).group(1) response = authenticated_client.post( f"/meters/{meter_id}/edit", data={ "name": "Compteur historique", "meter_type": "photocopieur", "unit": "pages", "scope_type": "equipment", "scope_id": machine_id, "contract_role": "COPIER_BW", "parent_id": "", "csrf_token": token, }, follow_redirects=False, ) assert response.status_code == 302, response.get_data(as_text=True) assert db.session.get(Meter, meter_id).contract_role == "COPIER_BW" response = authenticated_client.post( f"/meters/{meter_id}/edit", data={ "name": "Compteur historique", "meter_type": "photocopieur", "unit": "pages", "scope_type": "equipment", "scope_id": machine_id, "contract_role": "NONE", "parent_id": "", "csrf_token": token, }, follow_redirects=False, ) assert response.status_code == 302 assert db.session.get(Meter, meter_id).contract_role == "NONE" def test_c4_final_copier_estimation_uses_school_days_not_calendar_days(app): with app.app_context(): supplier = _supplier("TEST_UI_C4_FINAL_SCHOOL_SUPPLIER") contract = create_contract( supplier_id=supplier.id, name="TEST_UI_C4_FINAL_SCHOOL", start_date=date(2026, 2, 9), end_date=date(2026, 12, 31), anniversary_month=2, anniversary_day=1, ) period = contract.periods[0] machine = _machine("TEST_UI_C4_FINAL_SCHOOL_MACHINE") db.session.add(ContractEquipmentAssignment( contract=contract, equipment=machine, entry_date=period.start_date, )) meter = _meter(machine, "N&B", current=2600) # 30 jours calendaires, dont 7 jours de fermeture en semaine : # 1 500 pages / 15 jours scolaires = 100 pages/jour scolaire. db.session.add(CollegeClosure( name="TEST_UI_C4_FINAL_VACANCES", start_date=date(2026, 1, 5), end_date=date(2026, 1, 13), closure_type="vacances", work_hours_type="none", )) db.session.flush() assert _school_days(date(2026, 1, 1), date(2026, 1, 30)) == 15 assert _school_days(date(2026, 2, 9), date(2026, 2, 14)) == 5 _reading(meter, 1000, date(2026, 1, 1)) _reading(meter, 2500, date(2026, 1, 31)) _reading(meter, 2600, date(2026, 2, 12)) usage = calculate_contract_usage(period) # 3 jours scolaires manquants en février : 300 pages estimées. assert usage["quality"] == "ESTIMATED" assert 290 <= usage["bw"] <= 310