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"]])