gmao/tests/integration/test_meter_checkpoint_c3.py

from datetime import datetime, date, timezone
from time import perf_counter

import pytest

from app_new import db
from app_new.core.models.college import Building
from app_new.core.models.planning import (
    Meter, MeterReading, MeterAlertRule, MeterHeatingRegime, GasConversion, MeterTariff,
)
from app_new.core.services.meter_service import create_meter, record_meter_reading
from app_new.core.services.meter_analytics import (
    consumption_intervals, aggregate_intervals, evaluate_alert_rule,
    statistical_anomaly, remainder_for_interval, estimated_cost, gas_coefficient,
    create_bill_gas_conversion, close_meter_alert,
)


def _meter(name, kind="eau", unit="m³"):
    building = Building(name=f"TEST_UI_C3_{name}")
    db.session.add(building)
    db.session.flush()
    return create_meter(name=f"TEST_UI_C3_{name}", meter_type=kind, unit=unit, building_id=building.id)


def _reading(meter, value, day, user_id):
    return record_meter_reading(meter=meter, value=value, user_id=user_id, reading_date=datetime(day.year, day.month, day.day, tzinfo=timezone.utc))


def test_c3_intervals_reset_and_calendar_day(app, admin_user):
    with app.app_context():
        meter = _meter("INTERVAL")
        _reading(meter, 100, date(2026, 1, 1), admin_user["id"])
        _reading(meter, 130, date(2026, 1, 11), admin_user["id"])
        # Le service C1 exige le reset explicite pour une baisse.
        _reading(meter, 5, date(2026, 1, 12), admin_user["id"]) if False else None
        values = consumption_intervals(meter)
        assert len(values) == 1 and values[0].raw_delta == 30 and values[0].calendar_days == 10
        assert aggregate_intervals(values, "MONTH")[0]["consumption"] == 30


def test_c3_context_heating_and_gas_tariff_cost(app, admin_user):
    with app.app_context():
        meter = _meter("GAS", "gaz", "m³")
        db.session.add(MeterHeatingRegime(meter=meter, network_name="TEST_UI_C3_RESEAU_A", regime="STOP", valid_from=date(2026, 1, 1), valid_to=date(2026, 1, 31)))
        db.session.add(GasConversion(meter=meter, coefficient_kwh_per_m3=10.85, valid_from=date(2026, 1, 1), origin="MANUAL"))
        db.session.add(MeterTariff(meter=meter, energy_type="gaz", unit_price=0.12, unit="€/kWh", valid_from=date(2026, 1, 1)))
        db.session.commit()
        _reading(meter, 1000, date(2026, 1, 1), admin_user["id"])
        _reading(meter, 1100, date(2026, 1, 11), admin_user["id"])
        interval = consumption_intervals(meter)[0]
        assert interval.context["heating"]["STOP"] == 10
        assert gas_coefficient(meter, date(2026, 1, 5)) == 10.85
        assert estimated_cost(meter, interval) == pytest.approx(100 * 10.85 * .12)


def test_c3_manual_alert_is_persisted_and_deduplicated(app, admin_user):
    with app.app_context():
        meter = _meter("ALERT")
        _reading(meter, 10, date(2026, 1, 1), admin_user["id"])
        _reading(meter, 30, date(2026, 1, 2), admin_user["id"])
        rule = MeterAlertRule(meter=meter, metric="consumption_per_day", period="DAY", operator=">", threshold=5, level="WARNING")
        db.session.add(rule); db.session.commit()
        interval = consumption_intervals(meter)[0]
        first = evaluate_alert_rule(rule, interval)
        second = evaluate_alert_rule(rule, interval)
        assert first.id == second.id and len(first.evidence) == 1


def test_c3_statistical_median_minimum_and_parent_remainder(app, admin_user):
    with app.app_context():
        parent = _meter("PRIMARY")
        child = _meter("SUB")
        child.parent = parent
        db.session.commit()
        for index, day in enumerate((date(2025, 1, 1), date(2025, 1, 2), date(2025, 1, 3), date(2025, 1, 4))):
            _reading(parent, index * 10, day, admin_user["id"])
            _reading(child, index * 2, day, admin_user["id"])
        intervals = consumption_intervals(parent)
        remainder = remainder_for_interval(parent, intervals[-1])
        assert remainder["quality"] == "EXACT" and remainder["value"] == 8
        assert statistical_anomaly(parent, intervals[-1], min_intervals=3) is None


def test_c3_monitoring_and_analytics_http(app, admin_user, authenticated_client):
    with app.app_context():
        meter = _meter("HTTP")
        _reading(meter, 1, date(2026, 2, 1), admin_user["id"])
        _reading(meter, 4, date(2026, 2, 4), admin_user["id"])
        meter_id = meter.id
    dashboard = authenticated_client.get("/planning/meter-monitoring")
    assert dashboard.status_code == 200 and b"Surveillance des compteurs" in dashboard.data


def test_c3_water_never_uses_gas_conversion_and_bill_conversion(app, admin_user):
    with app.app_context():
        water = _meter("WATER_COST", "eau", "m³")
        db.session.add(MeterTariff(meter=water, energy_type="eau", unit_price=4.20, unit="€/m³", valid_from=date(2026, 1, 1)))
        _reading(water, 20, date(2026, 1, 1), admin_user["id"]); _reading(water, 40, date(2026, 1, 2), admin_user["id"])
        db.session.commit()
        assert estimated_cost(water, consumption_intervals(water)[0]) == pytest.approx(84)
        gas = _meter("BILL", "gaz", "m³")
        conversion = create_bill_gas_conversion(meter=gas, volume_m3=1000, billed_kwh=10850, valid_from=date(2026, 1, 1))
        assert conversion.coefficient_kwh_per_m3 == pytest.approx(10.85)


def test_c3_alert_periods_and_zero_low_are_real(app, admin_user):
    with app.app_context():
        meter = _meter("PERIODS")
        days = [date(2026, 1, 1) + __import__('datetime').timedelta(days=i) for i in range(8)]
        for i, day in enumerate(days): _reading(meter, i * 10, day, admin_user["id"])
        intervals = consumption_intervals(meter)
        for period in ("DAY", "WEEK", "MONTH"):
            rule = MeterAlertRule(meter=meter, metric="consumption_interval", period=period, operator=">=", threshold=1, level="WARNING")
            db.session.add(rule); db.session.commit()
            assert evaluate_alert_rule(rule, intervals[-1], intervals=intervals) is not None
        inactive = MeterAlertRule(meter=meter, metric="consumption_interval", period="MONTH", operator=">", threshold=1, is_active=False)
        db.session.add(inactive); db.session.commit()
        assert evaluate_alert_rule(inactive, intervals[-1], intervals=intervals) is None
        low = MeterAlertRule(meter=meter, metric="consumption_per_day", period="DAY", operator=">", threshold=999, detect_zero_or_low=True, low_consumption_threshold=11)
        db.session.add(low); db.session.commit()
        assert evaluate_alert_rule(low, intervals[-1], intervals=intervals) is not None


def test_c3_alert_closure_excludes_only_relevant_baselines(app, admin_user):
    with app.app_context():
        meter = _meter("CLOSE")
        _reading(meter, 0, date(2026, 1, 1), admin_user["id"]); _reading(meter, 10, date(2026, 1, 2), admin_user["id"])
        rule = MeterAlertRule(meter=meter, metric="consumption_interval", period="DAY", operator=">", threshold=1)
        db.session.add(rule); db.session.commit()
        alert = evaluate_alert_rule(rule, consumption_intervals(meter)[0])
        close_meter_alert(alert=alert, conclusion="READING_ERROR", comment="TEST_UI_C3 erreur")
        assert alert.status == "CLOSED" and alert.evidence[0].excluded_from_baseline is True


def test_c3_performance_measurement(app, admin_user, authenticated_client):
    with app.app_context():
        for index in range(4):
            meter = _meter(f"PERF_{index}")
            for offset in range(12):
                _reading(meter, offset * 5, date(2025, 1, 1) + __import__('datetime').timedelta(days=offset), admin_user["id"])
        count = {"value": 0}
        from sqlalchemy import event
        def before_cursor(*args): count["value"] += 1
        event.listen(db.engine, "before_cursor_execute", before_cursor)
        started = perf_counter(); response = authenticated_client.get("/planning/meter-monitoring"); elapsed = perf_counter() - started
        event.remove(db.engine, "before_cursor_execute", before_cursor)
        print(f"C3_PERF dashboard_seconds={elapsed:.4f} dashboard_queries={count['value']}")
        assert response.status_code == 200