fix unexpected picam requests errors with exception handling

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+asyncio_mode = auto
+asyncio_default_fixture_loop_scope = function

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

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()

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