87 lines
2.6 KiB
Python
87 lines
2.6 KiB
Python
|
import pytest
|
||
|
|
|
||
|
|
from datetime import datetime
|
||
|
|
import os
|
||
|
|
|
||
|
|
# os.environ["DATABASE_URL"] = "sqlite:///./testDB.sqlite"
|
||
|
|
os.environ["DATABASE_URL"] = "sqlite://"
|
||
|
|
|
||
|
|
from srv import db
|
||
|
|
|
||
|
|
class Test_db:
|
||
|
|
_DB_URL = "sqlite://"
|
||
|
|
|
||
|
|
def setup(self):
|
||
|
|
self._db = db.EnergyDB(self._DB_URL)
|
||
|
|
|
||
|
|
def teardown(self):
|
||
|
|
pass
|
||
|
|
|
||
|
|
# --- helper functions
|
||
|
|
|
||
|
|
def _clearTable(self, tableName : str):
|
||
|
|
self._db.execute(f"DELETE FROM {tableName}")
|
||
|
|
|
||
|
|
# --- test functions
|
||
|
|
|
||
|
|
def test_url(self):
|
||
|
|
assert self._db.url() == self._DB_URL
|
||
|
|
|
||
|
|
def test_table(self):
|
||
|
|
assert self._db.table("energy") == self._db._tables["energy"]
|
||
|
|
assert self._db.table("channels") == self._db._tables["channels"]
|
||
|
|
|
||
|
|
def _test_table_unknownTable(self):
|
||
|
|
pass
|
||
|
|
|
||
|
|
def test_getChannels_emptyDatabase(self):
|
||
|
|
channels = self._db.getChannels()
|
||
|
|
assert channels["channels"] == []
|
||
|
|
|
||
|
|
def test_addChannels(self):
|
||
|
|
self._db.addChannels(["abc", "def", "ghi"])
|
||
|
|
result = self._db.getChannels()
|
||
|
|
assert type(result) == dict
|
||
|
|
channels = result["channels"]
|
||
|
|
assert channels[0] == {"name": "abc"}
|
||
|
|
assert channels[1] == {"name": "def"}
|
||
|
|
assert channels[2] == {"name": "ghi"}
|
||
|
|
|
||
|
|
def test_getChannelId(self):
|
||
|
|
self._db.addChannels(["abc", "def", "ghi"])
|
||
|
|
assert self._db.getChannelId("abc") == 1
|
||
|
|
assert self._db.getChannelId("def") == 2
|
||
|
|
assert self._db.getChannelId("ghi") == 3
|
||
|
|
|
||
|
|
def test_getChannelId_ExceptionIfChannelIsUnknown(self):
|
||
|
|
with pytest.raises(Exception):
|
||
|
|
assert self._db.getChannelId("jkl") == 0
|
||
|
|
|
||
|
|
def test_getChannelData_EmptyDatabase(self):
|
||
|
|
fromTime = datetime.now()
|
||
|
|
tillTime = datetime.now()
|
||
|
|
result = self._db.getChannelData([1], fromTime, tillTime)
|
||
|
|
assert list(result.keys()) == [1]
|
||
|
|
assert result[1] == []
|
||
|
|
|
||
|
|
def test_addChannelData(self):
|
||
|
|
data = [
|
||
|
|
{"timestamp": datetime.fromisoformat("2020-12-12T09:00:01"), "value": 900.01},
|
||
|
|
{"timestamp": datetime.fromisoformat("2020-12-12T09:05:02"), "value": 905.02},
|
||
|
|
{"timestamp": datetime.fromisoformat("2020-12-12T09:10:03"), "value": 910.03},
|
||
|
|
]
|
||
|
|
self._db.addChannelData(8, data)
|
||
|
|
result = self._db.getChannelData(
|
||
|
|
[8],
|
||
|
|
datetime.fromisoformat("2020-12-12T09:00:00"),
|
||
|
|
datetime.now()
|
||
|
|
)
|
||
|
|
assert isinstance(result, dict)
|
||
|
|
assert len(result) == 1
|
||
|
|
assert 8 in result
|
||
|
|
channelData = result[8]
|
||
|
|
assert len(channelData) == 3
|
||
|
|
assert channelData[0] == data[0]
|
||
|
|
assert channelData[1] == data[1]
|
||
|
|
assert channelData[2] == data[2]
|