tchellomello / python-amcrest / 198

pending completion

Build # 198

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/tchellomello/python-amcrest/commit/<a class=hub.com/tchellomello/python-amcrest/commit/<a class="double-link" href="https://git"><a class=hub.com/tchellomello/python-amcrest/commit/396224045b5a9b69391bbb26bb833110a5033422">396224045">&lt;a href=&quot;https://github.com/tchellomello/python-amcrest/commit/</a><a class="double-link" href="https://github.com/tchellomello/python-amcrest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/tchellomello/python-amcrest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/tchellomello/python-amcrest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/tchellomello/python-amcrest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/tchellomello/python-amcrest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/tchellomello/python-amcrest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/tchellomello/python-amcrest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/tchellomello/python-amcrest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/tchellomello/python-amcrest/commit/396224045b5a9b69391bbb26bb833110a5033422">396224045</a><a href="https://github.com/tchellomello/python-amcrest/commit/396224045b5a9b69391bbb26bb833110a5033422">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/tchellomello/python-amcrest/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/tchellomello/python-amcrest/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/tchellomello/python-amcrest/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/tchellomello/python-amcrest/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;... (continued)

Run Details

4 of 4 new or added lines in 1 file covered. (100.0%)

215 of 649 relevant lines covered (33.13%)

0.99 hits per line

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

28.57

/src/amcrest/http.py

# -*- coding: utf-8 -*-
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, version 2 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# vim:sw=4:ts=4:et
import logging
import requests

from requests.adapters import HTTPAdapter

from amcrest.utils import clean_url, pretty

from amcrest.audio import Audio
from amcrest.event import Event
from amcrest.log import Log
from amcrest.motion_detection import MotionDetection
from amcrest.nas import Nas
from amcrest.network import Network
from amcrest.ptz import Ptz
from amcrest.record import Record
from amcrest.snapshot import Snapshot
from amcrest.special import Special
from amcrest.storage import Storage
from amcrest.system import System
from amcrest.user_management import UserManagement
from amcrest.video import Video

from amcrest.config import TIMEOUT_HTTP_PROTOCOL, MAX_RETRY_HTTP_CONNECTION

_LOGGER = logging.getLogger(__name__)


# pylint: disable=too-many-ancestors
class Http(System, Network, MotionDetection, Snapshot,
           UserManagement, Event, Audio, Record, Video,
           Log, Ptz, Special, Storage, Nas):

    def __init__(self, host, port, user,
                 password, verbose=True, protocol='http',
                 retries_connection=None, timeout_protocol=None):

        self._host = clean_url(host)
        self._port = port
        self._user = user
        self._password = password
        self._verbose = verbose
        self._protocol = protocol
        self._base_url = self.__base_url()

        if timeout_protocol is None:
            self._timeout_protocol = TIMEOUT_HTTP_PROTOCOL
        else:
            self._timeout_protocol = timeout_protocol

        if retries_connection is None:
            self._retries_conn = MAX_RETRY_HTTP_CONNECTION
        else:
            self._retries_conn = retries_connection

        self._token = self._generate_token()
        self._set_name()

    def _generate_token(self):
        """Discover which authentication method to use.

        Latest firmwares requires HttpDigestAuth
        Older firmwares requires HttpBasicAuth
        """
        url = self.__base_url('magicBox.cgi?action=getMachineName')
        try:
            # try old firmware first to force 401 if fails
            self._authentication = 'basic'
            auth = requests.auth.HTTPBasicAuth(self._user, self._password)
            req = requests.get(url, auth=auth)
            req.raise_for_status()

        except requests.exceptions.HTTPError:
            # if 401, then try new digest method
            self._authentication = 'digest'
            auth = requests.auth.HTTPDigestAuth(self._user, self._password)
            req = requests.get(url, auth=auth)
            req.raise_for_status()

            # check if user passed
            if 'invalid' in req.text.lower() or \
               'error' in req.text.lower():
                _LOGGER.info("%s Invalid credentials", req.text)
                raise
        return auth

    def _set_name(self):
        """Set device name."""
        try:
            self._name = pretty(self.machine_name)
            self._serial = self.serial_number
        except AttributeError:
            self._name = None
            self._serial = None

    def __repr__(self):
        """Default object representation."""
        return "<{0}: {1}>".format(self._name, self._serial)

    def as_dict(self):
        """Callback for __dict__."""
        cdict = self.__dict__.copy()
        cdict['_token'] = '_redacted'
        return cdict

    # Base methods
    def __base_url(self, param=""):
        return '%s://%s:%s/cgi-bin/%s' % (self._protocol, self._host,
                                          str(self._port), param)

    def get_base_url(self):
        return self._base_url

    def command(self, cmd, retries=None, timeout_cmd=None):
        """
        Args:
            cmd - command to execute via http
            timeout_cmd - timeout, default 3sec
            retries - maximum number of retries each connection should attempt
        """
        if timeout_cmd is not None:
            self._timeout_protocol = timeout_cmd

        if retries is not None:
            self._retries_conn = retries

        session = requests.Session()
        session.mount('http://', HTTPAdapter(max_retries=self._retries_conn))
        session.mount('https://', HTTPAdapter(max_retries=self._retries_conn))

        url = self.__base_url(cmd)
        resp = None
        loop = 0
        while loop <= self._retries_conn:
            loop += 1
            _LOGGER.debug("Running query attempt %s", loop)
            try:
                resp = session.get(
                    url,
                    auth=self._token,
                    stream=True,
                    timeout=self._timeout_protocol,
                )
                resp.raise_for_status()
                break
            except requests.exceptions.HTTPError as error:
                _LOGGER.debug("Trying again due error %s", error)
                continue

            # keep the loop when 401
            if resp.status_code == 401:
                continue

        # if we got here, let's raise because it did not work
        if resp is None:
            raise ValueError('Something went wrong!!!!')

        _LOGGER.debug("Query worked. Exit code: <%s>", resp.status_code)
        return resp

    def command_audio(self, cmd, file_content, http_header,
                      timeout=None):
        url = self.__base_url(cmd)

        if timeout is not None:
            timeout = self._timeout_protocol

        try:
            requests.post(
                url,
                files=file_content,
                auth=self._token,
                headers=http_header,
                timeout=timeout
            )
        except requests.exceptions.ReadTimeout:
            pass

1	# -- coding: utf-8 --
2	#
3	# This program is free software; you can redistribute it and/or modify
4	# it under the terms of the GNU General Public License as published by
5	# the Free Software Foundation, version 2 of the License.
6	#
7	# This program is distributed in the hope that it will be useful,
8	# but WITHOUT ANY WARRANTY; without even the implied warranty of
9	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10	# GNU General Public License for more details.
11	#
12	# vim:sw=4:ts=4:et
13	import logging	3✔
14	import requests	3✔
15
16	from requests.adapters import HTTPAdapter	3✔
17
18	from amcrest.utils import clean_url, pretty	3✔
19
20	from amcrest.audio import Audio	3✔
21	from amcrest.event import Event	3✔
22	from amcrest.log import Log	3✔
23	from amcrest.motion_detection import MotionDetection	3✔
24	from amcrest.nas import Nas	3✔
25	from amcrest.network import Network	3✔
26	from amcrest.ptz import Ptz	3✔
27	from amcrest.record import Record	3✔
28	from amcrest.snapshot import Snapshot	3✔
29	from amcrest.special import Special	3✔
30	from amcrest.storage import Storage	3✔
31	from amcrest.system import System	3✔
32	from amcrest.user_management import UserManagement	3✔
33	from amcrest.video import Video	3✔
34
35	from amcrest.config import TIMEOUT_HTTP_PROTOCOL, MAX_RETRY_HTTP_CONNECTION	3✔
36
37	_LOGGER = logging.getLogger(__name__)	3✔
38
39
40	# pylint: disable=too-many-ancestors
41	class Http(System, Network, MotionDetection, Snapshot,	3✔
42	UserManagement, Event, Audio, Record, Video,
43	Log, Ptz, Special, Storage, Nas):
44
45	def __init__(self, host, port, user,	3✔
46	password, verbose=True, protocol='http',
47	retries_connection=None, timeout_protocol=None):
48
49	self._host = clean_url(host)	×
50	self._port = port	×
51	self._user = user	×
52	self._password = password	×
53	self._verbose = verbose	×
54	self._protocol = protocol	×
55	self._base_url = self.__base_url()	×
56
57	if timeout_protocol is None:	×
58	self._timeout_protocol = TIMEOUT_HTTP_PROTOCOL	×
59	else:
60	self._timeout_protocol = timeout_protocol	×
61
62	if retries_connection is None:	×
63	self._retries_conn = MAX_RETRY_HTTP_CONNECTION	×
64	else:
65	self._retries_conn = retries_connection	×
66
67	self._token = self._generate_token()	×
68	self._set_name()	×
69
70	def _generate_token(self):	3✔
71	"""Discover which authentication method to use.
72
73	Latest firmwares requires HttpDigestAuth
74	Older firmwares requires HttpBasicAuth
75	"""
76	url = self.__base_url('magicBox.cgi?action=getMachineName')	×
77	try:	×
78	# try old firmware first to force 401 if fails
79	self._authentication = 'basic'	×
80	auth = requests.auth.HTTPBasicAuth(self._user, self._password)	×
81	req = requests.get(url, auth=auth)	×
82	req.raise_for_status()	×
83
84	except requests.exceptions.HTTPError:	×
85	# if 401, then try new digest method
86	self._authentication = 'digest'	×
87	auth = requests.auth.HTTPDigestAuth(self._user, self._password)	×
88	req = requests.get(url, auth=auth)	×
89	req.raise_for_status()	×
90
91	# check if user passed
92	if 'invalid' in req.text.lower() or \	×
93	'error' in req.text.lower():
94	_LOGGER.info("%s Invalid credentials", req.text)	×
95	raise	×
96	return auth	×
97
98	def _set_name(self):	3✔
99	"""Set device name."""
100	try:	×
101	self._name = pretty(self.machine_name)	×
102	self._serial = self.serial_number	×
103	except AttributeError:	×
104	self._name = None	×
105	self._serial = None	×
106
107	def __repr__(self):	3✔
108	"""Default object representation."""
109	return "<{0}: {1}>".format(self._name, self._serial)	×
110
111	def as_dict(self):	3✔
112	"""Callback for __dict__."""
113	cdict = self.__dict__.copy()	×
114	cdict['_token'] = '_redacted'	×
115	return cdict	×
116
117	# Base methods
118	def __base_url(self, param=""):	3✔
119	return '%s://%s:%s/cgi-bin/%s' % (self._protocol, self._host,	×
120	str(self._port), param)
121
122	def get_base_url(self):	3✔
123	return self._base_url	×
124
125	def command(self, cmd, retries=None, timeout_cmd=None):	3✔
126	"""
127	Args:
128	cmd - command to execute via http
129	timeout_cmd - timeout, default 3sec
130	retries - maximum number of retries each connection should attempt
131	"""
132	if timeout_cmd is not None:	×
133	self._timeout_protocol = timeout_cmd	×
134
135	if retries is not None:	×
136	self._retries_conn = retries	×
137
138	session = requests.Session()	×
139	session.mount('http://', HTTPAdapter(max_retries=self._retries_conn))	×
140	session.mount('https://', HTTPAdapter(max_retries=self._retries_conn))	×
141
142	url = self.__base_url(cmd)	×
143	resp = None	×
144	loop = 0	×
145	while loop <= self._retries_conn:	×
146	loop += 1	×
147	_LOGGER.debug("Running query attempt %s", loop)	×
148	try:	×
149	resp = session.get(	×
150	url,
151	auth=self._token,
152	stream=True,
153	timeout=self._timeout_protocol,
154	)
155	resp.raise_for_status()	×
156	break	×
157	except requests.exceptions.HTTPError as error:	×
158	_LOGGER.debug("Trying again due error %s", error)	×
159	continue	×
160
161	# keep the loop when 401
162	if resp.status_code == 401:	×
163	continue	×
164
165	# if we got here, let's raise because it did not work
166	if resp is None:	×
167	raise ValueError('Something went wrong!!!!')	×
168
169	_LOGGER.debug("Query worked. Exit code: <%s>", resp.status_code)	×
170	return resp	×
171
172	def command_audio(self, cmd, file_content, http_header,	3✔
173	timeout=None):
174	url = self.__base_url(cmd)	×
175
176	if timeout is not None:	×
177	timeout = self._timeout_protocol	×
178
179	try:	×
180	requests.post(	×
181	url,
182	files=file_content,
183	auth=self._token,
184	headers=http_header,
185	timeout=timeout
186	)
187	except requests.exceptions.ReadTimeout:	×
188	pass	×

tchellomello / python-amcrest / 198

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