diff --git a/tests/integration/test_meter_checkpoint_c1.py b/tests/integration/test_meter_checkpoint_c1.py index 145adbe..51f9638 100644 --- a/tests/integration/test_meter_checkpoint_c1.py +++ b/tests/integration/test_meter_checkpoint_c1.py @@ -78,12 +78,63 @@ 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"]) + old_reading = 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 + with pytest.raises(MeterDomainError): + record_meter_reading(meter=old, value=13, user_id=admin_user["id"]) + assert MeterReading.query.filter_by(meter_id=old.id).count() == 1 + assert old.current_value == 12 + corrected = correct_meter_reading( + reading=old_reading, new_value=12.5, user_id=admin_user["id"], reason="TEST_UI_C1_CLOSE correction" + ) + assert corrected.value == 12.5 + record_meter_reading(meter=new, value=1, user_id=admin_user["id"]) + assert new.current_value == 1 + + +def test_replaced_meter_http_is_read_only_but_history_remains_accessible(app, admin_user, authenticated_client): + with app.app_context(): + _, building, _ = _equipment() + old = create_meter(name="TEST_UI_C1_CLOSE_OLD", building_id=building.id) + record_meter_reading(meter=old, value=20, user_id=admin_user["id"]) + new = replace_meter(old_meter=old, name="TEST_UI_C1_CLOSE_NEW", user_id=admin_user["id"]) + + response = authenticated_client.get(f"/meters/{old.id}") + assert response.status_code == 200 + assert b"Compteur remplac" in response.data + assert b"Nouveau relev\xc3\xa9" not in response.data + assert new.name.encode() in response.data + assert b"Historique des relev\xc3\xa9s" in response.data + + before = MeterReading.query.filter_by(meter_id=old.id).count() + response = authenticated_client.post( + f"/meters/{old.id}/reading", data={"value": "21"}, follow_redirects=True + ) + assert response.status_code == 200 + assert b"inactif ou remplac" in response.data + assert MeterReading.query.filter_by(meter_id=old.id).count() == before + + +def test_replaced_equipment_meter_json_route_returns_business_error(app, admin_user, authenticated_client): + with app.app_context(): + equipment, _, _ = _equipment() + old = create_meter(name="TEST_UI_C1_CLOSE_EQUIPMENT", equipment_id=equipment.id) + record_meter_reading(meter=old, value=30, user_id=admin_user["id"]) + replace_meter(old_meter=old, name="TEST_UI_C1_CLOSE_EQUIPMENT_NEW", user_id=admin_user["id"]) + before = MeterReading.query.filter_by(meter_id=old.id).count() + + response = authenticated_client.post( + f"/equipments/{equipment.id}/meters/reading", + json={"meter_id": old.id, "value": 31}, + ) + assert response.status_code == 400 + assert response.get_json()["success"] is False + assert "remplac" in response.get_json()["error"] + assert MeterReading.query.filter_by(meter_id=old.id).count() == before def _edit_payload(meter, building, **extra):