PartagesPublics
/
papi
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Better count

This commit is contained in:
Colin Goutte 2021-11-12 17:33:55 +01:00
parent f3be2997dc
commit f6fd13043e
4 changed files with 88 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@ -32,7 +32,7 @@ test: cleandb
.ONESHELL:
dtest: cleandb
poetry run pytest -s --pdb --cov=. --cov-report=term-missing:skip-covered .
poetry run pytest -s --ff --lf --pdb --cov=. --cov-report=term-missing:skip-covered .
.ONESHELL:
qtest: cleandb

103
papi/main.py
View File

@ -49,7 +49,6 @@ def sondeid2notifsemails(idsonde, mapping=None):
if mapping is None:
mapping = PROBES
try:
breakpoint()
section = mapping[idsonde]
except KeyError: # pragma: no cover
raise KeyError(f"{idsonde} non trouvé dans {list(mapping.keys())}")
@ -78,9 +77,6 @@ def atomic_db():
db.close()
sondes = {"test": schemas.SondeBase(identifiant="test", nom="Testlocal")}
def apply_config(configs: list[dict] = []):
with atomic_db() as db:
for config in configs:
@ -102,8 +98,8 @@ apply_config(configurations)
class Notifier:
@staticmethod
def __call__(idsonde, changes):
status = get_rapport(idsonde)
def __call__(db, idsonde, changes):
status = get_rapport(idsonde, db)
notifications[idsonde].append(
{"changes": changes, "status": [x for x in status]}
)
@ -138,10 +134,6 @@ class Notifier:
Notifier = Notifier()
for k, sonde in PROBES.items():
sondes[k] = schemas.SondeBase(
identifiant=sonde["identifiant_sonde"], nom=sonde["nom_sonde"]
)
def default_sample():
@ -154,22 +146,22 @@ def read_root():
return {"msg": "Hello World"}
@app.post("/sonde/", response_model=List[schemas.SondeBase])
@app.post("/sonde/", response_model=schemas.SondeBase)
def post_sonde(sonde: schemas.SondeBase, db: Session = Depends(get_db)):
db_sonde = crud.get_sonde(db, identifiant=sonde.identifiant)
if not db_sonde:
db_sonde = crud.create_sonde(
db, sonde.identifiant, sonde.nom or sonde.identifiant
)
sondes[sonde.identifiant] = schemas.SondeBase.from_orm(
db_sonde
) # Attention au typage ici
return list(sondes.values())
# sondes[sonde.identifiant] = schemas.SondeBase.from_orm(
# db_sonde
# ) # Attention au typage ici
return schemas.SondeBase.from_orm(db_sonde)
@app.get("/sonde/", response_model=List[schemas.SondeBase])
def list_sonde():
return [x for x in sondes.values()]
def list_sonde(db: Session = Depends(get_db)):
return crud.list_sonde(db)
@app.get("/notifications/{idsonde}")
@ -218,7 +210,7 @@ def post_sonde_error(
):
if not (sonde := crud.get_sonde(db, idsonde)):
if idsonde != "test":
return # pragma: no cover
raise LookupError(f"{idsonde} not found")
Notifier.error_sonde(idsonde, body["msg"])
@ -251,18 +243,18 @@ def post_sonde_data(
db: Session = Depends(get_db),
):
if not (sonde := crud.get_sonde(db, idsonde)):
raise LookupError(f"{idsonde} not found")
return
mesures_ = [x for x in sonde.mesures]
if not mesures_:
mesures_.append(
crud.create_mesure(db, sonde.sonde_id, content=default_sample())
)
mesures_.append(crud.create_mesure(db, sonde.sonde_id, content=body))
mesures[idsonde] = mesures_
if not sonde.mesures:
crud.create_mesure(db, sonde.sonde_id, content=default_sample())
crud.create_mesure(db, sonde.sonde_id, content=body)
last_two = crud.get_mesure(db, sonde_id=sonde.sonde_id, only_last=2)
date = body["date"]
previous = utils.prepare(mesures_[-2].content)
present = utils.prepare(mesures_[-1].content)
previous = utils.prepare(last_two[-2].content)
present = utils.prepare(last_two[-1].content)
all_channels = sorted(set((*previous["channels"], *present["channels"])))
diff = utils.compare(all_channels, previous, present)
@ -273,42 +265,49 @@ def post_sonde_data(
else ""
)
if content:
Notifier(idsonde, content)
Notifier(db, idsonde, content)
res = last_notif_text(request, idsonde)
html = res.body.decode("utf-8")
from papi.mail_sendermodel import sendmail
conf = PROBES[idsonde]
sumup = utils.sample2statuscount(present)
subject = " ; ".join(sumup)
sendmail(
htmlpart=html,
emails=conf["emails"],
subject=subject,
)
if idsonde in PROBES:
conf = PROBES[idsonde]
sumup = utils.sample2statuscount(present)
subject = " ; ".join(sumup)
sendmail(
htmlpart=html,
emails=conf["emails"],
subject=subject,
)
cnt = crud.get_mesure(db, sonde_id=sonde.sonde_id)
offset = 0 if "date" in cnt[0].content else 1
return {
"count": len(mesures_)
if "date" in mesures_[0].content.keys()
else len(mesures_) - 1,
"count": len(crud.get_mesure(db, sonde_id=sonde.sonde_id)) - offset,
"notify": content,
}
@app.get("/sonde/{idsonde}/rapport")
def get_rapport(idsonde):
try:
last = mesures[idsonde][-1]
except IndexError:
return
else:
last = last.content
last = utils.prepare(last)
@app.get("/mesures/{idsonde}/")
def list_sonde_mesures(idsonde, db: Session = Depends(get_db)):
if not (sonde := crud.get_sonde(db, idsonde)):
raise LookupError(f"{idsonde} not found")
return [x for x in sonde.mesures if "date" in x.content]
for name, data in sorted(
last["channels"].items(), key=lambda text: float(text[0].split("#")[0])
):
yield f'{name.split("#", 1)[1]} {data["status"]}'
@app.get("/sonde/{idsonde}/rapport")
def get_rapport(idsonde, db: Session = Depends(get_db)):
if not (sonde := crud.get_sonde(db, idsonde)):
raise LookupError
last = crud.get_mesure(db, sonde_id=sonde.sonde_id, only_last=1)[0]
last = last.content
last = utils.prepare(last)
for name, data in sorted(
last["channels"].items(), key=lambda text: float(text[0].split("#")[0])
):
yield f'{name.split("#", 1)[1]} {data["status"]}'
@app.get("/sonde/{idsonde}/historique")

4
papi/sqlapp/crud.py
View File

@ -9,6 +9,10 @@ def get_sonde(db: Session, identifiant: str):
return q.first()
def list_sonde(db: Session):
return db.query(models.Sonde).all()
def create_sonde(db: Session, identifiant: str, nom: str):
db_sonde = models.Sonde(identifiant=identifiant, nom=nom)
db.add(db_sonde)

36
tests/test_papi.py
View File

@ -7,7 +7,6 @@ from fastapi.testclient import TestClient
from papi.main import app, get_db
from papi.main import sondes
import papi.main
from papi import main
@ -44,10 +43,32 @@ app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
probes_fixture = [
# {"identifiant": "masonde_1", "nom": "client1"},
# {"identifiant": "masonde_001", "nom": "client1"},
# {"identifiant": "masonde_histo", "nom": "client_histo"},
]
posted = []
def setup():
for probe_data in probes_fixture:
data = tuple(probe_data.items())
# breakpoint()
if data not in posted:
# response = client.post("/sonde/", json=probe_data)
# assert response.ok
# posted.append(data)
pass
def test_version():
assert __version__ == "0.1.0"
# setup()
def test_read_main():
response = client.get("/")
assert response.status_code == 200
@ -55,10 +76,13 @@ def test_read_main():
def test_creer_sonde():
sonde1 = {"identifiant": "masonde_001", "nom": "client1"}
response = client.post("/sonde/", json=sonde1)
assert response.status_code == 200
assert sonde1 in sondes.values()
# FIXME: test code
# breakpoint()
# assert sonde1 in sondes.values()
def test_lister_sondes():
@ -72,13 +96,17 @@ def test_post_data():
id_ = "masonde_001"
from papi.main import mesures
curlen = len(mesures[id_])
curlen = len(client.get(f"mesures/{id_}/").json())
data = testutils.probe_sample_body()
response = client.post(f"/sonde/{id_}/", json=data)
assert response.status_code == 200
assert response.json()["count"] == curlen + 1
assert (
response.json()["count"]
== curlen + 1
== len(client.get(f"mesures/{id_}/").json())
)
def test_sample_report():