note-samples/python-route-endpoint/test_dbstore.py at main · blues/note-samples · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
import pytest
import dbstore

inMemFile = ":memory:"

measurementTable = "measurements"
alertTable = "alerts"

def test_db_store_constructor():
    s = dbstore.dbstore(file=inMemFile)

    assert(s != None)


def test_dbStore_connect():
    s = dbstore.dbstore(file=inMemFile)
    s.connect()

    assert s._connection is not None

def test_dbStore_connect_whenConnectionIsOpen():
    s = dbstore.dbstore(file=inMemFile)
    s.connect()
    c = s._connection
    s.connect()
    assert s._connection == c


def test_dbStore_close_whenConnectionIsOpen():
    s = dbstore.dbstore(file=inMemFile)
    s.connect()
    assert s._connection is not None

    s.close()
    assert s._connection is None

def test_dbStore_close_whenConnectionIsClosed():
    s = dbstore.dbstore(file=inMemFile)
    assert s._connection is None
    s.close()
    assert s._connection is None


def test_dbStore_createTables():
    s = dbstore.dbstore(file="inMemFile")
    s.connect()
    s.createTables()
    for n in [measurementTable, alertTable]:
        s._cursor.execute(f"SELECT count(name) FROM sqlite_master WHERE type='table' AND name='{n}';")
        isTable = s._cursor.fetchone()[0]==1
        assert isTable


timestampTestData = "2021-04-29T23:25:44Z"

def generateConnectedInMemDb() -> dbstore.dbstore:
    s = dbstore.dbstore(file=inMemFile)
    s.connect()
    s.createTables()
    return s

def test_addMeasurement():
    s = generateConnectedInMemDb()

    deviceId = "dev:xxxxxxxxxxxx"
    measurementType = "sensor1"
    timestamp = timestampTestData
    value = 3.14
    units = "units1"

    s.addMeasurement(deviceId, timestamp, measurementType, value, units)

    c = s._cursor.execute(f'SELECT * from {measurementTable}')
    row = c.fetchone()


    assert row[0]  == deviceId
    assert row[1]  == timestamp
    assert row[2]  == measurementType
    assert row[3]  == value
    assert row[4]  == units


def test_addAlert():
    s = generateConnectedInMemDb()

    deviceId = "dev:xxxxxxxxxxxx"
    alertType = "overfill"
    timestamp = timestampTestData
    message = "message 1"

    s.addAlert(deviceId, timestamp, alertType, message)

    c = s._cursor.execute(f'SELECT * from {alertTable}')
    row = c.fetchone()


    assert row[0]  == deviceId
    assert row[1]  == timestamp
    assert row[2]  == alertType
    assert row[3]  == message


def test_getAlerts():
    s = generateConnectedInMemDb()

    deviceId = "dev:xxxxxxxxxxxx"
    alertType = "overfill"
    timestamp = timestampTestData
    message = "message 1"

    s.addAlert(deviceId, timestamp, alertType, message)
    s.addAlert(deviceId, timestamp, alertType, message)

    a = s.getAlerts()

    e = [{"deviceId":deviceId,"timestamp":timestamp,"type":alertType,"message":message},
         {"deviceId":deviceId,"timestamp":timestamp,"type":alertType,"message":message},]

    assert a == e

def test_getAlerts_noAlertsStored():
    s = generateConnectedInMemDb()
    a = s.getAlerts()
    assert a == []

def test_getAlerts_withLimit():
    s = generateConnectedInMemDb()

    deviceId = "dev:xxxxxxxxxxxx"
    alertType = "overfill"
    timestamp = timestampTestData
    message = "message 1"

    s.addAlert(deviceId, timestamp, alertType, message)
    s.addAlert(deviceId, timestamp, alertType, message)

    a = s.getAlerts(limit=1)

    e = [{"deviceId":deviceId,"timestamp":timestamp,"type":alertType,"message":message}]

    assert a == e