papi/tests/test_papi.py

from unittest import TestCase
import pytest


from fastapi import FastAPI
from fastapi.testclient import TestClient

from papi.main import app, get_db

from papi.main import sondes
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)


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_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))
        r = client.get(f"/sonde/{id_sonde}/rapport")
        r = client.post(f"/sonde/{id_sonde}/", json={})

        assert r.ok