From e38d956aa96165d9bfcec54f87932312ce65611b Mon Sep 17 00:00:00 2001 From: oliver Date: Sun, 2 Feb 2025 22:29:39 +0100 Subject: [PATCH 1/3] fix unexpected picam requests errors with exception handling --- camera/camera.py | 240 +++++++++++++++++++++++++++++++------------ pytest.ini | 3 + requirements.txt | 1 + test/test_camera.py | 216 +++++++++++++++++++++----------------- test/test_camera2.py | 135 ++++++++++++++++++++++++ test/test_camera3.py | 124 ++++++++++++++++++++++ 6 files changed, 559 insertions(+), 160 deletions(-) create mode 100644 pytest.ini create mode 100644 test/test_camera2.py create mode 100644 test/test_camera3.py diff --git a/camera/camera.py b/camera/camera.py index f6fe39f..8e73289 100644 --- a/camera/camera.py +++ b/camera/camera.py @@ -10,6 +10,7 @@ import aiohttp import requests + from astral import LocationInfo from astral.sun import sun from blinkpy.auth import Auth @@ -19,12 +20,20 @@ from config.config_util import Configuration, DefaultCam from config.data_class import Camera_Task, Message_Task +logging.basicConfig( + level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s" +) logger: logging.Logger = logging.getLogger(name="camera") class Camera: - def __init__(self, config: Configuration, loop, camera_task_queue_async: asyncio.Queue, - message_task_queue: queue.Queue) -> None: + def __init__( + self, + config: Configuration, + loop, + camera_task_queue_async: asyncio.Queue, + message_task_queue: queue.Queue, + ) -> None: """ Initializes a new instance of the Camera class. @@ -51,11 +60,12 @@ def __init__(self, config: Configuration, loop, camera_task_queue_async: asyncio async def start(self) -> None: """ - Initializes the camera, starts a blink session if enabled, and processes various camera tasks asynchronously. + Initializes the camera, starts a blink session if enabled, and processes + various camera tasks asynchronously. """ logger.debug(msg="thread camera start") - if (self.config.blink_enabled): + if self.config.blink_enabled: self.logger.debug("start blink session") self.session = aiohttp.ClientSession() self.logger.debug("add session to blink") @@ -80,13 +90,16 @@ async def start(self) -> None: self.logger.info(f"processing task.photo: {task.photo}") await self.choose_cam(task) - elif task.picam_photo: - self.logger.info(f"processing task.picam_photo: {task.picam_photo}") + self.logger.info( + f"processing task.picam_photo: {task.picam_photo}" + ) await self._picam_foto_helper(task) elif task.blink_photo: - self.logger.info(f"processing task.blink_photo: {task.blink_photo}") + self.logger.info( + f"processing task.blink_photo: {task.blink_photo}" + ) await self._blink_foto_helper(task) elif task.blink_mfa: @@ -96,18 +109,33 @@ async def start(self) -> None: result = await self.save_blink_config() if result: self.message_task_queue.put( - Message_Task(reply=True, chat_id=task.chat_id, message=task.message, - data_text="blink MFA added")) + Message_Task( + reply=True, + chat_id=task.chat_id, + message=task.message, + data_text="blink MFA added", + ) + ) else: self.message_task_queue.put( - Message_Task(reply=True, chat_id=task.chat_id, message=task.message, - data_text="an error occured during blink MFA " - "processing")) + Message_Task( + reply=True, + chat_id=task.chat_id, + message=task.message, + data_text="an error occured during blink MFA " + "processing", + ) + ) else: self.message_task_queue.put( - Message_Task(reply=True, chat_id=task.chat_id, message=task.message, - data_text="an error occured during blink MFA " - "processing")) + Message_Task( + reply=True, + chat_id=task.chat_id, + message=task.message, + data_text="an error occured during blink MFA " + "processing", + ) + ) except Exception as err: self.logger.error("Error: {0}".format(err)) pass @@ -116,7 +144,8 @@ async def start(self) -> None: async def choose_cam(self, task: Camera_Task): """ - Asynchronously chooses a camera based on various conditions such as daylight detection, night vision, and default camera type. + Asynchronously chooses a camera based on various conditions such as + daylight detection, night vision, and default camera type. Args: self: The Camera object. @@ -127,7 +156,7 @@ async def choose_cam(self, task: Camera_Task): """ self.logger.debug("choose camera") # check if daylight detection is enabled or not take default - if (self.config.enable_detect_daylight): + if self.config.enable_detect_daylight: self.logger.debug("daylight detection is enabled") # detect daylight = true or night = false @@ -171,7 +200,8 @@ async def choose_cam(self, task: Camera_Task): async def _picam_foto_helper(self, task: Camera_Task) -> bool: """ - A helper function for the PiCam to take and download a photo, log success and errors, and manage the message queue. + A helper function for the PiCam to take and download a photo, log + success and errors, and manage the message queue. Parameters: task (Camera_Task): The task object associated with the photo-taking process. @@ -179,7 +209,7 @@ async def _picam_foto_helper(self, task: Camera_Task) -> bool: Returns: bool: True if the photo was successfully taken and downloaded, False otherwise. """ - if (self.config.picam_enabled): + if self.config.picam_enabled: self.logger.debug("_picam_foto_helper - picam enabled") result = self.picam_request_take_foto() if result: @@ -190,7 +220,9 @@ async def _picam_foto_helper(self, task: Camera_Task) -> bool: self.put_msg_queue_photo(task) else: self.logger.error("picam snapshot download error") - self.put_msg_queue_error(task, "an error occured during PiCam foto download ") + self.put_msg_queue_error( + task, "an error occured during PiCam foto download " + ) else: self.logger.error("picam snapshot error") self.put_msg_queue_error(task, "an error occured during PiCam snapshot") @@ -209,9 +241,13 @@ async def _blink_foto_helper(self, task: Camera_Task) -> bool: Returns: bool: True if the photo was successfully taken and processed, False otherwise. - This function checks if the Blink camera is enabled in the configuration. If it is, it takes a photo using the Blink camera and processes the result. If the photo was successfully taken, it puts the photo in the message queue. If the photo was not taken successfully, it puts an error message in the message queue. If the Blink camera is not enabled, it logs a message and returns False. + This function checks if the Blink camera is enabled in the configuration. If it is, + it takes a photo using the Blink camera and processes the result. If the photo was + successfully taken, it puts the photo in the message queue. If the photo was not taken + successfully, it puts an error message in the message queue. If the Blink camera is not + enabled, it logs a message and returns False. """ - if (self.config.blink_enabled): + if self.config.blink_enabled: self.logger.debug("_blink_foto_helper - blink enabled") result = await self.blink_snapshot() if result: @@ -219,7 +255,9 @@ async def _blink_foto_helper(self, task: Camera_Task) -> bool: self.put_msg_queue_photo(task) else: self.logger.error("blink snapshot error detected") - self.put_msg_queue_error(task, "an error occured during pocessing Blink foto task") + self.put_msg_queue_error( + task, "an error occured during pocessing Blink foto task" + ) else: self.logger.info("_blink_foto_helper - blink disabled") return False @@ -236,18 +274,22 @@ async def _check_picam_result(self, task: Camera_Task, result: bool) -> bool: Returns: bool: The result of the picam function. - This function checks the result of the picam function. If the result is False, it logs a debug message. - If the blink camera is enabled in the configuration, it calls the _blink_foto_helper function with the task. + This function checks the result of the picam function. If the result is False, + it logs a debug message. + If the blink camera is enabled in the configuration, it calls the _blink_foto_helper + function with the task. If the blink camera is not enabled, it logs an error message. """ self.logger.debug("picam _check_picam_result") if not result: self.logger.debug("_check_picam_result FALSE") - if (self.config.blink_enabled): + if self.config.blink_enabled: self.logger.error("_check_picam_result - second try now with blink") return await self._blink_foto_helper(task) else: - self.logger.error("_check_picam_result - blink not enabled for second try") + self.logger.error( + "_check_picam_result - blink not enabled for second try" + ) return result async def _check_blink_result(self, task: Camera_Task, result: bool) -> bool: @@ -261,18 +303,22 @@ async def _check_blink_result(self, task: Camera_Task, result: bool) -> bool: Returns: bool: The result of the picam function. - This function checks the result of the picam function. If the result is False, it logs a debug message. - If the blink camera is enabled in the configuration, it calls the _picam_foto_helper function with the task. + This function checks the result of the picam function. If the result is False, + it logs a debug message. + If the blink camera is enabled in the configuration, it calls the _picam_foto_helper + function with the task. If the blink camera is not enabled, it logs an error message. """ self.logger.debug("blink _check_blink_result") if not result: self.logger.debug("blink _check_blink_result FALSE") - if (self.config.picam_enabled): + if self.config.picam_enabled: self.logger.error("_check_blink_result - second try now with picam") return await self._picam_foto_helper(task) else: - self.logger.error("_check_blink_result - picam not enabled for second try") + self.logger.error( + "_check_blink_result - picam not enabled for second try" + ) return result def put_msg_queue_photo(self, task: Camera_Task): @@ -287,11 +333,21 @@ def put_msg_queue_photo(self, task: Camera_Task): """ if task.reply: self.message_task_queue.put( - Message_Task(photo=True, filename=self.config.photo_image_path, chat_id=task.chat_id, - message=task.message)) + Message_Task( + photo=True, + filename=self.config.photo_image_path, + chat_id=task.chat_id, + message=task.message, + ) + ) else: self.message_task_queue.put( - Message_Task(photo=True, filename=self.config.photo_image_path, chat_id=task.chat_id)) + Message_Task( + photo=True, + filename=self.config.photo_image_path, + chat_id=task.chat_id, + ) + ) def put_msg_queue_error(self, task: Camera_Task, message: str): """ @@ -306,9 +362,17 @@ def put_msg_queue_error(self, task: Camera_Task, message: str): """ if task.reply: self.message_task_queue.put( - Message_Task(reply=True, data_text=message, chat_id=task.chat_id, message=task.message)) + Message_Task( + reply=True, + data_text=message, + chat_id=task.chat_id, + message=task.message, + ) + ) else: - self.message_task_queue.put(Message_Task(send=True, data_text=message, chat_id=task.chat_id)) + self.message_task_queue.put( + Message_Task(send=True, data_text=message, chat_id=task.chat_id) + ) def detect_daylight(self) -> bool: """ @@ -316,8 +380,10 @@ def detect_daylight(self) -> bool: """ loc = LocationInfo(name="Berlin", region="Germany", timezone="Europe/Berlin") time_now: datetime = datetime.now(tz=timezone.utc) - s: dict[str, datetime] = sun(observer=loc.observer, date=time_now, tzinfo=loc.timezone) - daylight: bool = (s['sunrise'] <= time_now <= (s['sunset'])) + s: dict[str, datetime] = sun( + observer=loc.observer, date=time_now, tzinfo=loc.timezone + ) + daylight: bool = s["sunrise"] <= time_now <= (s["sunset"]) self.logger.info(msg=f"Is daylight detected: {daylight}") return daylight @@ -329,8 +395,10 @@ async def blink_snapshot(self) -> bool: bool: True if the snapshot was successfully taken and saved, False otherwise. """ self.logger.info( - msg="i'll take a snapshot from blink cam {0} and store it here {1}".format(self.config.blink_name, - self.config.photo_image_path)) + msg="i'll take a snapshot from blink cam {0} and store it here {1}".format( + self.config.blink_name, self.config.photo_image_path + ) + ) try: await self.blink.refresh(force=True) self.logger.debug("create a camera instance") @@ -362,9 +430,17 @@ async def read_blink_config(self): """ Asynchronously reads the Blink configuration file and authenticates with Blink. - This function checks if the Blink configuration file exists. If it does, it loads the configuration file using the `json_load` function, creates an `Auth` object with the loaded configuration, sets the `auth` attribute of the `blink` object to the created `Auth` object, logs a message indicating that the authentication was done using the file, and sets the `authentication_success` variable to `True`. + This function checks if the Blink configuration file exists. If it does, it loads + the configuration file using the `json_load` function, creates an `Auth` object + with the loaded configuration, sets the `auth` attribute of the `blink` object to + the created `Auth` object, logs a message indicating that the authentication was + done using the file, and sets the `authentication_success` variable to `True`. - If the Blink configuration file does not exist, it logs a message indicating that the file was not found and that a 2FA authentication token is required. It creates an `Auth` object with the provided Blink username and password, sets the `auth` attribute of the `blink` object to the created `Auth` object, and sets the `authentication_success` variable to `None`. + If the Blink configuration file does not exist, it logs a message indicating that + the file was not found and that a 2FA authentication token is required. It creates + an `Auth` object with the provided Blink username and password, sets the `auth` + attribute of the `blink` object to the created `Auth` object, and sets the + `authentication_success` variable to `None`. Parameters: self (object): The instance of the class. @@ -373,14 +449,26 @@ async def read_blink_config(self): None """ if os.path.exists(self.config.blink_config_file): - self.blink.auth = Auth(await json_load(self.config.blink_config_file), no_prompt=True, session=self.session) + self.blink.auth = Auth( + await json_load(self.config.blink_config_file), + no_prompt=True, + session=self.session, + ) self.logger.info("blink aut with file done") - authentication_success = True + # authentication_success = True else: - self.logger.info("no blink_config.json found - 2FA " + "authentication token required") - self.blink.auth = Auth({"username": self.config.blink_username, "password": self.config.blink_password}, - no_prompt=True, session=self.session) - authentication_success = None + self.logger.info( + "no blink_config.json found - 2FA " + "authentication token required" + ) + self.blink.auth = Auth( + { + "username": self.config.blink_username, + "password": self.config.blink_password, + }, + no_prompt=True, + session=self.session, + ) + # authentication_success = None async def save_blink_config(self) -> bool: """ @@ -429,17 +517,32 @@ def picam_request_take_foto(self) -> bool: """ self.logger.info(msg="take a PiCam snapshot") self.logger.debug(msg=f"post url={self.config.picam_url}") - payload: dict[str, any] = {"rotation": self.config.picam_rotation, "width": self.config.picam_image_width, - "filename": self.config.picam_image_filename, "height": self.config.picam_image_hight, - "exposure": self.config.picam_exposure, "iso": self.config.picam_iso, } + payload: dict[str, any] = { + "rotation": self.config.picam_rotation, + "width": self.config.picam_image_width, + "filename": self.config.picam_image_filename, + "height": self.config.picam_image_hight, + "exposure": self.config.picam_exposure, + "iso": self.config.picam_iso, + } self.logger.debug(msg=payload) headers: dict[str, str] = {"content-type": "application/json"} self.logger.debug(msg=headers) - response: requests.Response = requests.post(url=self.config.picam_url, data=json.dumps(obj=payload), - headers=headers) - self.logger.debug(msg="make a snapshot ended with http status {}".format(response.status_code)) - - return True if response.status_code == 200 else False + try: + response: requests.Response = requests.post( + url=self.config.picam_url, data=json.dumps(obj=payload), headers=headers + ) + response.raise_for_status() + self.logger.debug( + msg="make a snapshot ended with http status {}".format( + response.status_code + ) + ) + return True if response.status_code == 200 else False + + except Exception as e: + self.logger.error("Exception: {}".format(e)) + return False def picam_request_download_foto(self) -> bool: """ @@ -457,10 +560,21 @@ def picam_request_download_foto(self) -> bool: logger.debug(msg="deleting already existing file before hand") os.remove(path=self.config.photo_image_path) - with open(self.config.photo_image_path, "wb") as file: - response: requests.Response = requests.get( - url=self.config.picam_url + "?filename=" + self.config.picam_image_filename) - file.write(response.content) - self.logger.debug(msg="downloading foto ended with status {}".format(response.status_code)) - self.logger.debug(msg="end downloading foto") - return True if response.status_code == 200 else False + try: + with open(self.config.photo_image_path, "wb") as file: + response: requests.Response = requests.get( + url=self.config.picam_url + + "?filename=" + + self.config.picam_image_filename + ) + response.raise_for_status() + file.write(response.content) + self.logger.debug( + msg="downloading foto ended with status {}".format(response.status_code) + ) + self.logger.debug(msg="end downloading foto") + return True if response.status_code == 200 else False + + except Exception as e: + self.logger.error("Fehler: {}".format(e)) + return False diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..2d371ae --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +asyncio_mode = auto +asyncio_default_fixture_loop_scope = function \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index f2374b1..1907a29 100755 --- a/requirements.txt +++ b/requirements.txt @@ -24,6 +24,7 @@ sphinx-rtd-theme==2.0.0 # pyup: ignore ghp-import==2.1.0 pytest==8.3.4 +pytest-asyncio==0.25.3 pytest-cov==6.0.0 mock==5.1.0 # coverage==7.60 diff --git a/test/test_camera.py b/test/test_camera.py index e76de4f..6ce4fa9 100644 --- a/test/test_camera.py +++ b/test/test_camera.py @@ -1,100 +1,122 @@ -import unittest -from unittest.mock import patch, MagicMock, AsyncMock, call -from camera.camera import Camera -from config.config_util import Configuration, DefaultCam -from config.data_class import Camera_Task, Message_Task import asyncio -import queue -import logging - -class TestCamera(unittest.TestCase): - - def setUp(self): - self.config = Configuration() - self.loop = asyncio.new_event_loop() - self.camera_task_queue_async = AsyncMock(asyncio.Queue) - self.message_task_queue = MagicMock(queue.Queue) - self.camera = Camera(self.config, self.loop, self.camera_task_queue_async, self.message_task_queue) - - @patch('camera.aiohttp.ClientSession') - @patch('camera.Blink') - async def test_start(self, mock_blink, mock_client_session): - mock_blink_instance = mock_blink.return_value - mock_client_session_instance = mock_client_session.return_value - - self.config.blink_enabled = True - self.camera_task_queue_async.get = AsyncMock(return_value=None) - - await self.camera.start() - - self.camera.logger.debug.assert_called_with(msg="thread camera start") - mock_client_session.assert_called_once() - mock_blink_instance.start.assert_called_once() - self.assertFalse(self.camera.running) - mock_client_session_instance.close.assert_called_once() - - @patch('camera.Blink') - @patch('camera.ClientSession') - async def test_read_blink_config(self, mock_blink, mock_client_session): - mock_blink_instance = mock_blink.return_value - - with patch('os.path.exists', return_value=True): - with patch('camera.json_load', new_callable=AsyncMock): - await self.camera.read_blink_config() - self.camera.logger.info.assert_called_with("blink aut with file done") - self.assertTrue(mock_blink_instance.auth.no_prompt) - - @patch('camera.Blink') - @patch('camera.ClientSession') - async def test_save_blink_config(self, mock_blink, mock_client_session): - mock_blink_instance = mock_blink.return_value - - result = await self.camera.save_blink_config() - self.camera.logger.info.assert_called_with("saving blink authenticated session infos into config file") - self.assertTrue(result) - - @patch('camera.Blink') - @patch('camera.ClientSession') - async def test_blink_snapshot(self, mock_blink, mock_client_session): - mock_blink_instance = mock_blink.return_value - mock_camera = MagicMock() - mock_blink_instance.cameras.__getitem__.return_value = mock_camera - - result = await self.camera.blink_snapshot() - self.camera.logger.info.assert_called_with( - msg="i'll take a snapshot from blink cam {0} and store it here {1}".format( - self.config.blink_name, self.config.photo_image_path)) - self.assertTrue(result) - mock_camera.snap_picture.assert_called_once() - mock_camera.image_to_file.assert_called_once_with(self.config.photo_image_path) - - @patch('camera.camera.requests') - def test_picam_request_take_foto(self, mock_requests): - mock_response = MagicMock() - mock_response.status_code = 200 - mock_requests.post.return_value = mock_response - - mock_logger_info = MagicMock() - self.camera.logger.info = mock_logger_info - - result = self.camera.picam_request_take_foto() - mock_logger_info.assert_called_with(msg="take a PiCam snapshot") - self.assertTrue(result) - - - @patch('camera.camera.requests') - def test_picam_request_download_foto(self, mock_requests): - mock_response = MagicMock() - mock_response.status_code = 200 - mock_requests.get.return_value = mock_response - - mock_logger_info = MagicMock() - self.camera.logger.info = mock_logger_info +import pytest +from unittest.mock import MagicMock, patch +from config.data_class import Camera_Task, Message_Task +from config.config_util import Configuration +from camera.camera import Camera - with patch('builtins.open', unittest.mock.mock_open()): - result = self.camera.picam_request_download_foto() - mock_logger_info.assert_called_with(msg="downloading PiCam foto") - self.assertTrue(result) -if __name__ == '__main__': - unittest.main() +@pytest.fixture +def camera_setup(): + """Fixture für die Kamera-Setup-Konfiguration""" + config = Configuration() + config.picam_enabled = True + loop = asyncio.new_event_loop() # Geändert von get_event_loop() zu new_event_loop() + asyncio.set_event_loop(loop) # Setze den neuen Loop als aktuellen Loop + camera_queue = asyncio.Queue() + message_queue = MagicMock() + camera = Camera(config, loop, camera_queue, message_queue) + + yield camera, camera_queue, message_queue + + # Cleanup + loop.close() + + +@pytest.mark.asyncio +async def test_start_picam_photo_success(camera_setup): + # Setup + camera, camera_queue, message_queue = camera_setup # Entfernt await + + # Mock die Hilfsmethoden + camera.picam_request_take_foto = MagicMock(return_value=True) + camera.picam_request_download_foto = MagicMock(return_value=True) + + # Mock die Session für die start-Methode + camera.session = MagicMock() + camera.session.close = MagicMock() + + # Erstelle Test-Task + test_task = Camera_Task(picam_photo=True, chat_id=123456) + await camera_queue.put(test_task) + await camera_queue.put(None) # Stoppsignal + + # Ausführen der start-Methode + await camera.start() + + # Überprüfungen + camera.picam_request_take_foto.assert_called_once() + camera.picam_request_download_foto.assert_called_once() + + # Überprüfe Message Queue + message_queue.put.assert_called_once() + call_args = message_queue.put.call_args[0][0] + assert isinstance(call_args, Message_Task) + assert call_args.photo is True + assert call_args.chat_id == 123456 + + +@pytest.mark.asyncio +async def test_start_picam_photo_take_foto_failure(camera_setup): + # Setup + camera, camera_queue, message_queue = camera_setup # Entfernt await + + # Mock die Hilfsmethoden + camera.picam_request_take_foto = MagicMock(return_value=False) + camera.picam_request_download_foto = MagicMock(return_value=True) + + # Mock die Session für die start-Methode + camera.session = MagicMock() + camera.session.close = MagicMock() + + # Erstelle Test-Task + test_task = Camera_Task(picam_photo=True, chat_id=123456) + await camera_queue.put(test_task) + await camera_queue.put(None) # Stoppsignal + + # Ausführen der start-Methode + await camera.start() + + # Überprüfungen + camera.picam_request_take_foto.assert_called_once() + camera.picam_request_download_foto.assert_not_called() + + # Überprüfe Message Queue für Fehlermeldung + message_queue.put.assert_called_once() + call_args = message_queue.put.call_args[0][0] + assert isinstance(call_args, Message_Task) + assert call_args.send is True + assert "error" in call_args.data_text.lower() + + +@pytest.mark.asyncio +async def test_start_picam_photo_download_failure(camera_setup): + # Setup + camera, camera_queue, message_queue = camera_setup # Entfernt await + + # Mock die Hilfsmethoden + camera.picam_request_take_foto = MagicMock(return_value=True) + camera.picam_request_download_foto = MagicMock(return_value=False) + + # Mock die Session für die start-Methode + camera.session = MagicMock() + camera.session.close = MagicMock() + + # Erstelle Test-Task + test_task = Camera_Task(picam_photo=True, chat_id=123456) + await camera_queue.put(test_task) + await camera_queue.put(None) # Stoppsignal + + # Ausführen der start-Methode + await camera.start() + + # Überprüfungen + camera.picam_request_take_foto.assert_called_once() + camera.picam_request_download_foto.assert_called_once() + + # Überprüfe Message Queue für Fehlermeldung + message_queue.put.assert_called_once() + call_args = message_queue.put.call_args[0][0] + assert isinstance(call_args, Message_Task) + assert call_args.send is True + assert "error" in call_args.data_text.lower() \ No newline at end of file diff --git a/test/test_camera2.py b/test/test_camera2.py new file mode 100644 index 0000000..5610c16 --- /dev/null +++ b/test/test_camera2.py @@ -0,0 +1,135 @@ +import pytest +import asyncio +import queue +from unittest.mock import Mock, patch, AsyncMock +from datetime import datetime, timezone +from config.config_util import Configuration, DefaultCam +from config.data_class import Camera_Task, Message_Task +from camera.camera import Camera + +@pytest.fixture +def config(): + config = Mock(spec=Configuration) + config.blink_enabled = True + config.picam_enabled = True + config.enable_detect_daylight = False + config.default_camera_type = DefaultCam.BLINK + config.photo_image_path = "/tmp/test.jpg" + config.blink_name = "test_camera" + return config + + +@pytest.fixture +def camera(config): + loop = asyncio.get_event_loop() + camera_queue = asyncio.Queue() + message_queue = queue.Queue() + return Camera(config, loop, camera_queue, message_queue) + + +@pytest.mark.asyncio +async def test_blink_foto_helper_success(camera): + task = Camera_Task(photo=True, chat_id=123) + + with patch.object(camera, 'blink_snapshot', new_callable=AsyncMock) as mock_snapshot: + mock_snapshot.return_value = True + + result = await camera._blink_foto_helper(task) + + assert result == True + mock_snapshot.assert_called_once() + assert not camera.message_task_queue.empty() + + +@pytest.mark.asyncio +async def test_blink_foto_helper_failure(camera): + task = Camera_Task(photo=True, chat_id=123) + + with patch.object(camera, 'blink_snapshot', new_callable=AsyncMock) as mock_snapshot: + mock_snapshot.return_value = False + + result = await camera._blink_foto_helper(task) + + assert result == False + mock_snapshot.assert_called_once() + assert not camera.message_task_queue.empty() + + +@pytest.mark.asyncio +async def test_picam_foto_helper_success(camera): + task = Camera_Task(photo=True, chat_id=123) + + with patch.object(camera, 'picam_request_take_foto') as mock_take_foto: + with patch.object(camera, 'picam_request_download_foto') as mock_download_foto: + mock_take_foto.return_value = True + mock_download_foto.return_value = True + + result = await camera._picam_foto_helper(task) + + assert result == True + mock_take_foto.assert_called_once() + mock_download_foto.assert_called_once() + assert not camera.message_task_queue.empty() + + +def test_detect_daylight(camera): + with patch('camera.camera.sun') as mock_sun: + # Simuliere Tageslicht + current_time = datetime.now(tz=timezone.utc) + mock_sun.return_value = { + 'sunrise': current_time.replace(hour=6), + 'sunset': current_time.replace(hour=20) + } + + result = camera.detect_daylight() + assert isinstance(result, bool) + + +@pytest.mark.asyncio +async def test_choose_cam_default_blink(camera): + task = Camera_Task(photo=True, chat_id=123) + camera.config.enable_detect_daylight = False + camera.config.default_camera_type = DefaultCam.BLINK + + with patch.object(camera, '_blink_foto_helper', new_callable=AsyncMock) as mock_blink: + mock_blink.return_value = True + + await camera.choose_cam(task) + + mock_blink.assert_called_once_with(task) + + +@pytest.mark.asyncio +async def test_choose_cam_default_picam(camera): + task = Camera_Task(photo=True, chat_id=123) + camera.config.enable_detect_daylight = False + camera.config.default_camera_type = DefaultCam.PICAM + + with patch.object(camera, '_picam_foto_helper', new_callable=AsyncMock) as mock_picam: + mock_picam.return_value = True + + await camera.choose_cam(task) + + mock_picam.assert_called_once_with(task) + + +def test_put_msg_queue_photo(camera): + task = Camera_Task(photo=True, chat_id=123, reply=True, message="test") + camera.put_msg_queue_photo(task) + + msg = camera.message_task_queue.get_nowait() + assert isinstance(msg, Message_Task) + assert msg.photo == True + assert msg.chat_id == 123 + + +def test_put_msg_queue_error(camera): + task = Camera_Task(photo=True, chat_id=123, reply=True, message="test") + error_message = "Test error" + camera.put_msg_queue_error(task, error_message) + + msg = camera.message_task_queue.get_nowait() + assert isinstance(msg, Message_Task) + assert msg.reply == True + assert msg.data_text == error_message + assert msg.chat_id == 123 \ No newline at end of file diff --git a/test/test_camera3.py b/test/test_camera3.py new file mode 100644 index 0000000..7e1c9d2 --- /dev/null +++ b/test/test_camera3.py @@ -0,0 +1,124 @@ +import asyncio +import queue +import unittest +from unittest.mock import patch, MagicMock, AsyncMock + +import requests +import logging + +from camera.camera import Camera +from config.config_util import Configuration + +logger: logging.Logger = logging.getLogger(name="test_camera") + + +class TestCamera(unittest.TestCase): + + def setUp(self): + self.config = Configuration() + self.loop = asyncio.new_event_loop() + self.camera_task_queue_async = AsyncMock(asyncio.Queue) + self.message_task_queue = MagicMock(queue.Queue) + self.camera = Camera(self.config, self.loop, self.camera_task_queue_async, self.message_task_queue) + + @patch('camera.aiohttp.ClientSession') + @patch('camera.Blink') + async def test_start(self, mock_blink, mock_client_session): + mock_blink_instance = mock_blink.return_value + mock_client_session_instance = mock_client_session.return_value + + self.config.blink_enabled = True + self.camera_task_queue_async.get = AsyncMock(return_value=None) + + await self.camera.start() + + self.camera.logger.debug.assert_called_with(msg="thread camera start") + mock_client_session.assert_called_once() + mock_blink_instance.start.assert_called_once() + self.assertFalse(self.camera.running) + mock_client_session_instance.close.assert_called_once() + + @patch('camera.Blink') + @patch('camera.ClientSession') + async def test_read_blink_config(self, mock_blink, mock_client_session): + mock_blink_instance = mock_blink.return_value + + with patch('os.path.exists', return_value=True): + with patch('camera.json_load', new_callable=AsyncMock): + await self.camera.read_blink_config() + self.camera.logger.info.assert_called_with("blink aut with file done") + self.assertTrue(mock_blink_instance.auth.no_prompt) + + @patch('camera.Blink') + @patch('camera.ClientSession') + async def test_save_blink_config(self, mock_blink, mock_client_session): + mock_blink_instance = mock_blink.return_value + + result = await self.camera.save_blink_config() + self.camera.logger.info.assert_called_with("saving blink authenticated session infos into config file") + self.assertTrue(result) + + @patch('camera.Blink') + @patch('camera.ClientSession') + async def test_blink_snapshot(self, mock_blink, mock_client_session): + mock_blink_instance = mock_blink.return_value + mock_camera = MagicMock() + mock_blink_instance.cameras.__getitem__.return_value = mock_camera + + result = await self.camera.blink_snapshot() + self.camera.logger.info.assert_called_with( + msg="i'll take a snapshot from blink cam {0} and store it here {1}".format( + self.config.blink_name, self.config.photo_image_path)) + self.assertTrue(result) + mock_camera.snap_picture.assert_called_once() + mock_camera.image_to_file.assert_called_once_with(self.config.photo_image_path) + + @patch('camera.camera.requests') + def test_picam_request_take_foto(self, mock_requests): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_requests.post.return_value = mock_response + + mock_logger_info = MagicMock() + self.camera.logger.info = mock_logger_info + + result = self.camera.picam_request_take_foto() + mock_logger_info.assert_called_with(msg="take a PiCam snapshot") + self.assertTrue(result) + + + @patch('camera.camera.requests') + def test_picam_request_take_foto_exception(self, mock_requests): + + mock_logger_info = MagicMock() + mock_logger_error = MagicMock() + self.camera.logger.info = mock_logger_info + self.camera.logger.error = mock_logger_error + + mock_response = mock_requests.post.return_value + mock_response.status_code = 404 + mock_response.raise_for_status.side_effect = requests.exceptions.HTTPError + + result = self.camera.picam_request_take_foto() + + self.assertFalse(result) + mock_logger_info.assert_called_with(msg="take a PiCam snapshot") + mock_logger_error.assert_called_with("Exception: ") + + @patch('camera.camera.requests') + def test_picam_request_download_foto(self, mock_requests): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_requests.get.return_value = mock_response + + mock_logger_info = MagicMock() + self.camera.logger.info = mock_logger_info + + with patch('builtins.open', unittest.mock.mock_open()): + result = self.camera.picam_request_download_foto() + mock_logger_info.assert_called_with(msg="downloading PiCam foto") + self.assertTrue(result) + + +if __name__ == '__main__': + unittest.main() From 6d712f57116644ccc04a9dbd8993d8604902399c Mon Sep 17 00:00:00 2001 From: oliver Date: Sun, 2 Feb 2025 22:51:24 +0100 Subject: [PATCH 2/3] updated readme - removed ring comparision - since there's for now no further plan --- README.md | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index eb37336..b68c7c1 100755 --- a/README.md +++ b/README.md @@ -86,6 +86,7 @@ Opening relais board can be buyed and must just be wired. - [ChangeLog](#changelog) - [Author info](#author-info) - [License](#license) + - [Security](#security) - [Contribution](#contribution) ## Long description @@ -116,15 +117,15 @@ The project offers the following functionality: ## Features advantage comparision FDIA, Blink, Ring and PiCamAPI -| Project with Product /
Features,Capabilities | FDIA with PiCamAPI Camera | FDIA with Blink Camera | FDIA with upcoming Ring Camera support - not there (in development branch feature/ring_camera_integration) | Blink only (no FDIA) | Ring only ( FDIA) | FDIA with no build HW module | -|----------------------------------------------------|------------------------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------|-------------------------------------|--------------------------------------|---------------------------------------------------| -| Open Door | ✅ with build HW module [Door opener board with relais:](#door-opener-board-with-relais) | ✅ with build HW module [Door opener board with relais:](#door-opener-board-with-relais) | ✅ with build HW module [Door opener board with relais:](#door-opener-board-with-relais) | ❌ | ❌ | ❌ | -| detect door bell ring | ✅ with build Hw module [Door bell detection board](#door-bell-detection-board) | ✅ with build Hw module [Door bell detection board](#door-bell-detection-board) | ✅ with build Hw module [Door bell detection board](#door-bell-detection-board) | ✅ | ✅ | ❌ | -| Notification (channel) | ✅ via Telegram group | ✅ via Telegram group | ✅ via Telegram group | ✅Blink App | ✅Ring App | ✅❌ Telegram but no HW module -> no detection | -| Multi user notification | ✅ | ✅ | ✅ | ❌ dependend on Account | ❌ | ✅ | -| Multi user door opening | ✅ | ✅ | ✅ | ❌ | ❌ | ❌ | -| GDPR data storage | ✅ no Internet facing data/nor Camaera exposure; Telegram chat group archival dependend on personal deletion interval | ✅Telegram chat group archival dependend on personal deletion interval - Cemera internet/clodu exposed | ✅Telegram chat group archival dependend on personal deletion interval - Cemera internet/clodu exposed | ❌ Camaera Internet / Cloud exposed | ❌ Camaera Internet / Cloud exposed | ✅ no data Exposure | -| possible local usage without Internet (door opening only) | ✅ | ✅opening - ❌ no camera snapshot | ✅opening - ❌ no camera snapshot | ❌ | ❌ | ✅ | +| Project with Product /
Features,Capabilities | FDIA with PiCamAPI Camera | FDIA with Blink Camera | Blink only (no FDIA) | FDIA with no build HW module | +|---|---|---|---|---| +| Open Door | ✅ with build HW module [Door opener board with relais:](#door-opener-board-with-relais) | ✅ with build HW module [Door opener board with relais:](#door-opener-board-with-relais) | ❌ |❌ | +| detect door bell ring | ✅ with build Hw module [Door bell detection board](#door-bell-detection-board) | ✅ with build Hw module [Door bell detection board](#door-bell-detection-board) | ✅ | ❌ | +| Notification (channel) | ✅ via Telegram group | ✅ via Telegram group | ✅Blink App | ✅❌ Telegram but no HW module -> no detection | +| Multi user notification | ✅ | ✅ | ❌ dependend on Account | ✅ | +| Multi user door opening | ✅ | ✅ | ❌ | ❌ | +| GDPR data storage | ✅ no Internet facing data/nor Camaera exposure; Telegram chat group archival dependend on personal deletion interval | ✅Telegram chat group archival dependend on personal deletion interval - Cemera internet/clodu exposed | ❌ Camaera Internet / Cloud exposed | ✅ no data Exposure | +| possible local usage without Internet (door opening only) | ✅ | ✅opening - ❌ no camera snapshot | ❌ | ✅ | ## Outlook/Ideas - Improvements plan From cfc11b84b0b67fcb28da43c20eb2e360c83391db Mon Sep 17 00:00:00 2001 From: oliver Date: Sun, 2 Feb 2025 23:27:40 +0100 Subject: [PATCH 3/3] removed logging settings from camera.py --- camera/camera.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/camera/camera.py b/camera/camera.py index 8e73289..ee5b978 100644 --- a/camera/camera.py +++ b/camera/camera.py @@ -20,12 +20,8 @@ from config.config_util import Configuration, DefaultCam from config.data_class import Camera_Task, Message_Task -logging.basicConfig( - level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s" -) logger: logging.Logger = logging.getLogger(name="camera") - class Camera: def __init__( self,