363 lines
16 KiB
Python
363 lines
16 KiB
Python
from fastapi.testclient import TestClient
|
|
import pytest
|
|
|
|
from datetime import datetime
|
|
import json
|
|
import os
|
|
import urllib.parse
|
|
|
|
#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
|
|
|
|
# os.environ["DATABASE_URL"] = "sqlite:///./testDB.sqlite"
|
|
os.environ["DATABASE_URL"] = "sqlite://"
|
|
|
|
from srv import energyDB
|
|
|
|
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):
|
|
response = self.client.put(
|
|
self._apiUrl("/channels"),
|
|
json = {"channels": self.testData["channels"]}
|
|
)
|
|
# self._dumpRequestAndResponse("_fillDatabase(channels)", response)
|
|
assert response.status_code == 200
|
|
|
|
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})")
|
|
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}")
|
|
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}")
|
|
print(f"response.reason: {response.reason}")
|
|
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()
|
|
bulkData = response.json()["bulk"]
|
|
assert len(bulkData) == 1
|
|
assert "channel_id" in bulkData[0]
|
|
assert "data" in bulkData[0]
|
|
assert bulkData[0]["channel_id"] == 1
|
|
channelData = bulkData[0]["data"]
|
|
assert len(channelData) == 0
|
|
|
|
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)
|
|
assert response.status_code == 200
|
|
|
|
def test_putChannels(self):
|
|
response = self.client.get(self._apiUrl("/channels/frequency/id"))
|
|
# self._dumpRequestAndResponse("test_getChannelId", response)
|
|
assert response.status_code == 500
|
|
|
|
response = self.client.put(
|
|
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)
|
|
assert response.status_code == 200
|
|
|
|
def test_getBulkData(self):
|
|
response = self.client.get(
|
|
self._apiUrl("/bulkData"),
|
|
json = {
|
|
"channel_ids": [1],
|
|
"fromTime": "0001-01-01T00:00:00"
|
|
}
|
|
)
|
|
# self._dumpRequestAndResponse("test_getBulkData", response)
|
|
assert response.status_code == 200
|
|
assert "bulk" in response.json()
|
|
bulkData = response.json()["bulk"]
|
|
assert len(bulkData) == 1
|
|
channelData = bulkData[0]
|
|
assert channelData["channel_id"] == 1
|
|
referenceData = self.testData["bulkdata"][0]
|
|
assert len(channelData["data"]) == len(referenceData["data"])
|
|
assert channelData["data"][0] == referenceData["data"][0]
|
|
assert channelData["data"][1] == referenceData["data"][1]
|
|
assert channelData["data"][2] == referenceData["data"][2]
|
|
assert channelData["data"][3] == referenceData["data"][3]
|
|
assert channelData["data"][4] == referenceData["data"][4]
|
|
assert channelData["data"][5] == referenceData["data"][5]
|
|
assert channelData["data"][6] == referenceData["data"][6]
|
|
assert channelData["data"][7] == referenceData["data"][7]
|
|
assert channelData["data"][8] == referenceData["data"][8]
|
|
assert channelData["data"][9] == referenceData["data"][9]
|
|
|
|
def test_putBulkData(self):
|
|
newData = [{
|
|
"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 },
|
|
]
|
|
}]
|
|
response = self.client.put(
|
|
self._apiUrl("/bulkData"),
|
|
json = {"bulk": newData}
|
|
)
|
|
# self._dumpRequestAndResponse("test_putBulkData", response)
|
|
assert response.status_code == 200
|
|
response = self.client.get(
|
|
self._apiUrl("/bulkData"),
|
|
json = {
|
|
"channel_ids": [3],
|
|
"fromTime": "2020-12-11T12:00:00",
|
|
"tillTime": "2020-12-11T12:59:59"
|
|
}
|
|
)
|
|
channelData = response.json()["bulk"][0]
|
|
assert channelData["channel_id"] == 3
|
|
assert len(channelData["data"]) == 2
|
|
assert channelData["data"][0] == newData[0]["data"][0]
|
|
assert channelData["data"][1] == newData[0]["data"][1]
|
|
|
|
# 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"] == ""
|