from unittest import TestCase import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from papi.main import app from papi.main import sondes from papi import __version__ from . import utils as testutils client = TestClient(app) def test_version(): assert __version__ == "0.1.0" 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 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(mesures[id_]) data = testutils.probe_sample_body() response = client.post(f"/sonde/{id_}/", json=data) assert response.status_code == 200 assert response.json()["count"] == curlen + 1 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 = 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 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" 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" 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_template_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)) r = client.get(f"/sonde/{id_sonde}/rapport") r = client.post(f"/sonde/{id_sonde}/", json={}) assert r.ok