test(meters): validate checkpoint C1 workflows

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