From 502c81bd158c442f3ea8c4826c7eeaa6d3773fe6 Mon Sep 17 00:00:00 2001 From: Robbie Van Gorkom Date: Fri, 8 May 2020 07:41:00 -0700 Subject: [PATCH] Adding proper cli script (for testing purposes). Fixing activate response. --- gogogate2_api/__init__.py | 37 ++++------- gogogate2_api/common.py | 25 ++++++-- pyproject.toml | 5 +- scripts/cli.py | 132 +++++++++++++++++++++++++------------- tests/common.py | 17 ++--- tests/test_common.py | 4 +- tests/test_init.py | 22 +------ 7 files changed, 132 insertions(+), 110 deletions(-) mode change 100644 => 100755 scripts/cli.py diff --git a/gogogate2_api/__init__.py b/gogogate2_api/__init__.py index ded0d45..179b64c 100644 --- a/gogogate2_api/__init__.py +++ b/gogogate2_api/__init__.py @@ -12,8 +12,10 @@ from typing_extensions import Final from .common import ( + ActivateResponse, DoorStatus, - GogoGate2Response, + InfoResponse, + element_to_activate_response, element_to_api_error, element_to_info_response, get_door_by_id, @@ -67,7 +69,6 @@ def __init__(self, host: str, username: str, password: str) -> None: self._password: Final = password self._api_url: Final = f"http://{host}/api.php" self._api_cipher: Final[ApiCipher] = ApiCipher(SHARED_SECRET) - self._api_code: Optional[str] = None def _request( self, action: str, arg1: Optional[str] = None, arg2: Optional[str] = None @@ -97,36 +98,24 @@ def _request( return root_element - def _handle_generic_response(self, element: Element) -> GogoGate2Response: - response = element_to_info_response(element) - self._api_code = response.apicode - return response - - @property - def api_code(self) -> Optional[str]: - """Get the current api code.""" - return self._api_code - - def clear_api_code(self) -> None: - """Clear the api key.""" - self._api_code = None - - def info(self) -> GogoGate2Response: + def info(self) -> InfoResponse: """Get info about the device and doors.""" - return self._handle_generic_response(self._request("info")) + return element_to_info_response(self._request("info")) - def activate(self, door_id: int) -> GogoGate2Response: + def activate( + self, door_id: int, api_code: Optional[str] = None + ) -> ActivateResponse: """Send a command to open/close/stop the door. Gogogate2 does not have a status for opening or closing. So running this method during an action will stop the door. It's recommended you use open_door() or close_door() as those methods check the status before running and run if needed.""" - if not self._api_code: - self.info() + if not api_code: + api_code = self.info().apicode - return self._handle_generic_response( - self._request("activate", str(door_id), self._api_code) + return element_to_activate_response( + self._request("activate", str(door_id), api_code) ) def _set_door_status(self, door_id: int, door_status: DoorStatus) -> bool: @@ -146,7 +135,7 @@ def _set_door_status(self, door_id: int, door_status: DoorStatus) -> bool: if door.status == door_status: return False - self.activate(door_id) + self.activate(door_id, response.apicode) return True def close_door(self, door_id: int) -> bool: diff --git a/gogogate2_api/common.py b/gogogate2_api/common.py index 89d2ec7..ff1e0ea 100644 --- a/gogogate2_api/common.py +++ b/gogogate2_api/common.py @@ -164,7 +164,7 @@ class Wifi(NamedTuple): signal: str -class GogoGate2Response(NamedTuple): +class InfoResponse(NamedTuple): """Response from gogogate2 api calls.""" user: str @@ -183,6 +183,12 @@ class GogoGate2Response(NamedTuple): wifi: Wifi +class ActivateResponse(NamedTuple): + """Response from gogogate2 activate calls.""" + + result: bool + + def element_or_none(element: Optional[Element], tag: str) -> Optional[Element]: """Get element from xml element.""" return None if element is None else element.find(tag) @@ -271,9 +277,9 @@ def door_or_raise(door_id: int, element: Element) -> Door: ) -def element_to_info_response(element: Element) -> GogoGate2Response: +def element_to_info_response(element: Element) -> InfoResponse: """Get response from xml element.""" - return GogoGate2Response( + return InfoResponse( user=element_text_or_raise(element, "user"), gogogatename=element_text_or_raise(element, "gogogatename"), model=element_text_or_raise(element, "model"), @@ -292,18 +298,25 @@ def element_to_info_response(element: Element) -> GogoGate2Response: ) -def get_door_by_id(door_id: int, response: GogoGate2Response) -> Optional[Door]: +def element_to_activate_response(element: Element) -> ActivateResponse: + """Get response from xml element.""" + return ActivateResponse( + result=element_text_or_raise(element, "result").lower() == "ok" + ) + + +def get_door_by_id(door_id: int, response: InfoResponse) -> Optional[Door]: """Get a door from a gogogate2 response.""" return next( iter([door for door in get_doors(response) if door.door_id == door_id]), None ) -def get_doors(response: GogoGate2Response) -> Tuple[Door, ...]: +def get_doors(response: InfoResponse) -> Tuple[Door, ...]: """Get a tuple of doors from a response.""" return (response.door1, response.door2, response.door3) -def get_configured_doors(response: GogoGate2Response) -> Tuple[Door, ...]: +def get_configured_doors(response: InfoResponse) -> Tuple[Door, ...]: """Get a tuple of configured doors from a response.""" return tuple([door for door in get_doors(response) if door.name]) diff --git a/pyproject.toml b/pyproject.toml index 047d1fa..8769b1e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "gogogate2_api" -version = "1.0.1" +version = "1.0.2" description = "Library for connecting to gogogate2 hubs" license = "MIT" @@ -27,6 +27,7 @@ typing-extensions = ">=3.7.4.2" [tool.poetry.dev-dependencies] bandit = "==1.6.2" black = "==19.10b0" +click = "==7.1.2" codespell = "==1.16.0" coverage = "==5.0.4" flake8 = "==3.7.8" @@ -75,7 +76,7 @@ line_length = 88 indent = " " # by default isort don't check module indexes not_skip = "__init__.py" -# will group `import x` and `from x import` of the same module. +# will group `import obj` and `from obj import` of the same module. force_sort_within_sections = true sections = "FUTURE,STDLIB,INBETWEENS,THIRDPARTY,FIRSTPARTY,LOCALFOLDER" default_section = "THIRDPARTY" diff --git a/scripts/cli.py b/scripts/cli.py old mode 100644 new mode 100755 index 6ae49b3..0b7040f --- a/scripts/cli.py +++ b/scripts/cli.py @@ -1,46 +1,86 @@ -# """CLI for gogogate device.""" -# from enum import Enum -# import pprint -# -# from gogogate2_api import GogoGate2Api -# -# -# def isnamedtupleinstance(x): -# _type = type(x) -# bases = _type.__bases__ -# if len(bases) != 1 or bases[0] != tuple: -# return False -# fields = getattr(_type, "_fields", None) -# if not isinstance(fields, tuple): -# return False -# return all(type(i) == str for i in fields) -# -# -# def unpack(obj): -# if isinstance(obj, dict): -# return {key: unpack(value) for key, value in obj.items()} -# elif isinstance(obj, list): -# return [unpack(value) for value in obj] -# elif isnamedtupleinstance(obj): -# return {key: unpack(value) for key, value in obj._asdict().items()} -# elif isinstance(obj, tuple): -# return tuple(unpack(value) for value in obj) -# elif isinstance(obj, Enum): -# return obj.value -# else: -# return obj -# -# -# def main() -> None: -# pretty_print = pprint.PrettyPrinter(indent=4) -# # API = GogoGate2Api("10.40.11.84", "admin", "4ZsdV9A4s0zhkp3%KoYQVXe$B$") -# api = GogoGate2Api("10.40.3.157", "admin", "OJa$52OaeSVA&b9W9WsR&yq!L1") -# # response = api.info() -# response = api.activate(11) -# # api._request("activate", 2, "FFFFF") -# # print('RRR', response) -# # print(type(unpack(response))) -# pretty_print.pprint(unpack(response)) -# -# -# main() +#!/usr/bin/env python3 +"""CLI for gogogate device.""" +from enum import Enum +import pprint +from typing import Any, Optional, cast + +import click +from gogogate2_api import GogoGate2Api + +API = "api" +PRETTY_PRINT = pprint.PrettyPrinter(indent=4) + + +def is_named_tuple(obj: Any) -> bool: + """Check if object is a named tuple.""" + _type = type(obj) + bases = _type.__bases__ + if len(bases) != 1 or bases[0] != tuple: + return False + fields = getattr(_type, "_fields", None) + if not isinstance(fields, tuple): + return False + return all(isinstance(field, str) for field in fields) + + +def unpack(obj: Any) -> Any: + """Convert object.""" + if isinstance(obj, dict): + return {key: unpack(value) for key, value in obj.items()} + if isinstance(obj, list): + return [unpack(value) for value in obj] + if is_named_tuple(obj): + return {key: unpack(value) for key, value in obj._asdict().items()} + if isinstance(obj, tuple): + return tuple(unpack(value) for value in obj) + if isinstance(obj, Enum): + return obj.value + return obj + + +@click.group() +@click.option("--host", required=True) +@click.option("--username", required=True) +@click.option("--password", required=True) +@click.pass_context +def cli( + ctx: Optional[click.core.Context] = None, + host: Optional[str] = None, + username: Optional[str] = None, + password: Optional[str] = None, +) -> None: + """Setup api.""" + ctx = cast(click.core.Context, ctx) + ctx.obj = { + API: GogoGate2Api(cast(str, host), cast(str, username), cast(str, password)) + } + + +@cli.command() +@click.pass_context +def info(ctx: click.core.Context) -> None: + """Get info from device.""" + api: GogoGate2Api = ctx.obj[API] + PRETTY_PRINT.pprint(unpack(api.info())) + + +@cli.command(name="open") +@click.argument("door_id", type=int, required=True) +@click.pass_context +def open_door(ctx: click.core.Context, door_id: int) -> None: + """Open the door.""" + api: GogoGate2Api = ctx.obj[API] + PRETTY_PRINT.pprint(unpack(api.open_door(door_id))) + + +@cli.command(name="close") +@click.argument("door_id", type=int, required=True) +@click.pass_context +def close_door(ctx: click.core.Context, door_id: int) -> None: + """Close the door.""" + api: GogoGate2Api = ctx.obj[API] + PRETTY_PRINT.pprint(unpack(api.close_door(door_id))) + + +if __name__ == "__main__": + cli() diff --git a/tests/common.py b/tests/common.py index 830cc10..921c419 100644 --- a/tests/common.py +++ b/tests/common.py @@ -66,13 +66,6 @@ def set_device_status(self, device_id: int, status: str) -> None: """Set the status of a device.""" self._devices[device_id]["status"] = status - # def set_device_temperature(self, device_id: int, temperature: str) -> None: - # """Set a device temperature.""" - # if temperature is None: - # del self._devices[device_id]["temperature"] - # else: - # self._devices[device_id]["temperature"] = temperature - # pylint: disable=too-many-return-statements def _handle_request(self, request: Any) -> tuple: # Simulate an HTTP error. @@ -120,8 +113,8 @@ def _handle_request(self, request: Any) -> tuple: if not door["name"]: return self._new_response( """ - OK - """ + OK + """ ) current_status = door["status"] @@ -131,7 +124,11 @@ def _handle_request(self, request: Any) -> tuple: else DoorStatus.OPENED.value ) - return self._info_response() + return self._new_response( + """ + OK + """ + ) def _device_to_xml_str(self, device_id: int) -> str: device_dict: dict = self._devices[device_id] diff --git a/tests/test_common.py b/tests/test_common.py index 0ef4003..74d5ad1 100644 --- a/tests/test_common.py +++ b/tests/test_common.py @@ -6,7 +6,7 @@ Door, DoorMode, DoorStatus, - GogoGate2Response, + InfoResponse, Network, Outputs, TagNotFoundException, @@ -104,7 +104,7 @@ def test_get_enabled_doors() -> None: temperature=None, ) - response = GogoGate2Response( + response = InfoResponse( user="user1", gogogatename="gogogatename1", model="", diff --git a/tests/test_init.py b/tests/test_init.py index 4f1ab10..aadece9 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -32,27 +32,9 @@ def test_activate() -> None: server = MockGogoGateServer("device1") api = GogoGate2Api(server.host, server.username, server.password) - api.clear_api_code() - assert api.api_code is None - - response = api.activate(1) - assert api.api_code == server.api_code - door1 = get_door_by_id(1, response) - assert door1 - assert door1.status == DoorStatus.OPENED - - api.clear_api_code() - assert api.api_code is None - response = api.activate(1) - door1 = get_door_by_id(1, response) - assert door1 - assert door1.status == DoorStatus.CLOSED - - with pytest.raises(ApiError) as exinfo: - api.activate(11) - assert exinfo.value.code == 5 - assert exinfo.value.message == "Error: invalid door" + assert response + assert response.result @responses.activate