energyDB/test/test_EnergyDB.py

from fastapi.testclient import TestClient
import pytest

from datetime import datetime
import json
import urllib.parse

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):
        # 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"] == ""