Merge pull request #97 from Kyeong6/feat/#96

feat: 음식 이미지 분석 API 성능 테스트 진행을 위한 테스트 자동화 코드 구현

.gitignore

@@ -204,6 +204,7 @@ redis.conf
 
 # test/data
 server/test/image
+server/test/test_image
 
 # food.csv(대용량)
 server/data/food.csv

server/core/config_local.py

@@ -27,6 +27,9 @@ class Settings:
     DOCKER_DATA_PATH = os.getenv("DOCKER_DATA_PATH")
     PROMPT_PATH = os.getenv("PROMPT_PATH")
 
+    # Test
+    TEST_PATH = os.getenv("TEST_PATH")
+
     # Log
     LOG_PATH = os.getenv("LOG_PATH")
 

server/routers/food_image_analysis.py

@@ -127,4 +127,56 @@ def remaning_requests_check(member_id: int = Depends(get_current_member)):
         "error": None
     }
 
-    return response
+    return response
+
+
+# # 음식 이미지 분석 API 평가 테스트
+# @router.post("/image", responses=analyze_food_image_responses)
+# async def analyze_food_image(file: UploadFile = File(...), member_id: int = Depends(get_current_member)):
+#     start_total = time.time()
+
+#     # 이미지 처리 및 Base64 변환
+#     image_base64 = await process_image_to_base64(file)
+
+#     # OpenAI 음식 감지 시간 측정
+#     start_analyze = time.time()
+#     detected_food_data = food_image_analyze(image_base64)
+#     end_analyze = time.time()
+#     analyze_time = round(end_analyze - start_analyze, 4)
+
+#     # JSON 변환 확인 및 오류 방지
+#     if isinstance(detected_food_data, str):
+#         try:
+#             detected_food_data = json.loads(detected_food_data)
+#         except json.JSONDecodeError as e:
+#             raise ValueError(f"Failed to parse JSON: {e}")
+
+#     if not isinstance(detected_food_data, list):
+#         raise ValueError("Unexpected response format, expected a list of dicts")
+
+#     # 유사도 분석 시간 측정
+#     start_search = time.time()
+#     food_info = []
+#     for food in detected_food_data:
+#         if isinstance(food, dict) and "food_name" in food:
+#             similar_foods = search_similar_food(food["food_name"])
+#             food_info.append({
+#                 "detected_food": food["food_name"],
+#                 "similar_foods": similar_foods
+#             })
+#         else:
+#             print(f"Skipping invalid food item: {food}")
+#     end_search = time.time()
+#     search_time = round(end_search - start_search, 4)
+
+#     total_time = round(time.time() - start_total, 4)
+
+#     return {
+#         "success": True,
+#         "food_image_analyze_time": analyze_time,
+#         "search_similar_time": search_time,
+#         "total_time": total_time,
+#         "response": {
+#             "food_info": food_info
+#         }
+#     }

server/test/test_food_image_analysis.py

@@ -1,7 +1,8 @@
 import os
 import sys
-import pytest
+import time
 import redis
+import pandas as pd
 from fastapi.testclient import TestClient
 
 # Root directory를 Project Root로 설정: server directory 
@@ -24,67 +25,98 @@
     decode_responses=True
 )
 
+# 테스트 진행을 위한 Rate limit 초기화
 def reset_rate_limit(user_id: int):
     redis_key = f"rate_limit:{user_id}"
     redis_client.delete(redis_key)
 
-def analyze_food_image(base64_data):
+# 테스트 진행을 위한 API 요청
+def analyze_food_image(image_path):
     headers = {
         "Authorization": f"Bearer {settings.TEST_TOKEN}"
     }
-
-    response = client.post("/v1/ai/food_image_analysis/", headers=headers, json={"food_image": base64_data})
+
+    # 파일을 multipart 형식으로 업로드
+    with open(image_path, "rb") as img:
+        files = {"file": (image_path, img, "image/jpeg")}
+
+        response = client.post("/ai/v1/food_image_analysis/image", headers=headers, files=files)
+
     if response.status_code == 200:
         return response.json()
-    elif response.status_code == 400 or response.status_code == 429:  # 429 상태 코드 추가
+    elif response.status_code == 400 or response.status_code == 429:
         raise ValueError("하루 요청 제한을 초과했습니다.")
     else:
         raise Exception(f"Failed with status code: {response.status_code}, Error: {response.text}")
 
-@pytest.fixture
-def load_base64_data():
-    with open("./test/image/파스타_768_1364.txt", "r") as file:
-        return file.read().strip()
-
-@pytest.fixture(autouse=True)
-def reset_rate_limit_before_tests():
-    reset_rate_limit(user_id=4)  
-
-def test_food_image_analysis(load_base64_data):
-    response = analyze_food_image(load_base64_data)
-
-    assert "success" in response, "응답에 'success' 필드가 없습니다."
-    assert response["success"] is True, "요청이 실패했습니다."
-    assert "response" in response, "응답에 'response' 필드가 없습니다."
-    assert "error" in response, "응답에 'error' 필드가 없습니다."
-
-    response_data = response["response"]
-    assert "remaining_requests" in response_data, "'remaining_requests' 필드가 없습니다."
-    assert isinstance(response_data["remaining_requests"], int), "'remaining_requests'는 정수형이어야 합니다."
-    assert "food_info" in response_data, "'food_info' 필드가 없습니다."
-    assert isinstance(response_data["food_info"], list), "'food_info' 필드는 리스트 형식이어야 합니다."
-
-    for food_data in response_data["food_info"]:
-        assert "detected_food" in food_data, "'detected_food' 필드가 응답에 없습니다."
-        assert "similar_foods" in food_data, "'similar_foods' 필드가 응답에 없습니다."
-        assert isinstance(food_data["similar_foods"], list), "'similar_foods' 필드는 리스트 형식이 아닙니다."
-
-        for similar_food in food_data["similar_foods"]:
-            if similar_food.get("food_name") is None and similar_food.get("food_pk") is None:
-                print("유사한 음식이 없는 경우를 확인했습니다.")
-            else:
-                assert "food_name" in similar_food, "유사 음식에 'food_name' 필드가 없습니다."
-                assert "food_pk" in similar_food, "유사 음식에 'food_pk' 필드가 없습니다."
-
-def test_rate_limit_exceeded(load_base64_data):
-    rate_limit = settings.RATE_LIMIT
-
-    for _ in range(rate_limit):
-        response = analyze_food_image(load_base64_data)
-        assert response["success"], "Rate limit 내의 요청이 실패했습니다."
-        remaining_requests = response["response"]["remaining_requests"]
-        assert remaining_requests >= 0, "'remaining_requests'가 0 이상이어야 합니다."
-
-    with pytest.raises(ValueError) as excinfo:
-        analyze_food_image(load_base64_data)
-    assert "하루 요청 제한을 초과했습니다." in str(excinfo.value), "Rate limit 초과 예외가 발생하지 않았습니다."
+# 테스트할 이미지 목록
+image_dir = os.path.join(settings.TEST_PATH, '/test_image/')
+image_files = sorted(
+    [f for f in os.listdir(image_dir) if f.endswith(('.jpeg', '.jpg'))],
+    key=lambda x: int(os.path.splitext(x)[0])
+)
+
+# CSV 파일 경로
+output_csv = os.path.join(settings.TEST_PATH, '/test_image/test_result.csv')
+
+# 테스트 결과 저장할 리스트
+test_results = []
+
+# Rate limit 초기화
+reset_rate_limit(user_id=1)
+
+# 음식 이미지 처리 및 API 테스트
+for idx, image_file in enumerate(image_files, start=1):
+    image_path = os.path.join(image_dir, image_file)
+    order = f"{idx}-1"
+    read_food = "None"
+
+    start_total = time.time()
+    try:
+        response = analyze_food_image(image_path)
+        end_total = time.time()
+
+        total_time = round(end_total - start_total, 4)
+        analyze_time = response.get("food_image_analyze_time", 0)
+        search_time = response.get("search_similar_time", 0)
+
+        if response.get("success") and "response" in response:
+            food_info = response["response"].get("food_info", [])
+
+            for j, food_item in enumerate(food_info):
+                # 감지된 음식 존재하지 않으면 N/A 설정
+                detected = food_item.get("detected_food", "N/A")
+
+                # None 값 제거 후 유사 음식 리스트 구성
+                similar_foods_list = [
+                    food["food_name"] for food in food_item.get("similar_foods", []) 
+                    if food["food_name"] is not None
+                ]
+
+                # 유사한 음식 존재하지 않으면 N/A 설정
+                similar_foods = ",".join(similar_foods_list) if similar_foods_list else "N/A"
+
+                test_results.append([
+                    f"{idx}-{j + 1}", # 1-1, 1-2 형식
+                    read_food,
+                    total_time,
+                    analyze_time,
+                    search_time,
+                    detected,
+                    similar_foods
+                ])
+        else:
+            print(f"API returned unexpected response: {response}")
+            test_results.append([f"{idx}-1", read_food, 0, 0, 0, "ERROR", ""])
+
+    except Exception as e:
+        print(f"Error processing {image_file}: {e}")
+        test_results.append([f"{idx}-1", read_food, 0, 0, 0, "ERROR", str(e)])
+
+# Dataframe 생성 및 저장
+columns = ["ORDER", "READ_FOOD", "ANALYZE_FOOD_IMAGE(sec)", "FOOD_IMAGE_ANALYZE(sec)", "SEARCH_SIMILAR(sec)", "DETECTED", "SIMILAR"]
+df = pd.DataFrame(test_results, columns=columns)
+
+# 최종적으로 csv로 저장
+df.to_csv(output_csv, index=False)
+print(f"테스트 및 결과 저장 완료")