230 lines
12 KiB
Python
230 lines
12 KiB
Python
|
from datetime import date, time
|
||
|
|
from time import perf_counter
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from sqlalchemy import event
|
||
|
|
|
||
|
|
from app_new import db
|
||
|
|
from app_new.core.models.college import Building, HousingUnit
|
||
|
|
from app_new.core.models.planning import (
|
||
|
|
Meter, MeterReading, MeterReadingOccurrence, MeterReadingRoundOccurrence,
|
||
|
|
MeterReadingSchedule, WorkSchedule,
|
||
|
|
)
|
||
|
|
from app_new.core.services.meter_reading_planning import (
|
||
|
|
CANCELLED, TREATED_READING, TREATED_WITHOUT_READING, add_round_member,
|
||
|
|
create_or_update_schedule, create_round, generate_occurrences,
|
||
|
|
operational_date_for, record_occurrence_reading, record_occurrence_without_reading,
|
||
|
|
)
|
||
|
|
from app_new.core.services.meter_service import create_meter, replace_meter
|
||
|
|
|
||
|
|
|
||
|
|
def _meter(name, building=None, housing=None):
|
||
|
|
if building is None:
|
||
|
|
building = Building(name=f"TEST_UI_C2_{name}_BUILDING")
|
||
|
|
db.session.add(building)
|
||
|
|
db.session.flush()
|
||
|
|
return create_meter(name=f"TEST_UI_C2_{name}", building_id=building.id if housing is None else None, housing_unit_id=housing.id if housing else None)
|
||
|
|
|
||
|
|
|
||
|
|
def _schedule(meter, admin_user, frequency, reference, **kwargs):
|
||
|
|
return create_or_update_schedule(
|
||
|
|
meter=meter, frequency=frequency, reference_date=reference,
|
||
|
|
assigned_to_id=admin_user["id"], **kwargs,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_c2_frequencies_and_generation_are_idempotent(app, admin_user):
|
||
|
|
with app.app_context():
|
||
|
|
frequencies = {
|
||
|
|
"WEEKLY": date(2026, 10, 5), "MONTHLY": date(2026, 10, 1),
|
||
|
|
"SEMI_ANNUAL": date(2026, 10, 1), "ANNUAL": date(2026, 10, 1),
|
||
|
|
"FIXED_ANNUAL_DATE": date(2026, 10, 1),
|
||
|
|
}
|
||
|
|
schedules = []
|
||
|
|
for frequency, reference in frequencies.items():
|
||
|
|
meter = _meter(f"FREQ_{frequency}")
|
||
|
|
schedules.append(_schedule(meter, admin_user, frequency, reference, fixed_month=10, fixed_day=1))
|
||
|
|
db.session.commit()
|
||
|
|
first = generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2027, 10, 10))
|
||
|
|
counts = {schedule.id: MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).count() for schedule in schedules}
|
||
|
|
second = generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2027, 10, 10))
|
||
|
|
assert first
|
||
|
|
assert second == []
|
||
|
|
assert all(MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).count() == counts[schedule.id] for schedule in schedules)
|
||
|
|
assert counts[schedules[0].id] == 53
|
||
|
|
assert counts[schedules[1].id] == 13
|
||
|
|
assert counts[schedules[2].id] == 3
|
||
|
|
assert counts[schedules[3].id] == 2
|
||
|
|
assert counts[schedules[4].id] == 2
|
||
|
|
|
||
|
|
|
||
|
|
def test_c2_target_and_operational_dates_and_housing_calendar(app, admin_user):
|
||
|
|
with app.app_context():
|
||
|
|
building = Building(name="TEST_UI_C2_CALENDAR_BUILDING")
|
||
|
|
housing = HousingUnit(name="TEST_UI_C2_HOUSING", building=building)
|
||
|
|
db.session.add_all([building, housing])
|
||
|
|
db.session.flush()
|
||
|
|
school_meter = _meter("SCHOOL_DATE", building=building)
|
||
|
|
schedule = _schedule(school_meter, admin_user, "FIXED_ANNUAL_DATE", date(2026, 10, 3))
|
||
|
|
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(2026, 10, 3), end_date=date(2026, 10, 3), schedule_ids=[schedule.id])
|
||
|
|
occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).one()
|
||
|
|
assert occurrence.target_date == date(2026, 10, 3)
|
||
|
|
assert occurrence.operational_date == date(2026, 10, 2)
|
||
|
|
|
||
|
|
housing_meter = _meter("HOUSING_DATE", housing=housing)
|
||
|
|
housing_schedule = _schedule(housing_meter, admin_user, "ANNUAL", date(2026, 10, 3))
|
||
|
|
generate_occurrences(start_date=date(2026, 10, 3), end_date=date(2026, 10, 3), schedule_ids=[housing_schedule.id])
|
||
|
|
housing_occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=housing_schedule.id).one()
|
||
|
|
assert housing_occurrence.operational_date == date(2026, 10, 3)
|
||
|
|
|
||
|
|
|
||
|
|
def test_c2_processing_with_reading_and_without_reading_never_creates_zero(app, admin_user):
|
||
|
|
with app.app_context():
|
||
|
|
meter = _meter("PROCESSING")
|
||
|
|
schedule = _schedule(meter, admin_user, "MONTHLY", date(2026, 10, 1))
|
||
|
|
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2026, 10, 1), schedule_ids=[schedule.id])
|
||
|
|
occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).one()
|
||
|
|
processed = record_occurrence_reading(occurrence_id=occurrence.id, value=42, user_id=admin_user["id"])
|
||
|
|
assert processed.status == TREATED_READING
|
||
|
|
assert MeterReading.query.filter_by(meter_id=meter.id).count() == 1
|
||
|
|
|
||
|
|
other = _meter("WITHOUT")
|
||
|
|
other_schedule = _schedule(other, admin_user, "MONTHLY", date(2026, 10, 1))
|
||
|
|
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2026, 10, 1), schedule_ids=[other_schedule.id])
|
||
|
|
other_occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=other_schedule.id).one()
|
||
|
|
without = record_occurrence_without_reading(
|
||
|
|
occurrence_id=other_occurrence.id, reason="OCCUPANT_REFUSED", user_id=admin_user["id"], comment="TEST_UI_C2 refus"
|
||
|
|
)
|
||
|
|
assert without.status == TREATED_WITHOUT_READING
|
||
|
|
assert without.comment == "TEST_UI_C2 refus"
|
||
|
|
assert MeterReading.query.filter_by(meter_id=other.id).count() == 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_c2_http_occurrence_actions_use_shared_service(app, admin_user, authenticated_client):
|
||
|
|
with app.app_context():
|
||
|
|
meter = _meter("HTTP_READING")
|
||
|
|
schedule = _schedule(meter, admin_user, "MONTHLY", date(2026, 10, 1))
|
||
|
|
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2026, 10, 1), schedule_ids=[schedule.id])
|
||
|
|
occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).one()
|
||
|
|
|
||
|
|
response = authenticated_client.post(
|
||
|
|
f"/planning/meter-occurrences/{occurrence.id}/reading",
|
||
|
|
data={"value": "123.5", "notes": "TEST_UI_C2_HTTP"},
|
||
|
|
)
|
||
|
|
assert response.status_code == 302
|
||
|
|
db.session.refresh(occurrence)
|
||
|
|
assert occurrence.status == TREATED_READING
|
||
|
|
assert MeterReading.query.filter_by(meter_id=meter.id).count() == 1
|
||
|
|
|
||
|
|
other = _meter("HTTP_WITHOUT")
|
||
|
|
other_schedule = _schedule(other, admin_user, "MONTHLY", date(2026, 10, 1))
|
||
|
|
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2026, 10, 1), schedule_ids=[other_schedule.id])
|
||
|
|
other_occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=other_schedule.id).one()
|
||
|
|
response = authenticated_client.post(
|
||
|
|
f"/planning/meter-occurrences/{other_occurrence.id}/without-reading",
|
||
|
|
data={"reason": "READING_UNKNOWN", "comment": "TEST_UI_C2_HTTP inconnu"},
|
||
|
|
)
|
||
|
|
assert response.status_code == 302
|
||
|
|
db.session.refresh(other_occurrence)
|
||
|
|
assert other_occurrence.status == TREATED_WITHOUT_READING
|
||
|
|
assert other_occurrence.no_reading_reason == "READING_UNKNOWN"
|
||
|
|
assert MeterReading.query.filter_by(meter_id=other.id).count() == 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_c2_open_occurrence_stays_visible_as_overdue_in_my_day(app, admin_user, authenticated_client):
|
||
|
|
with app.app_context():
|
||
|
|
target = date(2026, 10, 5)
|
||
|
|
meter = _meter("MY_DAY")
|
||
|
|
_schedule(meter, admin_user, "WEEKLY", date(2026, 9, 28), duration_minutes=5)
|
||
|
|
db.session.add(WorkSchedule(user_id=admin_user["id"], day_of_week=0, start_time=time(8), end_time=time(17), is_active=True))
|
||
|
|
db.session.commit()
|
||
|
|
response = authenticated_client.get(f"/planning/my-day?date={target.isoformat()}")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert b"Relev\xc3\xa9 compteur" in response.data
|
||
|
|
assert b"Saisir le relev\xc3\xa9" in response.data
|
||
|
|
|
||
|
|
|
||
|
|
def test_c2_replaced_meter_cancels_open_occurrences(app, admin_user):
|
||
|
|
with app.app_context():
|
||
|
|
meter = _meter("REPLACED")
|
||
|
|
schedule = _schedule(meter, admin_user, "MONTHLY", date(2026, 10, 1))
|
||
|
|
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2026, 10, 1), schedule_ids=[schedule.id])
|
||
|
|
old_occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).one()
|
||
|
|
replace_meter(old_meter=meter, name="TEST_UI_C2_REPLACEMENT", user_id=admin_user["id"])
|
||
|
|
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2027, 1, 1))
|
||
|
|
db.session.refresh(old_occurrence)
|
||
|
|
assert old_occurrence.status == CANCELLED
|
||
|
|
assert MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id, status="À FAIRE").count() == 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_c2_round_order_progress_and_resume(app, admin_user, authenticated_client):
|
||
|
|
with app.app_context():
|
||
|
|
first = _meter("ROUND_1")
|
||
|
|
second = _meter("ROUND_2")
|
||
|
|
first_schedule = _schedule(first, admin_user, "MONTHLY", date(2026, 10, 1))
|
||
|
|
second_schedule = _schedule(second, admin_user, "MONTHLY", date(2026, 10, 1))
|
||
|
|
round_ = create_round(name="TEST_UI_C2_TOURNEE", default_assigned_to_id=admin_user["id"])
|
||
|
|
add_round_member(round_=round_, schedule=first_schedule, position=1, commit=False)
|
||
|
|
add_round_member(round_=round_, schedule=second_schedule, position=2, commit=False)
|
||
|
|
db.session.commit()
|
||
|
|
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2026, 10, 1))
|
||
|
|
round_occurrence = MeterReadingRoundOccurrence.query.filter_by(round_id=round_.id, operational_date=date(2026, 10, 1)).one()
|
||
|
|
assert [item.schedule.meter.name for item in sorted(round_occurrence.occurrences, key=lambda item: item.schedule.meter.name)]
|
||
|
|
response = authenticated_client.get(f"/planning/meter-rounds/{round_.id}?date=2026-10-01")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert b"TEST_UI_C2_TOURNEE" in response.data
|
||
|
|
assert b"0 / 2" in response.data
|
||
|
|
record_occurrence_without_reading(
|
||
|
|
occurrence_id=round_occurrence.occurrences[0].id, reason="READING_UNKNOWN", user_id=admin_user["id"]
|
||
|
|
)
|
||
|
|
db.session.refresh(round_occurrence)
|
||
|
|
assert round_occurrence.status == "EN COURS"
|
||
|
|
|
||
|
|
|
||
|
|
def test_c2_permissions_keep_configuration_admin_only(app, client):
|
||
|
|
response = client.get('/planning/meter-schedules/new')
|
||
|
|
assert response.status_code in (302, 403)
|
||
|
|
|
||
|
|
|
||
|
|
def test_c2_planning_sql_smoke_is_measured(app, admin_user, authenticated_client):
|
||
|
|
with app.app_context():
|
||
|
|
schedules = []
|
||
|
|
for index in range(3):
|
||
|
|
meter = _meter(f"PERF_{index}")
|
||
|
|
schedules.append(_schedule(meter, admin_user, "MONTHLY", date(2026, 10, 1)))
|
||
|
|
round_ = create_round(name="TEST_UI_C2_PERF_ROUND", default_assigned_to_id=admin_user["id"])
|
||
|
|
for position, schedule in enumerate(schedules, 1):
|
||
|
|
add_round_member(round_=round_, schedule=schedule, position=position, commit=False)
|
||
|
|
db.session.commit()
|
||
|
|
|
||
|
|
query_count = 0
|
||
|
|
|
||
|
|
def count_queries(*_args, **_kwargs):
|
||
|
|
nonlocal query_count
|
||
|
|
query_count += 1
|
||
|
|
|
||
|
|
event.listen(db.engine, "before_cursor_execute", count_queries)
|
||
|
|
try:
|
||
|
|
start = perf_counter()
|
||
|
|
daily = authenticated_client.get("/planning/my-day?date=2026-10-01")
|
||
|
|
daily_seconds = perf_counter() - start
|
||
|
|
daily_queries = query_count
|
||
|
|
query_count = 0
|
||
|
|
start = perf_counter()
|
||
|
|
round_response = authenticated_client.get(
|
||
|
|
f"/planning/meter-rounds/{round_.id}?date=2026-10-01"
|
||
|
|
)
|
||
|
|
round_seconds = perf_counter() - start
|
||
|
|
round_queries = query_count
|
||
|
|
finally:
|
||
|
|
event.remove(db.engine, "before_cursor_execute", count_queries)
|
||
|
|
|
||
|
|
print(
|
||
|
|
f"C2_PERF my_day_seconds={daily_seconds:.4f} my_day_queries={daily_queries} "
|
||
|
|
f"round_seconds={round_seconds:.4f} round_queries={round_queries}"
|
||
|
|
)
|
||
|
|
assert daily.status_code == 200
|
||
|
|
assert round_response.status_code == 200
|