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:///./energyDB.sqlite" from srv import energyDB class Test_energyDb: restApiRoot = "/energy/v1/" 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( 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_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"] == ""