6148597193

Committed 11 Sep 2023 03:29PM UTC coverage: 66.139%. First build

Build # 6148597193

Build Type

push

github

Committed by

sdb9696

Commit Message

Add motion detection enabled switch

Run Details

33 of 33 new or added lines in 4 files covered. (100.0%)

418 of 632 relevant lines covered (66.14%)

0.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.63

/ring_doorbell/auth.py

# coding: utf-8
# vim:sw=4:ts=4:et:
"""Python Ring Auth Class."""
from uuid import uuid4 as uuid
from json import dumps as json_dumps
from requests_oauthlib import OAuth2Session
from oauthlib.oauth2 import LegacyApplicationClient, TokenExpiredError
from ring_doorbell.const import OAuth, API_VERSION, TIMEOUT


class Auth:
    """A Python Auth class for Ring"""

    def __init__(self, user_agent, token=None, token_updater=None, hardware_id=None):
        """
        :type token: Optional[Dict[str, str]]
        :type token_updater: Optional[Callable[[str], None]]
        """
        self.user_agent = user_agent

        self.hardware_id = hardware_id
        if self.hardware_id is None:
            self.hardware_id = str(uuid())

        self.token_updater = token_updater
        self._oauth = OAuth2Session(
            client=LegacyApplicationClient(client_id=OAuth.CLIENT_ID), token=token
        )

    def fetch_token(self, username, password, otp_code=None):
        """Initial token fetch with username/password & 2FA
        :type username: str
        :type password: str
        :type otp_code: str
        """
        headers = {"User-Agent": self.user_agent, "hardware_id": self.hardware_id}

        if otp_code:
            headers["2fa-support"] = "true"
            headers["2fa-code"] = otp_code

        token = self._oauth.fetch_token(
            OAuth.ENDPOINT,
            username=username,
            password=password,
            scope=OAuth.SCOPE,
            headers=headers,
        )

        if self.token_updater is not None:
            self.token_updater(token)

        return token

    def refresh_tokens(self):
        """Refreshes the auth tokens"""
        token = self._oauth.refresh_token(
            OAuth.ENDPOINT, headers={"User-Agent": self.user_agent}
        )

        if self.token_updater is not None:
            self.token_updater(token)

        return token

    def get_hardware_id(self):
        """Get hardware ID."""
        return self.hardware_id

    def query(
        self, url, method="GET", extra_params=None, data=None, json=None, timeout=None
    ):
        """Query data from Ring API."""
        if timeout is None:
            timeout = TIMEOUT

        params = {"api_version": API_VERSION}

        if extra_params:
            params.update(extra_params)

        kwargs = {
            "params": params,
            "headers": {"User-Agent": self.user_agent},
            "timeout": timeout,
        }

        if method == "POST":
            if json is not None:
                kwargs["json"] = json
            if data is not None:
                kwargs["data"] = data

        if method == "PATCH":
            # PATCH method of requests library does not have a json argument
            if json is not None:
                kwargs["data"] = json_dumps(json)
            if data is not None:
                kwargs["data"] = data

        try:
            req = getattr(self._oauth, method.lower())(url, **kwargs)
        except TokenExpiredError:
            self._oauth.token = self.refresh_tokens()
            req = getattr(self._oauth, method.lower())(url, **kwargs)

        req.raise_for_status()

        return req

1	# coding: utf-8
2	# vim:sw=4:ts=4:et:
3	"""Python Ring Auth Class."""	1✔
4	from uuid import uuid4 as uuid	1✔
5	from json import dumps as json_dumps	1✔
6	from requests_oauthlib import OAuth2Session	1✔
7	from oauthlib.oauth2 import LegacyApplicationClient, TokenExpiredError	1✔
8	from ring_doorbell.const import OAuth, API_VERSION, TIMEOUT	1✔
9
10
11	class Auth:	1✔
12	"""A Python Auth class for Ring"""
13
14	def __init__(self, user_agent, token=None, token_updater=None, hardware_id=None):	1✔
15	"""
16	:type token: Optional[Dict[str, str]]
17	:type token_updater: Optional[Callable[[str], None]]
18	"""
19	self.user_agent = user_agent	1✔
20
21	self.hardware_id = hardware_id	1✔
22	if self.hardware_id is None:	1✔
23	self.hardware_id = str(uuid())	1✔
24
25	self.token_updater = token_updater	1✔
26	self._oauth = OAuth2Session(	1✔
27	client=LegacyApplicationClient(client_id=OAuth.CLIENT_ID), token=token
28	)
29
30	def fetch_token(self, username, password, otp_code=None):	1✔
31	"""Initial token fetch with username/password & 2FA
32	:type username: str
33	:type password: str
34	:type otp_code: str
35	"""
36	headers = {"User-Agent": self.user_agent, "hardware_id": self.hardware_id}	1✔
37
38	if otp_code:	1✔
39	headers["2fa-support"] = "true"	×
40	headers["2fa-code"] = otp_code	×
41
42	token = self._oauth.fetch_token(	1✔
43	OAuth.ENDPOINT,
44	username=username,
45	password=password,
46	scope=OAuth.SCOPE,
47	headers=headers,
48	)
49
50	if self.token_updater is not None:	1✔
51	self.token_updater(token)	×
52
53	return token	1✔
54
55	def refresh_tokens(self):	1✔
56	"""Refreshes the auth tokens"""
57	token = self._oauth.refresh_token(	×
58	OAuth.ENDPOINT, headers={"User-Agent": self.user_agent}
59	)
60
61	if self.token_updater is not None:	×
62	self.token_updater(token)	×
63
64	return token	×
65
66	def get_hardware_id(self):	1✔
67	"""Get hardware ID."""
68	return self.hardware_id	1✔
69
70	def query(	1✔
71	self, url, method="GET", extra_params=None, data=None, json=None, timeout=None
72	):
73	"""Query data from Ring API."""
74	if timeout is None:	1✔
75	timeout = TIMEOUT	1✔
76
77	params = {"api_version": API_VERSION}	1✔
78
79	if extra_params:	1✔
80	params.update(extra_params)	1✔
81
82	kwargs = {	1✔
83	"params": params,
84	"headers": {"User-Agent": self.user_agent},
85	"timeout": timeout,
86	}
87
88	if method == "POST":	1✔
89	if json is not None:	1✔
90	kwargs["json"] = json	1✔
91	if data is not None:	1✔
92	kwargs["data"] = data	1✔
93
94	if method == "PATCH":	1✔
95	# PATCH method of requests library does not have a json argument
96	if json is not None:	1✔
97	kwargs["data"] = json_dumps(json)	1✔
98	if data is not None:	1✔
99	kwargs["data"] = data	×
100
101	try:	1✔
102	req = getattr(self._oauth, method.lower())(url, **kwargs)	1✔
103	except TokenExpiredError:	×
104	self._oauth.token = self.refresh_tokens()	×
105	req = getattr(self._oauth, method.lower())(url, **kwargs)	×
106
107	req.raise_for_status()	1✔
108
109	return req	1✔

sdb9696 / python-ring-doorbell / 6148597193

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous