Test pra
This commit is contained in:
parent
d1cebb444c
commit
f61d3ac0cf
16
papi/main.py
16
papi/main.py
|
@ -144,8 +144,22 @@ def post_sonde_error(
|
|||
):
|
||||
if not (sonde := crud.get_sonde(db, idsonde)):
|
||||
return
|
||||
|
||||
Notifier.error_sonde(idsonde)
|
||||
# create fake sample
|
||||
last = list(crud.get_mesure(db, sonde.sonde_id, only_last=1))[0]
|
||||
from copy import copy
|
||||
|
||||
api_error = copy(last.content)
|
||||
from datetime import datetime
|
||||
|
||||
date = str(datetime.now())
|
||||
for channel in api_error["channels"]:
|
||||
channel["status"] = "perte contact api"
|
||||
|
||||
api_error["date"] = date
|
||||
|
||||
crud.create_mesure(db, sonde.sonde_id, api_error)
|
||||
|
||||
return
|
||||
|
||||
|
||||
|
|
|
@ -25,11 +25,11 @@ def create_mesure(db: Session, sonde_id: int, content: dict):
|
|||
return db_mesure
|
||||
|
||||
|
||||
def get_mesure(db: Session, sonde_id: int, only_last: int =0):
|
||||
def get_mesure(db: Session, sonde_id: int, only_last: int = 0):
|
||||
q = db.query(models.Mesure).filter(models.Mesure.sonde_id == sonde_id)
|
||||
if not only_last:
|
||||
return q.all()
|
||||
else:
|
||||
q = q.order_by(models.Mesure.mesure_id.desc()).limit(only_last)
|
||||
|
||||
return reversed(q.all()) # order by id desc limit 1
|
||||
return list(reversed(q.all())) # order by id desc limit 1
|
||||
|
|
|
@ -105,16 +105,16 @@ def test_historique():
|
|||
|
||||
assert client.post("/sonde/", json=histo).ok
|
||||
|
||||
data = testutils.probe_sample_body(
|
||||
data_ok = testutils.probe_sample_body(
|
||||
channel_name="test_historique", channel_id=0, status="ok"
|
||||
)
|
||||
response = client.post(f"/sonde/{id_sonde}/", json=data)
|
||||
response = client.post(f"/sonde/{id_sonde}/", json=data_ok)
|
||||
assert response.ok
|
||||
|
||||
debut_supervision = client.get(f"/sonde/{id_sonde}/historique")
|
||||
debut_supervision = debut_supervision.json()
|
||||
assert len(debut_supervision) == 1
|
||||
assert debut_supervision[0] == f"{data['date']} test_historique absent ok"
|
||||
assert debut_supervision[0] == f"{data_ok['date']} test_historique absent ok"
|
||||
|
||||
data = testutils.probe_sample_body(
|
||||
channel_name="test_historique", channel_id=0, status="error"
|
||||
|
@ -137,6 +137,20 @@ def test_historique():
|
|||
perte_supervision = perte_supervision.json()
|
||||
assert perte_supervision[-1] == f"{data['date']} test_historique error absent"
|
||||
|
||||
# perte api et repise
|
||||
from papi.main import notifications
|
||||
|
||||
notifs = notifications[id_sonde]
|
||||
|
||||
response = client.post(f"/sonde/{id_sonde}/", json=data_ok)
|
||||
|
||||
perte_api = client.post(f"/sonde/{id_sonde}/error", json={})
|
||||
|
||||
perte_api_rapport = client.get(f"/sonde/{id_sonde}/rapport")
|
||||
assert "perte contact api" in perte_api_rapport.text
|
||||
assert perte_api.ok
|
||||
response = client.post(f"/sonde/{id_sonde}/", json=data_ok)
|
||||
|
||||
|
||||
def test_historique_rendering():
|
||||
idsonde = "masonde_001"
|
||||
|
|
Loading…
Reference in New Issue