Merge pull request #47 from stuertz/reactivate_test_api
test: reactivate commented test
This commit is contained in:
commit
661e9e2af4
299
test_api.py
299
test_api.py
|
@ -6,8 +6,10 @@ endpoint requests with various parameters.
|
|||
# pylint: disable=duplicate-code
|
||||
|
||||
import asyncio
|
||||
from datetime import datetime
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
|
||||
from aiohttp import ClientSession
|
||||
|
@ -22,88 +24,241 @@ CONSOLE: logging.Logger = logging.getLogger("console")
|
|||
CONSOLE.addHandler(logging.StreamHandler(sys.stdout))
|
||||
CONSOLE.setLevel(logging.INFO)
|
||||
|
||||
TESTAUTHENTICATE = False
|
||||
TESTAPIMETHODS = False
|
||||
TESTAPIENDPOINTS = False
|
||||
TESTAPIFROMJSON = False
|
||||
|
||||
|
||||
def _out(jsondata):
|
||||
CONSOLE.info(json.dumps(jsondata, indent=2))
|
||||
|
||||
|
||||
async def test_api_methods(myapi) -> None:
|
||||
_system = list(myapi.sites.values())[0]
|
||||
siteid = _system["site_info"]["site_id"]
|
||||
devicesn = _system["solarbank_info"]["solarbank_list"][0]["device_sn"]
|
||||
_out(await myapi.get_site_list())
|
||||
_out(await myapi.get_homepage())
|
||||
_out(await myapi.get_bind_devices())
|
||||
_out(await myapi.get_user_devices())
|
||||
_out(await myapi.get_charging_devices())
|
||||
_out(await myapi.get_auto_upgrade())
|
||||
_out(await myapi.get_scene_info(siteId=siteid))
|
||||
_out(await myapi.get_wifi_list(siteId=siteid))
|
||||
_out(await myapi.get_solar_info(solarbankSn=devicesn))
|
||||
_out(await myapi.get_device_parm(siteId=siteid, paramType="4"))
|
||||
_out(
|
||||
await myapi.get_power_cutoff(
|
||||
siteId=siteid,
|
||||
deviceSn=devicesn,
|
||||
)
|
||||
)
|
||||
_out(
|
||||
await myapi.get_device_load(
|
||||
siteId=siteid,
|
||||
deviceSn=devicesn,
|
||||
)
|
||||
)
|
||||
|
||||
_out(
|
||||
await myapi.energy_analysis(
|
||||
siteId=siteid,
|
||||
deviceSn=devicesn,
|
||||
rangeType="week",
|
||||
startDay=datetime.fromisoformat("2023-10-10"),
|
||||
endDay=datetime.fromisoformat("2023-10-10"),
|
||||
devType="solar_production",
|
||||
)
|
||||
)
|
||||
_out(
|
||||
await myapi.energy_analysis(
|
||||
siteId=siteid,
|
||||
deviceSn=devicesn,
|
||||
rangeType="week",
|
||||
startDay=datetime.fromisoformat("2023-10-10"),
|
||||
endDay=datetime.fromisoformat("2023-10-10"),
|
||||
devType="solarbank",
|
||||
)
|
||||
)
|
||||
_out(
|
||||
await myapi.energy_daily(
|
||||
siteId=siteid,
|
||||
deviceSn=devicesn,
|
||||
startDay=datetime.fromisoformat("2024-01-10"),
|
||||
numDays=10,
|
||||
)
|
||||
)
|
||||
_out(await myapi.home_load_chart(siteId=siteid))
|
||||
|
||||
|
||||
async def test_api_endpoints(myapi) -> None:
|
||||
_system = list(myapi.sites.values())[0]
|
||||
siteid = _system["site_info"]["site_id"]
|
||||
devicesn = _system["solarbank_info"]["solarbank_list"][0]["device_sn"]
|
||||
_out((await myapi.request("post", api._API_ENDPOINTS["homepage"], json={})))
|
||||
_out((await myapi.request("post", api._API_ENDPOINTS["site_list"], json={})))
|
||||
_out((await myapi.request("post", api._API_ENDPOINTS["bind_devices"], json={})))
|
||||
_out((await myapi.request("post", api._API_ENDPOINTS["user_devices"], json={})))
|
||||
_out((await myapi.request("post", api._API_ENDPOINTS["charging_devices"], json={})))
|
||||
_out((await myapi.request("post", api._API_ENDPOINTS["get_auto_upgrade"], json={})))
|
||||
_out(
|
||||
(
|
||||
await myapi.request(
|
||||
"post",
|
||||
api._API_ENDPOINTS["site_detail"],
|
||||
json={"site_id": siteid},
|
||||
)
|
||||
)
|
||||
)
|
||||
_out(
|
||||
(
|
||||
await myapi.request(
|
||||
"post",
|
||||
api._API_ENDPOINTS["wifi_list"],
|
||||
json={"site_id": siteid},
|
||||
)
|
||||
)
|
||||
)
|
||||
_out(
|
||||
(
|
||||
await myapi.request(
|
||||
"post",
|
||||
api._API_ENDPOINTS["get_site_price"],
|
||||
json={"site_id": siteid},
|
||||
)
|
||||
)
|
||||
)
|
||||
_out(
|
||||
(
|
||||
await myapi.request(
|
||||
"post",
|
||||
api._API_ENDPOINTS["solar_info"],
|
||||
json={
|
||||
"site_id": siteid,
|
||||
"solarbank_sn": devicesn,
|
||||
},
|
||||
)
|
||||
)
|
||||
)
|
||||
_out(
|
||||
(
|
||||
await myapi.request(
|
||||
"post",
|
||||
api._API_ENDPOINTS["get_cutoff"],
|
||||
json={
|
||||
"site_id": siteid,
|
||||
"device_sn": devicesn,
|
||||
},
|
||||
)
|
||||
)
|
||||
)
|
||||
_out(
|
||||
(
|
||||
await myapi.request(
|
||||
"post",
|
||||
api._API_ENDPOINTS["get_device_fittings"],
|
||||
json={
|
||||
"site_id": siteid,
|
||||
"device_sn": devicesn,
|
||||
},
|
||||
)
|
||||
)
|
||||
)
|
||||
_out(
|
||||
(
|
||||
await myapi.request(
|
||||
"post",
|
||||
api._API_ENDPOINTS["get_device_load"],
|
||||
json={
|
||||
"site_id": siteid,
|
||||
"device_sn": devicesn,
|
||||
},
|
||||
)
|
||||
)
|
||||
)
|
||||
_out(
|
||||
(
|
||||
await myapi.request(
|
||||
"post",
|
||||
api._API_ENDPOINTS["get_device_parm"],
|
||||
json={
|
||||
"site_id": siteid,
|
||||
"param_type": "4",
|
||||
},
|
||||
)
|
||||
)
|
||||
)
|
||||
_out(
|
||||
(
|
||||
await myapi.request(
|
||||
"post",
|
||||
api._API_ENDPOINTS["compatible_process"],
|
||||
json={"solarbank_sn": devicesn},
|
||||
)
|
||||
)
|
||||
)
|
||||
_out(
|
||||
(
|
||||
await myapi.request(
|
||||
"post",
|
||||
api._API_ENDPOINTS["home_load_chart"],
|
||||
json={"site_id": siteid},
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
async def test_api_from_json_files(myapi) -> None:
|
||||
myapi.testDir(os.path.join(os.path.dirname(__file__), "examples", "example1"))
|
||||
await myapi.update_sites(fromFile=True)
|
||||
await myapi.update_device_details(fromFile=True)
|
||||
_out(myapi.sites)
|
||||
_out(myapi.devices)
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
"""Create the aiohttp session and run the example."""
|
||||
CONSOLE.info("Testing Solix API:")
|
||||
try:
|
||||
async with ClientSession() as websession:
|
||||
# Update your account credentials in api.credentials.py or directly in this file for testing
|
||||
# Both files are added to .gitignore to avoid local changes being comitted to git
|
||||
myapi = api.AnkerSolixApi(
|
||||
common.user(),
|
||||
common.password(),
|
||||
common.country(),
|
||||
websession,
|
||||
_LOGGER,
|
||||
)
|
||||
async with ClientSession() as websession:
|
||||
myapi = api.AnkerSolixApi(
|
||||
common.user(),
|
||||
common.password(),
|
||||
common.country(),
|
||||
websession,
|
||||
_LOGGER,
|
||||
)
|
||||
|
||||
if TESTAUTHENTICATE:
|
||||
# show login response
|
||||
# new = await myapi.async_authenticate(restart=True) # enforce new login data from server
|
||||
# new = await myapi.async_authenticate() # receive new or load cached login data
|
||||
# if new:
|
||||
# CONSOLE.info("Received Login response:")
|
||||
# else:
|
||||
# CONSOLE.info("Cached Login response:")
|
||||
# CONSOLE.info(json.dumps(myapi._login_response, indent=2)) # show used login response for API reqests
|
||||
new = await myapi.async_authenticate(
|
||||
restart=True
|
||||
) # enforce new login data from server
|
||||
new = (
|
||||
await myapi.async_authenticate()
|
||||
) # receive new or load cached login data
|
||||
if new:
|
||||
CONSOLE.info("Received Login response:")
|
||||
else:
|
||||
CONSOLE.info("Cached Login response:")
|
||||
_out(myapi._login_response) # show used login response for API reqests
|
||||
|
||||
# test site api methods
|
||||
# test site api methods
|
||||
await myapi.update_sites()
|
||||
await myapi.update_device_details()
|
||||
CONSOLE.info("System Overview:")
|
||||
_out(myapi.sites)
|
||||
CONSOLE.info("Device Overview:")
|
||||
_out(myapi.devices)
|
||||
|
||||
await myapi.update_sites()
|
||||
await myapi.update_device_details()
|
||||
CONSOLE.info("System Overview:")
|
||||
CONSOLE.info(json.dumps(myapi.sites, indent=2))
|
||||
CONSOLE.info("Device Overview:")
|
||||
CONSOLE.info(json.dumps(myapi.devices, indent=2))
|
||||
if TESTAPIMETHODS:
|
||||
await test_api_methods(myapi)
|
||||
|
||||
# test api methods
|
||||
# CONSOLE.info(json.dumps(await myapi.get_site_list(), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.get_homepage(), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.get_bind_devices(), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.get_user_devices(), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.get_charging_devices(), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.get_auto_upgrade(), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.get_scene_info(siteId='efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.get_wifi_list(siteId='efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.get_solar_info(siteId='efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'), indent=2)) # json parameters unknown: site_id not sifficient, or works only with Anker Inverters?
|
||||
# CONSOLE.info(json.dumps(await myapi.get_device_parm(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",paramType="4"), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.get_power_cutoff(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",deviceSn="9JVB42LJK8J0P5RY"), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.get_device_load(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",deviceSn="9JVB42LJK8J0P5RY"), indent=2))
|
||||
if TESTAPIENDPOINTS:
|
||||
await test_api_endpoints(myapi)
|
||||
|
||||
# CONSOLE.info(json.dumps(await myapi.energy_analysis(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",deviceSn="9JVB42LJK8J0P5RY",rangeType="week",startDay=datetime.fromisoformat("2023-10-10"),endDay=datetime.fromisoformat("2023-10-10"),devType="solar_production"), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.energy_analysis(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",deviceSn="9JVB42LJK8J0P5RY",rangeType="week",startDay=datetime.fromisoformat("2023-10-10"),endDay=datetime.fromisoformat("2023-10-10"),devType="solarbank"), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.energy_daily(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",deviceSn="9JVB42LJK8J0P5RY",startDay=datetime.fromisoformat("2024-01-10"),numDays=10), indent=2))
|
||||
# CONSOLE.info(json.dumps(await myapi.home_load_chart(siteId='efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'), indent=2))
|
||||
# """
|
||||
|
||||
# # test api endpoints directly
|
||||
# """
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["homepage"],json={})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["site_list"],json={})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["bind_devices"],json={})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["user_devices"],json={})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["charging_devices"],json={})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_auto_upgrade"],json={})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["site_detail"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["wifi_list"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_site_price"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["solar_info"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', "device_sn": "9JVB42LJK8J0P5RY"})), indent=2)) # json parameters unknown: May need site_id and device_sn of inverter in system?
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_cutoff"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', "device_sn": "9JVB42LJK8J0P5RY"})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_device_fittings"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', "device_sn": "9JVB42LJK8J0P5RY"})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_device_load"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', "device_sn": "9JVB42LJK8J0P5RY"})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_device_parm"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', "param_type": "4"})), indent=2))
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["compatible_process"],json={})), indent=2)) # json parameters unknown
|
||||
# CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["home_load_chart"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'})), indent=2))
|
||||
|
||||
# test api from json files
|
||||
# myapi.testDir(os.path.join(os.path.dirname(__file__), "examples", "example1"))
|
||||
# await myapi.update_sites(fromFile=True)
|
||||
# await myapi.update_device_details(fromFile=True)
|
||||
# CONSOLE.info(json.dumps(myapi.sites,indent=2))
|
||||
# CONSOLE.info(json.dumps(myapi.devices,indent=2))
|
||||
|
||||
except Exception as exception: # pylint: disable=broad-exception-caught
|
||||
CONSOLE.error("%s: %s", type(exception), exception)
|
||||
if TESTAPIFROMJSON:
|
||||
await test_api_from_json_files(myapi)
|
||||
|
||||
|
||||
# run async main
|
||||
|
|
Loading…
Reference in New Issue