From 2d8219e2f268c56ce864f6d3529b2f300fd7f8f3 Mon Sep 17 00:00:00 2001 From: Rosi2143 Date: Sat, 18 Nov 2023 09:39:23 +0100 Subject: [PATCH] python formatter fix some flake8 warnings and typos add new pytests to cover changed lines Signed-off-by: Rosi2143 --- blinkpy/api.py | 4 +- blinkpy/auth.py | 6 +-- blinkpy/blinkpy.py | 13 ++++--- blinkpy/camera.py | 69 +++++++++++++++++++++------------- blinkpy/helpers/util.py | 7 +++- blinkpy/sync_module.py | 60 +++++++++++++++++------------ tests/test_camera_functions.py | 38 ++++++++++++++++++- tests/test_sync_module.py | 23 ++++++++++++ tox.ini | 2 +- 9 files changed, 158 insertions(+), 64 deletions(-) diff --git a/blinkpy/api.py b/blinkpy/api.py index 576568c0..4a9b612e 100644 --- a/blinkpy/api.py +++ b/blinkpy/api.py @@ -304,7 +304,7 @@ async def request_camera_sensors(blink, network, camera_id): :param blink: Blink instance. :param network: Sync module network id. - :param camera_id: Camera ID of camera to request sesnor info from. + :param camera_id: Camera ID of camera to request sensor info from. """ url = f"{blink.urls.base_url}/network/{network}/camera/{camera_id}/signals" return await http_get(blink, url) @@ -476,7 +476,7 @@ async def http_get( async def http_post(blink, url, is_retry=False, data=None, json=True, timeout=TIMEOUT): """Perform an http post request. - :param url: URL to perfom post request. + :param url: URL to perform post request. :param is_retry: Is this part of a re-auth attempt? :param data: str body for post request :param json: Return json response? TRUE/False diff --git a/blinkpy/auth.py b/blinkpy/auth.py index ca95321f..09eab0a0 100644 --- a/blinkpy/auth.py +++ b/blinkpy/auth.py @@ -25,7 +25,7 @@ def __init__(self, login_data=None, no_prompt=False, session=None): - username - password :param no_prompt: Should any user input prompts - be supressed? True/FALSE + be suppressed? True/FALSE """ if login_data is None: login_data = {} @@ -152,8 +152,8 @@ async def query( is_retry=False, timeout=TIMEOUT, ): - """Perform server requests.""" - """ + """Perform server requests. + :param url: URL to perform request :param data: Data to send :param headers: Headers to send diff --git a/blinkpy/blinkpy.py b/blinkpy/blinkpy.py index 8bfa4566..3e681517 100644 --- a/blinkpy/blinkpy.py +++ b/blinkpy/blinkpy.py @@ -58,7 +58,7 @@ def __init__( Useful for preventing motion_detected property from de-asserting too quickly. :param no_owls: Disable searching for owl entries (blink mini cameras \ - only known entity). Prevents an uneccessary API call \ + only known entity). Prevents an unnecessary API call \ if you don't have these in your network. """ self.auth = Auth(session=session) @@ -101,7 +101,7 @@ async def refresh(self, force=False, force_cache=False): # Prevents rapid clearing of motion detect property self.last_refresh = int(time.time()) last_refresh = datetime.datetime.fromtimestamp(self.last_refresh) - _LOGGER.debug(f"last_refresh={last_refresh}") + _LOGGER.debug("last_refresh = %s", last_refresh) return True return False @@ -128,8 +128,9 @@ async def start(self): # Initialize last_refresh to be just before the refresh delay period. self.last_refresh = int(time.time() - self.refresh_rate * 1.05) _LOGGER.debug( - f"Initialized last_refresh to {self.last_refresh} == " - f"{datetime.datetime.fromtimestamp(self.last_refresh)}" + "Initialized last_refresh to %s == %s", + self.last_refresh, + datetime.datetime.fromtimestamp(self.last_refresh), ) return await self.setup_post_verify() @@ -167,12 +168,13 @@ async def setup_sync_module(self, name, network_id, cameras): await self.sync[name].start() async def get_homescreen(self): - """Get homecreen information.""" + """Get homescreen information.""" if self.no_owls: _LOGGER.debug("Skipping owl extraction.") self.homescreen = {} return self.homescreen = await api.request_homescreen(self) + _LOGGER.debug("homescreen = %s", util.json_dumps(self.homescreen)) async def setup_owls(self): """Check for mini cameras.""" @@ -234,6 +236,7 @@ async def setup_camera_list(self): response = await api.request_camera_usage(self) try: for network in response["networks"]: + _LOGGER.info("network = %s", util.json_dumps(network)) camera_network = str(network["network_id"]) if camera_network not in all_cameras: all_cameras[camera_network] = [] diff --git a/blinkpy/camera.py b/blinkpy/camera.py index 9b848a26..bc5c6d80 100644 --- a/blinkpy/camera.py +++ b/blinkpy/camera.py @@ -20,7 +20,7 @@ class BlinkCamera: """Class to initialize individual camera.""" def __init__(self, sync): - """Initiailize BlinkCamera.""" + """Initialize BlinkCamera.""" self.sync = sync self.name = None self.camera_id = None @@ -76,7 +76,7 @@ def battery(self): @property def temperature_c(self): - """Return temperature in celcius.""" + """Return temperature in celsius.""" try: return round((self.temperature - 32) / 9.0 * 5.0, 1) except TypeError: @@ -170,7 +170,7 @@ async def get_thumbnail(self, url=None): if not url: url = self.thumbnail if not url: - _LOGGER.warning(f"Thumbnail URL not available: self.thumbnail={url}") + _LOGGER.warning("Thumbnail URL not available: self.thumbnail=%s", url) return None return await api.http_get( self.sync.blink, @@ -185,7 +185,7 @@ async def get_video_clip(self, url=None): if not url: url = self.clip if not url: - _LOGGER.warning(f"Video clip URL not available: self.clip={url}") + _LOGGER.warning("Video clip URL not available: self.clip=%s", url) return None return await api.http_get( self.sync.blink, @@ -240,7 +240,7 @@ def extract_config_info(self, config): self.product_type = config.get("type", None) async def get_sensor_info(self): - """Retrieve calibrated temperatue from special endpoint.""" + """Retrieve calibrated temperature from special endpoint.""" resp = await api.request_camera_sensors( self.sync.blink, self.network_id, self.camera_id ) @@ -248,7 +248,14 @@ async def get_sensor_info(self): self.temperature_calibrated = resp["temp"] except (TypeError, KeyError): self.temperature_calibrated = self.temperature - _LOGGER.warning("Could not retrieve calibrated temperature.") + _LOGGER.warning( + "Could not retrieve calibrated temperature response %s.", resp + ) + _LOGGER.warning( + "for network_id (%s) and camera_id (%s)", + self.network_id, + self.camera_id, + ) async def update_images(self, config, force_cache=False, expire_clips=True): """Update images for camera.""" @@ -278,7 +285,7 @@ async def update_images(self, config, force_cache=False, expire_clips=True): new_thumbnail = urljoin(self.sync.urls.base_url, thumb_string) else: - _LOGGER.warning("Could not find thumbnail for camera %s", self.name) + _LOGGER.warning("Could not find thumbnail for camera %s.", self.name) try: self.motion_detected = self.sync.motion[self.name] @@ -288,7 +295,7 @@ async def update_images(self, config, force_cache=False, expire_clips=True): clip_addr = None try: - def timest(record): + def timesort(record): rec_time = record["time"] iso_time = datetime.datetime.fromisoformat(rec_time) stamp = int(iso_time.timestamp()) @@ -298,7 +305,7 @@ def timest(record): len(self.sync.last_records) > 0 and len(self.sync.last_records[self.name]) > 0 ): - last_records = sorted(self.sync.last_records[self.name], key=timest) + last_records = sorted(self.sync.last_records[self.name], key=timesort) for rec in last_records: clip_addr = rec["clip"] self.clip = f"{self.sync.urls.base_url}{clip_addr}" @@ -310,17 +317,21 @@ def timest(record): self.recent_clips.append(recent) if len(self.recent_clips) > 0: _LOGGER.debug( - f"Found {len(self.recent_clips)} recent clips for {self.name}" + "Found %s recent clips for %s", + len(self.recent_clips), + self.name, ) _LOGGER.debug( - f"Most recent clip for {self.name} was created at " - f"{self.last_record}: {self.clip}" + "Most recent clip for %s was created at %s : %s", + self.name, + self.last_record, + self.clip, ) except (KeyError, IndexError): ex = traceback.format_exc() trace = "".join(traceback.format_stack()) - _LOGGER.error(f"Error getting last records for '{self.name}': {ex}") - _LOGGER.debug(f"\n{trace}") + _LOGGER.error("Error getting last records for '%s': %s", self.name, ex) + _LOGGER.debug("\n%s", trace) # If the thumbnail or clip have changed, update the cache update_cached_image = False @@ -356,12 +367,13 @@ async def expire_recent_clips(self, delta=datetime.timedelta(hours=1)): to_keep.append(clip) num_expired = len(self.recent_clips) - len(to_keep) if num_expired > 0: - _LOGGER.info(f"Expired {num_expired} clips from '{self.name}'") + _LOGGER.info("Expired %s clips from '%s'", num_expired, self.name) self.recent_clips = copy.deepcopy(to_keep) if len(self.recent_clips) > 0: _LOGGER.info( - f"'{self.name}' has {len(self.recent_clips)} " - "clips available for download" + "'%s' has %s clips available for download", + self.name, + len(self.recent_clips), ) for clip in self.recent_clips: url = clip["clip"] @@ -369,7 +381,7 @@ async def expire_recent_clips(self, delta=datetime.timedelta(hours=1)): await api.http_post(self.sync.blink, url) async def get_liveview(self): - """Get livewview rtsps link.""" + """Get liveview rtsps link.""" response = await api.request_camera_liveview( self.sync.blink, self.sync.network_id, self.camera_id ) @@ -384,8 +396,8 @@ async def image_to_file(self, path): _LOGGER.debug("Writing image from %s to %s", self.name, path) response = await self.get_media() if response and response.status == 200: - async with open(path, "wb") as imgfile: - await imgfile.write(await response.read()) + async with open(path, "wb") as imagefile: + await imagefile.write(await response.read()) else: _LOGGER.error("Cannot write image to file, response %s", response.status) @@ -425,7 +437,7 @@ async def save_recent_clips( created=created_at, name=to_alphanumeric(self.name) ) path = os.path.join(output_dir, file_name) - _LOGGER.debug(f"Saving {clip_addr} to {path}") + _LOGGER.debug("Saving %s to %s", clip_addr, path) media = await self.get_video_clip(clip_addr) if media and media.status == 200: async with open(path, "wb") as clip_file: @@ -434,19 +446,22 @@ async def save_recent_clips( try: # Remove recent clip from the list once the download has finished. self.recent_clips.remove(clip) - _LOGGER.debug(f"Removed {clip} from recent clips") + _LOGGER.debug("Removed %s from recent clips", clip) except ValueError: ex = traceback.format_exc() - _LOGGER.error(f"Error removing clip from list: {ex}") + _LOGGER.error("Error removing clip from list: %s", ex) trace = "".join(traceback.format_stack()) - _LOGGER.debug(f"\n{trace}") + _LOGGER.debug("\n%s", trace) if len(recent) == 0: - _LOGGER.info(f"No recent clips to save for '{self.name}'.") + _LOGGER.info("No recent clips to save for '%s'.", self.name) else: _LOGGER.info( - f"Saved {num_saved} of {len(recent)} recent clips from " - f"'{self.name}' to directory {output_dir}" + "Saved %s of %s recent clips from '%s' to directory %s", + num_saved, + len(recent), + self.name, + output_dir, ) diff --git a/blinkpy/helpers/util.py b/blinkpy/helpers/util.py index 2f1c16cd..9da0629c 100644 --- a/blinkpy/helpers/util.py +++ b/blinkpy/helpers/util.py @@ -37,8 +37,13 @@ async def json_save(data, file_name): await json_file.write(json.dumps(data, indent=4)) +def json_dumps(json_in, indent=2): + """Return a well formated json string.""" + return json.dumps(json_in, indent=indent) + + def gen_uid(size, uid_format=False): - """Create a random sring.""" + """Create a random string.""" if uid_format: token = ( f"BlinkCamera_{secrets.token_hex(4)}-" diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index ea1fcfb8..05d68a4b 100644 --- a/blinkpy/sync_module.py +++ b/blinkpy/sync_module.py @@ -9,7 +9,12 @@ from requests.structures import CaseInsensitiveDict from blinkpy import api from blinkpy.camera import BlinkCamera, BlinkCameraMini, BlinkDoorbell -from blinkpy.helpers.util import time_to_seconds, backoff_seconds, to_alphanumeric +from blinkpy.helpers.util import ( + time_to_seconds, + backoff_seconds, + to_alphanumeric, + json_dumps, +) from blinkpy.helpers.constants import ONLINE _LOGGER = logging.getLogger(__name__) @@ -184,6 +189,7 @@ async def update_cameras(self, camera_type=BlinkCamera): try: _LOGGER.debug("Updating cameras") for camera_config in self.camera_list: + _LOGGER.debug("Updating camera_config %s", json_dumps(camera_config)) if "name" not in camera_config: break blink_camera_type = camera_config.get("type", "") @@ -208,10 +214,11 @@ async def update_cameras(self, camera_type=BlinkCamera): def get_unique_info(self, name): """Extract unique information for Minis and Doorbells.""" try: - for camera_type in self.type_key_map: - type_key = self.type_key_map[camera_type] + for type_key in self.type_key_map.values(): for device in self.blink.homescreen[type_key]: + _LOGGER.debug("checking device %s", device) if device["name"] == name: + _LOGGER.debug("Found unique_info %s", device) return device except (TypeError, KeyError): pass @@ -277,18 +284,19 @@ async def check_new_videos(self): try: interval = self.blink.last_refresh - self.motion_interval * 60 last_refresh = datetime.datetime.fromtimestamp(self.blink.last_refresh) - _LOGGER.debug(f"last_refresh = {last_refresh}") - _LOGGER.debug(f"interval={interval}") + _LOGGER.debug("last_refresh = %s", last_refresh) + _LOGGER.debug("interval = %s", interval) except TypeError: # This is the first start, so refresh hasn't happened yet. # No need to check for motion. ex = traceback.format_exc() _LOGGER.error( - "Error calculating interval " - f"(last_refresh={self.blink.last_refresh}): {ex}" + "Error calculating interval (last_refresh = %s): %s", + self.blink.last_refresh, + ex, ) trace = "".join(traceback.format_stack()) - _LOGGER.debug(f"\n{trace}") + _LOGGER.debug("\n%s", trace) _LOGGER.info("No new videos since last refresh.") return False @@ -324,15 +332,17 @@ async def check_new_videos(self): except KeyError: last_refresh = datetime.datetime.fromtimestamp(self.blink.last_refresh) _LOGGER.debug( - f"No new videos for {entry} since last refresh at {last_refresh}." + "No new videos for %s since last refresh at %s.", + entry, + last_refresh, ) # Process local storage if active and if the manifest is ready. last_manifest_read = datetime.datetime.fromisoformat( self._local_storage["last_manifest_read"] ) - _LOGGER.debug(f"last_manifest_read = {last_manifest_read}") - _LOGGER.debug(f"Manifest ready? {self.local_storage_manifest_ready}") + _LOGGER.debug("last_manifest_read = %s", last_manifest_read) + _LOGGER.debug("Manifest ready? %s", self.local_storage_manifest_ready) if self.local_storage and self.local_storage_manifest_ready: _LOGGER.debug("Processing updated manifest") manifest = self._local_storage["manifest"] @@ -349,17 +359,20 @@ async def check_new_videos(self): iso_timestamp = item.created_at.isoformat() _LOGGER.debug( - f"Checking '{item.name}': clip_time={iso_timestamp}, " - f"manifest_read={last_manifest_read}" + "Checking '%s': clip_time = %s, manifest_read = %s", + item.name, + iso_timestamp, + last_manifest_read, ) # Exit the loop once there are no new videos in the list. if not self.check_new_video_time(iso_timestamp, last_manifest_read): _LOGGER.info( "No new local storage videos since last manifest " - f"read at {last_read_local}." + "read at %s.", + last_read_local, ) break - _LOGGER.debug(f"Found new item in local storage manifest: {item}") + _LOGGER.debug("Found new item in local storage manifest: %s", item) name = item.name clip_url = item.url(last_manifest_id) await item.prepare_download(self.blink) @@ -375,8 +388,8 @@ async def check_new_videos(self): datetime.datetime.utcnow() - datetime.timedelta(seconds=10) ).isoformat() self._local_storage["last_manifest_read"] = last_manifest_read - _LOGGER.debug(f"Updated last_manifest_read to {last_manifest_read}") - _LOGGER.debug(f"Last clip time was {last_clip_time}") + _LOGGER.debug("Updated last_manifest_read to %s", last_manifest_read) + _LOGGER.debug("Last clip time was %s", last_clip_time) # We want to keep the last record when no new motion was detected. for camera in self.cameras: # Check if there are no new records, indicating motion. @@ -389,8 +402,8 @@ async def check_new_videos(self): return True def check_new_video_time(self, timestamp, reference=None): - """Check if video has timestamp since last refresh.""" - """ + """Check if video has timestamp since last refresh. + :param timestamp ISO-formatted timestamp string :param reference ISO-formatted reference timestamp string """ @@ -450,14 +463,15 @@ async def update_local_storage_manifest(self): num_added = len(self._local_storage["manifest"]) - num_stored if num_added > 0: _LOGGER.info( - f"Found {num_added} new clip(s) in local storage " - f"manifest id={manifest_id}" + "Found %s new clip(s) in local storage manifest id = %s", + num_added, + manifest_id, ) except (TypeError, KeyError): ex = traceback.format_exc() - _LOGGER.error(f"Could not extract clips list from response: {ex}") + _LOGGER.error("Could not extract clips list from response: %s", ex) trace = "".join(traceback.format_stack()) - _LOGGER.debug(f"\n{trace}") + _LOGGER.debug("\n%s", trace) self._local_storage["manifest_stale"] = True return None diff --git a/tests/test_camera_functions.py b/tests/test_camera_functions.py index f90a9582..c1b7df49 100644 --- a/tests/test_camera_functions.py +++ b/tests/test_camera_functions.py @@ -130,9 +130,13 @@ async def test_no_thumbnails(self, mock_resp): [ ( "WARNING:blinkpy.camera:Could not retrieve calibrated " - "temperature." + f"temperature response {mock_resp.return_value}." ), - ("WARNING:blinkpy.camera:Could not find thumbnail for camera new"), + ( + f"WARNING:blinkpy.camera:for network_id ({config['network_id']}) " + f"and camera_id ({self.camera.camera_id})" + ), + ("WARNING:blinkpy.camera:Could not find thumbnail for camera new."), ], ) @@ -374,3 +378,33 @@ async def test_save_recent_clips(self, mock_clip, mock_open, mock_resp): f"'{self.camera.name}' to directory /tmp/", ) assert mock_open.call_count == 2 + + def remove_clip(self): + """Remove all clips to raise an exception on second removal.""" + self[0] *= 0 + return mresp.MockResponse({}, 200, raw_data="raw data") + + @mock.patch("blinkpy.camera.open", create=True) + @mock.patch( + "blinkpy.camera.BlinkCamera.get_video_clip", + create=True, + side_effect=remove_clip, + ) + async def test_save_recent_clips_exception(self, mock_clip, mock_open, mock_resp): + """Test corruption in recent clip list.""" + self.camera.recent_clips = [] + now = datetime.datetime.now() + self.camera.recent_clips.append( + { + "time": (now - datetime.timedelta(minutes=20)).isoformat(), + "clip": [self.camera.recent_clips], + }, + ) + with self.assertLogs(level="ERROR") as dl_log: + await self.camera.save_recent_clips() + print(f"Output = {dl_log.output}") + self.assertTrue( + "ERROR:blinkpy.camera:Error removing clip from list:" + in "\t".join(dl_log.output) + ) + assert mock_open.call_count == 1 diff --git a/tests/test_sync_module.py b/tests/test_sync_module.py index a0f1a03f..9b4c5a5d 100644 --- a/tests/test_sync_module.py +++ b/tests/test_sync_module.py @@ -1,5 +1,6 @@ """Tests camera and system functions.""" import datetime +import logging from unittest import IsolatedAsyncioTestCase from unittest import mock import aiofiles @@ -16,6 +17,10 @@ import tests.mock_responses as mresp from .test_api import COMMAND_RESPONSE, COMMAND_COMPLETE +_LOGGER = logging.getLogger(__name__) +logging.basicConfig(filename="blinkpy_test.log", level=logging.DEBUG) +_LOGGER.setLevel(logging.DEBUG) + @mock.patch("blinkpy.auth.Auth.query") class TestBlinkSyncModule(IsolatedAsyncioTestCase): @@ -76,6 +81,24 @@ def test_bad_arm(self, mock_resp) -> None: self.assertEqual(self.blink.sync["test"].arm, None) self.assertFalse(self.blink.sync["test"].available) + def test_get_unique_info_valid_device(self, mock_resp) -> None: + """Check that we get the correct info.""" + device = { + "enabled": True, + "name": "doorbell1", + } + self.blink.homescreen = {"doorbells": [device], "owls": []} + self.assertEqual(self.blink.sync["test"].get_unique_info("doorbell1"), device) + + def test_get_unique_info_invalid_device(self, mock_resp) -> None: + """Check what happens if the devide does not exist.""" + device = { + "enabled": True, + "name": "doorbell1", + } + self.blink.homescreen = {"doorbells": [device], "owls": []} + self.assertEqual(self.blink.sync["test"].get_unique_info("doorbell2"), None) + async def test_get_events(self, mock_resp) -> None: """Test get events function.""" mock_resp.return_value = {"event": True} diff --git a/tox.ini b/tox.ini index dc228048..3d0e66c5 100644 --- a/tox.ini +++ b/tox.ini @@ -31,7 +31,7 @@ deps = basepython = python3 commands = ruff check blinkpy tests blinkapp - black --check --diff blinkpy tests blinkapp + black --check --color --diff blinkpy tests blinkapp rst-lint README.rst CHANGES.rst CONTRIBUTING.rst [testenv:build]