test(apps): verifie les contraintes du schema data
This commit is contained in:
Meryemel-gham
@@ -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_readings_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 = 'readings'"
|
||||
)
|
||||
|
||||
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(
|
||||
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(
|
||||
alert_id=str(uuid4()),
|
||||
site_id=data_site,
|
||||
source="api_mock",
|
||||
timestamp=MOMENT,
|
||||
type="spike",
|
||||
severity="high",
|
||||
message="Test",
|
||||
raw_data={},
|
||||
)
|
||||
.returning(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)
|
||||
Reference in New Issue
Block a user