From f2538895ae10963a3e7393acffa475394e7fa7a1 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 23 Aug 2026 22:55:21 +0000 Subject: [PATCH] test(meters): validate checkpoint C1 workflows --- tests/integration/test_meter_checkpoint_c1.py | 86 +++++++++++++++++++ 1 file changed, 86 insertions(+) create mode 100644 tests/integration/test_meter_checkpoint_c1.py diff --git a/tests/integration/test_meter_checkpoint_c1.py b/tests/integration/test_meter_checkpoint_c1.py new file mode 100644 index 0000000..51aaa19 --- /dev/null +++ b/tests/integration/test_meter_checkpoint_c1.py @@ -0,0 +1,86 @@ +from datetime import datetime, timezone + +import pytest + +from app_new import db +from app_new.core.models.college import Building, HousingUnit, Room, Zone +from app_new.core.models.equipment import Equipment, EquipmentCategory +from app_new.core.models.planning import Meter, MeterReading, MeterReadingCorrection +from app_new.core.services.meter_service import ( + MeterDomainError, correct_meter_reading, create_meter, record_meter_reading, + replace_meter, +) + + +def _equipment(): + category = EquipmentCategory(name="TEST_UI_C1_CATEGORY") + building = Building(name="TEST_UI_C1_BUILDING") + room = Room(name="TEST_UI_C1_ROOM", code="TEST_UI_C1_ROOM", building=building) + equipment = Equipment(name="TEST_UI_C1_EQUIPMENT", category=category, room=room) + db.session.add_all([category, building, room, equipment]) + db.session.flush() + return equipment, building, room + + +def test_meter_supports_all_patrimony_scopes_and_multiple_roots(app): + with app.app_context(): + equipment, building, room = _equipment() + zone = Zone(name="TEST_UI_C1_ZONE", building=building) + housing = HousingUnit(name="TEST_UI_C1_HOUSING", building=building) + db.session.add_all([zone, housing]) + db.session.flush() + meters = [ + create_meter(name="Equipment", equipment_id=equipment.id, commit=False), + create_meter(name="Building", building_id=building.id, commit=False), + create_meter(name="Zone", zone_id=zone.id, commit=False), + create_meter(name="Room", room_id=room.id, commit=False), + create_meter(name="Housing", housing_unit_id=housing.id, commit=False), + create_meter(name="Other root", building_id=building.id, commit=False), + ] + db.session.commit() + assert {meter.building_id or meter.zone_id or meter.room_id or meter.housing_unit_id or meter.equipment_id for meter in meters} + assert all(Meter.query.get(meter.id) is not None for meter in meters) + + +def test_meter_hierarchy_rejects_self_and_deep_cycles(app): + with app.app_context(): + _, building, _ = _equipment() + root = create_meter(name="Root", building_id=building.id) + child = create_meter(name="Child", building_id=building.id, parent=root) + grandchild = create_meter(name="Grandchild", building_id=building.id, parent=child) + with pytest.raises(MeterDomainError): + root.parent = grandchild + from app_new.core.services.meter_service import validate_meter_scope + validate_meter_scope(root) + with pytest.raises(MeterDomainError): + root.parent = root + from app_new.core.services.meter_service import validate_meter_scope + validate_meter_scope(root) + root.parent = None + + +def test_reading_reset_correction_and_current_value_are_consistent(app, admin_user): + with app.app_context(): + _, building, _ = _equipment() + meter = create_meter(name="Water", meter_type="eau", unit="m³", building_id=building.id) + first = record_meter_reading(meter=meter, value=100, user_id=admin_user["id"], reading_date=datetime(2026, 1, 1, tzinfo=timezone.utc)) + with pytest.raises(MeterDomainError): + record_meter_reading(meter=meter, value=90, user_id=admin_user["id"]) + reset = record_meter_reading(meter=meter, value=10, user_id=admin_user["id"], is_reset=True, reset_reason="TEST_UI_C1 remise à zéro") + correction = correct_meter_reading(reading=first, new_value=101, user_id=admin_user["id"], reason="TEST_UI_C1 correction") + assert reset.is_reset is True + assert correction.value == 101 + assert meter.current_value == 10 + assert MeterReadingCorrection.query.filter_by(reading_id=first.id).one().old_value == 100 + + +def test_replacement_keeps_old_history_and_links_new_meter(app, admin_user): + with app.app_context(): + _, building, _ = _equipment() + old = create_meter(name="Old", building_id=building.id) + record_meter_reading(meter=old, value=12, user_id=admin_user["id"]) + new = replace_meter(old_meter=old, name="New", user_id=admin_user["id"]) + assert old.status == "replaced" + assert old.is_active is False + assert new.replaces_meter_id == old.id + assert MeterReading.query.filter_by(meter_id=old.id).count() == 1