test(planning): cover C2 fix workflows

tests/integration/test_meter_checkpoint_c2.py

@@ -10,10 +10,14 @@ from app_new.core.models.planning import (
     Meter, MeterReading, MeterReadingOccurrence, MeterReadingRoundOccurrence,
     MeterReadingSchedule, WorkSchedule,
 )
+from app_new.core.models.user import User
+from app_new.core.models.rbac import Role, UserRole
 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,
+    create_or_update_schedule, create_round,
+    generate_occurrences,
     operational_date_for, record_occurrence_reading, record_occurrence_without_reading,
+    MeterDomainError,
 )
 from app_new.core.services.meter_service import create_meter, replace_meter
 
@@ -33,6 +37,18 @@ def _schedule(meter, admin_user, frequency, reference, **kwargs):
     )
 
 
+def _test_user(suffix):
+    username = f"TEST_UI_C2_FIX_{suffix}"
+    user = User(username=username, email=f"{username.lower()}@gmao.local", full_name=username, role="technicien", is_active=True)
+    user.set_password("test-c2-fix-password")
+    db.session.add(user)
+    db.session.flush()
+    role = Role.query.filter_by(slug="technicien", is_active=True).first()
+    db.session.add(UserRole(user_id=user.id, role_id=role.id))
+    db.session.flush()
+    return user
+
+
 def test_c2_frequencies_and_generation_are_idempotent(app, admin_user):
     with app.app_context():
         frequencies = {
@@ -181,6 +197,106 @@ def test_c2_round_order_progress_and_resume(app, admin_user, authenticated_clien
         )
         db.session.refresh(round_occurrence)
         assert round_occurrence.status == "EN COURS"
+        record_occurrence_reading(
+            occurrence_id=round_occurrence.occurrences[1].id, value=10, user_id=admin_user["id"]
+        )
+        db.session.refresh(round_occurrence)
+        assert round_occurrence.status == "TERMINÉE"
+        db.session.expire_all()
+        response = authenticated_client.get(f"/planning/meter-rounds/{round_.id}?date=2026-10-01")
+        assert response.status_code == 200
+        assert b"2 / 2" in response.data
+        assert response.data.index(b"ROUND_1") < response.data.index(b"ROUND_2")
+
+
+def test_c2_fix_direct_round_generates_future_target_on_operational_date(app, admin_user, authenticated_client):
+    with app.app_context():
+        meter = _meter("DIRECT_OPERATIONAL_DATE")
+        schedule = create_or_update_schedule(
+            meter=meter, frequency="FIXED_ANNUAL_DATE", reference_date=date(2023, 10, 1),
+            fixed_month=10, fixed_day=1, assigned_to_id=None,
+        )
+        db.session.add(WorkSchedule(user_id=admin_user["id"], day_of_week=4, start_time=time(8), end_time=time(17), is_active=True))
+        round_ = create_round(name="TEST_UI_C2_FIX_DIRECT_ROUND", default_assigned_to_id=admin_user["id"])
+        add_round_member(round_=round_, schedule=schedule, position=1)
+        db.session.commit()
+
+        # Aucun appel préalable à Ma journée ni génération couvrant le 01/10.
+        assert MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).count() == 0
+        response = authenticated_client.get(f"/planning/meter-rounds/{round_.id}?date=2023-09-29")
+        assert response.status_code == 200
+        assert b"TEST_UI_C2_DIRECT_OPERATIONAL_DATE" in response.data
+        occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).one()
+        assert occurrence.target_date == date(2023, 10, 1)
+        assert occurrence.operational_date == date(2023, 9, 29)
+        assert occurrence.round_occurrence.round_id == round_.id
+        assert occurrence.assigned_to_id == admin_user["id"]
+        second_response = authenticated_client.get(f"/planning/meter-rounds/{round_.id}?date=2023-09-29")
+        assert second_response.status_code == 200
+        assert MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).count() == 1
+
+
+def test_c2_fix_schedule_assignment_precedes_round_default(app, admin_user):
+    with app.app_context():
+        individual = _test_user("INDIVIDUAL")
+        meter = _meter("ASSIGNMENT_PRIORITY")
+        schedule = create_or_update_schedule(
+            meter=meter, frequency="MONTHLY", reference_date=date(2026, 10, 1),
+            assigned_to_id=individual.id,
+        )
+        round_ = create_round(name="TEST_UI_C2_FIX_PRIORITY", default_assigned_to_id=admin_user["id"])
+        add_round_member(round_=round_, schedule=schedule, position=1)
+        db.session.commit()
+        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()
+        assert occurrence.assigned_to_id == individual.id
+
+
+def test_c2_fix_double_processing_is_rejected_without_duplicate_reading(app, admin_user):
+    with app.app_context():
+        meter = _meter("DOUBLE_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()
+        record_occurrence_reading(occurrence_id=occurrence.id, value=5, user_id=admin_user["id"])
+        with pytest.raises(MeterDomainError, match="déjà traitée ou annulée"):
+            record_occurrence_reading(occurrence_id=occurrence.id, value=6, user_id=admin_user["id"])
+        assert MeterReading.query.filter_by(meter_id=meter.id).count() == 1
+
+        other = _meter("DOUBLE_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()
+        record_occurrence_without_reading(occurrence_id=other_occurrence.id, reason="INACCESSIBLE", user_id=admin_user["id"])
+        with pytest.raises(MeterDomainError, match="déjà traitée ou annulée"):
+            record_occurrence_reading(occurrence_id=other_occurrence.id, value=9, user_id=admin_user["id"])
+        assert MeterReading.query.filter_by(meter_id=other.id).count() == 0
+
+
+def test_c2_fix_user_isolation_http(app):
+    with app.app_context():
+        user_a = _test_user("USER_A")
+        user_b = _test_user("USER_B")
+        meter = _meter("USER_ISOLATION")
+        schedule = create_or_update_schedule(
+            meter=meter, frequency="MONTHLY", reference_date=date(2026, 10, 1), assigned_to_id=user_a.id,
+        )
+        db.session.commit()
+        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()
+        client = app.test_client()
+        response = client.post('/auth/login', data={'username': user_b.username, 'password': 'test-c2-fix-password'})
+        assert response.status_code == 302
+        assert client.get(f"/planning/meter-occurrences/{occurrence.id}").status_code == 403
+        assert client.post(f"/planning/meter-occurrences/{occurrence.id}/reading", data={'value': '12'}).status_code == 302
+        db.session.refresh(occurrence)
+        assert occurrence.status == "À FAIRE"
+        client.get('/auth/logout')
+        response = client.post('/auth/login', data={'username': user_a.username, 'password': 'test-c2-fix-password'})
+        assert response.status_code == 302
+        assert client.post(f"/planning/meter-occurrences/{occurrence.id}/without-reading", data={'reason': 'READING_UNKNOWN'}).status_code == 302
+        db.session.refresh(occurrence)
+        assert occurrence.status == TREATED_WITHOUT_READING
 
 
 def test_c2_permissions_keep_configuration_admin_only(app, client):