142 lines
6.4 KiB
Python
142 lines
6.4 KiB
Python
from fastapi.testclient import TestClient
|
|
import pytest
|
|
|
|
from datetime import datetime
|
|
import json
|
|
import urllib.parse
|
|
|
|
from srv import energyDB
|
|
|
|
class Test_energyDb:
|
|
|
|
bulkTestData = [
|
|
{
|
|
"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
|
|
|
|
def _test_bulkData_put(self):
|
|
# response = self.client.put("/energy/bulkData", json=self.bulkTestData);
|
|
response = self.client.put("/energy/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_bulkData_get(self):
|
|
# 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(
|
|
"/energy/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(
|
|
"/energy/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("/energy/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"] == ""
|