test_tsp.py 3.22 KB
Newer Older
lcn's avatar
lcn committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118
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', [
    {
        "cid": 44,
        "tsp_id": 1
    },
])
@pytest.mark.asyncio
async def test_real_time(data):
    """
    TSP信息-实时参数
    """
    resp_str, status = await AioHttpUtils().post(
        url=f"{HTTP_PREFIX}/unify-api/tsp-water/tsp/real-time",
        data=data,
        timeout=10,
        headers={"Authorization": f"Bearer {TOKEN}"}
    )
    assert status == 200
    resp_str = json.loads(resp_str)
    assert all([resp_str["data"]["pm2_5"], resp_str["data"]["pm10"],
                resp_str["data"]["tsp"]])


@pytest.mark.parametrize('data', [
    {
        "cid": 44,
        "tsp_id": 1,
        "start": "2021-04-26 00:00:00",
        "end": "2021-04-26 23:59:59"
    },
])
@pytest.mark.asyncio
async def test_tsp_history(data):
    """TSP信息-历史曲线"""
    resp_str, status = await AioHttpUtils().post(
        url=f"{HTTP_PREFIX}/unify-api/tsp-water/tsp/tsp-history",
        data=data,
        timeout=10,
        headers={"Authorization": f"Bearer {TOKEN}"}
    )
    assert status == 200
    resp_str = json.loads(resp_str)
    assert all([resp_str["data"]["pm2_5"], resp_str["data"]["pm10"],
                resp_str["data"]["tsp"]])


@pytest.mark.parametrize('data', [
    {
        "cid": 44,
        "tsp_id": 1,
        "start": "2021-04-26 00:00:00",
        "end": "2021-04-26 23:59:59"
    },
])
@pytest.mark.asyncio
async def test_tsp_index_statistics(data):
    """TSP信息-指标统计"""
    resp_str, status = await AioHttpUtils().post(
        url=f"{HTTP_PREFIX}/unify-api/tsp-water/tsp"
            f"/tsp-index-statistics",
        data=data,
        timeout=10,
        headers={"Authorization": f"Bearer {TOKEN}"}
    )
    assert status == 200
    resp_str = json.loads(resp_str)
    assert all([resp_str["data"]["pm2_5"], resp_str["data"]["pm10"],
                resp_str["data"]["tsp"]])


@pytest.mark.parametrize('data', [
    {
        "cid": 44,
    },
])
@pytest.mark.asyncio
async def test_day_env(data):
    """当日环境"""
    resp_str, status = await AioHttpUtils().post(
        url=f"{HTTP_PREFIX}/unify-api/tsp-water/tsp/day-env",
        data=data,
        timeout=10,
        headers={"Authorization": f"Bearer {TOKEN}"}
    )
    assert status == 200
    resp_str = json.loads(resp_str)
    assert all([resp_str["data"]["pm2_5"], resp_str["data"]["pm10"],
                resp_str["data"]["tsp"]])


@pytest.mark.parametrize('data', [
    {
        "cid": 44,
        "tsp_id": 1,
        "start": "2021-04-26 00:00:00",
        "end": "2021-04-26 23:59:59"
    },
])
@pytest.mark.asyncio
async def test_stat_analysis(data):
    """统计分析-扬尘"""
    resp_str, status = await AioHttpUtils().post(
        url=f"{HTTP_PREFIX}/unify-api/tsp-water/tsp/stat-analysis",
        data=data,
        timeout=10,
        headers={"Authorization": f"Bearer {TOKEN}"}
    )
    assert status == 200
    resp_str = json.loads(resp_str)
    assert all([resp_str["data"]["pm2_5"], resp_str["data"]["pm10"],
                resp_str["data"]["water"],  resp_str["data"]["fog_gun"],
                resp_str["data"]["time_slots"]])