Skip to content

Commit

Permalink
Use OverkizServer class instead of API_URL (#201)
Browse files Browse the repository at this point in the history
* Use OverkizServer type

* Fix test

* Fix test2

* Bugfixes
  • Loading branch information
iMicknl authored Oct 20, 2021
1 parent 8d745d1 commit af3b6db
Show file tree
Hide file tree
Showing 3 changed files with 16 additions and 28 deletions.
16 changes: 2 additions & 14 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -23,26 +23,14 @@ pip install pyhoma
import asyncio
import time

from pyhoma.const import SUPPORTED_SERVERS
from pyhoma.client import TahomaClient

SUPPORTED_ENDPOINTS = {
"Cozytouch": "https://ha110-1.overkiz.com/enduser-mobile-web/enduserAPI/",
"eedomus": "https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/",
"Hi Kumo": "https://ha117-1.overkiz.com/enduser-mobile-web/enduserAPI/",
"Rexel Energeasy Connect": "https://ha112-1.overkiz.com/enduser-mobile-web/enduserAPI/",
"Somfy Connexoon IO": "https://tahomalink.com/enduser-mobile-web/enduserAPI/",
"Somfy Connexoon RTS": "https://ha201-1.overkiz.com/enduser-mobile-web/enduserAPI/",
"Somfy TaHoma": "https://tahomalink.com/enduser-mobile-web/enduserAPI/",
"Somfy (Australia)": "https://ha201-1.overkiz.com/enduser-mobile-web/enduserAPI/",
"Somfy (Europe)": "https://tahomalink.com/enduser-mobile-web/enduserAPI/",
"Somfy (North America)": "https://ha401-1.overkiz.com/enduser-mobile-web/enduserAPI/",
}

USERNAME = ""
PASSWORD = ""

async def main() -> None:
async with TahomaClient(USERNAME, PASSWORD, api_url=SUPPORTED_ENDPOINTS["Somfy (Europe)"]) as client:
async with TahomaClient(USERNAME, PASSWORD, server=SUPPORTED_SERVERS["somfy_europe"]) as client:
try:
await client.login()
except Exception as exception: # pylint: disable=broad-except
Expand Down
25 changes: 12 additions & 13 deletions pyhoma/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,15 +43,14 @@
Execution,
Gateway,
HistoryExecution,
OverkizServer,
Place,
Scenario,
State,
)

JSON = Union[Dict[str, Any], List[Dict[str, Any]]]

DEFAULT_SERVER = SUPPORTED_SERVERS["somfy_europe"]


async def relogin(invocation: dict[str, Any]) -> None:
await invocation["args"][0].login()
Expand All @@ -68,21 +67,21 @@ def __init__(
self,
username: str,
password: str,
api_url: str = DEFAULT_SERVER.endpoint,
server: OverkizServer,
session: ClientSession = None,
) -> None:
"""
Constructor
:param username: the username for Tahomalink.com
:param password: the password for Tahomalink.com
:param api_url: optional custom api url
:param server: OverkizServer
:param session: optional ClientSession
"""

self.username = username
self.password = password
self.api_url = api_url
self.server = server

self.devices: list[Device] = []
self.gateways: list[Gateway] = []
Expand Down Expand Up @@ -118,12 +117,12 @@ async def login(
"""

# CozyTouch authentication using jwt
if self.api_url == SUPPORTED_SERVERS["atlantic_cozytouch"].endpoint:
if self.server == SUPPORTED_SERVERS["atlantic_cozytouch"]:
jwt = await self.cozytouch_login()
payload = {"jwt": jwt}

# Nexity authentication using ssoToken
elif self.api_url == SUPPORTED_SERVERS["nexity"].endpoint:
elif self.server == SUPPORTED_SERVERS["nexity"]:
sso_token = await self.nexity_login()
user_id = self.username.replace("@", "_-_") # Replace @ for _-_
payload = {"ssoToken": sso_token, "userId": user_id}
Expand Down Expand Up @@ -441,27 +440,27 @@ async def execute_scenario(self, oid: str) -> str:
response = await self.__post(f"exec/{oid}")
return response["execId"]

async def __get(self, endpoint: str) -> Any:
async def __get(self, path: str) -> Any:
"""Make a GET request to the TaHoma API"""
async with self.session.get(f"{self.api_url}{endpoint}") as response:
async with self.session.get(f"{self.server.endpoint}{path}") as response:
await self.check_response(response)
return await response.json()

async def __post(
self, endpoint: str, payload: JSON | None = None, data: JSON | None = None
self, path: str, payload: JSON | None = None, data: JSON | None = None
) -> Any:
"""Make a POST request to the TaHoma API"""
async with self.session.post(
f"{self.api_url}{endpoint}",
f"{self.server.endpoint}{path}",
data=data,
json=payload,
) as response:
await self.check_response(response)
return await response.json()

async def __delete(self, endpoint: str) -> None:
async def __delete(self, path: str) -> None:
"""Make a DELETE request to the TaHoma API"""
async with self.session.delete(f"{self.api_url}{endpoint}") as response:
async with self.session.delete(f"{self.server.endpoint}{path}") as response:
await self.check_response(response)

@staticmethod
Expand Down
3 changes: 2 additions & 1 deletion tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,14 +7,15 @@
from pytest import fixture

from pyhoma.client import TahomaClient
from pyhoma.const import SUPPORTED_SERVERS

CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))


class TestTahomaClient:
@fixture
def api(self):
return TahomaClient("username", "password")
return TahomaClient("username", "password", SUPPORTED_SERVERS["somfy_europe"])

@pytest.mark.asyncio
async def test_get_devices_basic(self, api):
Expand Down

0 comments on commit af3b6db

Please sign in to comment.