diff --git a/apps/backend/alembic/env.py b/apps/backend/alembic/env.py index 7b09cae..a1a4adc 100644 --- a/apps/backend/alembic/env.py +++ b/apps/backend/alembic/env.py @@ -19,7 +19,7 @@ config = context.config if config.config_file_name is not None: fileConfig(config.config_file_name) -config.set_main_option("sqlalchemy.url", get_settings().database_url) +config.set_main_option("sqlalchemy.url", get_settings().database_url.replace("%", "%%")) target_metadata = Base.metadata diff --git a/apps/backend/alembic/versions/e6d2026091501_create_data_schema.py b/apps/backend/alembic/versions/e6d2026091501_create_data_schema.py new file mode 100644 index 0000000..8fb3694 --- /dev/null +++ b/apps/backend/alembic/versions/e6d2026091501_create_data_schema.py @@ -0,0 +1,218 @@ +"""Création des six tables Data et de l'hypertable reading. + +Revision ID: e6d2026091501 +Revises: 821f71be74c0 +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision = "e6d2026091501" +down_revision = "821f71be74c0" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "dataset", + sa.Column("dataset_id", sa.BigInteger(), autoincrement=True, nullable=False), + sa.Column("dataset_name", sa.Text(), nullable=False), + sa.Column("archive_sha256", sa.String(length=64), nullable=False), + sa.Column("storage_uri", sa.Text(), nullable=False), + sa.Column("source_timezone", sa.Text(), nullable=True), + sa.Column( + "metadata", postgresql.JSONB(none_as_null=True, astext_type=sa.Text()), nullable=False + ), + sa.CheckConstraint("dataset_id > 0", name="ck_dataset_positive_id"), + sa.PrimaryKeyConstraint("dataset_id"), + sa.UniqueConstraint("archive_sha256", name="uq_dataset_archive_sha256"), + ) + op.create_table( + "site", + sa.Column("site_id", sa.Text(), nullable=False), + sa.Column("site_name", sa.Text(), nullable=False), + sa.Column("site_type", sa.Text(), nullable=False), + sa.Column("location", sa.Text(), nullable=True), + sa.Column("capacity_kw", sa.Double(), nullable=True), + sa.Column("status", sa.Text(), nullable=True), + sa.PrimaryKeyConstraint("site_id"), + ) + op.create_table( + "prediction", + sa.Column("prediction_id", sa.BigInteger(), autoincrement=True, nullable=False), + sa.Column("site_id", sa.Text(), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column("target_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("target_metric", sa.Text(), nullable=False), + sa.Column("period_minutes", sa.Integer(), nullable=True), + sa.Column("predicted_value", sa.Double(), nullable=True), + sa.Column("model_reference", sa.Text(), nullable=False), + sa.Column("status", sa.Text(), nullable=False), + sa.Column("failure_reason", sa.Text(), nullable=True), + sa.CheckConstraint( + "(status = 'available' AND predicted_value IS NOT NULL AND failure_reason IS NULL) OR (status IN ('insufficient_data', 'error') AND predicted_value IS NULL AND failure_reason IS NOT NULL)", + name="ck_prediction_status", + ), + sa.CheckConstraint( + "target_metric <> 'consumption_kwh' OR period_minutes IS NOT NULL", + name="ck_prediction_energy_period", + ), + sa.CheckConstraint( + "target_metric IN ('consumption_kwh', 'consumption_kw')", name="ck_prediction_metric" + ), + sa.CheckConstraint( + "period_minutes IS NULL OR period_minutes > 0", name="ck_prediction_period" + ), + sa.ForeignKeyConstraint( + ["site_id"], ["site.site_id"], name="fk_prediction_site", ondelete="RESTRICT" + ), + sa.PrimaryKeyConstraint("prediction_id"), + sa.UniqueConstraint("prediction_id", "site_id", name="uq_prediction_id_site"), + ) + op.create_index( + "ix_prediction_site_target", "prediction", ["site_id", "target_at"], unique=False + ) + op.create_table( + "reading", + sa.Column("reading_id", sa.BigInteger(), autoincrement=True, nullable=False), + sa.Column("site_id", sa.Text(), nullable=False), + sa.Column("timestamp", sa.DateTime(timezone=True), nullable=False), + sa.Column("source", sa.Text(), nullable=False), + sa.Column("dataset_id", sa.BigInteger(), nullable=True), + sa.Column("consumption_kw", sa.Double(), nullable=True), + sa.Column("consumption_kwh", sa.Double(), nullable=True), + sa.Column("consumption_euros", sa.Numeric(precision=14, scale=2), nullable=True), + sa.Column("voltage_v", sa.Double(), nullable=True), + sa.Column("current_a", sa.Double(), nullable=True), + sa.Column("power_factor", sa.Double(), nullable=True), + sa.Column("temperature_celsius", sa.Double(), nullable=True), + sa.Column("humidity_percent", sa.Double(), nullable=True), + sa.Column("solar_irradiance_wm2", sa.Double(), nullable=True), + sa.Column("is_working_hours", sa.Boolean(), nullable=True), + sa.Column("data_quality", sa.Text(), nullable=True), + sa.Column("null_reasons", postgresql.ARRAY(sa.Text()), nullable=True), + sa.Column( + "imputed_values", postgresql.JSONB(none_as_null=True, astext_type=sa.Text()), nullable=True + ), + sa.Column("imputation_method", sa.Text(), nullable=True), + sa.Column( + "ingested_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column( + "raw_data", postgresql.JSONB(none_as_null=True, astext_type=sa.Text()), nullable=False + ), + sa.CheckConstraint( + "(source = 'csv' AND dataset_id IS NOT NULL) OR (source IN ('api_current', 'api_history') AND dataset_id IS NULL)", + name="ck_reading_dataset_source", + ), + sa.CheckConstraint( + "data_quality IS NULL OR data_quality IN ('good', 'partial', 'degraded', 'critical')", + name="ck_reading_quality", + ), + sa.CheckConstraint( + "source IN ('csv', 'api_current', 'api_history')", name="ck_reading_source" + ), + sa.CheckConstraint( + "(imputed_values IS NULL AND imputation_method IS NULL) OR (imputed_values IS NOT NULL AND imputation_method IS NOT NULL)", + name="ck_reading_imputation", + ), + sa.ForeignKeyConstraint( + ["dataset_id"], ["dataset.dataset_id"], name="fk_reading_dataset", ondelete="RESTRICT" + ), + sa.ForeignKeyConstraint( + ["site_id"], ["site.site_id"], name="fk_reading_site", ondelete="RESTRICT" + ), + sa.PrimaryKeyConstraint("reading_id", "timestamp"), + ) + op.create_index("ix_reading_dataset_id", "reading", ["dataset_id"], unique=False) + op.create_index( + "ix_reading_site_timestamp", "reading", ["site_id", "timestamp"], unique=False + ) + op.create_index( + "uq_reading_source", + "reading", + ["site_id", "timestamp", "source", sa.literal_column("coalesce(dataset_id, 0)")], + unique=True, + ) + op.execute( + "SELECT create_hypertable('reading', by_range('timestamp'), create_default_indexes => FALSE)" + ) + op.create_table( + "alert", + sa.Column("alert_id", sa.BigInteger(), autoincrement=True, nullable=False), + sa.Column("source_alert_id", sa.Text(), nullable=False), + sa.Column("site_id", sa.Text(), nullable=False), + sa.Column("source", sa.Text(), nullable=False), + sa.Column("timestamp", sa.DateTime(timezone=True), nullable=False), + sa.Column("type", sa.Text(), nullable=False), + sa.Column("severity", sa.Text(), nullable=False), + sa.Column("message", sa.Text(), nullable=False), + sa.Column("value", sa.Double(), nullable=True), + sa.Column("threshold", sa.Double(), nullable=True), + sa.Column("metric", sa.Text(), nullable=True), + sa.Column("prediction_id", sa.BigInteger(), nullable=True), + sa.Column( + "raw_data", postgresql.JSONB(none_as_null=True, astext_type=sa.Text()), nullable=False + ), + sa.CheckConstraint( + "severity IN ('low', 'medium', 'high', 'critical')", name="ck_alert_severity" + ), + sa.CheckConstraint("source IN ('api_mock', 'enervision')", name="ck_alert_source"), + sa.CheckConstraint( + "type IN ('spike', 'threshold', 'anomaly', 'outage', 'sensor')", name="ck_alert_type" + ), + sa.ForeignKeyConstraint( + ["prediction_id", "site_id"], + ["prediction.prediction_id", "prediction.site_id"], + name="fk_alert_prediction_site", + ondelete="RESTRICT", + ), + sa.ForeignKeyConstraint( + ["site_id"], ["site.site_id"], name="fk_alert_site", ondelete="RESTRICT" + ), + sa.PrimaryKeyConstraint("alert_id"), + sa.UniqueConstraint( + "source", "site_id", "source_alert_id", name="uq_alert_source_reference" + ), + ) + op.create_index("ix_alert_site_timestamp", "alert", ["site_id", "timestamp"], unique=False) + op.create_table( + "recommendation", + sa.Column("recommendation_id", sa.BigInteger(), autoincrement=True, nullable=False), + sa.Column("alert_id", sa.BigInteger(), nullable=False), + sa.Column("action", sa.Text(), nullable=False), + sa.Column("explanation", sa.Text(), nullable=False), + sa.Column("rule_reference", sa.Text(), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.ForeignKeyConstraint( + ["alert_id"], ["alert.alert_id"], name="fk_recommendation_alert", ondelete="RESTRICT" + ), + sa.PrimaryKeyConstraint("recommendation_id"), + sa.UniqueConstraint("alert_id", "rule_reference", name="uq_recommendation_alert_rule"), + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + op.drop_table("recommendation") + op.drop_table("alert") + op.drop_table("reading") + op.drop_table("prediction") + op.drop_table("site") + op.drop_table("dataset") diff --git a/apps/backend/app/models/__init__.py b/apps/backend/app/models/__init__.py index 9e65265..10a5ecb 100644 --- a/apps/backend/app/models/__init__.py +++ b/apps/backend/app/models/__init__.py @@ -2,8 +2,20 @@ # --autogenerate`, qui générerait alors un drop de sa table. from app.models.audit_log import AuditLog +from app.models.energy import Alert, Dataset, Prediction, Reading, Recommendation, Site from app.models.login_attempt import LoginAttempt from app.models.refresh_token import RefreshToken from app.models.user import AppUser -__all__ = ["AppUser", "AuditLog", "LoginAttempt", "RefreshToken"] +__all__ = [ + "Alert", + "AppUser", + "AuditLog", + "Dataset", + "LoginAttempt", + "Prediction", + "Reading", + "Recommendation", + "RefreshToken", + "Site", +] diff --git a/apps/backend/app/models/energy.py b/apps/backend/app/models/energy.py new file mode 100644 index 0000000..285ad26 --- /dev/null +++ b/apps/backend/app/models/energy.py @@ -0,0 +1,210 @@ +"""Tables du modèle de données EnerVision (CSV, API Mock et résultats ML).""" + +from datetime import datetime +from decimal import Decimal +from typing import Any + +from sqlalchemy import ( + BigInteger, + Boolean, + CheckConstraint, + DateTime, + Double, + ForeignKey, + ForeignKeyConstraint, + Index, + Integer, + Numeric, + String, + Text, + UniqueConstraint, + func, + text, +) +from sqlalchemy.dialects.postgresql import ARRAY, JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.base import Base + + +class Dataset(Base): + __tablename__ = "dataset" + __table_args__ = ( + CheckConstraint("dataset_id > 0", name="ck_dataset_positive_id"), + UniqueConstraint("archive_sha256", name="uq_dataset_archive_sha256"), + ) + + dataset_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + dataset_name: Mapped[str] = mapped_column(Text) + archive_sha256: Mapped[str] = mapped_column(String(64)) + storage_uri: Mapped[str] = mapped_column(Text) + source_timezone: Mapped[str | None] = mapped_column(Text) + # "metadata" est réservé par SQLAlchemy ; le nom SQL reste inchangé. + dataset_metadata: Mapped[dict[str, Any]] = mapped_column("metadata", JSONB(none_as_null=True)) + + +class Site(Base): + __tablename__ = "site" + + site_id: Mapped[str] = mapped_column(Text, primary_key=True) + site_name: Mapped[str] = mapped_column(Text) + site_type: Mapped[str] = mapped_column(Text) + location: Mapped[str | None] = mapped_column(Text) + capacity_kw: Mapped[float | None] = mapped_column(Double) + status: Mapped[str | None] = mapped_column(Text) + + +class Reading(Base): + __tablename__ = "reading" + __table_args__ = ( + CheckConstraint( + "source IN ('csv', 'api_current', 'api_history')", name="ck_reading_source" + ), + CheckConstraint( + "(source = 'csv' AND dataset_id IS NOT NULL) OR " + "(source IN ('api_current', 'api_history') AND dataset_id IS NULL)", + name="ck_reading_dataset_source", + ), + CheckConstraint( + "data_quality IS NULL OR data_quality IN ('good', 'partial', 'degraded', 'critical')", + name="ck_reading_quality", + ), + CheckConstraint( + "(imputed_values IS NULL AND imputation_method IS NULL) OR " + "(imputed_values IS NOT NULL AND imputation_method IS NOT NULL)", + name="ck_reading_imputation", + ), + Index("ix_reading_site_timestamp", "site_id", "timestamp"), + Index("ix_reading_dataset_id", "dataset_id"), + ) + + reading_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + site_id: Mapped[str] = mapped_column( + Text, ForeignKey("site.site_id", name="fk_reading_site", ondelete="RESTRICT") + ) + timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True), primary_key=True) + source: Mapped[str] = mapped_column(Text) + dataset_id: Mapped[int | None] = mapped_column( + BigInteger, + ForeignKey("dataset.dataset_id", name="fk_reading_dataset", ondelete="RESTRICT"), + ) + consumption_kw: Mapped[float | None] = mapped_column(Double) + consumption_kwh: Mapped[float | None] = mapped_column(Double) + consumption_euros: Mapped[Decimal | None] = mapped_column(Numeric(14, 2)) + voltage_v: Mapped[float | None] = mapped_column(Double) + current_a: Mapped[float | None] = mapped_column(Double) + power_factor: Mapped[float | None] = mapped_column(Double) + temperature_celsius: Mapped[float | None] = mapped_column(Double) + humidity_percent: Mapped[float | None] = mapped_column(Double) + solar_irradiance_wm2: Mapped[float | None] = mapped_column(Double) + is_working_hours: Mapped[bool | None] = mapped_column(Boolean) + data_quality: Mapped[str | None] = mapped_column(Text) + null_reasons: Mapped[list[str] | None] = mapped_column(ARRAY(Text)) + imputed_values: Mapped[dict[str, Any] | None] = mapped_column(JSONB(none_as_null=True)) + imputation_method: Mapped[str | None] = mapped_column(Text) + ingested_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + raw_data: Mapped[dict[str, Any]] = mapped_column(JSONB(none_as_null=True)) + + +Index( + "uq_reading_source", + Reading.site_id, + Reading.timestamp, + Reading.source, + func.coalesce(Reading.dataset_id, text("0")), + unique=True, +) + + +class Prediction(Base): + __tablename__ = "prediction" + __table_args__ = ( + UniqueConstraint("prediction_id", "site_id", name="uq_prediction_id_site"), + Index("ix_prediction_site_target", "site_id", "target_at"), + CheckConstraint( + "target_metric IN ('consumption_kwh', 'consumption_kw')", + name="ck_prediction_metric", + ), + CheckConstraint( + "period_minutes IS NULL OR period_minutes > 0", name="ck_prediction_period" + ), + CheckConstraint( + "target_metric <> 'consumption_kwh' OR period_minutes IS NOT NULL", + name="ck_prediction_energy_period", + ), + CheckConstraint( + "(status = 'available' AND predicted_value IS NOT NULL AND failure_reason IS NULL) OR " + "(status IN ('insufficient_data', 'error') AND predicted_value IS NULL " + "AND failure_reason IS NOT NULL)", + name="ck_prediction_status", + ), + ) + + prediction_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + site_id: Mapped[str] = mapped_column( + Text, ForeignKey("site.site_id", name="fk_prediction_site", ondelete="RESTRICT") + ) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + target_at: Mapped[datetime] = mapped_column(DateTime(timezone=True)) + target_metric: Mapped[str] = mapped_column(Text) + period_minutes: Mapped[int | None] = mapped_column(Integer) + predicted_value: Mapped[float | None] = mapped_column(Double) + model_reference: Mapped[str] = mapped_column(Text) + status: Mapped[str] = mapped_column(Text) + failure_reason: Mapped[str | None] = mapped_column(Text) + + +class Alert(Base): + __tablename__ = "alert" + __table_args__ = ( + UniqueConstraint("source", "site_id", "source_alert_id", name="uq_alert_source_reference"), + Index("ix_alert_site_timestamp", "site_id", "timestamp"), + ForeignKeyConstraint( + ["prediction_id", "site_id"], + ["prediction.prediction_id", "prediction.site_id"], + name="fk_alert_prediction_site", + ondelete="RESTRICT", + ), + CheckConstraint("source IN ('api_mock', 'enervision')", name="ck_alert_source"), + CheckConstraint( + "type IN ('spike', 'threshold', 'anomaly', 'outage', 'sensor')", name="ck_alert_type" + ), + CheckConstraint( + "severity IN ('low', 'medium', 'high', 'critical')", name="ck_alert_severity" + ), + ) + + alert_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + source_alert_id: Mapped[str] = mapped_column(Text) + site_id: Mapped[str] = mapped_column( + Text, ForeignKey("site.site_id", name="fk_alert_site", ondelete="RESTRICT") + ) + source: Mapped[str] = mapped_column(Text) + timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True)) + type: Mapped[str] = mapped_column(Text) + severity: Mapped[str] = mapped_column(Text) + message: Mapped[str] = mapped_column(Text) + value: Mapped[float | None] = mapped_column(Double) + threshold: Mapped[float | None] = mapped_column(Double) + metric: Mapped[str | None] = mapped_column(Text) + prediction_id: Mapped[int | None] = mapped_column(BigInteger) + raw_data: Mapped[dict[str, Any]] = mapped_column(JSONB(none_as_null=True)) + + +class Recommendation(Base): + __tablename__ = "recommendation" + __table_args__ = ( + UniqueConstraint("alert_id", "rule_reference", name="uq_recommendation_alert_rule"), + ) + + recommendation_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + alert_id: Mapped[int] = mapped_column( + BigInteger, + ForeignKey("alert.alert_id", name="fk_recommendation_alert", ondelete="RESTRICT"), + ) + action: Mapped[str] = mapped_column(Text) + explanation: Mapped[str] = mapped_column(Text) + rule_reference: Mapped[str] = mapped_column(Text) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) diff --git a/apps/backend/tests/db/test_data_schema.py b/apps/backend/tests/db/test_data_schema.py new file mode 100644 index 0000000..c564042 --- /dev/null +++ b/apps/backend/tests/db/test_data_schema.py @@ -0,0 +1,261 @@ +from collections.abc import AsyncIterator +from datetime import UTC, datetime +from uuid import uuid4 + +import pytest +from sqlalchemy import insert, select, text +from sqlalchemy.engine import make_url +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncConnection, create_async_engine + +from app.core.config import get_settings +from app.models.energy import Alert, Dataset, Prediction, Reading, Recommendation, Site + +pytestmark = pytest.mark.integration +MOMENT = datetime(2024, 1, 1, tzinfo=UTC) + + +@pytest.fixture +async def data_connection() -> AsyncIterator[AsyncConnection]: + url = make_url(get_settings().database_url) + if url.database != "enervision_test": + pytest.fail("Ces tests exigent DATABASE_URL vers enervision_test.") + engine = create_async_engine(url) + try: + async with engine.connect() as connection: + transaction = await connection.begin() + try: + yield connection + finally: + await transaction.rollback() + finally: + await engine.dispose() + + +@pytest.fixture +async def data_site(data_connection: AsyncConnection) -> str: + site_id = f"TEST-{uuid4()}" + await data_connection.execute( + insert(Site).values(site_id=site_id, site_name="Site de test", site_type="office") + ) + return site_id + + +async def test_reading_is_a_time_hypertable_when_migrated( + data_connection: AsyncConnection, +) -> None: + query = text( + "SELECT column_name FROM timescaledb_information.dimensions " + "WHERE hypertable_schema = 'public' AND hypertable_name = 'reading'" + ) + + result = await data_connection.execute(query) + + assert result.scalars().all() == ["timestamp"] + + +async def test_reading_preserves_null_and_zero_when_inserted( + data_connection: AsyncConnection, data_site: str +) -> None: + statement = insert(Reading).values( + site_id=data_site, + timestamp=MOMENT, + source="api_current", + consumption_kw=None, + consumption_kwh=0, + data_quality="partial", + null_reasons=["sensor_failure"], + raw_data={"consumption_kw": None}, + imputed_values=None, + imputation_method=None, + ) + + await data_connection.execute(statement) + result = ( + await data_connection.execute( + select( + Reading.consumption_kw, + Reading.consumption_kwh, + Reading.raw_data, + Reading.imputed_values, + ).where(Reading.site_id == data_site) + ) + ).one() + + assert tuple(result) == (None, 0, {"consumption_kw": None}, None) + + +@pytest.mark.parametrize("source", ["csv", "api_current", "api_history"]) +async def test_duplicate_reading_is_rejected_when_key_matches( + data_connection: AsyncConnection, data_site: str, source: str +) -> None: + dataset_id = None + if source == "csv": + dataset_id = ( + await data_connection.execute( + insert(Dataset.__table__) + .values( + dataset_name="Archive de test", + archive_sha256=uuid4().hex + uuid4().hex, + storage_uri="test://archive", + metadata={}, + ) + .returning(Dataset.dataset_id) + ) + ).scalar_one() + statement = insert(Reading).values( + site_id=data_site, + timestamp=MOMENT, + source=source, + dataset_id=dataset_id, + raw_data={}, + ) + await data_connection.execute(statement) + + with pytest.raises(IntegrityError): + async with data_connection.begin_nested(): + await data_connection.execute(statement) + + +@pytest.mark.parametrize( + "changes", + [ + {"source": "csv"}, + {"source": "unknown"}, + {"site_id": "UNKNOWN-SITE"}, + {"data_quality": "unknown"}, + {"imputed_values": {"consumption_kw": 12}}, + {"imputation_method": "mean-v1"}, + ], + ids=[ + "csv_sans_dataset", + "source_inconnue", + "site_absent", + "qualite_inconnue", + "imputation_sans_methode", + "methode_sans_imputation", + ], +) +async def test_invalid_reading_is_rejected_when_constraints_fail( + data_connection: AsyncConnection, data_site: str, changes: dict[str, object] +) -> None: + values: dict[str, object] = { + "site_id": data_site, + "timestamp": MOMENT, + "source": "api_current", + "raw_data": {}, + } + values.update(changes) + + with pytest.raises(IntegrityError): + async with data_connection.begin_nested(): + await data_connection.execute(insert(Reading).values(**values)) + + +async def test_prediction_requires_period_when_energy_is_predicted( + data_connection: AsyncConnection, data_site: str +) -> None: + statement = insert(Prediction).values( + site_id=data_site, + target_at=MOMENT, + target_metric="consumption_kwh", + predicted_value=12, + status="available", + model_reference="test-model/1", + ) + + with pytest.raises(IntegrityError): + async with data_connection.begin_nested(): + await data_connection.execute(statement) + + +async def test_unavailable_prediction_preserves_null_when_inserted( + data_connection: AsyncConnection, data_site: str +) -> None: + statement = ( + insert(Prediction) + .values( + site_id=data_site, + target_at=MOMENT, + target_metric="consumption_kw", + status="insufficient_data", + failure_reason="Historique trop court", + model_reference="test-model/1", + ) + .returning(Prediction.predicted_value) + ) + + value = (await data_connection.execute(statement)).scalar_one() + + assert value is None + + +async def test_alert_rejects_prediction_when_site_differs( + data_connection: AsyncConnection, data_site: str +) -> None: + other_site = f"TEST-{uuid4()}" + await data_connection.execute( + insert(Site).values(site_id=other_site, site_name="Autre site", site_type="office") + ) + prediction_id = ( + await data_connection.execute( + insert(Prediction) + .values( + site_id=data_site, + target_at=MOMENT, + target_metric="consumption_kw", + predicted_value=12, + status="available", + model_reference="test-model/1", + ) + .returning(Prediction.prediction_id) + ) + ).scalar_one() + + with pytest.raises(IntegrityError): + async with data_connection.begin_nested(): + await data_connection.execute( + insert(Alert).values( + source_alert_id=str(uuid4()), + site_id=other_site, + source="enervision", + timestamp=MOMENT, + type="spike", + severity="high", + message="Test", + prediction_id=prediction_id, + raw_data={}, + ) + ) + + +async def test_recommendation_is_unique_when_alert_and_rule_match( + data_connection: AsyncConnection, data_site: str +) -> None: + alert_id = ( + await data_connection.execute( + insert(Alert) + .values( + source_alert_id=str(uuid4()), + site_id=data_site, + source="api_mock", + timestamp=MOMENT, + type="spike", + severity="high", + message="Test", + raw_data={}, + ) + .returning(Alert.alert_id) + ) + ).scalar_one() + statement = insert(Recommendation).values( + alert_id=alert_id, + action="Vérifier la consommation", + explanation="Pic détecté", + rule_reference="spike-v1", + ) + await data_connection.execute(statement) + + with pytest.raises(IntegrityError): + async with data_connection.begin_nested(): + await data_connection.execute(statement)