Adding proper cli script (for testing purposes).

Fixing activate response.

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:

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])

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"

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()

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(
                 """
-                <result>OK</result>
-            """
+                    <result>OK</result>
+                """
             )
 
         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(
+            """
+                <result>OK</result>
+            """
+        )
 
     def _device_to_xml_str(self, device_id: int) -> str:
         device_dict: dict = self._devices[device_id]

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="",

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