255 lines
6.8 KiB
Python
255 lines
6.8 KiB
Python
from unittest import TestCase
|
|
import pytest
|
|
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from papi.main import app, get_db
|
|
|
|
import papi.main
|
|
from papi import main
|
|
|
|
|
|
from papi import __version__
|
|
from . import utils as testutils
|
|
|
|
from papi.sqlapp.database import Base
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
|
|
SQLALCHEMY_DATABASE_URL = "sqlite:///./test_sql_app.db"
|
|
|
|
engine = create_engine(
|
|
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
|
|
)
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
|
|
def override_get_db():
|
|
try:
|
|
db = TestingSessionLocal()
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
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
|
|
assert response.json() == {"msg": "Hello World"}
|
|
|
|
|
|
def test_creer_sonde():
|
|
|
|
sonde1 = {"identifiant": "masonde_001", "nom": "client1"}
|
|
response = client.post("/sonde/", json=sonde1)
|
|
assert response.status_code == 200
|
|
# FIXME: test code
|
|
# breakpoint()
|
|
# assert sonde1 in sondes.values()
|
|
|
|
|
|
def test_lister_sondes():
|
|
sonde1 = {"identifiant": "masonde_001", "nom": "client1"}
|
|
response = client.get("/sonde/")
|
|
assert response.status_code == 200
|
|
assert sonde1 in response.json()
|
|
|
|
|
|
def test_post_data():
|
|
id_ = "masonde_001"
|
|
from papi.main import mesures
|
|
|
|
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
|
|
== len(client.get(f"mesures/{id_}/").json())
|
|
)
|
|
|
|
|
|
def test_sample_report():
|
|
id_ = "masonde_001"
|
|
|
|
data = testutils.probe_sample_body()
|
|
response = client.post(f"/sonde/{id_}/", json=data)
|
|
assert response.status_code == 200
|
|
|
|
report = client.get(f"/sonde/{id_}/rapport")
|
|
rjson = report.json()
|
|
assert len(rjson) == 1
|
|
assert rjson[0] == "Test channel ok"
|
|
|
|
data["channels"][0]["status"] = "error"
|
|
response = client.post(f"/sonde/{id_}/", json=data)
|
|
|
|
report = client.get(f"/sonde/{id_}/rapport")
|
|
rjson = report.json()
|
|
assert len(rjson) == 1
|
|
assert rjson[0] == "Test channel error"
|
|
|
|
|
|
def test_historique():
|
|
id_sonde = "masonde_histo"
|
|
histo = {"identifiant": id_sonde, "nom": "Test historique"}
|
|
|
|
assert client.post("/sonde/", json=histo).ok
|
|
|
|
data_ok = testutils.probe_sample_body(
|
|
channel_name="test_historique", channel_id=0, status="ok"
|
|
)
|
|
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_ok['date']} test_historique absent -> ok"
|
|
|
|
data = testutils.probe_sample_body(
|
|
channel_name="test_historique", channel_id=0, status="error"
|
|
)
|
|
response = client.post(f"/sonde/{id_sonde}/", json=data)
|
|
assert response.ok
|
|
|
|
ok_erreur_supervision = client.get(f"/sonde/{id_sonde}/historique")
|
|
ok_erreur_supervision = ok_erreur_supervision.json()
|
|
assert ok_erreur_supervision[-1] == f"{data['date']} test_historique ok -> error"
|
|
|
|
data = testutils.probe_sample_body(
|
|
channel_name="test_historique", channel_id=0, status="error"
|
|
)
|
|
data["channels"] = []
|
|
response = client.post(f"/sonde/{id_sonde}/", json=data)
|
|
assert response.ok
|
|
|
|
perte_supervision = client.get(f"/sonde/{id_sonde}/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"
|
|
notifs_text = client.get(f"/sonde/{idsonde}/historique/text")
|
|
assert notifs_text.ok
|
|
content = notifs_text
|
|
assert f"absent" in content.text.lower()
|
|
|
|
|
|
def test_onchange_notification():
|
|
id_sonde = "masonde_notif"
|
|
histo = {"identifiant": id_sonde, "nom": "Test notification"}
|
|
assert client.post("/sonde/", json=histo).ok
|
|
from papi.main import notifications
|
|
|
|
notiflen = len(notifications[id_sonde])
|
|
assert notiflen == 0
|
|
|
|
data = testutils.probe_sample_body(
|
|
channel_name="test_historique", channel_id=0, status="ok"
|
|
)
|
|
response = client.post(f"/sonde/{id_sonde}/", json=data)
|
|
assert response.ok
|
|
assert len(notifications[id_sonde]) == notiflen + 1
|
|
|
|
jresp = response.json()
|
|
|
|
assert jresp["notify"]
|
|
assert jresp["notify"][0].split(" ", 1)[1] == "test_historique absent -> ok"
|
|
|
|
data = testutils.probe_sample_body(
|
|
channel_name="test_historique", channel_id=0, status="ok"
|
|
)
|
|
|
|
response = client.post(f"/sonde/{id_sonde}/", json=data)
|
|
assert response.ok
|
|
|
|
jresp = response.json()
|
|
assert not jresp["notify"]
|
|
|
|
notifs = client.get(f"/notifications/{id_sonde}/")
|
|
assert notifs.ok
|
|
assert len(notifs.json()) == 1
|
|
|
|
|
|
def test_notif_rendering():
|
|
idsonde = "masonde_001"
|
|
notifs_text = client.get(f"/notifications/{idsonde}/text")
|
|
assert notifs_text.ok
|
|
content = notifs_text
|
|
assert f"{idsonde}@fqdn" in content.text.lower()
|
|
|
|
|
|
class CodeCoverageTestCase(TestCase):
|
|
|
|
""" Get covergage to 100% """
|
|
|
|
def test_cover(self):
|
|
with pytest.raises(ValueError):
|
|
testutils.probe_sample_body(status="erruer")
|
|
|
|
def test_list_unregistred_probe(self):
|
|
from random import choices
|
|
import string
|
|
|
|
id_sonde = "".join(choices(string.ascii_lowercase, k=8))
|
|
with pytest.raises(LookupError):
|
|
r = client.get(f"/sonde/{id_sonde}/rapport")
|
|
with pytest.raises(LookupError):
|
|
r = client.post(f"/sonde/{id_sonde}/", json={})
|