201 lines
8.7 KiB
Python
201 lines
8.7 KiB
Python
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
|