Improve test coverage

examples/line_chatbot/line_bot.py

@@ -2,9 +2,10 @@
 
 import logging
 
-from flask import abort
-from flask import Flask
-from flask import request
+from fastapi import FastAPI
+from fastapi import HTTPException
+from fastapi import Request
+from fastapi.responses import PlainTextResponse
 from linebot import LineBotApi
 from linebot import WebhookHandler
 from linebot.exceptions import InvalidSignatureError
@@ -13,58 +14,78 @@
 from linebot.models import TextMessage
 from linebot.models import TextSendMessage
 
-app: Flask = Flask(__name__)
+# Create a FastAPI application instance
+app = FastAPI()
 
 # Initialise the LINE Bot API and WebhookHandler with access token and secret
 line_bot_api: LineBotApi = LineBotApi('YOUR_LINE_CHANNEL_ACCESS_TOKEN')
 handler: WebhookHandler = WebhookHandler('YOUR_LINE_CHANNEL_SECRET')
 
 
-@app.route('/webhook', methods=['POST'])
-def callback() -> str:
+@app.post('/webhook', response_class=PlainTextResponse)
+async def callback(request: Request) -> str:
     """
-    Handle the incoming webhook requests from LINE.
+    Handle incoming webhook requests from LINE.
+
+    Args:
+        request (Request): The HTTP request object
+            containing the webhook data.
 
     Returns:
-        str: The response msg to indicate successful handling of the request.
+        str: A plain text response
+            indicating successful handling of the request.
+
+    Raises:
+        HTTPException: If the request is invalid
+            or an error occurs during processing.
     """
+    # Extract the signature and body from the request
     signature: str = request.headers.get('X-Line-Signature', '')
-    body: str = request.get_data(as_text=True)
+    body: bytes = await request.body()
+    body_text: str = body.decode('utf-8') if body else ''
 
     # Check if signature or body is missing
-    if not signature or not body:
-        logging.warning('Received invalid request')
-        abort(400)  # Bad Request
+    if not signature or not body_text:
+        logging.warning(
+            'Received invalid request: signature or body is missing',
+        )
+        raise HTTPException(status_code=400, detail='Bad Request')
 
     try:
-        # Handle the request with the provided body and signature
-        handler.handle(body, signature)
+        # Process the incoming message using the LINE handler
+        handler.handle(body_text, signature)
     except InvalidSignatureError:
-        # Log and abort if the signature is invalid
         logging.error('Invalid signature error')
-        abort(400)  # Bad Request
+        raise HTTPException(status_code=400, detail='Invalid Signature Error')
     except Exception as e:
-        # Log and abort if any other unexpected error occurs
         logging.error(f"Unexpected error: {e}")
-        abort(500)  # Internal Server Error
+        raise HTTPException(status_code=500, detail='Internal Server Error')
 
+    # Return success response
     return 'OK'
 
 
 @handler.add(MessageEvent, message=TextMessage)
 def handle_text_message(event: MessageEvent) -> None:
     """
-    Handle the incoming text message event from the user.
+    Handle incoming text messages from users.
 
     Args:
-        event (MessageEvent): Event object containing details of the message.
+        event (MessageEvent): The LINE message event
+            containing the user's message.
+
+    Raises:
+        LineBotApiError: If an error occurs
+            while sending a reply message.
+        Exception: If an unexpected error occurs during processing.
 
     Returns:
         None
     """
+    # Extract the user's message
     user_message: str = event.message.text
 
-    # Check if the user's message is empty or contains only whitespace
+    # Check if the message is empty or contains only whitespace
     if not user_message.strip():
         logging.warning('Received empty user message')
         return
@@ -75,16 +96,17 @@ def handle_text_message(event: MessageEvent) -> None:
 
         # Send the response back to the user via LINE
         line_bot_api.reply_message(
-            event.reply_token, TextSendMessage(text=assistant_response),
+            event.reply_token,
+            TextSendMessage(text=assistant_response),
         )
     except LineBotApiError as e:
-        # Log the error if the LINE Bot API fails
         logging.error(f"Error responding to message: {e}")
     except Exception as e:
-        # Log any other unexpected errors
         logging.error(f"Unexpected error: {e}")
 
 
 if __name__ == '__main__':
-    # Run the Flask application on the specified host and port
-    app.run(host='0.0.0.0', port=8000)
+    import uvicorn
+
+    # Start the FastAPI application using uvicorn
+    uvicorn.run(app, host='127.0.0.1', port=8000)

src/drawing_manager.py

@@ -9,7 +9,7 @@
 from PIL import ImageFont
 from shapely.geometry import Polygon
 
-from .lang_config import LANGUAGES
+from src.lang_config import LANGUAGES
 
 
 class DrawingManager:

tests/examples/YOLO_server_api/backend/detection_test.py

@@ -255,48 +255,66 @@ async def test_check_containment(self) -> None:
         result = await check_containment(index1, index2, datas)
         self.assertIsInstance(result, set)
 
-    async def test_remove_overlapping_labels_line_134(self) -> None:
-        """
-        測試移除重疊的標籤，覆蓋到 datas.pop(index) 這一行。
-        """
-        datas = [
-            [0, 0, 100, 100, 0.9, 0],   # 'hardhat' 標籤
-            [0, 0, 100, 100, 0.8, 2],   # 'no_hardhat' 標籤，完全重疊
-            [200, 200, 300, 300, 0.85, 7], # 'safety_vest' 標籤
-            [200, 200, 300, 300, 0.7, 4]   # 'no_safety_vest' 標籤，完全重疊
-        ]
-        result = await remove_overlapping_labels(datas.copy())
-        self.assertEqual(len(result), 2)
-        remaining_labels = [d[5] for d in result]
-        self.assertListEqual(remaining_labels, [0, 7])
-
     async def test_remove_completely_contained_labels_line_340(self) -> None:
         """
-        測試移除完全包含的標籤，覆蓋到 datas.pop(index) 這一行。
+        Test the `remove_completely_contained_labels` function to ensure it
+        removes labels that are completely contained within another label.
         """
+        # Test data representing bounding boxes and their corresponding labels
         datas = [
-            [50, 50, 150, 150, 0.9, 0],   # 'hardhat' 標籤
-            [70, 70, 130, 130, 0.8, 2],   # 'no_hardhat' 標籤，完全包含在 'hardhat' 中
-            [200, 200, 300, 300, 0.85, 7], # 'safety_vest' 標籤
-            [220, 220, 280, 280, 0.7, 4]   # 'no_safety_vest' 標籤，完全包含在 'safety_vest' 中
+            [50, 50, 150, 150, 0.9, 0],   # 'hardhat' label
+            # 'no_hardhat' label, fully contained within 'hardhat'
+            [70, 70, 130, 130, 0.8, 2],
+            [200, 200, 300, 300, 0.85, 7],  # 'safety_vest' label
+            # 'no_safety_vest' label, fully contained within 'safety_vest'
+            [220, 220, 280, 280, 0.7, 4],
         ]
+
+        # Call the function and pass a copy of the data
         result = await remove_completely_contained_labels(datas.copy())
+
+        # Assert the length of the resulting list is as expected
         self.assertEqual(len(result), 2)
+
+        # Extract remaining labels and verify they match the expected values
         remaining_labels = [d[5] for d in result]
         self.assertListEqual(remaining_labels, [0, 7])
 
     async def test_check_containment_elif_condition(self) -> None:
         """
-        測試 `check_containment` 函數中的 `elif` 條件，覆蓋相應的代碼行。
+        Test the `check_containment` function's `elif` condition to ensure
+        that the appropriate lines of code are covered.
+
+        This test simulates a scenario where one bounding box
+        (`index1`) is completely contained within another bounding box
+        (`index2`). The function should correctly identify that
+        `index1` is contained and return the set containing `index1`.
+
+        Args:
+            None
+
+        Returns:
+            None
         """
-        index1 = 0
-        index2 = 1
-        datas = [
-            [70, 70, 130, 130, 0.9, 1],  # index1，被 index2 完全包含
-            [50, 50, 150, 150, 0.85, 2], # index2
+        # Define the indices of the bounding boxes to test
+        index1: int = 0
+        index2: int = 1
+
+        # Define the test data representing bounding boxes
+        datas: list[list[float | int]] = [
+            [70, 70, 130, 130, 0.9, 1],  # Bounding box for index1
+            [50, 50, 150, 150, 0.85, 2],  # Bounding box for index2
         ]
-        result = await check_containment(index1, index2, datas)
-        self.assertSetEqual(result, {0})
+
+        # Call the `check_containment` function with the test data
+        result: set[int] = await check_containment(index1, index2, datas)
+
+        # Assert that the result matches the expected output
+        # In this case, `index1` is contained within `index2`
+        self.assertSetEqual(
+            result, {0},
+            'The function did not correctly identify containment.',
+        )
 
 
 if __name__ == '__main__':

tests/examples/YOLO_server_api/backend/routers_test.py

@@ -449,7 +449,9 @@ async def mock_update_model_file_func(model, temp_path):
         # Simulate ValueError in update_model_file
         async def mock_update_model_file_raise_value_error(model, temp_path):
             raise ValueError('Invalid model')
-        mock_update_model_file.side_effect = mock_update_model_file_raise_value_error
+        mock_update_model_file.side_effect = (
+            mock_update_model_file_raise_value_error
+        )
 
         with self.override_jwt_credentials({'role': 'admin', 'id': 1}):
             response = self.client.post(
@@ -463,7 +465,7 @@ async def mock_update_model_file_raise_value_error(model, temp_path):
 
         # Simulate OSError during file operation
         with patch('pathlib.Path.open', side_effect=OSError('Disk error')):
-            mock_update_model_file.side_effect = mock_update_model_file_func  # Reset side effect
+            mock_update_model_file.side_effect = mock_update_model_file_func
             with self.override_jwt_credentials({'role': 'admin', 'id': 1}):
                 response = self.client.post(
                     '/api/model_file_update', data=data, files=files,