robin
/
anker-solix-api
mirror of https://github.com/thomluther/anker-solix-api.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refactor: tests only executed on demand, but nolonger commented to avoid rot

This commit is contained in:
Jan Stürtz 2024-02-22 16:47:09 +01:00
parent 1a21a7a2ee
commit 6592698f01
No known key found for this signature in database
GPG Key ID: 8183CC01FD23D435
1 changed files with 213 additions and 306 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

519
test_api.py
View File

@ -24,13 +24,203 @@ 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:")
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(),
@ -39,319 +229,36 @@ async def main() -> None:
_LOGGER,
)
# 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
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:")
_out(myapi._login_response) # show used login response for API reqests
# test site api methods
await myapi.update_sites()
await myapi.update_device_details()
CONSOLE.info("System Overview:")
CONSOLE.info(json.dumps(myapi.sites, indent=2))
_system = list(myapi.sites.values())[0]
siteid = _system["site_info"]["site_id"]
devicesn = _system["solarbank_info"]["solarbank_list"][0]["device_sn"]
_out(myapi.sites)
CONSOLE.info("Device Overview:")
CONSOLE.info(json.dumps(myapi.devices, indent=2))
_out(myapi.devices)
# 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=siteid),
indent=2,
)
)
CONSOLE.info(
json.dumps(
await myapi.get_wifi_list(siteId=siteid),
indent=2,
)
)
CONSOLE.info(
json.dumps(
await myapi.get_solar_info(solarbankSn=devicesn),
indent=2,
)
)
CONSOLE.info(
json.dumps(
await myapi.get_device_parm(siteId=siteid, paramType="4"),
indent=2,
)
)
CONSOLE.info(
json.dumps(
await myapi.get_power_cutoff(
siteId=siteid,
deviceSn=devicesn,
),
indent=2,
)
)
CONSOLE.info(
json.dumps(
await myapi.get_device_load(
siteId=siteid,
deviceSn=devicesn,
),
indent=2,
)
)
if TESTAPIMETHODS:
await test_api_methods(myapi)
CONSOLE.info(
json.dumps(
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",
),
indent=2,
)
)
CONSOLE.info(
json.dumps(
await myapi.energy_analysis(
siteId=siteid,
deviceSn=devicesn,
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=siteid,
deviceSn=devicesn,
startDay=datetime.fromisoformat("2024-01-10"),
numDays=10,
),
indent=2,
)
)
CONSOLE.info(
json.dumps(
await myapi.home_load_chart(siteId=siteid),
indent=2,
)
)
if TESTAPIENDPOINTS:
await test_api_endpoints(myapi)
# # 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": siteid},
)
),
indent=2,
)
)
CONSOLE.info(
json.dumps(
(
await myapi.request(
"post",
api._API_ENDPOINTS["wifi_list"],
json={"site_id": siteid},
)
),
indent=2,
)
)
CONSOLE.info(
json.dumps(
(
await myapi.request(
"post",
api._API_ENDPOINTS["get_site_price"],
json={"site_id": siteid},
)
),
indent=2,
)
)
CONSOLE.info(
json.dumps(
(
await myapi.request(
"post",
api._API_ENDPOINTS["solar_info"],
json={
"site_id": siteid,
"solarbank_sn": devicesn,
},
)
),
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": siteid,
"device_sn": devicesn,
},
)
),
indent=2,
)
)
CONSOLE.info(
json.dumps(
(
await myapi.request(
"post",
api._API_ENDPOINTS["get_device_fittings"],
json={
"site_id": siteid,
"device_sn": devicesn,
},
)
),
indent=2,
)
)
CONSOLE.info(
json.dumps(
(
await myapi.request(
"post",
api._API_ENDPOINTS["get_device_load"],
json={
"site_id": siteid,
"device_sn": devicesn,
},
)
),
indent=2,
)
)
CONSOLE.info(
json.dumps(
(
await myapi.request(
"post",
api._API_ENDPOINTS["get_device_parm"],
json={
"site_id": siteid,
"param_type": "4",
},
)
),
indent=2,
)
)
CONSOLE.info(
json.dumps(
(
await myapi.request(
"post",
api._API_ENDPOINTS["compatible_process"],
json={"solarbank_sn": devicesn},
)
),
indent=2,
)
) # json parameters unknown
CONSOLE.info(
json.dumps(
(
await myapi.request(
"post",
api._API_ENDPOINTS["home_load_chart"],
json={"site_id": siteid},
)
),
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))
if TESTAPIFROMJSON:
await test_api_from_json_files(myapi)
# run async main