Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add call to account status endpoint before attempting to login to fix authentication. #38

Open
wants to merge 8 commits into
base: master
Choose a base branch
from
Open

Add call to account status endpoint before attempting to login to fix authentication. #38

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
6ffd4de
add account stats GET before login POST
bfayers Feb 12, 2021
b41eadf
set verify=False because dyson run self signed certs
bfayers Feb 12, 2021
ac5566f
use the headers for the status check, maybe unnecesarry; also use jso…
bfayers Feb 12, 2021
8ac02d0
some changes to bring style in line with existing code as well as to …
bfayers Feb 12, 2021
6307f4c
Make changes as suggested by shenxn
bfayers Feb 12, 2021
3965d81
break up the format on line 57
bfayers Feb 13, 2021
f720d24
Updated login tests to work with call to userstatus
bfayers Feb 27, 2021
dfe77cf
Lint the changes for CI
bfayers Feb 27, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 20 additions & 1 deletion libpurecool/dyson.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
Expand Down
200 changes: 123 additions & 77 deletions tests/test_dyson_account.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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(
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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",
},
]
)

Expand All @@ -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",
},
]
)

Expand All @@ -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))
Expand All @@ -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")