import json import pytest from unify_api.tests.constants_t import TOKEN, HTTP_PREFIX from pot_libs.aiohttp_util.aiohttp_utils import AioHttpUtils @pytest.mark.parametrize('data', [ { "page_size": 20, "page_num": 1, "filter": { "equals": [ { "field": "point_id", "value": 175 } ], "ranges": [ { "field": "datetime", "start": "2020-07-22 00:00:00", "end": "2020-07-22 23:59:59" } ], "in_groups": [ { "field": "location_id", "group": [ 259, 260, 261, 262, 263 ] } ] }, "sort": { "field": "datetime", "direction": "desc" } }, ]) @pytest.mark.asyncio async def test_adio_history(data): """安全监测历史曲线""" resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/adio/adio/adio-history", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) len_temperature_1 = len(resp_str["data"]["temperature"][0]["value_slots"]) len_temperature_2 = len(resp_str["data"]["temperature"][1]["value_slots"]) len_temperature_3 = len(resp_str["data"]["temperature"][2]["value_slots"]) len_time_slots = len(resp_str["data"]["time_slots"]) assert len(list({len_temperature_1, len_temperature_2, len_temperature_3, len_time_slots})) == 1 @pytest.mark.parametrize('data', [ { "filter": { "equals": [], "ranges": [], "in_groups": [ { "field": "location_id", "group": [765, 766, 767, 768, 769] } ] } }, ]) @pytest.mark.asyncio async def test_adio_current(data): """安全监测实时参数""" resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/adio/adio/adio-current", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert resp_str["data"]["residual_current"] and \ resp_str["data"]["temperature"] @pytest.mark.parametrize('data', [ { "page_size": 20, "page_num": 1, "filter": { "equals": [ { "field": "point_id", "value": 327 } ], "ranges": [ { "field": "datetime", "start": "2021-07-02 00:00:00", "end": "2021-07-02 23:59:59" } ], "in_groups": [ { "field": "location_id", "group": [765, 766, 767, 768, 769] } ] }, "sort": { "field": "datetime", "direction": "desc" } }, ]) @pytest.mark.asyncio async def test_adio_index(data): """ 安全监测指标统计 温度和漏电流的最高、最低、平均值 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/adio/adio/adio-index", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert resp_str["data"]["adio_indexes"] @pytest.mark.parametrize('data', [ { "location_list": [988, 989, 990, 991, 992], "cid": 118, }, ]) @pytest.mark.asyncio async def test_adio_card_service(data): """ 安全监测-卡片信息-level """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/adio/adio-card/adio-card", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert resp_str["data"]["feeder"][0]["point_id"] == 450