From ddef82ae8ec11e33f58f4847f30e58c85cb3fad6 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 24 Aug 2026 09:56:23 +0000 Subject: [PATCH] test(planning): cover C2 fix workflows --- tests/integration/test_meter_checkpoint_c2.py | 118 +++++++++++++++++- 1 file changed, 117 insertions(+), 1 deletion(-) diff --git a/tests/integration/test_meter_checkpoint_c2.py b/tests/integration/test_meter_checkpoint_c2.py index 5c9150b..ce01a12 100644 --- a/tests/integration/test_meter_checkpoint_c2.py +++ b/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):