test_electric.py 7 KB
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"]])