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": 73 } ]) @pytest.mark.asyncio async def test_count_info(data): """ 获取首页统计信息 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info/count-info", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["max_temperature"], resp_str["data"]["max_residual_current"], resp_str["data"]["temperature_qr"], resp_str["data"]["residual_current_qr"], resp_str["data"]["electric_info"]]) @pytest.mark.parametrize('data', [ { "cid": 73 } ]) @pytest.mark.asyncio async def test_electric_economic_index(data): """ 获取首页用电经济指数 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info" f"/electric-economic-index", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["power_factor"], resp_str["data"]["md_space"], resp_str["data"]["pcvf"], resp_str["data"]["power_save"], resp_str["data"]["avg_price"]]) @pytest.mark.parametrize('data', [ { "cid": 73 } ]) @pytest.mark.asyncio async def test_zd_info_factory(data): """ 工厂版首页统计信息-智电U """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info" f"/zd-info-factory", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["total_monitor"], resp_str["data"]["safe_operation_days"], resp_str["data"]["total_power"], resp_str["data"]["total_alarm"], resp_str["data"]["co2"]]) @pytest.mark.parametrize('data', [ {"cid": 73}, {"cid": 90}, ]) @pytest.mark.asyncio async def test_risk_cost(data): """ 工厂版首页统计信息-智电U """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info" f"/risk-cost", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert resp_str["data"]["risk_data"] @pytest.mark.parametrize('data', [ {"cid": 73}, {"cid": 90}, ]) @pytest.mark.asyncio async def test_info_yang_chen(data): """ 工厂版首页统计信息-扬尘 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info" f"/info-yang-chen", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["total_point"], resp_str["data"]["air_quality"], resp_str["data"]["total_water"], resp_str["data"]["safe_operation_days"]]) @pytest.mark.parametrize('data', [ {"cid": 44}, {"cid": 90}, ]) @pytest.mark.asyncio async def test_info_yang_chen_map(data): """ 工厂版首页统计信息-扬尘 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info" f"/info-yang-chen-map", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert resp_str["data"]["safety_index"] @pytest.mark.parametrize('data', [ {"cid": 44}, ]) @pytest.mark.asyncio async def test_electric_economic_index_new(data): """ 获取首页用电经济指数 """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info" f"/electric-economic-index-new", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["power_factor"], resp_str["data"]["md_space"], resp_str["data"]["pcvf"], resp_str["data"]["avg_price"], resp_str["data"]["power_save"]]) @pytest.mark.parametrize('data', [ {"cid": 44}, {"cid": 90}, ]) @pytest.mark.asyncio async def test_rank_type_ranking(data): """ 首页报警统计-等级类型排名-智电U """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info" f"/rank-type-ranking", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["alarm_level_cnt"], resp_str["data"]["alarm_content_cnt"], resp_str["data"]["ranking"]]) @pytest.mark.parametrize('data', [ {"cid": 44}, {"cid": 90}, ]) @pytest.mark.asyncio async def test_condition_monitor(data): """ 首页状态监测-智电U """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info" f"/condition-monitor", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["safe"], resp_str["data"]["power"]]) @pytest.mark.parametrize('data', [ {"cid": 90}, ]) @pytest.mark.asyncio async def test_alarm_price_costtl(data): """ 首页-今日报警平均电价等-智电U """ resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info" f"/alarm-price-costtl", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["today_alarm"], resp_str["data"]["avg_price"]]) @pytest.mark.parametrize('data', [ {"cid": 90}, ]) @pytest.mark.asyncio async def test_alarm_safe_index(data): """首页-用电安全指数等""" resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info" f"/alarm-safe-index", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["first_alarm_cnt"], resp_str["data"]["second_alarm_cnt"], resp_str["data"]["alarm_score"], resp_str["data"]["electric_use_score"], resp_str["data"]["safety_ratio"]]) @pytest.mark.parametrize('data', [ {"cid": 44}, {"cid": 90}, ]) @pytest.mark.asyncio async def test_health_status_index(data): """首页-健康指数等""" resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info" f"/health-status-index", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["health_index"], resp_str["data"]["health_status"], resp_str["data"]["v_dev"], resp_str["data"]["freq_dev"], resp_str["data"]["ubl"], resp_str["data"]["costtl"]]) @pytest.mark.parametrize('data', [ {"cid": 118}, ]) @pytest.mark.asyncio async def test_all_index_info(data): """首页-全部指数""" resp_str, status = await AioHttpUtils().post( url=f"{HTTP_PREFIX}/unify-api/home-page/count-info" f"/all-index-info", data=data, timeout=10, headers={"Authorization": f"Bearer {TOKEN}"} ) assert status == 200 resp_str = json.loads(resp_str) assert all([resp_str["data"]["health_index"], resp_str["data"]["safe_index"], resp_str["data"]["economic_index"]])