From c6e947736c311bb94bc79b3508286b4d5e9410a0 Mon Sep 17 00:00:00 2001 From: Robert Van Gorkom Date: Wed, 29 Apr 2020 15:50:44 -0700 Subject: [PATCH] Fixing several bugs that would result in crashes. (#7) * Fixing bugs that caused crashes. Adding tests. * Adding log error. --- .gitignore | 2 + README.md | 15 +++ pygogogate2/__init__.py | 18 +++- setup.py | 2 + tests/__init__.py | 0 tests/test_init.py | 214 ++++++++++++++++++++++++++++++++++++++++ 6 files changed, 247 insertions(+), 4 deletions(-) create mode 100644 tests/__init__.py create mode 100644 tests/test_init.py diff --git a/.gitignore b/.gitignore index 7bbc71c..b3d2a18 100644 --- a/.gitignore +++ b/.gitignore @@ -99,3 +99,5 @@ ENV/ # mypy .mypy_cache/ + +.idea diff --git a/README.md b/README.md index 0387abf..209a40b 100644 --- a/README.md +++ b/README.md @@ -49,3 +49,18 @@ def open_device(self, device_id): ### Disclaimer The code here is based off of an unsupported API from [Gogogate2](https://www.gogogate.com/) and is subject to change without notice. The authors claim no responsibility for damages to your garage door or property by use of the code within. + +# Development +``` +# Setup the virtual environemnt. +python -m venv venv + +# Enter venv +source venv/bin/activate + +# Install dependencies. +pip install -r requirements.txt + +# Test code +python setup.py test +``` \ No newline at end of file diff --git a/pygogogate2/__init__.py b/pygogogate2/__init__.py index c0875da..9d62e9c 100644 --- a/pygogogate2/__init__.py +++ b/pygogogate2/__init__.py @@ -37,6 +37,7 @@ def __init__(self, username, password, ip_address): self._logged_in = False self._device_states = {} self.cipher = AESCipher(self.APP_ID) + self.apicode = None def make_request(self, command): try: @@ -77,7 +78,12 @@ def get_devices(self): garage_doors = [] try: - self.apicode = devices.find('apicode').text + apicode_element = devices.find('apicode') + if apicode_element is None: + self.logger.error('Gogogate2 - Invalid username or password provided.') + return False + + self.apicode = apicode_element.text self._device_states = {} for doorNum in range(1, 4): door = devices.find('door' + str(doorNum)) @@ -99,7 +105,7 @@ def get_devices(self): print(ex) return False else: - return False; + return False def get_status(self, device_id): @@ -120,11 +126,15 @@ def get_temperature(self, device_id): if devices != False: for device in devices: if device['door'] == device_id: + temp = device.get('temperature') + if temp is None: + return False + # gogogate returns '-1000000' when the door does not have a value - if device['temperature'] == "-1000000": + if temp == "-1000000": return 0.0 else: - celcius = float(device['temperature']) + celcius = float(temp) return celcius return False diff --git a/setup.py b/setup.py index 0cea2f5..722ecd2 100644 --- a/setup.py +++ b/setup.py @@ -15,6 +15,8 @@ packages=['pygogogate2'], package_dir={'pygogogate2': 'pygogogate2'}, data_files = [('',['LICENSE'])], + test_suite='tests', + tests_require=['requests-mock'], install_requires=[ 'pycryptodomex' ] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_init.py b/tests/test_init.py new file mode 100644 index 0000000..3e3e17b --- /dev/null +++ b/tests/test_init.py @@ -0,0 +1,214 @@ +"""Tests for gogogate2 api.""" +import json +import unittest +from urllib.parse import parse_qs, urlparse + +import requests +import requests_mock + +from pygogogate2 import AESCipher, Gogogate2API + + +class MockGogoGateServer: + def __init__(self): + self.username: str = "username1" + self.password: str = "password1" + self.api_code: str = "api_code1" + self.http_status: int = 200 + self._requests_mocker = requests_mock.Mocker() + self._cipher = AESCipher(Gogogate2API.APP_ID) + self._devices = { + 1: { + "permission": "yes", + "name": "Gate", + "mode": "garage", + "status": "closed", + "sensor": "yes", + "sensorid": "WIRE", + "camera": "no", + "events": 1234, + }, + 2: { + "permission": "yes", + "name": "Garage", + "mode": "garage", + "status": "opened", + "sensor": "yes", + "sensorid": "WIRE", + "camera": "no", + "events": 4321, + "temperature": 13, + }, + } + + def setUp(self) -> None: + self._requests_mocker.start() + self._requests_mocker.get("http://localhost/api.php", text=self._handle_request) + + def tearDown(self) -> None: + self._requests_mocker.stop() + + def set_device_status(self, device_id: int, status: str) -> None: + self._devices[device_id]["status"] = status + + def set_device_temperature(self, device_id: int, temperature: str) -> None: + if temperature is None: + del self._devices[device_id]["temperature"] + else: + self._devices[device_id]["temperature"] = temperature + + def _handle_request(self, request: requests.Request, context) -> str: + context.status_code = self.http_status + + # Simulate an HTTP error. + if context.status_code != 200: + return "" + + # Parse the request. + query = parse_qs(urlparse(request.url).query) + data = query["data"][0] + decrypted = self._cipher.decrypt(data.encode("utf-8")) + payload = json.loads(decrypted) + + username = payload[0] + password = payload[1] + command = payload[2] + + # Validate credentials. + if username != self.username or password != self.password: + return self._new_response( + """ + + 01 + Error: wrong login or password + + """ + ) + + # Handle activation command. + if command == "activate": + device_id = int(payload[3]) + api_code = payload[4] + + if api_code != self.api_code: + context.status_code = 401 + return self._new_response("") + + current_status = self._devices[device_id]["status"] + self._devices[device_id]["status"] = ( + "closed" if current_status == "opened" else "opened" + ) + + return self._new_response( + """ + ok + """ + ) + + # handle info command. + if command == "info": + self.api_code = "fjsll33" + return self._new_response( + f""" + {self.username} + home + GGG2 + 1.5 + 0 + abcdefg12345.my-gogogate.com + 260\n + {self.api_code} + + {self._device_to_xml_str(1)} + + + {self._device_to_xml_str(2)} + + + yes + + garage + undefined + no + no + 0 + + + off + off + off + + + 127.0.1.1 + + + + 61% + -67 dBm + + """ + ) + + return self._new_response("") + + def _device_to_xml_str(self, device_id: int) -> str: + device_dict = self._devices[device_id] + return "\n".join( + [f"<{key}>{value}" for key, value in device_dict.items()] + ) + + def _new_response(self, xml_str: str) -> str: + return self._cipher.encrypt( + f""" + + {xml_str} + + """ + ) + + +class TestApi(unittest.TestCase): + def setUp(self) -> None: + self.server = MockGogoGateServer() + self.server.setUp() + self.api = Gogogate2API(self.server.username, self.server.password, "localhost") + + def tearDown(self) -> None: + self.server.tearDown() + + def test_http_error(self) -> None: + self.server.http_status = 503 + assert self.api.get_devices() is False + + def test_auth_failed(self) -> None: + self.server.username = "a" + self.api.username = "b" + self.server.password = "a" + self.api.password = "b" + assert self.api.get_devices() is False + + self.server.username = "a" + self.api.username = "a" + assert self.api.get_devices() is False + + def test_open_close(self) -> None: + assert self.api.open_device(1) + assert self.api.get_status(1) == "open" + assert self.api.close_device(1) + assert self.api.get_status(1) == "closed" + + assert self.api.get_status(2) == "open" + assert self.api.open_device(2) is False + + def test_get_temperature(self) -> None: + assert self.api.get_temperature(1) is False + assert self.api.get_temperature(2) == 13 + + self.server.set_device_temperature(1, 10) + assert self.api.get_temperature(1) == 10 + + self.server.set_device_temperature(2, -1000000) + assert self.api.get_temperature(2) == 0.0 + + self.api.username = "invalid" + assert self.api.get_temperature(2) is False