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', [ { "filter": { "equals": [{"field": "point_id", "value": 53}], "ranges": [ { "field": "datetime", "start": "2020-07-30 00:00:00", "end": "2020-07-30 23:59:59" } ] } }, ]) @pytest.mark.asyncio async def test_elec_history(data): """ 用电监测-实时监测-历史曲线 包含负载率、有功/无功、功率因素、电压、电流、频率曲线 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/electric/electric/elec-history", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert len(resp_str["data"]["lf"]) == len(resp_str["data"]["costtl"]) assert all([resp_str["data"]["ctnum"], resp_str["data"]["power"], resp_str["data"]["costtl"], resp_str["data"]["u"], resp_str["data"]["i"], resp_str["data"]["freq"], resp_str["data"]["time_slots"]]) @pytest.mark.parametrize('data', [ { "filter": { "equals": [{"field": "point_id", "value": 53}], "ranges": [], "in_groups": [] } }, ]) @pytest.mark.asyncio async def test_elec_current(data): """ 用电监测-实时监测-实时参数 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/electric/electric/elec-current", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 @pytest.mark.parametrize('data', [ { "filter": { "equals": [{"field": "point_id", "value": 53}], "ranges": [{ "field": "datetime", "start": "2020-07-30 00:00:00", "end": "2020-07-30 23:59:59" }] }, "sort": {"field": "datetime", "direction": "desc"} }, ]) @pytest.mark.asyncio async def test_elec_index(data): """ 指标统计-指标统计-常规参数+电能质量 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/electric/electric/elec-index", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["ctnum"], resp_str["data"]["common_indexes"]]) @pytest.mark.parametrize('data', [ { "filter": { "equals": [{"field": "point_id", "value": 53}], "ranges": [{ "field": "datetime", "start": "2020-07-30 00:00:00", "end": "2020-07-30 23:59:59" }] } }, ]) @pytest.mark.asyncio async def test_qual_history(data): """ 电能质量-历史曲线 包含电压偏差、频率偏差、三相不平衡、谐波畸变率曲线 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/electric/electric/qual-history", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["ctnum"], resp_str["data"]["voltage_dev"], resp_str["data"]["freq_dev"], resp_str["data"]["time_slots"], resp_str["data"]["voltage_harmonic"], resp_str["data"]["current_harmonic"], resp_str["data"]["three_phase_unbalance"]]) @pytest.mark.parametrize('data', [ { "cid": 43, "filter": { "equals": [{"field": "point_id", "value": 53}], "ranges": [], "in_groups": [] } }, ]) @pytest.mark.asyncio async def test_qual_current(data): """ 用电监测-实时监测-实时参数 包含用电健康指数、电压偏差、频率偏差、三相不平衡度、谐波畸变率 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/electric/electric/qual-current", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["ctnum"], resp_str["data"]["real_time"]]) @pytest.mark.parametrize('data', [ { "cid": 44, "storeys": [1, 2, 3], "product": 4 }, ]) @pytest.mark.asyncio async def test_elec_current_storeys(data): """ 用电监测-实时监测-楼层 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/electric/electric" f"/elec-current-storeys", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert len(resp_str["data"]["elec_data"]) @pytest.mark.parametrize('data', [ { "cid": 44, "storeys": [1, 2, 3], "product": 4 }, ]) @pytest.mark.asyncio async def test_qual_current_storeys(data): """ 电能质量-实时参数-楼层 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/electric/electric" f"/qual-current-storeys", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert len(resp_str["data"]["qual_data"]) @pytest.mark.parametrize('data', [ { "cid": 44, "point_list": [439, 440, 441, 442, 443, 444, 445, 446, 447] }, ]) @pytest.mark.asyncio async def test_elec_card_level(data): """ 用电监测-卡片信息-level """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/electric/electric" f"/elec-card-level", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["feeder"], resp_str["data"]["transformer"], resp_str["data"]["power_dist"]]) @pytest.mark.parametrize('data', [ { "cid": 44, "point_list": [439, 440, 441, 442, 443, 444, 445, 446, 447] }, ]) @pytest.mark.asyncio async def test_qual_current_level(data): """ 电能质量-卡片信息-level """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/electric/electric" f"/qual-current-level", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["feeder"], resp_str["data"]["transformer"], resp_str["data"]["power_dist"]])