diff --git a/HISTORY.md b/HISTORY.md index 675fcca..290fcc0 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -8,6 +8,11 @@ dev - \[Short description of non-trivial change.\] +0.1.0 (09-02-2020) +----------------- +- Initial version + + 0.0.2-alpha1 (05-02-2020) ------------------------- diff --git a/kodaksmarthome/__version__.py b/kodaksmarthome/__version__.py index e7ec702..44260b8 100644 --- a/kodaksmarthome/__version__.py +++ b/kodaksmarthome/__version__.py @@ -7,7 +7,7 @@ __title__ = "python-kodaksmarthome" __description__ = "Python Kodak SmartHome API" __url__ = "https://github.com/kairoaraujo/kodak_smarthome_api" -__version__ = "0.0.2-alpha1" +__version__ = "0.1.0" __author__ = "Kairo de Araujo" __author_email__ = "kairo@dearaujo.nl" __license__ = "Apache 2.0" diff --git a/kodaksmarthome/api.py b/kodaksmarthome/api.py index 4f39244..ec4e392 100644 --- a/kodaksmarthome/api.py +++ b/kodaksmarthome/api.py @@ -49,24 +49,126 @@ def __init__(self, username, password, region="EU"): else: self.region_url = _URLS(SUPPORTED_REGIONS[region]) - def _get_options(self): + def _http_request(self, method, url, headers=None, data=None, params=None): + + try: + if method == "POST": + http_response = self.http_session.post( + url, headers=headers, data=data, params=params + ) + + elif method == "OPTIONS": + http_response = self.http_session.options( + url, headers=headers, data=data, params=params + ) + + elif method == "GET": + http_response = self.http_session.get( + url, headers=headers, data=data, params=params + ) + + else: + raise AttributeError(f"Invalid Method {method}") + + except requests.exceptions.ConnectionError as err: + raise ConnectionError(str(err)) + + status_code = http_response.status_code + content_type = None + response_json = None + response_text = http_response.text + error = None + error_description = None + + if "Content-Type" in http_response.headers: + content_type = http_response.headers["Content-Type"] + + if content_type and "application/json" in content_type: + response_json = http_response.json() + + if "error" in response_json: + error = response_json["error"] + + if "error_description" in response_json: + error_description = response_json["error_description"] + + if status_code == HTTP_CODE.OK: + if response_json: + self.is_connected = True + return response_json + + elif response_json is None and method == "OPTIONS": + return True + + else: + self.is_connected = False + raise TypeError("Unexpected response format") + + elif status_code == HTTP_CODE.UNAUTHORIZED: + if error == "invalid_grant": + self.is_connected = False + + raise ConnectionError(error_description) + + elif ( + type(error) == dict + and "reason" in error + and error["reason"] == "authError" + and self.is_connected + ): + self.is_connected = False + return True + + elif ( + type(error) == dict + and "reason" in error + and error["reason"] == "authError" + and self.is_connected is False + ): + if "message" in error: + raise ConnectionError(error["message"]) + + elif ( + "msg" in response_json + and "Access Denied" in response_json["msg"] + and self.is_connected + ): + self.is_connected = False + return True + + elif ( + "msg" in response_json + and "Access Denied" in response_json["msg"] + and self.is_connected is False + ): + self.is_connected = False + raise ConnectionError(response_json["msg"]) + + else: + self.is_connected = False + + raise ConnectionError("Unexpected 401 error " + response_text) + + else: + self.is_connected = False + raise ConnectionError( + "Unexpected HTTP CODE error " + response_text + ) + + def _options(self): """ Verify the connection with Kodak Smart Home portal :return: boolean result :rtype: bool """ - options_response = self.http_session.options( - self.region_url.URL_TOKEN, headers=HTTP_HEADERS_BASIC + options_response = self._http_request( + "OPTIONS", self.region_url.URL_TOKEN, headers=HTTP_HEADERS_BASIC ) - if options_response.status_code != HTTP_CODE.OK: - raise ConnectionError( - "HTTP CODE " + str(options_response.status_code) - ) - return True + return options_response - def _get_token(self): + def _token(self): """ Get Kodak Smart Home Portal Token @@ -89,29 +191,21 @@ def _get_token(self): + f"model={HTTP_CLIENT_MODEL}" ) - token_response = self.http_session.post( + token_response = self._http_request( + "POST", self.region_url.URL_TOKEN, headers=HTTP_HEADERS_AUTH, data=token_payload, ) - if token_response.status_code != HTTP_CODE.OK: - raise ConnectionError( - "HTTP CODE " - + str(token_response.status_code) - + "DETAILS " - + str(token_response.text) - ) - - token_json = token_response.json() - self.token_info["access_token"] = token_json["access_token"] - self.token_info["token_type"] = token_json["token_type"] - self.token_info["refresh_token"] = token_json["refresh_token"] - self.token_info["expires_in"] = token_json["expires_in"] - self.token_info["scope"] = token_json["scope"] - self.account_info = token_json["account_info"] - self.web_urls = token_json["web_urls"] - self.token_info["access_token"] = token_json["access_token"] + self.token_info["access_token"] = token_response["access_token"] + self.token_info["token_type"] = token_response["token_type"] + self.token_info["refresh_token"] = token_response["refresh_token"] + self.token_info["expires_in"] = token_response["expires_in"] + self.token_info["scope"] = token_response["scope"] + self.account_info = token_response["account_info"] + self.web_urls = token_response["web_urls"] + self.token_info["access_token"] = token_response["access_token"] self.token = self.token_info["access_token"] return True @@ -124,27 +218,19 @@ def _authentication(self): :rtype: bool """ auth_payload = f"username=&password={self.token}&rememberme=false" - auth_response = self.http_session.post( + auth_response = self._http_request( + "POST", self.region_url.URL_AUTH, headers=HTTP_HEADERS_AUTH, data=auth_payload, ) - if auth_response.status_code != HTTP_CODE.OK: - raise ConnectionError( - "HTTP CODE " - + str(auth_response.status_code) - + "DETAILS " - + str(auth_response.text) - ) - - auth_json = auth_response.json() self.cookie = self.http_session.cookies["JSESSIONID"] - self.user_id = auth_json["data"]["id"] + self.user_id = auth_response["data"]["id"] return True - def _devices(self): + def _get_devices(self): """ Get all devices available in Kodak Smart Home Portal @@ -152,21 +238,19 @@ def _devices(self): :rtype: list """ parameters = {"access_token": f"{self.token}"} - devices_response = self.http_session.get( + devices_response = self._http_request( + "GET", self.region_url.URL_DEVICES, headers=HTTP_HEADERS_BASIC, params=parameters, ) - if devices_response.status_code != HTTP_CODE.OK: - raise ConnectionError( - "HTTP CODE " - + str(devices_response.status_code) - + "DETAILS " - + str(devices_response.text) - ) - devices_json = devices_response.json() - self.devices = devices_json["data"] + if self.is_connected is False: + self.connect() + + return self.devices + + self.devices = devices_response["data"] return self.devices @@ -178,11 +262,8 @@ def _get_events(self): :rtype: list """ for device in self.devices: - device_id = device['device_id'] - device_events = { - 'device_id': device_id, - 'events': list() - } + device_id = device["device_id"] + device_events = {"device_id": device_id, "events": list()} pages = 1 events_pages = 1 while pages <= events_pages: @@ -193,28 +274,26 @@ def _get_events(self): + f"page={pages}" ) - events_response = self.http_session.get( - url_events, headers=HTTP_HEADERS_BASIC + events_response = self._http_request( + "GET", url_events, headers=HTTP_HEADERS_BASIC ) - if events_response.status_code != HTTP_CODE.OK: - raise ConnectionError( - "HTTP CODE " - + str(events_response.status_code) - + "DETAILS " - + str(events_response.text) - ) + if self.is_connected is False: + self.connect() - events_json = events_response.json() - events_pages = events_json["data"]["total_pages"] - if events_json["data"]["total_events"] == 0: + return self.events + + events_pages = events_response["data"]["total_pages"] + if events_response["data"]["total_events"] == 0: continue - events = events_json["data"]["events"] + + events = events_response["data"]["events"] for event in events: - if event not in device_events['events']: - device_events['events'].append(event) + if event not in device_events["events"]: + device_events["events"].append(event) pages += 1 + self.events.append(device_events) return self.events @@ -227,16 +306,26 @@ def connect(self): :exception: ``ConnectionError`` """ try: - self._get_options() - self._get_token() + self._options() + self._token() self._authentication() - self._devices() + self._get_devices() self._get_events() - self.is_connected = True except requests.exceptions.ConnectionError as err: raise ConnectionError(str(err)) + def update(self): + """ + Update the device list and events data + + :return: True + :rtype: bool + :exception: ``ConnectionError`` + """ + self._get_devices() + self._get_events() + def disconnect(self): """ Disconnect from Kodak Smart Portal @@ -244,12 +333,12 @@ def disconnect(self): :return: None :exception: ``ConnectionError`` """ - self.http_session.get(self.region_url.URL_LOGOUT) + self._http_request("GET", self.region_url.URL_LOGOUT) self.http_session.close() self.is_connected = False @property - def list_devices(self): + def get_devices(self): """ List all registered devices in Kodak Smart Portal and its details. @@ -296,7 +385,7 @@ def get_events_device(self, device_id=None): return self.events else: - if device_id in [d['device_id'] for d in self.devices]: + if device_id in [d["device_id"] for d in self.devices]: device_events = list( filter(lambda d: d["device_id"] == device_id, self.events) ) @@ -308,7 +397,7 @@ def get_events_device(self, device_id=None): return None def _filter_event_type( - self, device_id=None, event_type=DEVICE_EVENT_MOTION + self, device_id=None, event_type=DEVICE_EVENT_MOTION ): """ Filter events from device by event type. @@ -354,7 +443,8 @@ def _filter_event_type( def get_motion_events(self, device_id=None): """ - List all motion devices events from specific device + List all motion devices events from specific device sorted by + creation date. :return: list of motion devices events :exception: ``ConnectionError`` @@ -369,7 +459,7 @@ def get_motion_events(self, device_id=None): return list() - return events + return sorted(events, key=lambda e: e["created_date"]) else: raise ConnectionError( @@ -378,7 +468,8 @@ def get_motion_events(self, device_id=None): def get_battery_events(self, device_id=None): """ - List all battery devices events from specific device + List all battery devices events from specific device, sorted by + creation date. :return: list of battery devices events :exception: ``ConnectionError`` @@ -393,7 +484,7 @@ def get_battery_events(self, device_id=None): return list() - return events + return sorted(events, key=lambda e: e["created_date"]) else: raise ConnectionError( @@ -402,7 +493,8 @@ def get_battery_events(self, device_id=None): def get_sound_events(self, device_id=None): """ - List all sound devices events from specific device + List all sound devices events from specific device sorted by + creation date. :return: list of sound devices events :exception: ``ConnectionError`` @@ -417,7 +509,7 @@ def get_sound_events(self, device_id=None): return list() - return events + return sorted(events, key=lambda e: e["created_date"]) else: raise ConnectionError( diff --git a/tests/conftest.py b/tests/conftest.py index f8f57a3..aa70249 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,18 +3,32 @@ # # Copyright 2019, 2020 Kairo de Araujo # +import json import pytest from unittest import mock class MockRequestsResponse: - def __init__(self, json_data, status_code): + def __init__(self, json_data, status_code, headers): + self._headers = headers self.json_data = json_data - self.status_code = status_code + self._status_code = status_code def json(self): return self.json_data + @property + def text(self): + return json.dumps(self.json_data) + + @property + def headers(self): + return self._headers + + @property + def status_code(self): + return self._status_code + @pytest.fixture def requests_session_mock_ok(): diff --git a/tests/json_responses.py b/tests/json_responses.py index 9fe9bb6..450f0ba 100644 --- a/tests/json_responses.py +++ b/tests/json_responses.py @@ -4,7 +4,7 @@ # Copyright 2019, Kairo de Araujo # -auth_body = { +auth_response = { "status": 200, "msg": "Success", "data": { diff --git a/tests/test_api.py b/tests/test_api.py index 839701f..e2c3be2 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -4,12 +4,17 @@ # Copyright 2019, 2020 Kairo de Araujo # import pytest -from requests import exceptions +import requests from unittest import mock from kodaksmarthome.api import KodakSmartHome +from kodaksmarthome.constants import HTTP_CODE from tests.conftest import MockRequestsResponse -from tests.json_responses import devices_response, events_response +from tests.json_responses import ( + auth_response, + devices_response, + events_response, +) class TestKodakSmartHome: @@ -20,143 +25,415 @@ def test_unsupported_region(self): @mock.patch("kodaksmarthome.api.requests") -def test__get_options(mock_requests, requests_session_mock_ok): +def test__http_request_methods(mock_requests): + + mocked_response_get_post = MockRequestsResponse( + {"key": "value"}, HTTP_CODE.OK, {"Content-Type": "application/json"} + ) + mocked_response_options = MockRequestsResponse( + None, HTTP_CODE.OK, {"Content-Type": "somethingelse"} + ) + + mock_requests.Session.return_value = mock.MagicMock( + get=mock.MagicMock(return_value=mocked_response_get_post), + post=mock.MagicMock(return_value=mocked_response_get_post), + options=mock.MagicMock(return_value=mocked_response_options), + ) - mock_requests.Session.return_value = requests_session_mock_ok test_ksh = KodakSmartHome("fake_user", "fake_pass") - assert test_ksh._get_options() + assert test_ksh._http_request("GET", "http://fake=url") == {"key": "value"} + + assert test_ksh._http_request("POST", "http://fake=url") == { + "key": "value" + } + + assert test_ksh._http_request("OPTIONS", "http://fake=url") + + with pytest.raises(TypeError) as exception_msg: + test_ksh._http_request("INVALID", "http://fake=url") + assert "Invalid Method INVALID" in str(exception_msg.value) @mock.patch("kodaksmarthome.api.requests") -def test__get_options_exception(mock_requests, requests_session_mock_error): +def test__http_request_session_exception(mock_requests): + + mock_requests.Session.return_value = mock.MagicMock( + get=mock.MagicMock(side_effect=requests.exceptions.ConnectionError), + post=mock.MagicMock(side_effect=requests.exceptions.ConnectionError), + options=mock.MagicMock( + side_effect=requests.exceptions.ConnectionError + ), + ) - mock_requests.Session.return_value = requests_session_mock_error test_ksh = KodakSmartHome("fake_user", "fake_pass") - with pytest.raises(ConnectionError): - test_ksh._get_options() + with pytest.raises(TypeError): + test_ksh._http_request("GET", "http://fake=url") @mock.patch("kodaksmarthome.api.requests") -def test__get_token(mock_requests, requests_session_mock_ok): +def test__http_request_200_withou_body_methods_get_post(mock_requests): + + no_response = None + mocked_response_get_post = MockRequestsResponse( + no_response, HTTP_CODE.OK, {"Content-Type": "NOT_OK"} + ) + + mock_requests.Session.return_value = mock.MagicMock( + get=mock.MagicMock(return_value=mocked_response_get_post), + post=mock.MagicMock(return_value=mocked_response_get_post), + ) - mock_requests.Session.return_value = requests_session_mock_ok test_ksh = KodakSmartHome("fake_user", "fake_pass") - assert test_ksh._get_token() + with pytest.raises(TypeError) as exception_msg: + assert test_ksh._http_request("GET", "http://fake=url") + assert test_ksh.is_connected is False + assert "Unexpected response format" in str(exception_msg.value) @mock.patch("kodaksmarthome.api.requests") -def test__get_token_exception(mock_requests, requests_session_mock_error): +def test__http_request_401_invalid_grant(mock_requests): + + invalid_grant_response = { + "error": "invalid_grant", + "error_description": " ... invalid grant ...", + } + mocked_response_get_post = MockRequestsResponse( + invalid_grant_response, + HTTP_CODE.UNAUTHORIZED, + {"Content-Type": "application/json"}, + ) + + mock_requests.Session.return_value = mock.MagicMock( + get=mock.MagicMock(return_value=mocked_response_get_post), + post=mock.MagicMock(return_value=mocked_response_get_post), + ) - mock_requests.Session.return_value = requests_session_mock_error test_ksh = KodakSmartHome("fake_user", "fake_pass") - with pytest.raises(ConnectionError): - test_ksh._get_token() + with pytest.raises(ConnectionError) as exception_msg: + assert test_ksh._http_request("GET", "http://fake=url") + assert test_ksh.is_connected is False + assert "invalid grant" in str(exception_msg.value) @mock.patch("kodaksmarthome.api.requests") -def test__authentication(mock_requests, requests_session_mock_ok): +def test__http_request_401_auth_error_is_connected(mock_requests): + + invalid_grant_response = { + "error": {"reason": "authError", "message": "auth error"} + } + mocked_response_get_post = MockRequestsResponse( + invalid_grant_response, + HTTP_CODE.UNAUTHORIZED, + {"Content-Type": "application/json"}, + ) + + mock_requests.Session.return_value = mock.MagicMock( + get=mock.MagicMock(return_value=mocked_response_get_post), + post=mock.MagicMock(return_value=mocked_response_get_post), + ) - mock_requests.Session.return_value = requests_session_mock_ok test_ksh = KodakSmartHome("fake_user", "fake_pass") + test_ksh.is_connected = True - assert test_ksh._authentication() + assert test_ksh._http_request("GET", "http://fake=url") + assert test_ksh.is_connected is False @mock.patch("kodaksmarthome.api.requests") -def test__authentication_exception(mock_requests, requests_session_mock_error): +def test__http_request_401_auth_error_is_connected_false(mock_requests): + + auth_error_response = { + "error": {"reason": "authError", "message": "auth error"} + } + mocked_response_get_post = MockRequestsResponse( + auth_error_response, + HTTP_CODE.UNAUTHORIZED, + {"Content-Type": "application/json"}, + ) + + mock_requests.Session.return_value = mock.MagicMock( + get=mock.MagicMock(return_value=mocked_response_get_post), + post=mock.MagicMock(return_value=mocked_response_get_post), + ) - mock_requests.Session.return_value = requests_session_mock_error test_ksh = KodakSmartHome("fake_user", "fake_pass") + test_ksh.is_connected = False - with pytest.raises(ConnectionError): - test_ksh._authentication() + with pytest.raises(ConnectionError) as exception_msg: + assert test_ksh._http_request("GET", "http://fake=url") + assert test_ksh.is_connected is False + assert "auth error" in str(exception_msg.value) @mock.patch("kodaksmarthome.api.requests") -def test__devices(mock_requests, requests_session_mock_ok): +def test__http_request_401_access_denied_is_connected(mock_requests): + + access_denied_response = {"msg": "Access Denied"} + mocked_response_get_post = MockRequestsResponse( + access_denied_response, + HTTP_CODE.UNAUTHORIZED, + {"Content-Type": "application/json"}, + ) - mocked_response = MockRequestsResponse(devices_response, 200) mock_requests.Session.return_value = mock.MagicMock( - get=mock.MagicMock(return_value=mocked_response) + get=mock.MagicMock(return_value=mocked_response_get_post), + post=mock.MagicMock(return_value=mocked_response_get_post), ) + test_ksh = KodakSmartHome("fake_user", "fake_pass") + test_ksh.is_connected = True - test_devices = test_ksh._devices() + assert test_ksh._http_request("GET", "http://fake=url") + assert test_ksh.is_connected is False - assert test_devices == devices_response["data"] + +@mock.patch("kodaksmarthome.api.requests") +def test__http_request_401_access_denied_is_connected_false(mock_requests): + + access_denied_response = {"msg": "Access Denied"} + mocked_response_get_post = MockRequestsResponse( + access_denied_response, + HTTP_CODE.UNAUTHORIZED, + {"Content-Type": "application/json"}, + ) + + mock_requests.Session.return_value = mock.MagicMock( + get=mock.MagicMock(return_value=mocked_response_get_post), + post=mock.MagicMock(return_value=mocked_response_get_post), + ) + + test_ksh = KodakSmartHome("fake_user", "fake_pass") + test_ksh.is_connected = False + + with pytest.raises(ConnectionError) as exception_msg: + assert test_ksh._http_request("GET", "http://fake=url") + assert test_ksh.is_connected is False + assert "Access Denied" in str(exception_msg.value) + + +@mock.patch("kodaksmarthome.api.requests") +def test__http_request_401_unexpected(mock_requests): + + unexpected_response = {"not_mapped": "Strange Error"} + mocked_response_get_post = MockRequestsResponse( + unexpected_response, + HTTP_CODE.UNAUTHORIZED, + {"Content-Type": "application/json"}, + ) + + mock_requests.Session.return_value = mock.MagicMock( + get=mock.MagicMock(return_value=mocked_response_get_post), + post=mock.MagicMock(return_value=mocked_response_get_post), + ) + + test_ksh = KodakSmartHome("fake_user", "fake_pass") + test_ksh.is_connected = True + + with pytest.raises(ConnectionError) as exception_msg: + assert test_ksh._http_request("GET", "http://fake=url") + assert test_ksh.is_connected is False + assert "Unexpected 401 error" in str(exception_msg.value) @mock.patch("kodaksmarthome.api.requests") -def test__devices_exception(mock_requests, requests_session_mock_error): +def test__http_request_not_200_401_http_code(mock_requests): + + mocked_response_get_post = MockRequestsResponse( + None, HTTP_CODE.INTERNAL_SERVER_ERROR, "None" + ) + + mock_requests.Session.return_value = mock.MagicMock( + get=mock.MagicMock(return_value=mocked_response_get_post), + post=mock.MagicMock(return_value=mocked_response_get_post), + ) + + test_ksh = KodakSmartHome("fake_user", "fake_pass") + test_ksh.is_connected = True + + with pytest.raises(ConnectionError) as exception_msg: + assert test_ksh._http_request("GET", "http://fake=url") + assert test_ksh.is_connected is False + assert "Unexpected HTTP CODE error" in str(exception_msg.value) + + +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +def test__options(mock__http_request): - mock_requests.Session.return_value = requests_session_mock_error + mock__http_request.return_value = True + test_ksh = KodakSmartHome("fake_user", "fake_pass") + + assert test_ksh._options() + + +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +def test__options_exception(mock__http_request): + + mock__http_request.side_effect = [ConnectionError] + test_ksh = KodakSmartHome("fake_user", "fake_pass") + + with pytest.raises(ConnectionError): + test_ksh._options() + + +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +def test__token(mock__http_request): + + mock__http_request.return_value = { + "access_token": "access_token", + "token_type": "token_type", + "refresh_token": "refresh_token", + "expires_in": "expires_in", + "scope": "scope", + "account_info": "account_info", + "web_urls": "web_urls", + } + + test_ksh = KodakSmartHome("fake_user", "fake_pass") + + assert test_ksh._token() + assert test_ksh.token == "access_token" + + +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +def test__get_token_exception(mock__http_request): + + mock__http_request.side_effect = [ConnectionError] test_ksh = KodakSmartHome("fake_user", "fake_pass") with pytest.raises(ConnectionError): - test_ksh._devices() + test_ksh._token() @mock.patch("kodaksmarthome.api.requests") -def test__get_events(mock_requests): +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +def test__authentication(mock_requests, mock__http_request): + mocked_response = MockRequestsResponse( + devices_response, HTTP_CODE.OK, None + ) - mocked_response = MockRequestsResponse(events_response, 200) + mock__http_request.return_value = auth_response mock_requests.Session.return_value = mock.MagicMock( get=mock.MagicMock(return_value=mocked_response) ) + + test_ksh = KodakSmartHome("fake_user", "fake_pass") + + assert test_ksh._authentication() + + +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +def test__authentication_exception(mock__http_request): + mock__http_request.side_effect = [ConnectionError] + test_ksh = KodakSmartHome("fake_user", "fake_pass") + + with pytest.raises(ConnectionError): + test_ksh._authentication() + + +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +def test__get_devices(mock__http_request): + + mock__http_request.return_value = devices_response + test_ksh = KodakSmartHome("fake_user", "fake_pass") + test_ksh.is_connected = True + + test_devices = test_ksh._get_devices() + + assert test_devices == devices_response["data"] + + +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +@mock.patch("kodaksmarthome.api.KodakSmartHome.connect") +def test__get_devices_is_connected_false(mock__http_request, mock_connect): + + mock__http_request.return_value = [devices_response, devices_response] + mock_connect.return_value = devices_response + test_ksh = KodakSmartHome("fake_user", "fake_pass") + test_ksh.is_connected = False + + test_devices = test_ksh._get_devices() + + assert test_devices == [] + + +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +def test__get_devices_exception(mock__http_request): + mock__http_request.side_effect = [ConnectionError] + test_ksh = KodakSmartHome("fake_user", "fake_pass") + test_ksh.is_connected = True + + with pytest.raises(ConnectionError): + test_ksh._get_devices() + + +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +def test__get_events(mock__http_request): + + mock__http_request.return_value = events_response test_ksh = KodakSmartHome("fake_user", "fake_pass") test_ksh.devices = devices_response["data"]["devices"] - test_ksh.token = "abcdef0123456789" + test_ksh.is_connected = True test_events = test_ksh._get_events() expected_result = [ { "device_id": devices_response["data"]["devices"][0]["device_id"], - "events": events_response["data"]["events"] + "events": events_response["data"]["events"], } ] assert test_events == expected_result -@mock.patch("kodaksmarthome.api.requests") -def test__get_events_none(mock_requests): +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +@mock.patch("kodaksmarthome.api.KodakSmartHome.connect") +def test__get_events_is_connected_false(mock__http_request, mock_connect): + + mock__http_request.return_value = events_response + mock_connect.return_value = True + test_ksh = KodakSmartHome("fake_user", "fake_pass") + test_ksh.devices = devices_response["data"]["devices"] + test_ksh.is_connected = False + test_events = test_ksh._get_events() + + assert test_events == [] + + +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +def test__get_events_none(mock__http_request): events_response_none = { - "status": 200, + "status": HTTP_CODE.OK, "msg": "Success", "total_pages": 0, - "data": { - 'total_events': 0, - 'total_pages': 0, - "events": [], - } + "data": {"total_events": 0, "total_pages": 0, "events": []}, } - mocked_response = MockRequestsResponse(events_response_none, 200) - mock_requests.Session.return_value = mock.MagicMock( - get=mock.MagicMock(return_value=mocked_response) - ) + + mock__http_request.return_value = events_response_none + test_ksh = KodakSmartHome("fake_user", "fake_pass") test_ksh.devices = devices_response["data"]["devices"] test_ksh.token = "abcdef0123456789" + test_ksh.is_connected = True test_events = test_ksh._get_events() expected_result = [ { "device_id": devices_response["data"]["devices"][0]["device_id"], - "events": events_response_none["data"]["events"] + "events": events_response_none["data"]["events"], } ] assert test_events == expected_result -@mock.patch("kodaksmarthome.api.requests") -def test__get_events_exception(mock_requests, requests_session_mock_error): +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +def test__get_events_exception(mock__http_request): - mock_requests.Session.return_value = requests_session_mock_error + mock__http_request.side_effect = [ConnectionError] test_ksh = KodakSmartHome("fake_user", "fake_pass") test_ksh.devices = devices_response["data"]["devices"] test_ksh.token = "abcdef0123456789" @@ -165,39 +442,72 @@ def test__get_events_exception(mock_requests, requests_session_mock_error): test_ksh._get_events() -@mock.patch("kodaksmarthome.api.requests") -def test_connect(mock_requests, requests_session_mock_ok): +@mock.patch("kodaksmarthome.api.KodakSmartHome._options") +@mock.patch("kodaksmarthome.api.KodakSmartHome._token") +@mock.patch("kodaksmarthome.api.KodakSmartHome._authentication") +@mock.patch("kodaksmarthome.api.KodakSmartHome._get_devices") +@mock.patch("kodaksmarthome.api.KodakSmartHome._get_events") +def test_connect( + mock__options, + mock__token, + mock__authentication, + mock__get_devices, + mock__get_events, +): + mock__options.return_value = True + mock__token.return_value = True + mock__authentication.return_value = True + mock__get_devices.return_value = True + mock__get_events.return_value = True - mock_requests.Session.return_value = requests_session_mock_ok test_ksh = KodakSmartHome("fake_user", "fake_pass") - test_ksh.connect() - assert test_ksh.is_connected + assert test_ksh.connect() is None -@mock.patch("kodaksmarthome.api.requests") -def test_disconnect(mock_requests, requests_session_mock_ok): +@mock.patch("kodaksmarthome.api.KodakSmartHome._options") +def test_connect_exception(mock__options,): + mock__options.side_effect = requests.exceptions.ConnectionError + test_ksh = KodakSmartHome("fake_user", "fake_pass") - mock_requests.Session.return_value = requests_session_mock_ok + with pytest.raises(ConnectionError): + test_ksh.connect() + + +@mock.patch("kodaksmarthome.api.KodakSmartHome._http_request") +def test_disconnect(mock__http_request): + + mock__http_request.return_value = True test_ksh = KodakSmartHome("fake_user", "fake_pass") test_ksh.disconnect() assert test_ksh.is_connected is False -def test_list_devices(): +@mock.patch("kodaksmarthome.api.KodakSmartHome._get_devices") +@mock.patch("kodaksmarthome.api.KodakSmartHome._get_events") +def test_update(mock__get_devices, mock__get_events): + mock__get_devices.return_value = True + mock__get_events.return_value = True + + test_ksh = KodakSmartHome("fake_user", "fake_pass") + + assert test_ksh.update() is None + + +def test_get_devices(): test_ksh = KodakSmartHome("fake_user", "fake_pass") test_ksh.is_connected = True test_ksh.devices = devices_response["data"] - assert test_ksh.list_devices == devices_response["data"] + assert test_ksh.get_devices == devices_response["data"] -def test_list_devices_disconnected(): +def test_get_devices_disconnected(): test_ksh = KodakSmartHome("fake_user", "fake_pass") with pytest.raises(ConnectionError): - test_ksh.list_devices() + test_ksh.get_devices def test_get_events(): @@ -211,9 +521,10 @@ def test_get_events(): def test_get_events_disconnected(): test_ksh = KodakSmartHome("fake_user", "fake_pass") + test_ksh.is_connected = False with pytest.raises(ConnectionError): - test_ksh.get_events() + test_ksh.get_events def test__filter_event_type(): @@ -222,7 +533,7 @@ def test__filter_event_type(): test_ksh.events = [ { "device_id": devices_response["data"]["devices"][0]["device_id"], - "events": events_response["data"]["events"] + "events": events_response["data"]["events"], } ] test_ksh.devices = devices_response["data"]["devices"] @@ -238,7 +549,7 @@ def test__filter_event_type_none(): test_ksh.events = [ { "device_id": devices_response["data"]["devices"][0]["device_id"], - "events": events_response["data"]["events"] + "events": events_response["data"]["events"], } ] test_ksh.devices = devices_response["data"]["devices"] @@ -262,7 +573,7 @@ def test_get_motion_events(): test_ksh.events = [ { "device_id": devices_response["data"]["devices"][0]["device_id"], - "events": events_response["data"]["events"] + "events": events_response["data"]["events"], } ] test_ksh.devices = devices_response["data"]["devices"] @@ -276,7 +587,7 @@ def test_get_motion_events_invalid_device_id(): test_ksh.events = [ { "device_id": devices_response["data"]["devices"][0]["device_id"], - "events": events_response["data"]["events"] + "events": events_response["data"]["events"], } ] test_ksh.devices = devices_response["data"]["devices"] @@ -297,7 +608,7 @@ def test_get_battery_events(): test_ksh.events = [ { "device_id": devices_response["data"]["devices"][0]["device_id"], - "events": events_response["data"]["events"] + "events": events_response["data"]["events"], } ] test_ksh.devices = devices_response["data"]["devices"] @@ -318,7 +629,7 @@ def test_get_battery_events_invalid_device_id(): test_ksh.events = [ { "device_id": devices_response["data"]["devices"][0]["device_id"], - "events": events_response["data"]["events"] + "events": events_response["data"]["events"], } ] test_ksh.devices = devices_response["data"]["devices"] @@ -332,7 +643,7 @@ def test_get_sound_events(): test_ksh.events = [ { "device_id": devices_response["data"]["devices"][0]["device_id"], - "events": events_response["data"]["events"] + "events": events_response["data"]["events"], } ] test_ksh.devices = devices_response["data"]["devices"] @@ -353,7 +664,7 @@ def test_get_sound_events_invalid_device_id(): test_ksh.events = [ { "device_id": devices_response["data"]["devices"][0]["device_id"], - "events": events_response["data"]["events"] + "events": events_response["data"]["events"], } ] test_ksh.devices = devices_response["data"]["devices"] diff --git a/tox.ini b/tox.ini index d2f4f30..22ef81d 100644 --- a/tox.ini +++ b/tox.ini @@ -11,4 +11,4 @@ commands = flake8 --ignore=F401,W503 deps = -r{toxinidir}/requirements.txt -r{toxinidir}/dev-requirements.txt -commands = coverage run -m pytest --pdb -vv -s tests \ No newline at end of file +commands = coverage run -m pytest --pdb -vvv -s tests \ No newline at end of file