johanleroy/ENI-projet-piscine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(backend): supprime les vulnerabilites Sonar du Dockerfile et allege les tests d'exception

Browse Source
This commit is contained in:
Dorian
2026-09-21 14:08:04 +02:00
parent 44f3416ffe
commit 2adfdf0eb0
10 changed files with 312 additions and 284 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/Dockerfile
+5 -4
View File
@@ -11,13 +11,14 @@ WORKDIR /app
RUN --mount=type=cache,target=/root/.cache/uv \ RUN --mount=type=cache,target=/root/.cache/uv \
--mount=type=bind,source=uv.lock,target=uv.lock \ --mount=type=bind,source=uv.lock,target=uv.lock \
--mount=type=bind,source=pyproject.toml,target=pyproject.toml \ --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
uv sync --locked --no-install-project --no-dev uv sync --locked --no-install-project --no-dev --no-build
# Le projet lui-meme n'est pas installe (pas de second `uv sync`) : il tourne depuis /app, le
# repertoire de travail, et rien ne lit ses metadonnees. L'installer imposerait de le construire
# (backend hatchling), donc de retirer `--no-build` de l'etape ci-dessus, qui garantit que
# l'installation des dependances n'execute aucun script de build (regle Sonar docker:S8541).
COPY . /app COPY . /app
RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --locked --no-dev
FROM python:3.14-slim AS runtime FROM python:3.14-slim AS runtime
apps/backend/tests/db/test_data_schema.py
+29 -19
View File
@@ -112,8 +112,10 @@ async def test_duplicate_reading_is_rejected_when_key_matches(
) )
await data_connection.execute(statement) await data_connection.execute(statement)
savepoint = data_connection.begin_nested()
with pytest.raises(IntegrityError): with pytest.raises(IntegrityError):
async with data_connection.begin_nested(): async with savepoint:
await data_connection.execute(statement) await data_connection.execute(statement)
@@ -147,9 +149,12 @@ async def test_invalid_reading_is_rejected_when_constraints_fail(
} }
values.update(changes) values.update(changes)
statement = insert(Reading).values(**values)
savepoint = data_connection.begin_nested()
with pytest.raises(IntegrityError): with pytest.raises(IntegrityError):
async with data_connection.begin_nested(): async with savepoint:
await data_connection.execute(insert(Reading).values(**values)) await data_connection.execute(statement)
async def test_prediction_requires_period_when_energy_is_predicted( async def test_prediction_requires_period_when_energy_is_predicted(
@@ -164,8 +169,10 @@ async def test_prediction_requires_period_when_energy_is_predicted(
model_reference="test-model/1", model_reference="test-model/1",
) )
savepoint = data_connection.begin_nested()
with pytest.raises(IntegrityError): with pytest.raises(IntegrityError):
async with data_connection.begin_nested(): async with savepoint:
await data_connection.execute(statement) await data_connection.execute(statement)
@@ -212,21 +219,22 @@ async def test_alert_rejects_prediction_when_site_differs(
) )
).scalar_one() ).scalar_one()
statement = 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={},
)
savepoint = data_connection.begin_nested()
with pytest.raises(IntegrityError): with pytest.raises(IntegrityError):
async with data_connection.begin_nested(): async with savepoint:
await data_connection.execute( await data_connection.execute(statement)
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( async def test_recommendation_is_unique_when_alert_and_rule_match(
@@ -256,6 +264,8 @@ async def test_recommendation_is_unique_when_alert_and_rule_match(
) )
await data_connection.execute(statement) await data_connection.execute(statement)
savepoint = data_connection.begin_nested()
with pytest.raises(IntegrityError): with pytest.raises(IntegrityError):
async with data_connection.begin_nested(): async with savepoint:
await data_connection.execute(statement) await data_connection.execute(statement)
apps/backend/tests/etl/test_historical_import.py
+245 -239
View File
@@ -1,239 +1,245 @@
import hashlib import hashlib
import json import json
import pandas as pd import pandas as pd
import pytest import pytest
from app.etl.historical_import import ( from app.etl.historical_import import (
SOURCE_NAME, SOURCE_NAME,
build_reading_batch, build_reading_batch,
classify_quality, classify_quality,
compute_sha256, compute_sha256,
load_metadata, load_metadata,
normalize_timestamps, normalize_timestamps,
validate_source, validate_source,
) )
def make_metadata() -> dict: def make_metadata() -> dict:
return { return {
"total_records": 2, "total_records": 2,
"sites": { "sites": {
"SITE001": {}, "SITE001": {},
}, },
} }
def make_dataframe() -> pd.DataFrame: def make_dataframe() -> pd.DataFrame:
return pd.DataFrame( return pd.DataFrame(
[ [
{ {
"timestamp": "2023-01-01 00:00:00", "timestamp": "2023-01-01 00:00:00",
"site_id": "SITE001", "site_id": "SITE001",
"site_type": "office", "site_type": "office",
"site_name": "Site 1", "site_name": "Site 1",
"consumption_kwh": 10.5, "consumption_kwh": 10.5,
"consumption_euros": 2.5, "consumption_euros": 2.5,
"temperature_celsius": 20.0, "temperature_celsius": 20.0,
"humidity_percent": 50.0, "humidity_percent": 50.0,
"solar_irradiance_wm2": 0.0, "solar_irradiance_wm2": 0.0,
"hour": 0, "hour": 0,
"day_of_week": 6, "day_of_week": 6,
"day_name": "Sunday", "day_name": "Sunday",
"month": 1, "month": 1,
"is_weekend": True, "is_weekend": True,
"is_working_hours": False, "is_working_hours": False,
}, },
{ {
"timestamp": "2023-01-01 01:00:00", "timestamp": "2023-01-01 01:00:00",
"site_id": "SITE001", "site_id": "SITE001",
"site_type": "office", "site_type": "office",
"site_name": "Site 1", "site_name": "Site 1",
"consumption_kwh": 11.0, "consumption_kwh": 11.0,
"consumption_euros": 2.7, "consumption_euros": 2.7,
"temperature_celsius": 19.5, "temperature_celsius": 19.5,
"humidity_percent": 52.0, "humidity_percent": 52.0,
"solar_irradiance_wm2": 0.0, "solar_irradiance_wm2": 0.0,
"hour": 1, "hour": 1,
"day_of_week": 6, "day_of_week": 6,
"day_name": "Sunday", "day_name": "Sunday",
"month": 1, "month": 1,
"is_weekend": True, "is_weekend": True,
"is_working_hours": False, "is_working_hours": False,
}, },
] ]
) )
def test_compute_sha256(tmp_path): def test_compute_sha256(tmp_path):
file_path = tmp_path / "dataset.csv" file_path = tmp_path / "dataset.csv"
content = b"hello-enervision" content = b"hello-enervision"
file_path.write_bytes(content) file_path.write_bytes(content)
expected = hashlib.sha256(content).hexdigest() expected = hashlib.sha256(content).hexdigest()
assert compute_sha256(file_path) == expected assert compute_sha256(file_path) == expected
def test_load_metadata(tmp_path): def test_load_metadata(tmp_path):
metadata_path = tmp_path / "metadata.json" metadata_path = tmp_path / "metadata.json"
metadata = { metadata = {
"total_records": 2, "total_records": 2,
"sites": { "sites": {
"SITE001": {}, "SITE001": {},
}, },
} }
metadata_path.write_text( metadata_path.write_text(
json.dumps(metadata), json.dumps(metadata),
encoding="utf-8", encoding="utf-8",
) )
assert load_metadata(metadata_path) == metadata assert load_metadata(metadata_path) == metadata
def test_validate_source_accepts_valid_dataset(): def test_validate_source_accepts_valid_dataset():
frame = make_dataframe() frame = make_dataframe()
validate_source( validate_source(
frame, frame,
make_metadata(), make_metadata(),
) )
def test_validate_source_rejects_missing_column(): def test_validate_source_rejects_missing_column():
frame = make_dataframe().drop(columns=["consumption_kwh"]) frame = make_dataframe().drop(columns=["consumption_kwh"])
with pytest.raises( metadata = make_metadata()
ValueError,
match="Colonnes obligatoires absentes", with pytest.raises(
): ValueError,
validate_source( match="Colonnes obligatoires absentes",
frame, ):
make_metadata(), validate_source(
) frame,
metadata,
)
def test_validate_source_rejects_duplicates():
frame = make_dataframe()
def test_validate_source_rejects_duplicates():
frame.loc[1, "timestamp"] = frame.loc[ frame = make_dataframe()
0,
"timestamp", frame.loc[1, "timestamp"] = frame.loc[
] 0,
"timestamp",
with pytest.raises( ]
ValueError,
match="doublons", metadata = make_metadata()
):
validate_source( with pytest.raises(
frame, ValueError,
make_metadata(), match="doublons",
) ):
validate_source(
frame,
def test_validate_source_rejects_unknown_site(): metadata,
frame = make_dataframe() )
frame.loc[1, "site_id"] = "SITE999"
def test_validate_source_rejects_unknown_site():
with pytest.raises( frame = make_dataframe()
ValueError,
match="Sites incohérents", frame.loc[1, "site_id"] = "SITE999"
):
validate_source( metadata = make_metadata()
frame,
make_metadata(), with pytest.raises(
) ValueError,
match="Sites incohérents",
):
def test_normalize_timestamps_adds_timezone(): validate_source(
frame = make_dataframe() frame,
metadata,
normalized = normalize_timestamps( )
frame,
"UTC",
) def test_normalize_timestamps_adds_timezone():
frame = make_dataframe()
assert normalized["timestamp"].dt.tz is not None
normalized = normalize_timestamps(
assert "_source_timestamp" in normalized.columns frame,
"UTC",
)
def test_classify_quality_good():
row = make_dataframe().iloc[0].to_dict() assert normalized["timestamp"].dt.tz is not None
quality, reasons = classify_quality(row) assert "_source_timestamp" in normalized.columns
assert quality == "good"
assert reasons == [] def test_classify_quality_good():
row = make_dataframe().iloc[0].to_dict()
def test_classify_quality_degraded_when_consumption_missing(): quality, reasons = classify_quality(row)
row = make_dataframe().iloc[0].to_dict()
row["consumption_kwh"] = None assert quality == "good"
assert reasons == []
quality, reasons = classify_quality(row)
assert quality == "degraded" def test_classify_quality_degraded_when_consumption_missing():
row = make_dataframe().iloc[0].to_dict()
assert "missing:consumption_kwh" in reasons row["consumption_kwh"] = None
quality, reasons = classify_quality(row)
def test_build_reading_batch_respects_database_contract():
frame = normalize_timestamps( assert quality == "degraded"
make_dataframe(),
"UTC", assert "missing:consumption_kwh" in reasons
)
rows = build_reading_batch( def test_build_reading_batch_respects_database_contract():
frame.iloc[:1], frame = normalize_timestamps(
dataset_id=3, make_dataframe(),
) "UTC",
)
assert len(rows) == 1
rows = build_reading_batch(
row = rows[0] frame.iloc[:1],
dataset_id=3,
assert row["dataset_id"] == 3 )
# Important : assert len(rows) == 1
# contrainte ck_reading_dataset_source.
assert row["source"] == "csv" row = rows[0]
assert SOURCE_NAME == "csv"
assert row["dataset_id"] == 3
# Important :
# contrainte ck_reading_imputation. # Important :
assert row["imputed_values"] is None # contrainte ck_reading_dataset_source.
assert row["imputation_method"] is None assert row["source"] == "csv"
assert SOURCE_NAME == "csv"
assert row["data_quality"] == "good"
assert row["null_reasons"] == [] # Important :
# contrainte ck_reading_imputation.
assert row["imputed_values"] is None
def test_build_reading_batch_keeps_missing_values(): assert row["imputation_method"] is None
frame = make_dataframe()
assert row["data_quality"] == "good"
frame.loc[0, "temperature_celsius"] = None assert row["null_reasons"] == []
frame = normalize_timestamps(
frame, def test_build_reading_batch_keeps_missing_values():
"UTC", frame = make_dataframe()
)
frame.loc[0, "temperature_celsius"] = None
rows = build_reading_batch(
frame.iloc[:1], frame = normalize_timestamps(
dataset_id=3, frame,
) "UTC",
)
row = rows[0]
rows = build_reading_batch(
assert row["temperature_celsius"] is None frame.iloc[:1],
dataset_id=3,
assert "missing:temperature_celsius" in row["null_reasons"] )
# RAW ingestion : aucune imputation. row = rows[0]
assert row["imputed_values"] is None
assert row["imputation_method"] is None assert row["temperature_celsius"] is None
assert "missing:temperature_celsius" in row["null_reasons"]
# RAW ingestion : aucune imputation.
assert row["imputed_values"] is None
assert row["imputation_method"] is None
apps/backend/tests/repositories/test_audit_log.py
+3 -1
View File
@@ -49,8 +49,10 @@ async def test_the_database_refuses_to_mutate_the_audit_log(
) -> None: ) -> None:
await une_ligne(session) await une_ligne(session)
requete = text(instruction)
with pytest.raises(DBAPIError, match="ajout seul"): with pytest.raises(DBAPIError, match="ajout seul"):
await session.execute(text(instruction)) await session.execute(requete)
await session.rollback() await session.rollback()
apps/backend/tests/repositories/test_password_reset_token.py
+5 -2
View File
@@ -131,11 +131,14 @@ async def test_the_database_refuses_two_tokens_sharing_a_fingerprint(
user_agent=None, user_agent=None,
) )
empreinte = fingerprint_refresh(secret)
expiration = datetime.now(UTC) + DUREE
with pytest.raises(IntegrityError): with pytest.raises(IntegrityError):
await depot.create( await depot.create(
user_id=compte, user_id=compte,
token_hash=fingerprint_refresh(secret), token_hash=empreinte,
expires_at=datetime.now(UTC) + DUREE, expires_at=expiration,
client_ip=None, client_ip=None,
user_agent=None, user_agent=None,
) )
apps/backend/tests/repositories/test_refresh_token.py
+7 -3
View File
@@ -178,12 +178,16 @@ async def test_the_database_refuses_two_tokens_sharing_a_fingerprint(
user_agent=None, user_agent=None,
) )
famille = uuid.uuid4()
empreinte = fingerprint_refresh(secret)
expiration = datetime.now(UTC) + DUREE
with pytest.raises(IntegrityError): with pytest.raises(IntegrityError):
await depot.create( await depot.create(
user_id=compte, user_id=compte,
family_id=uuid.uuid4(), family_id=famille,
token_hash=fingerprint_refresh(secret), token_hash=empreinte,
expires_at=datetime.now(UTC) + DUREE, expires_at=expiration,
client_ip=None, client_ip=None,
user_agent=None, user_agent=None,
) )
apps/backend/tests/repositories/test_user.py
+5 -7
View File
@@ -31,14 +31,12 @@ async def test_the_database_refuses_an_email_written_in_upper_case(
) -> None: ) -> None:
saisie = adresse().upper() saisie = adresse().upper()
requete = text(
"insert into app_user (email, password_hash, role) values (:e, '$argon2id$x', 'lecteur')"
)
with pytest.raises(IntegrityError): with pytest.raises(IntegrityError):
await session.execute( await session.execute(requete, {"e": saisie})
text(
"insert into app_user (email, password_hash, role) "
"values (:e, '$argon2id$x', 'lecteur')"
),
{"e": saisie},
)
await session.rollback() await session.rollback()
apps/backend/tests/services/test_reading.py
+4 -6
View File
@@ -116,13 +116,11 @@ async def test_list_history_normalizes_naive_datetimes_to_utc() -> None:
async def test_list_history_raises_when_start_is_after_end() -> None: async def test_list_history_raises_when_start_is_after_end() -> None:
service = ReadingService(readings=FakeRepository([])) service = ReadingService(readings=FakeRepository([]))
debut = datetime(2026, 9, 2, tzinfo=UTC)
fin = datetime(2026, 9, 1, tzinfo=UTC)
with pytest.raises(FenetreInverseeError): with pytest.raises(FenetreInverseeError):
await service.list_history( await service.list_history(start=debut, end=fin, limit=500, offset=0)
start=datetime(2026, 9, 2, tzinfo=UTC),
end=datetime(2026, 9, 1, tzinfo=UTC),
limit=500,
offset=0,
)
async def test_list_history_raises_when_start_equals_end() -> None: async def test_list_history_raises_when_start_equals_end() -> None:
apps/backend/tests/services/test_user.py
+3 -1
View File
@@ -235,5 +235,7 @@ async def test_every_operation_refuses_an_unknown_account(action: str) -> None:
if action == "set_active": if action == "set_active":
arguments["is_active"] = False arguments["is_active"] = False
methode = getattr(attirail.service, action)
with pytest.raises(UserNotFoundError): with pytest.raises(UserNotFoundError):
await getattr(attirail.service, action)(**arguments) await methode(**arguments)
apps/backend/tests/test_cli.py
+6 -2
View File
@@ -19,13 +19,17 @@ def test_build_parser_reads_the_create_admin_arguments() -> None:
def test_build_parser_requires_a_subcommand() -> None: def test_build_parser_requires_a_subcommand() -> None:
parser = cli.build_parser()
with pytest.raises(SystemExit): with pytest.raises(SystemExit):
cli.build_parser().parse_args([]) parser.parse_args([])
def test_build_parser_requires_an_email() -> None: def test_build_parser_requires_an_email() -> None:
parser = cli.build_parser()
with pytest.raises(SystemExit): with pytest.raises(SystemExit):
cli.build_parser().parse_args(["create-admin"]) parser.parse_args(["create-admin"])
def test_read_password_generates_a_long_secret_when_asked( def test_read_password_generates_a_long_secret_when_asked(