diff --git a/libpurecool/dyson.py b/libpurecool/dyson.py index 79a818f..fb3ad4e 100644 --- a/libpurecool/dyson.py +++ b/libpurecool/dyson.py @@ -52,6 +52,25 @@ def login(self): urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) _LOGGER.debug("Disabling insecure request warnings since " "dyson are using a self signed certificate.") + # Must first check account status + accountstatus = requests.get( + "https://{0}/v1/userregistration/userstatus".format( + self._dyson_api_url + ), + params={"country": self._country, "email": self._email}, + headers=self._headers, + verify=False, + ) + # pylint: disable=no-member + if accountstatus.status_code == requests.codes.ok: + json_status = accountstatus.json() + if json_status['accountStatus'] != "ACTIVE": + # The account is not active + self._logged = False + return self._logged + else: + self._logged = False + return self._logged request_body = { "Email": self._email, @@ -61,7 +80,7 @@ def login(self): "https://{0}/v1/userregistration/authenticate?country={1}".format( self._dyson_api_url, self._country), headers=self._headers, - data=request_body, + json=request_body, verify=False ) # pylint: disable=no-member diff --git a/tests/test_dyson_account.py b/tests/test_dyson_account.py index 69d18fa..39a1346 100644 --- a/tests/test_dyson_account.py +++ b/tests/test_dyson_account.py @@ -4,12 +4,17 @@ from libpurecool.dyson_pure_cool import DysonPureCool from libpurecool.dyson_pure_hotcool import DysonPureHotCool -from libpurecool.dyson import DysonAccount, DysonPureCoolLink, \ - DysonPureHotCoolLink, Dyson360Eye, DysonNotLoggedException, \ - DYSON_API_USER_AGENT +from libpurecool.dyson import ( + DysonAccount, + DysonPureCoolLink, + DysonPureHotCoolLink, + Dyson360Eye, + DysonNotLoggedException, + DYSON_API_USER_AGENT, +) -API_HOST = 'appapi.cp.dyson.com' -API_CN_HOST = 'appapi.cp.dyson.cn' +API_HOST = "appapi.cp.dyson.com" +API_CN_HOST = "appapi.cp.dyson.cn" class MockResponse: @@ -22,58 +27,93 @@ def json(self, **kwargs): def _mocked_login_post_failed(*args, **kwargs): - url = 'https://{0}{1}?{2}={3}'.format(API_HOST, - '/v1/userregistration/authenticate', - 'country', - 'language') - payload = {'Password': 'password', 'Email': 'email'} - if args[0] == url and kwargs['data'] == payload and \ - kwargs['headers'] == {'User-Agent': DYSON_API_USER_AGENT}: - return MockResponse({ - 'Account': 'account', - 'Password': 'password' - }, 401) + url = "https://{0}{1}?{2}={3}".format( + API_HOST, "/v1/userregistration/authenticate", "country", "language" + ) + payload = {"Password": "password", "Email": "email"} + if ( + args[0] == url + and kwargs["data"] == payload + and kwargs["headers"] == {"User-Agent": DYSON_API_USER_AGENT} + ): + return MockResponse( + {"Account": "account", "Password": "password"}, + 401) + else: + raise Exception("Unknown call") + + +def _mock_gets(*args, **kwargs): + if "provisioningservice" in args[0]: + return _mocked_list_devices(*args, **kwargs) + else: + return _mocked_status_get(*args, **kwargs) + + +def _mocked_status_get(*args, **kwargs): + url = "https://{0}{1}".format(API_HOST, "/v1/userregistration/userstatus") + params = {"country": "language", "email": "email"} + if args[0] == url and kwargs["params"] == params: + return MockResponse({"accountStatus": "ACTIVE"}) + else: + raise Exception("Unknown call") + + +def _mocked_status_get_cn(*args, **kwargs): + url = "https://{0}{1}".format(API_CN_HOST, + "/v1/userregistration/userstatus") + params = {"country": "CN", "email": "email"} + if args[0] == url and kwargs["params"] == params: + return MockResponse( + {"accountStatus": "ACTIVE"} + ) + else: + raise Exception("Unknown call") + + +def _mocked_status_get_failed(*args, **kwargs): + url = "https://{0}{1}".format(API_HOST, "/v1/userregistration/userstatus") + params = {"country": "language", "email": "email"} + if args[0] == url and kwargs["params"] == params: + return MockResponse({"accountStatus": "INACTIVE"}) else: raise Exception("Unknown call") def _mocked_login_post(*args, **kwargs): - url = 'https://{0}{1}?{2}={3}'.format(API_HOST, - '/v1/userregistration/authenticate', - 'country', - 'language') - payload = {'Password': 'password', 'Email': 'email'} - if args[0] == url and kwargs['data'] == payload and \ - kwargs['headers'] == {'User-Agent': DYSON_API_USER_AGENT}: - return MockResponse({ - 'Account': 'account', - 'Password': 'password' - }) + url = "https://{0}{1}?{2}={3}".format( + API_HOST, "/v1/userregistration/authenticate", "country", "language" + ) + payload = {"Password": "password", "Email": "email"} + if ( + args[0] == url + and kwargs["json"] == payload + and kwargs["headers"] == {"User-Agent": DYSON_API_USER_AGENT} + ): + return MockResponse({"Account": "account", "Password": "password"}) else: raise Exception("Unknown call") def _mocked_login_post_cn(*args, **kwargs): - url = 'https://{0}{1}?{2}={3}'.format(API_CN_HOST, - '/v1/userregistration/authenticate', - 'country', - 'CN') - payload = {'Password': 'password', 'Email': 'email'} - if args[0] == url and kwargs['data'] == payload and \ - kwargs['headers'] == {'User-Agent': DYSON_API_USER_AGENT}: - return MockResponse({ - 'Account': 'account', - 'Password': 'password' - }) + url = "https://{0}{1}?{2}={3}".format( + API_CN_HOST, "/v1/userregistration/authenticate", "country", "CN" + ) + payload = {"Password": "password", "Email": "email"} + if ( + args[0] == url + and kwargs["json"] == payload + and kwargs["headers"] == {"User-Agent": DYSON_API_USER_AGENT} + ): + return MockResponse({"Account": "account", "Password": "password"}) else: raise Exception("Unknown call") def _mocked_list_devices(*args, **kwargs): - url = 'https://{0}{1}'.format(API_HOST, - '/v1/provisioningservice/manifest') - url_v2 = 'https://{0}{1}'.format(API_HOST, - '/v2/provisioningservice/manifest') + url = "https://{0}{1}".format(API_HOST, "/v1/provisioningservice/manifest") + url_v2 = "https://{0}{1}".format(API_HOST, + "/v2/provisioningservice/manifest") if args[0] == url: return MockResponse( @@ -85,11 +125,11 @@ def _mocked_list_devices(*args, **kwargs): "ScaleUnit": "SU01", "Version": "21.03.08", "LocalCredentials": "1/aJ5t52WvAfn+z+fjDuef86kQDQPefbQ6/" - "70ZGysII1Ke1i0ZHakFH84DZuxsSQ4KTT2v" - "bCm7uYeTORULKLKQ==", + "70ZGysII1Ke1i0ZHakFH84DZuxsSQ4KTT2v" + "bCm7uYeTORULKLKQ==", "AutoUpdate": True, "NewVersionAvailable": False, - "ProductType": "475" + "ProductType": "475", }, { "Active": False, @@ -98,11 +138,11 @@ def _mocked_list_devices(*args, **kwargs): "ScaleUnit": "SU02", "Version": "21.02.04", "LocalCredentials": "1/aJ5t52WvAfn+z+fjDuebkH6aWl2H5Q1vCq" - "CQSjJfENzMefozxWaDoW1yDluPsi09SGT5nW" - "MxqxtrfkxnUtRQ==", + "CQSjJfENzMefozxWaDoW1yDluPsi09SGT5nW" + "MxqxtrfkxnUtRQ==", "AutoUpdate": False, "NewVersionAvailable": True, - "ProductType": "455" + "ProductType": "455", }, { "Active": True, @@ -111,12 +151,12 @@ def _mocked_list_devices(*args, **kwargs): "ScaleUnit": "SU01", "Version": "21.03.08", "LocalCredentials": "1/aJ5t52WvAfn+z+fjDuef86kQDQPefbQ6/" - "70ZGysII1Ke1i0ZHakFH84DZuxsSQ4KTT2v" - "bCm7uYeTORULKLKQ==", + "70ZGysII1Ke1i0ZHakFH84DZuxsSQ4KTT2v" + "bCm7uYeTORULKLKQ==", "AutoUpdate": True, "NewVersionAvailable": False, - "ProductType": "N223" - } + "ProductType": "N223", + }, ] ) @@ -129,33 +169,33 @@ def _mocked_list_devices(*args, **kwargs): "Version": "02.05.001.0006", "AutoUpdate": True, "LocalCredentials": "1/aJ5t52WvAfn+z+fjDuef86kQDQPefbQ6/" - "70ZGysII1Ke1i0ZHakFH84DZuxsSQ4KTT2v" - "bCm7uYeTORULKLKQ==", + "70ZGysII1Ke1i0ZHakFH84DZuxsSQ4KTT2v" + "bCm7uYeTORULKLKQ==", "NewVersionAvailable": False, - "ProductType": "438" + "ProductType": "438", }, { "Serial": "DB1-US-DBD1231D", "Name": "device-3", "Version": "02.05.001.0006", "LocalCredentials": "1/aJ5t52WvAfn+z+fjDuebkH6aWl2H5Q1vCq" - "CQSjJfENzMefozxWaDoW1yDluPsi09SGT5nW" - "MxqxtrfkxnUtRQ==", + "CQSjJfENzMefozxWaDoW1yDluPsi09SGT5nW" + "MxqxtrfkxnUtRQ==", "AutoUpdate": True, "NewVersionAvailable": False, - "ProductType": "520" + "ProductType": "520", }, { "Serial": "CB1-US-DBD1231C", "Name": "device-4", "Version": "02.05.001.0006", "LocalCredentials": "1/aJ5t52WvAfn+z+fjDuebkH6aWl2H5Q1vCq" - "CQSjJfENzMefozxWaDoW1yDluPsi09SGT5nW" - "MxqxtrfkxnUtRQ==", + "CQSjJfENzMefozxWaDoW1yDluPsi09SGT5nW" + "MxqxtrfkxnUtRQ==", "AutoUpdate": True, "NewVersionAvailable": False, - "ProductType": "527" - } + "ProductType": "527", + }, ] ) @@ -167,40 +207,46 @@ def setUp(self): def tearDown(self): pass - @mock.patch('requests.post', side_effect=_mocked_login_post) - def test_connect_account(self, mocked_login): + @mock.patch("requests.get", side_effect=_mock_gets) + @mock.patch("requests.post", side_effect=_mocked_login_post) + def test_connect_account(self, mocked_login, mocked_status): dyson_account = DysonAccount("email", "password", "language") logged = dyson_account.login() + self.assertEqual(mocked_status.call_count, 1) self.assertEqual(mocked_login.call_count, 1) self.assertTrue(logged) - @mock.patch('requests.post', side_effect=_mocked_login_post_cn) - def test_connect_account_cn(self, mocked_login): + @mock.patch("requests.get", side_effect=_mocked_status_get_cn) + @mock.patch("requests.post", side_effect=_mocked_login_post_cn) + def test_connect_account_cn(self, mocked_login, mocked_status): dyson_account = DysonAccount("email", "password", "CN") logged = dyson_account.login() + self.assertEqual(mocked_status.call_count, 1) self.assertEqual(mocked_login.call_count, 1) self.assertTrue(logged) - @mock.patch('requests.post', side_effect=_mocked_login_post_failed) - def test_connect_account_failed(self, mocked_login): + @mock.patch("requests.get", side_effect=_mocked_status_get_failed) + @mock.patch("requests.post", side_effect=_mocked_login_post_failed) + def test_connect_account_failed(self, mocked_login, mocked_status): dyson_account = DysonAccount("email", "password", "language") logged = dyson_account.login() - self.assertEqual(mocked_login.call_count, 1) + self.assertEqual(mocked_status.call_count, 1) + self.assertEqual(mocked_login.call_count, 0) self.assertFalse(logged) def test_not_logged(self): dyson_account = DysonAccount("email", "password", "language") self.assertRaises(DysonNotLoggedException, dyson_account.devices) - @mock.patch('requests.get', side_effect=_mocked_list_devices) - @mock.patch('requests.post', side_effect=_mocked_login_post) + @mock.patch("requests.get", side_effect=_mock_gets) + @mock.patch("requests.post", side_effect=_mocked_login_post) def test_list_devices(self, mocked_login, mocked_list_devices): dyson_account = DysonAccount("email", "password", "language") dyson_account.login() self.assertEqual(mocked_login.call_count, 1) self.assertTrue(dyson_account.logged) devices = dyson_account.devices() - self.assertEqual(mocked_list_devices.call_count, 2) + self.assertEqual(mocked_list_devices.call_count, 3) self.assertEqual(len(devices), 6) self.assertTrue(isinstance(devices[0], DysonPureCoolLink)) self.assertTrue(isinstance(devices[1], DysonPureHotCoolLink)) @@ -211,8 +257,8 @@ def test_list_devices(self, mocked_login, mocked_list_devices): self.assertTrue(devices[0].active) self.assertTrue(devices[0].auto_update) self.assertFalse(devices[0].new_version_available) - self.assertEqual(devices[0].serial, 'device-id-1') - self.assertEqual(devices[0].name, 'device-1') - self.assertEqual(devices[0].version, '21.03.08') - self.assertEqual(devices[0].product_type, '475') - self.assertEqual(devices[0].credentials, 'password1') + self.assertEqual(devices[0].serial, "device-id-1") + self.assertEqual(devices[0].name, "device-1") + self.assertEqual(devices[0].version, "21.03.08") + self.assertEqual(devices[0].product_type, "475") + self.assertEqual(devices[0].credentials, "password1")