hk/energyDB
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
energyDB/test/test_EnergyDB.py

323 lines
14 KiB
Python
Raw Normal View History

Initial commit
2020-12-17 01:01:27 +01:00
from fastapi.testclient import TestClient
import pytest
from datetime import datetime
import json
Use the database URL given by the environment variable DATABASE_URL
2020-12-19 02:06:08 +00:00
import os
Initial commit
2020-12-17 01:01:27 +01:00
import urllib.parse
Use the database URL given by the environment variable DATABASE_URL
2020-12-19 02:06:08 +00:00
#TODO Use in-memory DB to test the case that there is no table
#TODO Add helper function to fill the in-memory DB before test
Add table "channels" + move some of the db functionas to module db.py
2020-12-21 23:52:18 +00:00
# os.environ["DATABASE_URL"] = "sqlite:///./testDB.sqlite"
os.environ["DATABASE_URL"] = "sqlite://"
Use the database URL given by the environment variable DATABASE_URL
2020-12-19 02:06:08 +00:00
Initial commit
2020-12-17 01:01:27 +01:00
from srv import energyDB
Add table "channels" + move some of the db functionas to module db.py
2020-12-21 23:52:18 +00:00
class Test_energyDB:
restApiRoot = "/energy/v1"
testData = {
"channels": (
{"name": "dc_power1"},
{"name": "daily_yield"},
{"name": "total_yield"},
),
"bulkdata": (
{
"channel_id": 1,
"data": (
{ "timestamp": "2020-12-11T12:00:22", "value": 1100.1 },
{ "timestamp": "2020-12-11T12:10:15", "value": 1109.2 },
{ "timestamp": "2020-12-11T12:20:13", "value": 1119.3 },
{ "timestamp": "2020-12-11T12:30:21", "value": 1131.4 },
{ "timestamp": "2020-12-11T12:40:08", "value": 1143.5 },
{ "timestamp": "2020-12-11T12:50:13", "value": 1152.6 },
{ "timestamp": "2020-12-11T13:00:11", "value": 1160.7 },
{ "timestamp": "2020-12-11T13:10:09", "value": 1169.8 },
{ "timestamp": "2020-12-11T13:20:10", "value": 1181.9 },
{ "timestamp": "2020-12-11T13:30:17", "value": 1190.0 },
)
},
{
"channel_id": 2,
"data": [
{ "timestamp": "2020-12-11T12:01:15", "value": 1200.1 },
{ "timestamp": "2020-12-11T12:21:28", "value": 1219.2 },
{ "timestamp": "2020-12-11T12:41:21", "value": 1243.3 },
{ "timestamp": "2020-12-11T13:01:16", "value": 1260.4 },
{ "timestamp": "2020-12-11T13:21:18", "value": 1281.5 },
]
}
)
}
# client = TestClient(energyDB)
def setup(self):
self.client = TestClient(energyDB)
def teardown(self):
self.client = None
# --- helper functions
def _apiUrl(self, sub_url : str):
if sub_url.startswith("/"):
return self.restApiRoot + sub_url
else:
return self.restApiRoot + "/" + sub_url
def _fillDatabase(self):
Introduce version number in RESP API root like "/energy/v1/"
2020-12-17 15:09:27 +00:00
response = self.client.put(
Add table "channels" + move some of the db functionas to module db.py
2020-12-21 23:52:18 +00:00
self._apiUrl("/channels"),
json = {"channels": self.testData["channels"]}
)
# self._dumpRequestAndResponse("_fillDatabase(channels)", response)
Initial commit
2020-12-17 01:01:27 +01:00
assert response.status_code == 200
Add table "channels" + move some of the db functionas to module db.py
2020-12-21 23:52:18 +00:00
response = self.client.put(
self._apiUrl("/bulkData"),
json = {"bulk": self.testData["bulkdata"]}
)
# self._dumpRequestAndResponse("_fillDatabase(data)", response)
assert response.status_code == 200
def _dumpRequestAndResponse(self, context : str, response):
print("\n")
print(f"---- request ({context})")
Initial commit
2020-12-17 01:01:27 +01:00
print(f"response.request.method: {response.request.method}")
print(f"response.request.url: {urllib.parse.unquote(response.request.url)}")
print(f"response.request.headers: {response.request.headers}")
Add table "channels" + move some of the db functionas to module db.py
2020-12-21 23:52:18 +00:00
try:
requestJson = json.loads(response.request.body)
print(f"response.request.body(json): {json.dumps(requestJson, indent=2)}")
except:
print(f"response.request.body(plain): {response.request.body}")
print(f"---- response ({context})")
print(f"response.status_code: {response.status_code}")
Initial commit
2020-12-17 01:01:27 +01:00
print(f"response.reason: {response.reason}")
Add table "channels" + move some of the db functionas to module db.py
2020-12-21 23:52:18 +00:00
try:
responseJson = json.loads(response.text)
print(f"response.text(json): {json.dumps(responseJson, indent=2)}")
except:
print(f"response.text(plain): {response.text}")
# --- test functions
def test_invalidRoute(self):
response = self.client.get("/")
assert response.status_code == 404
def test_getVersion(self):
response = self.client.get(self._apiUrl("/version"))
# self._dumpRequestAndResponse("test_getVersion", response)
assert response.status_code == 200
assert response.json()["version"] == "1.0"
def _test_getInfo(self):
response = self.client.get( self._apiUrl("/info" ))
# self._dumpRequestAndResponse("test_getInfo", response)
assert response.status_code == 200
def _test_getChannelsOfEmptyTable(self):
response = self.client.get(self._apiUrl("/channels"))
# self._dumpRequestAndResponse("test_getChannelsOfEmptyTable", response)
assert response.status_code == 200
assert response.json()["channels"] == []
def _test_getBulkDataOfEmptyTable(self):
response = self.client.get(
self._apiUrl("/bulkData"),
json = {
"channel_ids": [1],
"fromTime": "0001-01-01T00:00:00"
}
)
self._dumpRequestAndResponse("test_getBulkDataOfEmptyTable", response)
assert response.status_code == 200
assert "bulk" in response.json()
assert response.json()["bulk"] == [
{
"channel_id": 1,
"channel": None,
"data": [],
"msg": None
}
]
def test_fillDatabase(self):
self._fillDatabase()
response = self.client.get(self._apiUrl("/1/count"))
# self._dumpRequestAndResponse("test_fillDatabase", response)
assert response.status_code == 200
def test_getChannels(self):
response = self.client.get(self._apiUrl("/channels"))
# self._dumpRequestAndResponse("test_getChannels", response)
assert response.status_code == 200
def test_getChannelId(self):
response = self.client.get(self._apiUrl("/channels/total_yield/id"))
# self._dumpRequestAndResponse("test_getChannelId", response)
assert response.status_code == 200
response = self.client.get(self._apiUrl("/channels/dc_power1/id"))
# self._dumpRequestAndResponse("test_getChannelId", response)
assert response.status_code == 200
response = self.client.get(self._apiUrl("/channels/daily_yield/id"))
# self._dumpRequestAndResponse("test_getChannelId", response)
Initial commit
2020-12-17 01:01:27 +01:00
assert response.status_code == 200
Add table "channels" + move some of the db functionas to module db.py
2020-12-21 23:52:18 +00:00
def test_putChannels(self):
response = self.client.get(self._apiUrl("/channels/frequency/id"))
# self._dumpRequestAndResponse("test_getChannelId", response)
assert response.status_code == 500
Initial commit
2020-12-17 01:01:27 +01:00
response = self.client.put(
Add table "channels" + move some of the db functionas to module db.py
2020-12-21 23:52:18 +00:00
self._apiUrl("/channels"),
json = {"channels": [{"name": "frequency"}]}
)
# self._dumpRequestAndResponse("test_fillDatabase", response)
assert response.status_code == 200
response = self.client.get(self._apiUrl("/channels/frequency/id"))
# self._dumpRequestAndResponse("test_getChannelId", response)
Initial commit
2020-12-17 01:01:27 +01:00
assert response.status_code == 200
Add table "channels" + move some of the db functionas to module db.py
2020-12-21 23:52:18 +00:00
def _test_putBulkData(self):
response = self.client.put(
self._apiUrl("/bulkData"),
json = {"bulk": {
"channel_id": None,
"channel": "total_yield",
"data": [
{ "timestamp": "2020-12-11T12:01:20", "value": 120120.1 },
{ "timestamp": "2020-12-11T12:30:25", "value": 123025.2 },
]
}}
)
self._dumpRequestAndResponse("test_putBulkData", response)
Initial commit
2020-12-17 01:01:27 +01:00
assert response.status_code == 200
Add table "channels" + move some of the db functionas to module db.py
2020-12-21 23:52:18 +00:00
# def test_getRecordCount(self):
# response = self.client.get(self._apiUrl("/1/count"))
# self._dumpRequestAndResponse("test_getRecordCount", response)
# assert response.status_code == 200
# assert response.json()["info"] == {"columns"}
# def _test_bulkData_put(self):
# # response = self.client.put("/energy/bulkData", json=self.bulkTestData);
# response = self.client.put(
# self.restApiRoot + "bulkData",
# json={"bulk": self.bulkTestData})
# # print(f"dir(response): {dir(response)}")
# # print(f"dir(response.request): {dir(response.request)}")
# # print("---- request")
# # print(f"response.request.method: {response.request.method}")
# # print(f"response.request.url: {urllib.parse.unquote(response.request.url)}")
# # print(f"response.request.headers: {response.request.headers}")
# # requestJson = json.loads(response.request.body)
# # print(f"response.request.body: {json.dumps(requestJson, indent=2)}")
# print("---- response")
# print(f"response.reason: {response.reason}")
# responseJson = json.loads(response.text)
# print(f"response.text: {json.dumps(responseJson, indent=2)}")
# # print(f"response.text: {json.dumps(response.text, indent=2)}")
# assert response.status_code == 200
# def _test_getInfo2(self):
# response = self.client.get( self.restApiRoot + "info" )
# # print(f"dir(response): {dir(response)}")
# # print(f"dir(response.request): {dir(response.request)}")
# print("---- request")
# print(f"response.request.method: {response.request.method}")
# print(f"response.request.url: {urllib.parse.unquote(response.request.url)}")
# print(f"response.request.headers: {response.request.headers}")
# # print(f"dir(response.request): {dir(response.request)}")
# print("---- response")
# print(f"response.reason: {response.reason}")
# # print(f"response.text: {response.text}")
# responseJson = json.loads(response.text)
# print(f"response.text: {json.dumps(responseJson, indent=2)}")
# assert False #response.status_code == 404
# def _test_bulkData_get(self):
# print(f"DB_URL: {os.getenv('DATABASE_URL')}")
# # response = self.client.put("/energy/bulkData", json=self.bulkTestData);
# fromTimestamp = datetime.fromisoformat("2020-12-11T12:30:00")
# tillTimestamp = datetime.fromisoformat("2020-12-11T12:30:59")
# response = self.client.get(
# self.restApiRoot + "bulkData",
# json = {
# "channel_ids": [1, 2, 3],
# "fromTime": fromTimestamp.isoformat(),
# # "tillTime": tillTimestamp.isoformat()
# })
# # print(f"dir(response): {dir(response)}")
# # print(f"dir(response.request): {dir(response.request)}")
# print("---- request")
# print(f"response.request.method: {response.request.method}")
# print(f"response.request.url: {urllib.parse.unquote(response.request.url)}")
# print(f"response.request.headers: {response.request.headers}")
# print(f"dir(response.request): {dir(response.request)}")
# print(f"response.request.body: {response.request.body}")
# # requestJson = json.loads(response.request.body)
# # print(f"response.request.body: {json.dumps(requestJson, indent=2)}")
# print("---- response")
# print(f"response.reason: {response.reason}")
# responseJson = json.loads(response.text)
# print(f"response.text: {json.dumps(responseJson, indent=2)}")
# # print(f"response.text: {json.dumps(response.text, indent=2)}")
# assert response.status_code == 200
# @pytest.mark.skip(reason="Ignore me temporarily")
# def test_insert_energy(self):
# energyData = {
# "timestamp": datetime.now().isoformat(),
# "value": 234.5,
# }
# print(f"energyData: {energyData}")
# # response = self.client.put("/energies/1", json=energyData) #, headers={"X-Token": "coneofsilence"})
# response = self.client.put(
# self.restApiRoot + "2",
# # params=energyData,
# json=energyData) #, headers={"X-Token": "coneofsilence"})
# # print(f"dir(response): {dir(response)}")
# # print(f"dir(response.request): {dir(response.request)}")
# # print("---- request")
# # print(f"response.request.method: {response.request.method}")
# # print(f"response.request.url: {urllib.parse.unquote(response.request.url)}")
# # print(f"response.request.headers: {response.request.headers}")
# # print(f"response.request.body: {json.loads(response.request.body)}")
# # print("---- response")
# # print(f"response.reason: {response.reason}")
# # print(f"response.text: {json.loads(response.text)}")
# # print(f"request.header: \"{response.request.header}\"")
# # print(f"request.body: \"{response.request.body}\"")
# assert response.status_code == 200
# # assert response.json()["msg"] == ""
# @pytest.mark.skip(reason="Ignore me temporarily")
# def test_get_energy(self):
# response = self.client.get(self.restApiRoot + "1")
# # print(f"dir(response): {dir(response)}")
# # print(f"dir(response.request): {dir(response.request)}")
# # print("---- request")
# # print(f"response.request.method: {response.request.method}")
# # print(f"response.request.url: {urllib.parse.unquote(response.request.url)}")
# # print(f"response.request.headers: {response.request.headers}")
# # print("---- response")
# # print(f"response.reason: {response.reason}")
# responseJson = json.loads(response.text)
# print(f"response.text: {json.dumps(responseJson, indent=2)}")
# data = response.json()
# # print(f"data: {type(data)}")
# # for k,v in data.items():
# # print(f"key: {k} -> value: {v}")
# print(f"data of channel: {data['channel_id']}")
# for r in data["data"]:
# print(f"r: {r}")
# assert response.status_code == 200
# # assert response.json()["msg"] == ""
Reference in a new issue Copy permalink