From 2459e65bc640b2603c82778d8f6e619b0e3392c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=9E=D0=BB=D0=B8=D0=BC=D0=BE=D0=B2=20=D0=A4=D0=B0=D1=80?= =?UTF-8?q?=D0=B8=D0=B4=D1=83=D0=BD?= Date: Tue, 3 Nov 2026 02:07:19 +0300 Subject: [PATCH 1/3] endpoints --- backend/app/api/alerts.py | 13 +- backend/app/api/equipment.py | 34 +++--- backend/app/api/predict.py | 7 +- backend/app/core/errors.py | 8 ++ backend/app/schemas/equipment.py | 24 ++-- backend/app/services/equipment_service.py | 4 +- backend/app/services/mock_data.py | 4 +- backend/tests/test_endpoints.py | 142 ++++++++++++++++++++++ 8 files changed, 198 insertions(+), 38 deletions(-) create mode 100644 backend/app/core/errors.py create mode 100644 backend/tests/test_endpoints.py diff --git a/backend/app/api/alerts.py b/backend/app/api/alerts.py index bd81fba..5368a66 100644 --- a/backend/app/api/alerts.py +++ b/backend/app/api/alerts.py @@ -1,5 +1,6 @@ -from fastapi import APIRouter, HTTPException, Query +from fastapi import APIRouter, Query +from app.core.errors import NOT_FOUND, equipment_not_found from app.schemas.alert import AlertOut from app.services import alert_service, equipment_service @@ -7,12 +8,14 @@ @router.get("", response_model=list[AlertOut], summary="Список последних алертов") -def list_alerts(limit: int = Query(50, ge=1, le=500)): +def list_alerts(limit: int = Query(50, ge=1, le=500, description="Максимум записей")): return alert_service.list_alerts(limit=limit) -@router.get("/{equipment_id}", response_model=list[AlertOut], summary="Алерты по оборудованию") -def alerts_by_equipment(equipment_id: int, limit: int = Query(50, ge=1, le=500)): +@router.get("/{equipment_id}", response_model=list[AlertOut], responses=NOT_FOUND, + summary="Алерты по конкретному оборудованию") +def alerts_by_equipment(equipment_id: int, + limit: int = Query(50, ge=1, le=500, description="Максимум записей")): if equipment_service.get_equipment(equipment_id) is None: - raise HTTPException(status_code=404, detail="Оборудование не найдено") + raise equipment_not_found(equipment_id) return alert_service.list_alerts(equipment_id, limit) diff --git a/backend/app/api/equipment.py b/backend/app/api/equipment.py index 673f7c1..41c0d1f 100644 --- a/backend/app/api/equipment.py +++ b/backend/app/api/equipment.py @@ -1,14 +1,15 @@ from fastapi import APIRouter, HTTPException, Query +from app.core.errors import NOT_FOUND, equipment_not_found from app.schemas.equipment import EquipmentOut, ForecastOut, ReadingOut, StatusOut from app.services import equipment_service router = APIRouter(prefix="/equipment", tags=["Оборудование"]) -def _or_404(value): +def _found(value, equipment_id: int): if value is None: - raise HTTPException(status_code=404, detail="Оборудование не найдено") + raise equipment_not_found(equipment_id) return value @@ -17,25 +18,30 @@ def list_equipment(): return equipment_service.list_equipment() -@router.get("/{equipment_id}", response_model=EquipmentOut, summary="Информация об одной единице") +@router.get("/{equipment_id}", response_model=EquipmentOut, responses=NOT_FOUND, + summary="Информация об одной единице") def get_equipment(equipment_id: int): - return _or_404(equipment_service.get_equipment(equipment_id)) + return _found(equipment_service.get_equipment(equipment_id), equipment_id) -@router.get("/{equipment_id}/status", response_model=StatusOut, summary="Текущее состояние") +@router.get("/{equipment_id}/status", response_model=StatusOut, responses=NOT_FOUND, + summary="Текущее состояние: статус, RUL, вероятность дефекта, невязка") def get_status(equipment_id: int): - return _or_404(equipment_service.get_status(equipment_id)) + return _found(equipment_service.get_status(equipment_id), equipment_id) -@router.get("/{equipment_id}/history", response_model=list[ReadingOut], - summary="История показаний датчиков") +@router.get("/{equipment_id}/history", response_model=list[ReadingOut], responses=NOT_FOUND, + summary="История показаний датчиков за период") def get_history(equipment_id: int, - from_cycle: int | None = Query(None, ge=0), - to_cycle: int | None = Query(None, ge=0), - limit: int = Query(500, ge=1, le=5000)): - return _or_404(equipment_service.get_history(equipment_id, from_cycle, to_cycle, limit)) + from_cycle: int | None = Query(None, ge=0, description="С какого цикла (включительно)"), + to_cycle: int | None = Query(None, ge=0, description="По какой цикл (включительно)"), + limit: int = Query(500, ge=1, le=5000, description="Максимум записей (берутся последние)")): + if from_cycle is not None and to_cycle is not None and from_cycle > to_cycle: + raise HTTPException(status_code=400, detail="from_cycle не может быть больше to_cycle") + return _found(equipment_service.get_history(equipment_id, from_cycle, to_cycle, limit), equipment_id) -@router.get("/{equipment_id}/forecast", response_model=ForecastOut, summary="Прогноз деградации") +@router.get("/{equipment_id}/forecast", response_model=ForecastOut, responses=NOT_FOUND, + summary="Прогноз деградации: RUL по будущим циклам с доверительным интервалом") def get_forecast(equipment_id: int): - return _or_404(equipment_service.get_forecast(equipment_id)) + return _found(equipment_service.get_forecast(equipment_id), equipment_id) diff --git a/backend/app/api/predict.py b/backend/app/api/predict.py index 312cd59..8def7ab 100644 --- a/backend/app/api/predict.py +++ b/backend/app/api/predict.py @@ -1,5 +1,6 @@ -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter +from app.core.errors import NOT_FOUND, equipment_not_found from app.schemas.equipment import StatusOut from app.schemas.predict import PredictIn from app.services import predict_service @@ -7,9 +8,9 @@ router = APIRouter(prefix="/predict", tags=["Прогнозирование"]) -@router.post("", response_model=StatusOut, summary="Ручной запрос инференса") +@router.post("", response_model=StatusOut, responses=NOT_FOUND, summary="Ручной запрос инференса") def predict(body: PredictIn): result = predict_service.run_prediction(body.equipment_id) if result is None: - raise HTTPException(status_code=404, detail="Оборудование не найдено") + raise equipment_not_found(body.equipment_id) return result diff --git a/backend/app/core/errors.py b/backend/app/core/errors.py new file mode 100644 index 0000000..bc59196 --- /dev/null +++ b/backend/app/core/errors.py @@ -0,0 +1,8 @@ +from fastapi import HTTPException + + +def equipment_not_found(equipment_id: int) -> HTTPException: + return HTTPException(status_code=404, detail=f"Оборудование с id={equipment_id} не найдено") + + +NOT_FOUND = {404: {"description": "Оборудование не найдено"}} diff --git a/backend/app/schemas/equipment.py b/backend/app/schemas/equipment.py index a7f16cb..b3b79e4 100644 --- a/backend/app/schemas/equipment.py +++ b/backend/app/schemas/equipment.py @@ -1,6 +1,6 @@ from typing import Literal -from pydantic import BaseModel +from pydantic import BaseModel, Field Status = Literal["normal", "warning", "critical"] @@ -12,33 +12,33 @@ class EquipmentOut(BaseModel): unit_number: int current_cycle: int status: Status - predicted_rul: float + predicted_rul: float = Field(description="Прогноз остаточного ресурса, циклов") class StatusOut(BaseModel): equipment_id: int - status: Status + status: Status = Field(description="normal: RUL > 75, warning: 31–75, critical: ≤ 30 циклов") current_cycle: int - predicted_rul: float - defect_probability: float - residual: float + predicted_rul: float = Field(description="Прогноз остаточного ресурса, циклов") + defect_probability: float = Field(ge=0, le=1, description="Вероятность дефекта, 0–1") + residual: float = Field(description="Невязка: отклонение от физически ожидаемого поведения") class ReadingOut(BaseModel): cycle: int - sensors: dict[str, float] + sensors: dict[str, float] = Field(description="Показания 14 рабочих датчиков: код → значение") class ForecastPoint(BaseModel): cycle: int - rul: float - rul_low: float - rul_high: float + rul: float = Field(description="Ожидаемый остаточный ресурс на этом цикле") + rul_low: float = Field(description="Нижняя граница доверительного интервала") + rul_high: float = Field(description="Верхняя граница доверительного интервала") class ForecastOut(BaseModel): equipment_id: int current_cycle: int predicted_rul: float - predicted_failure_cycle: int - points: list[ForecastPoint] + predicted_failure_cycle: int = Field(description="Цикл, на котором ожидается отказ") + points: list[ForecastPoint] = Field(description="От текущего цикла до ожидаемого отказа, шаг — 1 цикл") diff --git a/backend/app/services/equipment_service.py b/backend/app/services/equipment_service.py index 3297815..51a1331 100644 --- a/backend/app/services/equipment_service.py +++ b/backend/app/services/equipment_service.py @@ -38,10 +38,10 @@ def get_forecast(equipment_id: int) -> dict | None: if e is None: return None rul, cycle = e["predicted_rul"], e["current_cycle"] - horizon = round(rul) + horizon = int(rul) points = [] for k in range(horizon + 1): - value, spread = rul - k, 2 + 0.12 * k + value, spread = rul - k, 2 + 0.12 * k points.append({"cycle": cycle + k, "rul": round(value, 1), "rul_low": round(max(0.0, value - spread), 1), "rul_high": round(value + spread, 1)}) return {"equipment_id": equipment_id, "current_cycle": cycle, "predicted_rul": rul, diff --git a/backend/app/services/mock_data.py b/backend/app/services/mock_data.py index 2163c94..ca0c819 100644 --- a/backend/app/services/mock_data.py +++ b/backend/app/services/mock_data.py @@ -1,5 +1,5 @@ import random -from datetime import UTC, datetime, timedelta +from datetime import UTC, datetime, timedelta, timezone from app.core.config import RUL_CLIP, RUL_CRITICAL, RUL_WARNING @@ -34,7 +34,7 @@ def _build() -> None: now = datetime.now(UTC) for i in range(1, 13): life = _rnd.randint(130, 300) - cut = int(life * _rnd.uniform(0.35, 0.97)) + cut = int(life * _rnd.uniform(0.35, 0.97)) READINGS[i] = [ {"cycle": c, "sensors": { code: round(base + trend * std * 3 * (c / life) ** 2 + _rnd.gauss(0, std * 0.4), 4) diff --git a/backend/tests/test_endpoints.py b/backend/tests/test_endpoints.py new file mode 100644 index 0000000..2b2936c --- /dev/null +++ b/backend/tests/test_endpoints.py @@ -0,0 +1,142 @@ +"""Тесты эндпоинтов из задачи #29: статус, история, прогноз, алерты (+ базовые equipment и predict).""" +from datetime import datetime + +from fastapi.testclient import TestClient +from main import app + +client = TestClient(app) +STATUSES = {"normal", "warning", "critical"} + + +def _ts(s: str) -> datetime: + return datetime.fromisoformat(s.replace("Z", "+00:00")) + + +# ---------- /equipment ---------- +def test_list_equipment(): + r = client.get("/equipment") + assert r.status_code == 200 + items = r.json() + assert len(items) > 0 + assert {"id", "name", "type", "unit_number", "current_cycle", "status", "predicted_rul"} <= set(items[0]) + assert all(e["status"] in STATUSES for e in items) + + +def test_get_equipment_and_404(): + assert client.get("/equipment/1").json()["id"] == 1 + r = client.get("/equipment/9999") + assert r.status_code == 404 and "9999" in r.json()["detail"] + + +# ---------- /equipment/{id}/status ---------- +def test_status_fields_and_ranges(): + s = client.get("/equipment/1/status").json() + assert {"equipment_id", "status", "current_cycle", "predicted_rul", + "defect_probability", "residual"} <= set(s) + assert s["status"] in STATUSES + assert s["predicted_rul"] >= 0 + assert 0 <= s["defect_probability"] <= 1 + + +def test_status_matches_equipment_list(): + item = client.get("/equipment").json()[0] + s = client.get(f"/equipment/{item['id']}/status").json() + assert (s["status"], s["predicted_rul"], s["current_cycle"]) == \ + (item["status"], item["predicted_rul"], item["current_cycle"]) + + +def test_status_404(): + assert client.get("/equipment/9999/status").status_code == 404 + + +# ---------- /equipment/{id}/history ---------- +def test_history_ascending_with_14_sensors(): + rows = client.get("/equipment/1/history").json() + cycles = [r["cycle"] for r in rows] + assert cycles == sorted(cycles) and len(set(cycles)) == len(cycles) + assert len(rows[0]["sensors"]) == 14 + assert "s1" not in rows[0]["sensors"] # константные датчики не отдаём + + +def test_history_limit_returns_latest(): + current = client.get("/equipment/1").json()["current_cycle"] + rows = client.get("/equipment/1/history", params={"limit": 5}).json() + assert len(rows) == 5 and rows[-1]["cycle"] == current + + +def test_history_period_filter(): + rows = client.get("/equipment/1/history", params={"from_cycle": 10, "to_cycle": 20}).json() + assert [r["cycle"] for r in rows] == list(range(10, 21)) + + +def test_history_invalid_params(): + assert client.get("/equipment/1/history", params={"from_cycle": 30, "to_cycle": 10}).status_code == 400 + assert client.get("/equipment/1/history", params={"limit": 0}).status_code == 422 + assert client.get("/equipment/1/history", params={"from_cycle": -1}).status_code == 422 + + +def test_history_404(): + assert client.get("/equipment/9999/history").status_code == 404 + + +# ---------- /equipment/{id}/forecast ---------- +def test_forecast_shape(): + f = client.get("/equipment/1/forecast").json() + pts = f["points"] + assert pts[0]["cycle"] == f["current_cycle"] + assert [p["cycle"] for p in pts] == list(range(f["current_cycle"], f["current_cycle"] + len(pts))) + assert pts[-1]["cycle"] == f["predicted_failure_cycle"] + + +def test_forecast_rul_decreases_and_stays_non_negative(): + pts = client.get("/equipment/1/forecast").json()["points"] + ruls = [p["rul"] for p in pts] + assert all(a >= b for a, b in zip(ruls, ruls[1:])) + assert all(p["rul"] >= 0 and p["rul_low"] >= 0 for p in pts) + + +def test_forecast_interval_contains_rul(): + for p in client.get("/equipment/1/forecast").json()["points"]: + assert p["rul_low"] <= p["rul"] <= p["rul_high"] + + +def test_forecast_all_equipment_non_negative(): + for e in client.get("/equipment").json(): + pts = client.get(f"/equipment/{e['id']}/forecast").json()["points"] + assert min(p["rul"] for p in pts) >= 0 + + +def test_forecast_404(): + assert client.get("/equipment/9999/forecast").status_code == 404 + + +# ---------- /alerts ---------- +def test_alerts_newest_first(): + alerts = client.get("/alerts").json() + assert len(alerts) > 0 + times = [_ts(a["triggered_at"]) for a in alerts] + assert times == sorted(times, reverse=True) + assert all(a["severity"] in {"info", "warning", "critical"} for a in alerts) + + +def test_alerts_limit(): + assert len(client.get("/alerts", params={"limit": 2}).json()) == 2 + assert client.get("/alerts", params={"limit": 0}).status_code == 422 + + +def test_alerts_by_equipment(): + equipment_id = client.get("/alerts").json()[0]["equipment_id"] + alerts = client.get(f"/alerts/{equipment_id}").json() + assert alerts and all(a["equipment_id"] == equipment_id for a in alerts) + + +def test_alerts_by_unknown_equipment_404(): + assert client.get("/alerts/9999").status_code == 404 + + +# ---------- /predict ---------- +def test_predict(): + r = client.post("/predict", json={"equipment_id": 1}) + assert r.status_code == 200 and r.json()["equipment_id"] == 1 + assert client.post("/predict", json={"equipment_id": 9999}).status_code == 404 + assert client.post("/predict", json={"equipment_id": 0}).status_code == 422 From 54bc5aa3a6ae10f8a145e3b5946c6f1b84f3ee88 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=9E=D0=BB=D0=B8=D0=BC=D0=BE=D0=B2=20=D0=A4=D0=B0=D1=80?= =?UTF-8?q?=D0=B8=D0=B4=D1=83=D0=BD?= Date: Tue, 3 Nov 2026 02:15:41 +0300 Subject: [PATCH 2/3] fixed linters --- backend/app/services/mock_data.py | 2 +- backend/tests/test_endpoints.py | 14 ++++---------- 2 files changed, 5 insertions(+), 11 deletions(-) diff --git a/backend/app/services/mock_data.py b/backend/app/services/mock_data.py index ca0c819..bad233b 100644 --- a/backend/app/services/mock_data.py +++ b/backend/app/services/mock_data.py @@ -1,5 +1,5 @@ import random -from datetime import UTC, datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from app.core.config import RUL_CLIP, RUL_CRITICAL, RUL_WARNING diff --git a/backend/tests/test_endpoints.py b/backend/tests/test_endpoints.py index 2b2936c..2f40570 100644 --- a/backend/tests/test_endpoints.py +++ b/backend/tests/test_endpoints.py @@ -1,4 +1,4 @@ -"""Тесты эндпоинтов из задачи #29: статус, история, прогноз, алерты (+ базовые equipment и predict).""" +import itertools from datetime import datetime from fastapi.testclient import TestClient @@ -9,10 +9,9 @@ def _ts(s: str) -> datetime: - return datetime.fromisoformat(s.replace("Z", "+00:00")) + return datetime.fromisoformat(s) -# ---------- /equipment ---------- def test_list_equipment(): r = client.get("/equipment") assert r.status_code == 200 @@ -28,7 +27,6 @@ def test_get_equipment_and_404(): assert r.status_code == 404 and "9999" in r.json()["detail"] -# ---------- /equipment/{id}/status ---------- def test_status_fields_and_ranges(): s = client.get("/equipment/1/status").json() assert {"equipment_id", "status", "current_cycle", "predicted_rul", @@ -49,13 +47,12 @@ def test_status_404(): assert client.get("/equipment/9999/status").status_code == 404 -# ---------- /equipment/{id}/history ---------- def test_history_ascending_with_14_sensors(): rows = client.get("/equipment/1/history").json() cycles = [r["cycle"] for r in rows] assert cycles == sorted(cycles) and len(set(cycles)) == len(cycles) assert len(rows[0]["sensors"]) == 14 - assert "s1" not in rows[0]["sensors"] # константные датчики не отдаём + assert "s1" not in rows[0]["sensors"] def test_history_limit_returns_latest(): @@ -79,7 +76,6 @@ def test_history_404(): assert client.get("/equipment/9999/history").status_code == 404 -# ---------- /equipment/{id}/forecast ---------- def test_forecast_shape(): f = client.get("/equipment/1/forecast").json() pts = f["points"] @@ -91,7 +87,7 @@ def test_forecast_shape(): def test_forecast_rul_decreases_and_stays_non_negative(): pts = client.get("/equipment/1/forecast").json()["points"] ruls = [p["rul"] for p in pts] - assert all(a >= b for a, b in zip(ruls, ruls[1:])) + assert all(a >= b for a, b in itertools.pairwise(ruls)) assert all(p["rul"] >= 0 and p["rul_low"] >= 0 for p in pts) @@ -110,7 +106,6 @@ def test_forecast_404(): assert client.get("/equipment/9999/forecast").status_code == 404 -# ---------- /alerts ---------- def test_alerts_newest_first(): alerts = client.get("/alerts").json() assert len(alerts) > 0 @@ -134,7 +129,6 @@ def test_alerts_by_unknown_equipment_404(): assert client.get("/alerts/9999").status_code == 404 -# ---------- /predict ---------- def test_predict(): r = client.post("/predict", json={"equipment_id": 1}) assert r.status_code == 200 and r.json()["equipment_id"] == 1 From b66d18f5950e6e5fee0c5acf92b2784f5c99aa20 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=9E=D0=BB=D0=B8=D0=BC=D0=BE=D0=B2=20=D0=A4=D0=B0=D1=80?= =?UTF-8?q?=D0=B8=D0=B4=D1=83=D0=BD?= Date: Tue, 3 Nov 2026 22:43:30 +0300 Subject: [PATCH 3/3] the end --- backend/tests/test_endpoints.py | 136 -------------------------------- 1 file changed, 136 deletions(-) delete mode 100644 backend/tests/test_endpoints.py diff --git a/backend/tests/test_endpoints.py b/backend/tests/test_endpoints.py deleted file mode 100644 index 2f40570..0000000 --- a/backend/tests/test_endpoints.py +++ /dev/null @@ -1,136 +0,0 @@ -import itertools -from datetime import datetime - -from fastapi.testclient import TestClient -from main import app - -client = TestClient(app) -STATUSES = {"normal", "warning", "critical"} - - -def _ts(s: str) -> datetime: - return datetime.fromisoformat(s) - - -def test_list_equipment(): - r = client.get("/equipment") - assert r.status_code == 200 - items = r.json() - assert len(items) > 0 - assert {"id", "name", "type", "unit_number", "current_cycle", "status", "predicted_rul"} <= set(items[0]) - assert all(e["status"] in STATUSES for e in items) - - -def test_get_equipment_and_404(): - assert client.get("/equipment/1").json()["id"] == 1 - r = client.get("/equipment/9999") - assert r.status_code == 404 and "9999" in r.json()["detail"] - - -def test_status_fields_and_ranges(): - s = client.get("/equipment/1/status").json() - assert {"equipment_id", "status", "current_cycle", "predicted_rul", - "defect_probability", "residual"} <= set(s) - assert s["status"] in STATUSES - assert s["predicted_rul"] >= 0 - assert 0 <= s["defect_probability"] <= 1 - - -def test_status_matches_equipment_list(): - item = client.get("/equipment").json()[0] - s = client.get(f"/equipment/{item['id']}/status").json() - assert (s["status"], s["predicted_rul"], s["current_cycle"]) == \ - (item["status"], item["predicted_rul"], item["current_cycle"]) - - -def test_status_404(): - assert client.get("/equipment/9999/status").status_code == 404 - - -def test_history_ascending_with_14_sensors(): - rows = client.get("/equipment/1/history").json() - cycles = [r["cycle"] for r in rows] - assert cycles == sorted(cycles) and len(set(cycles)) == len(cycles) - assert len(rows[0]["sensors"]) == 14 - assert "s1" not in rows[0]["sensors"] - - -def test_history_limit_returns_latest(): - current = client.get("/equipment/1").json()["current_cycle"] - rows = client.get("/equipment/1/history", params={"limit": 5}).json() - assert len(rows) == 5 and rows[-1]["cycle"] == current - - -def test_history_period_filter(): - rows = client.get("/equipment/1/history", params={"from_cycle": 10, "to_cycle": 20}).json() - assert [r["cycle"] for r in rows] == list(range(10, 21)) - - -def test_history_invalid_params(): - assert client.get("/equipment/1/history", params={"from_cycle": 30, "to_cycle": 10}).status_code == 400 - assert client.get("/equipment/1/history", params={"limit": 0}).status_code == 422 - assert client.get("/equipment/1/history", params={"from_cycle": -1}).status_code == 422 - - -def test_history_404(): - assert client.get("/equipment/9999/history").status_code == 404 - - -def test_forecast_shape(): - f = client.get("/equipment/1/forecast").json() - pts = f["points"] - assert pts[0]["cycle"] == f["current_cycle"] - assert [p["cycle"] for p in pts] == list(range(f["current_cycle"], f["current_cycle"] + len(pts))) - assert pts[-1]["cycle"] == f["predicted_failure_cycle"] - - -def test_forecast_rul_decreases_and_stays_non_negative(): - pts = client.get("/equipment/1/forecast").json()["points"] - ruls = [p["rul"] for p in pts] - assert all(a >= b for a, b in itertools.pairwise(ruls)) - assert all(p["rul"] >= 0 and p["rul_low"] >= 0 for p in pts) - - -def test_forecast_interval_contains_rul(): - for p in client.get("/equipment/1/forecast").json()["points"]: - assert p["rul_low"] <= p["rul"] <= p["rul_high"] - - -def test_forecast_all_equipment_non_negative(): - for e in client.get("/equipment").json(): - pts = client.get(f"/equipment/{e['id']}/forecast").json()["points"] - assert min(p["rul"] for p in pts) >= 0 - - -def test_forecast_404(): - assert client.get("/equipment/9999/forecast").status_code == 404 - - -def test_alerts_newest_first(): - alerts = client.get("/alerts").json() - assert len(alerts) > 0 - times = [_ts(a["triggered_at"]) for a in alerts] - assert times == sorted(times, reverse=True) - assert all(a["severity"] in {"info", "warning", "critical"} for a in alerts) - - -def test_alerts_limit(): - assert len(client.get("/alerts", params={"limit": 2}).json()) == 2 - assert client.get("/alerts", params={"limit": 0}).status_code == 422 - - -def test_alerts_by_equipment(): - equipment_id = client.get("/alerts").json()[0]["equipment_id"] - alerts = client.get(f"/alerts/{equipment_id}").json() - assert alerts and all(a["equipment_id"] == equipment_id for a in alerts) - - -def test_alerts_by_unknown_equipment_404(): - assert client.get("/alerts/9999").status_code == 404 - - -def test_predict(): - r = client.post("/predict", json={"equipment_id": 1}) - assert r.status_code == 200 and r.json()["equipment_id"] == 1 - assert client.post("/predict", json={"equipment_id": 9999}).status_code == 404 - assert client.post("/predict", json={"equipment_id": 0}).status_code == 422