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() 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) 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_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) assert response.status_code == 200 # 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"] == ""