251 lines
12 KiB
Python
251 lines
12 KiB
Python
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)
|
|
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):
|
|
payload = {
|
|
"name": meter.name,
|
|
"meter_type": meter.meter_type,
|
|
"unit": meter.unit,
|
|
"scope_type": "building",
|
|
"scope_id": str(building.id),
|
|
"parent_id": "",
|
|
}
|
|
payload.update(extra)
|
|
return payload
|
|
|
|
|
|
def test_meter_edit_http_changes_and_removes_parent(app, authenticated_client):
|
|
with app.app_context():
|
|
_, building, _ = _equipment()
|
|
zone = Zone(name="TEST_UI_C1_FIX_ZONE", building=building)
|
|
db.session.add(zone)
|
|
db.session.flush()
|
|
root = create_meter(name="TEST_UI_C1_FIX_ROOT", building_id=building.id)
|
|
child = create_meter(name="TEST_UI_C1_FIX_CHILD", building_id=building.id)
|
|
response = authenticated_client.get(f"/meters/{child.id}/edit")
|
|
assert response.status_code == 200
|
|
assert b"Compteur parent" in response.data
|
|
|
|
response = authenticated_client.post(
|
|
f"/meters/{child.id}/edit",
|
|
data=_edit_payload(child, building, parent_id=str(root.id)),
|
|
)
|
|
assert response.status_code == 302
|
|
db.session.refresh(child)
|
|
assert child.parent_id == root.id
|
|
|
|
response = authenticated_client.post(
|
|
f"/meters/{child.id}/edit",
|
|
data=_edit_payload(child, building, scope_type="zone", scope_id=str(zone.id), parent_id=str(root.id)),
|
|
)
|
|
assert response.status_code == 302
|
|
db.session.refresh(child)
|
|
assert child.zone_id == zone.id
|
|
assert child.building_id is None
|
|
|
|
response = authenticated_client.post(
|
|
f"/meters/{child.id}/edit",
|
|
data=_edit_payload(child, building, parent_id=""),
|
|
)
|
|
assert response.status_code == 302
|
|
db.session.refresh(child)
|
|
assert child.parent_id is None
|
|
|
|
|
|
def test_meter_edit_http_rejects_self_parent_and_deep_cycle(app, authenticated_client):
|
|
with app.app_context():
|
|
_, building, _ = _equipment()
|
|
root = create_meter(name="TEST_UI_C1_FIX_CYCLE_ROOT", building_id=building.id)
|
|
child = create_meter(name="TEST_UI_C1_FIX_CYCLE_CHILD", building_id=building.id, parent=root)
|
|
grandchild = create_meter(name="TEST_UI_C1_FIX_CYCLE_GRANDCHILD", building_id=building.id, parent=child)
|
|
|
|
response = authenticated_client.post(
|
|
f"/meters/{root.id}/edit",
|
|
data=_edit_payload(root, building, parent_id=str(root.id)),
|
|
)
|
|
assert response.status_code == 302
|
|
db.session.expire_all()
|
|
assert db.session.get(Meter, root.id).parent_id is None
|
|
|
|
response = authenticated_client.post(
|
|
f"/meters/{root.id}/edit",
|
|
data=_edit_payload(root, building, parent_id=str(grandchild.id)),
|
|
)
|
|
assert response.status_code == 302
|
|
db.session.expire_all()
|
|
assert db.session.get(Meter, root.id).parent_id is None
|
|
|
|
|
|
def test_meter_edit_http_types_status_replacement_and_c2_wording(app, authenticated_client, admin_user):
|
|
with app.app_context():
|
|
_, building, _ = _equipment()
|
|
old = create_meter(name="TEST_UI_C1_FIX_OLD", meter_type="eau", unit="m³", building_id=building.id)
|
|
new = replace_meter(old_meter=old, name="TEST_UI_C1_FIX_NEW", user_id=admin_user["id"])
|
|
|
|
response = authenticated_client.get(f"/meters/{old.id}")
|
|
assert response.status_code == 200
|
|
assert "État du compteur".encode() in response.data
|
|
assert "Remplacé".encode() in response.data
|
|
assert "Remplacé par".encode() in response.data
|
|
assert new.name.encode() in response.data
|
|
|
|
response = authenticated_client.get(f"/meters/{new.id}")
|
|
assert response.status_code == 200
|
|
assert "Remplace".encode() in response.data
|
|
assert old.name.encode() in response.data
|
|
|
|
response = authenticated_client.get(f"/meters/{old.id}/edit")
|
|
assert response.status_code == 200
|
|
for label in ("Eau", "Gaz", "Électricité", "Photocopieur", "Heures", "Cycles", "Distance"):
|
|
assert label.encode() in response.data
|
|
assert b"Le syst\xc3\xa8me vous rappellera" not in response.data
|
|
assert b"planification automatique des \xc3\xa9ch\xc3\xa9ances" in response.data
|
|
|
|
response = authenticated_client.post(
|
|
f"/meters/{old.id}/edit",
|
|
data=_edit_payload(old, building, meter_type="eau"),
|
|
)
|
|
assert response.status_code == 302
|
|
db.session.expire_all()
|
|
assert db.session.get(Meter, old.id).meter_type == "eau"
|
|
|
|
unknown = create_meter(name="TEST_UI_C1_FIX_UNKNOWN", meter_type="future_type", building_id=building.id)
|
|
response = authenticated_client.get(f"/meters/{unknown.id}/edit")
|
|
assert response.status_code == 200
|
|
assert b"Autre \xe2\x80\x94 future_type" in response.data
|