diff --git a/.github/workflows/ci-cd.yml b/.github/workflows/ci-cd.yml index 72be5de..062a41a 100644 --- a/.github/workflows/ci-cd.yml +++ b/.github/workflows/ci-cd.yml @@ -151,10 +151,6 @@ jobs: run: | flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics > flake8-results.txt 2>&1 || true LINT_ERRORS=$(grep -c "E9\|F63\|F7\|F82" flake8-results.txt 2>/dev/null || echo "0") - - # 안전하게 output 쓰기 - echo "lint_errors=${LINT_ERRORS}" >> $GITHUB_OUTPUT - echo "service_name=${{ matrix.service }}" >> $GITHUB_OUTPUT - name: Run tests with coverage - ${{ matrix.service }} id: test diff --git a/requirements-dev.txt b/requirements-dev.txt deleted file mode 100644 index 5f92985..0000000 --- a/requirements-dev.txt +++ /dev/null @@ -1,38 +0,0 @@ -annotated-types==0.7.0 -anyio==4.9.0 -black==25.1.0 -certifi==2025.7.14 -click==8.2.1 -colorama==0.4.6 -exceptiongroup==1.3.0 -fastapi==0.116.1 -flake8==7.3.0 -h11==0.16.0 -httpcore==1.0.9 -httpx==0.28.1 -idna==3.10 -iniconfig==2.1.0 -mccabe==0.7.0 -mypy==1.17.0 -mypy_extensions==1.1.0 -packaging==25.0 -pathspec==0.12.1 -platformdirs==4.3.8 -pluggy==1.6.0 -prometheus_client==0.22.1 -psutil==7.0.0 -pycodestyle==2.14.0 -pydantic==2.11.7 -pydantic_core==2.33.2 -pyflakes==3.4.0 -Pygments==2.19.2 -pytest==8.4.1 -python-dotenv==1.1.1 -python-json-logger==3.3.0 -python-multipart==0.0.20 -sniffio==1.3.1 -starlette==0.47.2 -tomli==2.2.1 -typing-inspection==0.4.1 -typing_extensions==4.14.1 -uvicorn==0.35.0 diff --git a/services/painting-surface-data-simulator-service/Dockerfile b/services/painting-surface-data-simulator-service/Dockerfile new file mode 100644 index 0000000..ff65044 --- /dev/null +++ b/services/painting-surface-data-simulator-service/Dockerfile @@ -0,0 +1,31 @@ +FROM python:3.10 + +# 작업 디렉토리 설정 +WORKDIR /app + +# 시스템 패키지 업데이트 및 필요한 패키지 설치 +RUN apt-get update && apt-get install -y \ + gcc \ + && rm -rf /var/lib/apt/lists/* + +# Python 의존성 파일 복사 및 설치 +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# 애플리케이션 코드 복사 +COPY app/ ./app/ + +# 환경 설정 파일 복사 +COPY .env ./ + +# 로그 디렉토리 생성 +RUN mkdir -p logs + +# 포트 노출 +EXPOSE 8012 + +# 환경 변수 설정 +ENV PYTHONPATH=/app + +# 애플리케이션 실행 +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8012"] diff --git a/services/painting-surface-data-simulator-service/README.md b/services/painting-surface-data-simulator-service/README.md new file mode 100644 index 0000000..3d2f768 --- /dev/null +++ b/services/painting-surface-data-simulator-service/README.md @@ -0,0 +1,404 @@ +# 도장 표면 결함 탐지 데이터 시뮬레이터 서비스 + +도장 표면의 결함을 탐지하기 위한 실시간 데이터 시뮬레이터 서비스입니다. Azure Blob Storage에서 도장 표면 이미지 데이터를 주기적으로 수집하여 결함 탐지 모델 서비스에 전송하고, 결과를 체계적으로 로깅합니다. + +## 🎯 주요 기능 + +- **🔄 자동화된 데이터 시뮬레이션**: 설정 가능한 간격으로 도장 표면 이미지 데이터 수집 및 모델 추론 +- **☁️ Azure Blob Storage 연동**: 클라우드 기반 이미지 데이터 관리 및 실시간 접근 +- **🤖 모델 서비스 통신**: 도장 표면 결함 탐지 모델과의 HTTP 통신 및 예측 요청 +- **📊 실시간 모니터링**: 결함 탐지 결과, 시스템 상태, 연결 상태 실시간 모니터링 +- **📝 체계적 로깅**: 결함 탐지, 정상 처리, 오류 상황을 JSON 형태로 체계적 기록 +- **⚡ 비동기 처리**: FastAPI 기반의 고성능 비동기 처리로 효율적인 데이터 시뮬레이션 + +## 🏗️ 아키텍처 + +``` +┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐ +│ Azure Blob │ │ Simulator │ │ Painting │ +│ Storage │◄──►│ Service │◄──►│ Surface │ +│ (Images) │ │ (Port 8012) │ │ Model │ +│ │ │ │ │ (Port 8002) │ +└─────────────────┘ └──────────────────┘ └─────────────────┘ + │ + ▼ + ┌──────────────────┐ + │ Logger │ + │ (JSON Files) │ + │ - logs/ │ + │ - errors/ │ + └──────────────────┘ +``` + +## 🚀 설치 및 설정 + +### 1. 의존성 설치 + +```bash +pip install -r requirements.txt +``` + +**핵심 의존성 패키지:** +- **FastAPI & Uvicorn**: 웹 서비스 프레임워크 +- **Azure Storage Blob**: 클라우드 스토리지 연동 +- **APScheduler**: 주기적 작업 스케줄링 +- **httpx**: 모델 서비스와의 HTTP 통신 +- **pydantic-settings**: 환경 변수 관리 + +### 2. 환경 변수 설정 + +`.env` 파일을 생성하고 다음 내용을 설정하세요: + +```env +# Azure Storage 설정 (필수) +AZURE_CONNECTION_STRING=DefaultEndpointsProtocol=https;AccountName=your_account;AccountKey=your_key;EndpointSuffix=core.windows.net +AZURE_CONTAINER_NAME=simulator-data +PAINTING_DATA_FOLDER=painting-surface + +# 스케줄러 설정 +SCHEDULER_INTERVAL_MINUTES=1 +BATCH_SIZE=10 + +# 모델 서비스 설정 +PAINTING_MODEL_URL=http://painting-model-service:8002 + +# 로깅 설정 +LOG_DIRECTORY=logs +LOG_FILENAME=painting_defect_detections.json +ERROR_LOG_FILENAME=painting_errors.json + +# HTTP 설정 +HTTP_TIMEOUT=30 +MAX_RETRIES=3 +``` + +### 3. 서비스 실행 + +```bash +# 개발 모드 +uvicorn app.main:app --host 0.0.0.0 --port 8012 --reload + +# 프로덕션 모드 +uvicorn app.main:app --host 0.0.0.0 --port 8012 +``` + +**포트 정보:** +- **시뮬레이터 서비스**: 포트 8012 +- **도장 표면 모델 서비스**: 포트 8002 (외부 연결) + +## 📡 API 엔드포인트 + +### 🏠 기본 정보 + +#### 서비스 정보 조회 +```http +GET / +``` +**응답 예시:** +```json +{ + "service": "Painting Surface Data Simulator Service", + "version": "1.0.0", + "status": "running", + "target_model": "painting-surface-defect-detection", + "scheduler_status": { + "is_running": true, + "scheduler_interval_minutes": 1, + "batch_size": 10 + }, + "azure_storage": { + "container": "simulator-data", + "data_folder": "painting-surface", + "connection_status": "connected" + } +} +``` + +#### 헬스 체크 +```http +GET /health +``` +**응답:** +```json +{ + "status": "healthy" +} +``` + +### 🎮 시뮬레이터 제어 + +#### 시뮬레이터 시작 +```http +POST /simulator/start +``` +**응답 예시:** +```json +{ + "message": "시뮬레이터가 시작되었습니다.", + "status": { + "is_running": true, + "scheduler_interval_minutes": 1, + "batch_size": 10 + } +} +``` + +#### 시뮬레이터 중지 +```http +POST /simulator/stop +``` +**응답 예시:** +```json +{ + "message": "시뮬레이터가 중지되었습니다.", + "status": { + "is_running": false, + "scheduler_interval_minutes": 1, + "batch_size": 10 + } +} +``` + +#### 시뮬레이터 상태 조회 +```http +GET /simulator/status +``` +**응답 예시:** +```json +{ + "is_running": true, + "scheduler_interval_minutes": 1, + "batch_size": 10, + "painting_surface_service_health": true +} +``` + +### 📊 로그 관리 + +#### 최근 로그 조회 +```http +GET /simulator/logs/recent +``` +**응답 예시:** +```json +{ + "logs": [ + { + "timestamp": "2024-01-01T12:00:00", + "service_name": "painting-surface", + "prediction": { + "status": "anomaly", + "defect_count": 2, + "total_count": 5, + "defect_ratio": 0.4, + "combined_logic": "총 5개 이미지 중 2개에서 결함 탐지 → 최종: anomaly" + }, + "original_data": { + "images": ["image1.jpg", "image2.jpg", "image3.jpg", "image4.jpg", "image5.jpg"] + } + } + ], + "total_count": 1 +} +``` + +### 🔗 연결 테스트 + +#### Azure Storage 연결 테스트 +```http +POST /test/azure-storage-connection +``` +**성공 응답:** +```json +{ + "status": "success", + "message": "Azure Storage 연결 성공", + "file_count": 15, + "sample_files": [ + "painting-surface/image1.jpg", + "painting-surface/image2.jpg", + "painting-surface/image3.jpg", + "painting-surface/image4.jpg", + "painting-surface/image5.jpg" + ] +} +``` + +**실패 응답:** +```json +{ + "status": "error", + "message": "Azure Storage 연결 실패: 연결 문자열이 올바르지 않습니다." +} +``` + +#### 모델 서비스 연결 테스트 +```http +POST /test/models-connection +``` +**성공 응답:** +```json +{ + "status": "success", + "service_name": "painting-surface-defect-detection", + "healthy": true, + "message": "도장 표면 결함탐지 서비스 연결 성공" +} +``` + +**실패 응답:** +```json +{ + "status": "error", + "service_name": "painting-surface-defect-detection", + "healthy": false, + "message": "도장 표면 결함탐지 서비스 연결 실패" +} +``` + +## ⚙️ 시뮬레이션 프로세스 + +### 1. **초기화 단계** +- Azure Storage 연결 및 인증 +- 도장 표면 결함 탐지 모델 서비스 헬스 체크 +- 로그 디렉토리 생성 및 설정 + +### 2. **데이터 수집 단계** +- Azure Blob Storage에서 `painting-surface/` 폴더 내 이미지 파일 목록 조회 +- 지원 형식: `.jpg`, `.jpeg`, `.png`, `.bmp` +- 순차적 이미지 인덱싱으로 데이터 시뮬레이션 + +### 3. **모델 추론 단계** +- 각 이미지를 Azure에서 다운로드 +- 파일 업로드 방식으로 모델 서비스에 전송 +- 결함 탐지 결과 수신 및 분석 + +### 4. **결과 처리 단계** +- 결함 탐지 여부에 따른 결과 분류 +- 상세 정보 로깅 (결함 개수, 총 이미지 수, 비율 등) +- 정상/이상 상태 판정 + +### 5. **로깅 및 모니터링** +- JSON 형태로 구조화된 로그 저장 +- 실시간 콘솔 출력 +- 에러 상황 별도 로그 파일 관리 + +## 📋 결함 탐지 결과 구조 + +### 이미지별 결과 +```json +{ + "timestamp": "2024-01-01T12:00:00", + "service_name": "painting-surface", + "prediction": { + "status": "anomaly", + "defect_count": 2, + "total_count": 5, + "defect_ratio": 0.4, + "combined_logic": "총 5개 이미지 중 2개에서 결함 탐지 → 최종: anomaly" + }, + "original_data": {...} +} +``` + +### 결합 결과 로직 +- **정상 (normal)**: 모든 이미지에서 결함 미탐지 +- **이상 (anomaly)**: 하나 이상의 이미지에서 결함 탐지 +- **결함 비율**: `defect_count / total_count` + +## 🐳 Docker 실행 + +### 이미지 빌드 +```bash +docker build -t painting-surface-data-simulator-service . +``` + +### 컨테이너 실행 +```bash +docker run -d -p 8012:8012 \ + -e AZURE_CONNECTION_STRING="your_connection_string" \ + -e AZURE_CONTAINER_NAME="simulator-data" \ + -e PAINTING_MODEL_URL="http://host.docker.internal:8002" \ + --name painting-data-simulator \ + painting-surface-data-simulator-service +``` + +## 🔍 모니터링 및 디버깅 + +### 시뮬레이터 상태 확인 +```bash +# 상태 조회 +curl http://localhost:8012/simulator/status + +# 헬스 체크 +curl http://localhost:8012/health + +# 최근 로그 +curl http://localhost:8012/simulator/logs/recent +``` + +### 로그 파일 구조 +``` +logs/ +├── painting_defect_detections.json # 결함 탐지 결과 로그 +└── painting_errors.json # 에러 로그 +``` + +### 연결 상태 확인 +```bash +# Azure Storage 연결 테스트 +curl -X POST http://localhost:8012/test/azure-storage-connection + +# 모델 서비스 연결 테스트 +curl -X POST http://localhost:8012/test/models-connection +``` + +## 🧪 개발 및 테스트 + +### 테스트 실행 +```bash +pytest tests/ +``` + +### 코드 품질 관리 +```bash +# 코드 포맷팅 +black app/ +isort app/ + +# 린팅 +flake8 app/ +mypy app/ +``` + +## ⚠️ 주의사항 + +1. **Azure Storage 연결**: `AZURE_CONNECTION_STRING` 환경 변수는 필수입니다. +2. **모델 서비스**: 도장 표면 결함 탐지 모델 서비스가 실행 중이어야 합니다. +3. **포트 충돌**: 포트 8012가 사용 가능한지 확인하세요. +4. **로그 디스크 공간**: 로그 파일이 지속적으로 증가하므로 디스크 공간을 모니터링하세요. + +## 🔧 설정 옵션 + +### 스케줄러 설정 +- **`SCHEDULER_INTERVAL_MINUTES`**: 데이터 수집 간격 (기본값: 1분) +- **`BATCH_SIZE`**: 한 번에 처리할 이미지 수 (기본값: 10개) + +### HTTP 설정 +- **`HTTP_TIMEOUT`**: 모델 서비스 요청 타임아웃 (기본값: 30초) +- **`MAX_RETRIES`**: 재시도 최대 횟수 (기본값: 3회) + +### 로깅 설정 +- **`LOG_DIRECTORY`**: 로그 저장 디렉토리 (기본값: `logs`) +- **`LOG_FILENAME`**: 결함 탐지 로그 파일명 +- **`ERROR_LOG_FILENAME`**: 에러 로그 파일명 + +## 📞 지원 및 문의 + +서비스 관련 문제나 개선 사항이 있으시면 개발팀에 문의해주세요. + +--- + +**버전**: 1.0.0 +**최종 업데이트**: 2024년 1월 +**라이선스**: 내부 사용 전용 diff --git a/services/painting-surface-data-simulator-service/app/config/settings.py b/services/painting-surface-data-simulator-service/app/config/settings.py new file mode 100644 index 0000000..ccbca23 --- /dev/null +++ b/services/painting-surface-data-simulator-service/app/config/settings.py @@ -0,0 +1,49 @@ +import os +from typing import Dict +from pydantic_settings import BaseSettings + + +class Settings(BaseSettings): + # Azure Storage 설정 + azure_connection_string: str + azure_container_name: str = "simulator-data" + + # 도장 표면 결함 감지 전용 설정 + painting_data_folder: str = "painting-surface" + + # 스케줄러 설정 + scheduler_interval_minutes: float = 0.5 + batch_size: int = 4 + + # 도장 표면 결함 감지 모델 서비스 설정 (로컬 실행용) + painting_model_url: str = "http://localhost:8002" + + # 백엔드 서비스 설정 (로컬 실행용) + backend_url: str = "http://localhost:8087" + + # 로그 설정 + log_directory: str = "logs" + log_filename: str = "painting_defect_detections.json" + error_log_filename: str = "painting_defect_detections.json" + + # HTTP 클라이언트 설정 + http_timeout: int = 30 + max_retries: int = 3 + + @property + def model_service_url(self) -> str: + """도장 표면 결함 감지 모델 서비스 URL""" + return self.painting_model_url + + @property + def backend_service_url(self) -> str: + """백엔드 서비스 URL""" + return self.backend_url + + model_config = { + "env_file": ".env", + "env_file_encoding": "utf-8" + } + + +settings = Settings() diff --git a/services/painting-surface-data-simulator-service/app/main.py b/services/painting-surface-data-simulator-service/app/main.py new file mode 100644 index 0000000..42010f4 --- /dev/null +++ b/services/painting-surface-data-simulator-service/app/main.py @@ -0,0 +1,90 @@ +from fastapi import FastAPI +from contextlib import asynccontextmanager +from app.config.settings import settings +from app.services.scheduler_service import simulator_scheduler +from app.routers import simulator_router +from app.routers import test_connection_router +from app.services.azure_storage import azure_storage +from app.services.model_client import painting_surface_model_client +import os + + +@asynccontextmanager +async def lifespan(app: FastAPI): + """애플리케이션 생명주기 관리""" + # 시작 시 + print("🚀 Painting Surface Defect Simulator Service 시작 중...") + + # 환경 변수 체크 + print("🔍 환경 변수 확인 중...") + print(f" Azure Connection String: {'✅ 설정됨' if settings.azure_connection_string else '❌ 설정되지 않음'}") + print(f" Azure Container: {settings.azure_container_name}") + print(f" Painting Data Folder: {settings.painting_data_folder}") + print(f" Backend URL: {settings.backend_service_url}") + + if not settings.azure_connection_string: + print("⚠️ AZURE_CONNECTION_STRING 환경 변수가 설정되지 않았습니다.") + print(" .env 파일을 생성하거나 환경 변수를 설정해주세요.") + print(" 예시: AZURE_CONNECTION_STRING=DefaultEndpointsProtocol=https;AccountName=...") + else: + # Azure Storage 연결 테스트 + try: + print("🔗 Azure Storage 연결 테스트 중...") + # 간단한 연결 테스트 - 파일 목록 조회 + test_files = await azure_storage.list_data_files() + print(f"✅ Azure Storage 연결 성공! ({len(test_files)}개 파일 발견)") + except Exception as e: + print(f"❌ Azure Storage 연결 실패: {e}") + print(" 연결 문자열과 계정 키를 확인해주세요.") + + # 로그 디렉토리 생성 + os.makedirs(settings.log_directory, exist_ok=True) + + print(f"📁 로그 디렉토리: {settings.log_directory}") + print(f"🔧 스케줄러 간격: {settings.scheduler_interval_minutes}분") + print(f"🎯 대상 서비스: 도장 표면 결함탐지 모델") + + yield + + # 종료 시 + print("🛑 Painting Surface Defect Simulator Service 종료 중...") + if simulator_scheduler.is_running: + await simulator_scheduler.stop() + + +# FastAPI 앱 생성 +app = FastAPI( + title="Painting Surface Defect Simulator Service", + description="도장 표면 결함 탐지 모델을 위한 실시간 데이터 시뮬레이터", + version="1.0.0", + lifespan=lifespan +) + +# 라우터 설정 +# 시뮬레이터 활성화/비활성화/상태확인 API 모음 +app.include_router(simulator_router.router, prefix="/simulator") +# Azure Storage 연결, 모델 서비스 연결 확인 API 모음 +app.include_router(test_connection_router.router, prefix="/test") + +# 아래는 서비스 기본 정보 확인과 서비스 헬스 체크 api 정의 +@app.get("/") +async def root(): + """서비스 정보""" + return { + "service": "Painting Surface Data Simulator Service", + "version": "1.0.0", + "status": "running", + "target_model": "painting-surface-defect-detection", + "scheduler_status": simulator_scheduler.get_status(), + "azure_storage": { + "container": settings.azure_container_name, + "data_folder": settings.painting_data_folder, + "connection_status": "connected" if hasattr(azure_storage, 'client') and azure_storage.client else "disconnected" + } + } + + +@app.get("/health") +async def health_check(): + """헬스 체크""" + return {"status": "healthy"} diff --git a/services/painting-surface-data-simulator-service/app/routers/simulator_router.py b/services/painting-surface-data-simulator-service/app/routers/simulator_router.py new file mode 100644 index 0000000..74df974 --- /dev/null +++ b/services/painting-surface-data-simulator-service/app/routers/simulator_router.py @@ -0,0 +1,77 @@ +from fastapi import APIRouter, HTTPException +from app.services.model_client import painting_surface_model_client +from app.services.scheduler_service import simulator_scheduler +from app.config.settings import settings + +import os +import json + +router = APIRouter() + + +@router.get("/status") +async def get_simulator_status(): + """시뮬레이터 상태 조회""" + status = simulator_scheduler.get_status() + + # 도장 표면 결함탐지 서비스 헬스 체크 추가 + if simulator_scheduler.is_running: + health_status = await painting_surface_model_client.health_check() + status["painting_surface_service_health"] = health_status + + return status + + +@router.post("/start") +async def start_simulator(): + """시뮬레이션 시작""" + try: + await simulator_scheduler.start() + return { + "message": "시뮬레이터가 시작되었습니다.", + "status": simulator_scheduler.get_status() + } + except Exception as e: + raise HTTPException(status_code=500, detail=f"시뮬레이터 시작 실패: {str(e)}") + + +@router.post("/stop") +async def stop_simulator(): + """시뮬레이션 중지""" + try: + await simulator_scheduler.stop() + return { + "message": "시뮬레이터가 중지되었습니다.", + "status": simulator_scheduler.get_status() + } + except Exception as e: + raise HTTPException(status_code=500, detail=f"시뮬레이터 중지 실패: {str(e)}") + + +@router.get("/logs/recent") +async def get_recent_logs(): + """최근 로그 조회""" + try: + log_file_path = os.path.join( + settings.log_directory, settings.log_filename) + + if not os.path.exists(log_file_path): + return {"logs": [], "message": "로그 파일이 없습니다."} + + # 최근 10개 로그만 반환 + logs = [] + with open(log_file_path, "r", encoding="utf-8") as f: + lines = f.readlines() + for line in lines[-10:]: # 최근 10개 + try: + logs.append(json.loads(line.strip())) + except: + continue + + return { + "logs": logs, + "total_count": len(lines) if 'lines' in locals() else 0 + } + + except Exception as e: + raise HTTPException(status_code=500, detail=f"로그 조회 실패: {str(e)}") diff --git a/services/painting-surface-data-simulator-service/app/routers/test_connection_router.py b/services/painting-surface-data-simulator-service/app/routers/test_connection_router.py new file mode 100644 index 0000000..549da1b --- /dev/null +++ b/services/painting-surface-data-simulator-service/app/routers/test_connection_router.py @@ -0,0 +1,47 @@ +from fastapi import APIRouter +from app.services.model_client import painting_surface_model_client +from app.services.azure_storage import azure_storage +from app.config.settings import settings +import pytest + +router = APIRouter() + +pytestmark = pytest.mark.asyncio + + +@router.post("/azure-storage-connection") +async def test_azure_connection(): + """Azure Storage 연결 테스트""" + try: + files = await azure_storage.list_data_files() + + return { + "status": "success", + "message": "Azure Storage 연결 성공", + "file_count": len(files), + "sample_files": files[:5] # 처음 5개 파일만 표시 + } + except Exception as e: + return { + "status": "error", + "message": f"Azure Storage 연결 실패: {str(e)}" + } + + +@router.post("/models-connection") +async def test_model_services(): + """도장 표면 결함탐지 서비스 연결 테스트""" + try: + is_healthy = await painting_surface_model_client.health_check() + + return { + "status": "success" if is_healthy else "error", + "service_name": "painting-surface-defect-detection", + "healthy": is_healthy, + "message": "도장 표면 결함탐지 서비스 연결 성공" if is_healthy else "도장 표면 결함탐지 서비스 연결 실패" + } + except Exception as e: + return { + "status": "error", + "message": f"도장 표면 결함탐지 서비스 테스트 실패: {str(e)}" + } diff --git a/services/painting-surface-data-simulator-service/app/services/azure_storage.py b/services/painting-surface-data-simulator-service/app/services/azure_storage.py new file mode 100644 index 0000000..a9516cf --- /dev/null +++ b/services/painting-surface-data-simulator-service/app/services/azure_storage.py @@ -0,0 +1,122 @@ +from typing import List, Dict, Any, Optional +import asyncio +from azure.storage.blob.aio import BlobServiceClient +from azure.core.exceptions import AzureError, ClientAuthenticationError +from app.config.settings import settings + + +class AzureStorageService: + def __init__(self): + self.connection_string = settings.azure_connection_string + self.container_name = settings.azure_container_name + + # 도장 표면 이미지 처리를 위한 인덱스 + self.image_index = 0 + + + + async def list_data_files(self) -> List[str]: + """데이터 파일 목록 조회""" + if not self.connection_string: + print("❌ Azure connection string이 설정되지 않았습니다.") + return [] + + try: + async with BlobServiceClient.from_connection_string(self.connection_string) as client: + container_client = client.get_container_client(self.container_name) + blob_list = [] + + # 도장 표면 이미지 폴더 검색 + prefix = f"{settings.painting_data_folder}/" + print(f"🔍 검색 중: {prefix}") + + async for blob in container_client.list_blobs(name_starts_with=prefix): + if blob.name.lower().endswith(('.jpg', '.jpeg', '.png', '.bmp')): + blob_list.append(blob.name) + print(f"📁 발견된 파일: {blob.name}") + + print(f"📊 총 {len(blob_list)}개의 이미지 파일 발견") + return sorted(blob_list) + + except ClientAuthenticationError as e: + print(f"❌ 인증 오류로 인한 파일 목록 조회 실패: {e}") + print(" Azure Storage 계정 키와 연결 문자열을 확인해주세요.") + return [] + except Exception as e: + print(f"❌ 도장 표면 이미지 파일 목록 조회 실패: {e}") + return [] + + async def read_image_data(self, blob_name: str) -> Optional[bytes]: + """이미지 파일 읽기""" + if not self.connection_string: + print("❌ Azure connection string이 설정되지 않았습니다.") + return None + + try: + async with BlobServiceClient.from_connection_string(self.connection_string) as client: + blob_client = client.get_blob_client( + container=self.container_name, + blob=blob_name + ) + + # 비동기로 blob 데이터 다운로드 + blob_data = await blob_client.download_blob() + content = await blob_data.readall() + + print(f"📁 이미지 읽기 성공: {blob_name} ({len(content)} bytes)") + return content + + except Exception as e: + print(f"❌ 이미지 읽기 실패 ({blob_name}): {e}") + return None + + async def get_recent_data_files(self) -> Dict[str, List[str]]: + """최근 N시간 내 도장 표면 이미지 파일들 반환""" + try: + all_files = await self.list_data_files() + + # painting-surface 전용 + process_files = { + "painting-surface": all_files # 모든 파일이 도장 표면 이미지 + } + + return process_files + + except Exception as e: + print(f"❌ 최근 도장 표면 이미지 파일 조회 실패: {e}") + return {"painting-surface": []} + + async def simulate_painting_surface_data(self) -> Optional[Dict[str, List[str]]]: + """도장 표면 결함 감지 이미지 데이터 시뮬레이션""" + try: + # 도장 표면 이미지 파일 목록 조회 + image_files = await self.list_data_files() + + if not image_files: + print("⚠️ 도장 표면 이미지 파일이 없습니다.") + return None + + # 순차적으로 이미지 처리 (배치 크기만큼) + batch_size = min(settings.batch_size, len(image_files)) + start_idx = self.image_index % len(image_files) + end_idx = min(start_idx + batch_size, len(image_files)) + + batch_files = image_files[start_idx:end_idx] + self.image_index = (self.image_index + batch_size) % len(image_files) + + print(f"📸 이미지 배치 처리: {len(batch_files)}개 (전체: {len(image_files)}개)") + + return { + "images": batch_files, + "total_images": len(image_files), + "batch_size": batch_size, + "current_batch": batch_files + } + + except Exception as e: + print(f"❌ 도장 표면 데이터 시뮬레이션 실패: {e}") + return None + + +# 글로벌 Azure Storage 서비스 인스턴스 +azure_storage = AzureStorageService() diff --git a/services/painting-surface-data-simulator-service/app/services/model_client.py b/services/painting-surface-data-simulator-service/app/services/model_client.py new file mode 100644 index 0000000..ad83f4a --- /dev/null +++ b/services/painting-surface-data-simulator-service/app/services/model_client.py @@ -0,0 +1,186 @@ +import httpx +import asyncio +from typing import Dict, Any, Optional, List +from app.config.settings import settings +from datetime import datetime +from app.services.azure_storage import azure_storage + +class PaintingSurfaceModelClient: + """도장 표면 결함탐지 전용 모델 클라이언트""" + + def __init__(self): + self.timeout = httpx.Timeout(settings.http_timeout) + self.max_retries = settings.max_retries + self.service_name = "painting-surface" + # AI 모델 대신 백엔드 URL 사용 + self.backend_url = settings.backend_service_url + + async def predict_painting_surface_data(self, image_files: List[str]) -> Optional[Dict[str, Any]]: + """도장 표면 이미지 데이터를 백엔드로 전송하여 결함 감지 요청""" + if not self.backend_url: + print(f"❌ Backend 서비스 URL을 찾을 수 없습니다.") + return None + + # 백엔드의 결함 감지 API 엔드포인트 사용 + predict_url = f"{self.backend_url}/api/painting-surface/defect-detection" + + results = {} + + # 각 이미지 파일에 대해 예측 요청 + print(f"🎨 도장 표면 이미지 예측 요청...") + image_results = [] + + for image_file in image_files: + try: + # Azure Storage에서 이미지 다운로드 + print(f"📥 이미지 다운로드 중: {image_file}") + image_data = await self._download_image_from_azure(image_file) + + if not image_data: + print(f"⚠️ 이미지 다운로드 실패: {image_file}") + continue + + # 백엔드로 결함 감지 요청 + result = await self._predict_with_backend(predict_url, image_data, image_file) + if result: + # 상세한 예측 결과 로깅 + self._log_detailed_prediction_result(image_file, result) + image_results.append(result) + + except Exception as e: + print(f"❌ 이미지 {image_file} 예측 실패: {e}") + continue + + if not image_results: + print("❌ 모든 이미지 예측이 실패했습니다.") + return None + + # 결과 조합 + combined_result = self._combine_painting_results(image_results) + results["images"] = image_results + results["combined"] = combined_result + + return results + + def _combine_painting_results(self, image_results: List[Dict[str, Any]]) -> Dict[str, Any]: + """도장 표면 이미지 결과 조합""" + if not image_results: + return { + "status": "error", + "message": "예측 결과를 받을 수 없습니다." + } + + # 결함이 하나라도 탐지되면 anomaly + defect_count = sum(1 for result in image_results if result.get('status') == 'defect') + total_count = len(image_results) + + if defect_count > 0: + final_status = "anomaly" + else: + final_status = "normal" + + return { + "status": final_status, + "defect_count": defect_count, + "total_count": total_count, + "defect_ratio": defect_count / total_count if total_count > 0 else 0, + "combined_logic": f"총 {total_count}개 이미지 중 {defect_count}개에서 결함 탐지 → 최종: {final_status}" + } + + async def health_check(self) -> bool: + """백엔드 서비스 헬스 체크""" + if not self.backend_url: + return False + + # 백엔드의 모델 헬스 체크 엔드포인트 사용 + health_url = f"{self.backend_url}/api/painting-surface/defect-detection/model-health" + + try: + async with httpx.AsyncClient(timeout=self.timeout) as client: + response = await client.get(health_url) + return response.status_code == 200 + except: + return False + + async def _download_image_from_azure(self, image_path: str) -> Optional[bytes]: + """Azure Storage에서 이미지 다운로드""" + try: + # azure_storage 서비스에서 이미지 데이터 읽기 + image_data = await azure_storage.read_image_data(image_path) + return image_data + except Exception as e: + print(f"❌ Azure Storage에서 이미지 다운로드 실패 ({image_path}): {e}") + return None + + async def _predict_with_backend(self, backend_url: str, image_data: bytes, image_file: str) -> Optional[Dict[str, Any]]: + """백엔드를 통해 결함 감지 요청""" + for attempt in range(self.max_retries): + try: + # multipart/form-data로 이미지 파일 업로드 + files = { + 'image': (image_file, image_data, 'image/jpeg') + } + + async with httpx.AsyncClient(timeout=self.timeout) as client: + response = await client.post(backend_url, files=files) + + if response.status_code == 200: + result = response.json() + print(f"✅ {image_file} 백엔드 응답 성공 (시도 {attempt + 1})") + + # 백엔드에서 이미 DB 저장 처리를 했으므로 + # 시뮬레이터에서는 결과만 확인 + if result.get("status") == "defect": + print(f"🚨 결함 감지됨: {image_file} - 백엔드에서 DB에 자동 저장됨") + else: + print(f"✅ 정상 상태: {image_file}") + + return result + else: + print(f"⚠️ {image_file} HTTP {response.status_code} (시도 {attempt + 1})") + + except httpx.TimeoutException: + print(f"⏰ {image_file} 타임아웃 (시도 {attempt + 1})") + except httpx.ConnectError: + print(f"🔌 {image_file} 연결 실패 (시도 {attempt + 1})") + except Exception as e: + print(f"❌ {image_file} 백엔드 통신 오류 (시도 {attempt + 1}): {e}") + + if attempt < self.max_retries - 1: + await asyncio.sleep(2 ** attempt) # 지수 백오프 + + print(f"❌ {image_file} 최대 재시도 횟수 초과") + return None + + def _log_detailed_prediction_result(self, image_file: str, result: Dict[str, Any]): + """상세한 예측 결과 로깅""" + print(f"\n🔍 {image_file} 상세 예측 결과:") + print(f" 📊 이미지 크기: {result.get('image_shape', 'N/A')}") + print(f" 🎯 신뢰도 임계값: {result.get('confidence_threshold', 'N/A')}") + + predictions = result.get('predictions', []) + if predictions: + print(f" ⚠️ 결함 탐지됨: {len(predictions)}개") + for i, pred in enumerate(predictions, 1): + print(f" 결함 {i}:") + print(f" 🏷️ 종류: {pred.get('class_name', 'N/A')}") + print(f" 📍 위치: {pred.get('bbox', 'N/A')}") + print(f" 📏 크기: {pred.get('area', 'N/A')} 픽셀²") + print(f" 🎯 신뢰도: {pred.get('confidence', 'N/A'):.3f}") + else: + print(f" ✅ 결함 없음 - 정상 상태") + + print(f" 🕒 예측 시간: {result.get('timestamp', 'N/A')}") + print(f" 🤖 모델 소스: {result.get('model_source', 'N/A')}") + + # 추가: 전체 응답 구조 로깅 + print(f" 🔍 전체 응답 구조:") + for key, value in result.items(): + if key != 'predictions': # predictions는 이미 위에서 처리됨 + print(f" {key}: {value}") + + print("-" * 60) + + +# 글로벌 도장 표면 모델 클라이언트 인스턴스 +painting_surface_model_client = PaintingSurfaceModelClient() diff --git a/services/painting-surface-data-simulator-service/app/services/scheduler_service.py b/services/painting-surface-data-simulator-service/app/services/scheduler_service.py new file mode 100644 index 0000000..710684d --- /dev/null +++ b/services/painting-surface-data-simulator-service/app/services/scheduler_service.py @@ -0,0 +1,175 @@ +from datetime import datetime +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.interval import IntervalTrigger +from app.config.settings import settings +from app.services.azure_storage import azure_storage +from app.services.model_client import painting_surface_model_client +from app.utils.logger import anomaly_logger + + +class SimulatorScheduler: + def __init__(self): + self.scheduler = AsyncIOScheduler() + self.is_running = False + + async def start(self): + """스케줄러 시작""" + if self.is_running: + print("⚠️ 스케줄러가 이미 실행 중입니다.") + return + + try: + print("🔧 스케줄러 시작 준비 중...") + + # 헬스 체크 + await self._initial_health_check() + + # 스케줄 작업 등록 + print("📅 스케줄 작업 등록 중...") + self.scheduler.add_job( + func=self._simulate_data_collection, + trigger=IntervalTrigger( + minutes=settings.scheduler_interval_minutes), + id='data_simulation', + name='Data Collection Simulation', + replace_existing=True + ) + print("✅ 스케줄 작업 등록 완료") + + print("🚀 스케줄러 시작 중...") + self.scheduler.start() + self.is_running = True + + print(f"🚀 시뮬레이터 시작! (간격: {settings.scheduler_interval_minutes}분)") + print(f"📊 대상 서비스: 도장 표면 결함탐지 모델") + print("-" * 60) + + except Exception as e: + print(f"❌ 스케줄러 시작 실패: {e}") + raise + + async def stop(self): + """스케줄러 중지""" + if not self.is_running: + print("⚠️ 스케줄러가 실행 중이 아닙니다.") + return + + self.scheduler.shutdown() + await azure_storage.disconnect() + self.is_running = False + print("🛑 시뮬레이터 중지됨") + + async def _initial_health_check(self): + """초기 헬스 체크""" + print("🔍 백엔드 서비스 헬스 체크 중...") + is_healthy = await painting_surface_model_client.health_check() + + status = "✅" if is_healthy else "❌" + print(f" {status} 백엔드 서비스") + + if not is_healthy: + raise Exception("백엔드 서비스가 비활성 상태입니다.") + + print(f"📈 활성 서비스: 1/1") + print("-" * 60) + + async def _simulate_data_collection(self): + """주기적 도장 표면 결함 감지 데이터 수집 및 백엔드 전송""" + try: + print(f"🔄 도장 표면 결함 감지 데이터 수집 시작 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + print("📊 Azure Storage에서 이미지 데이터 조회 중...") + + # Azure Blob에서 도장 표면 이미지 데이터 시뮬레이션 + simulated_data = await azure_storage.simulate_painting_surface_data() + + if not simulated_data: + print("⚠️ 수집할 데이터가 없습니다.") + return + + image_data = simulated_data["images"] + total_images = len(image_data) + print(f"📊 이미지 데이터: {total_images} 개") + + # 백엔드에 배치 시작 정보 전송 + await self._send_batch_info_to_backend(total_images) + + # 백엔드 서비스에 결함 감지 요청 (AI 모델 호출 대신) + print("🤖 백엔드 서비스에 결함 감지 요청 중...") + predictions = await painting_surface_model_client.predict_painting_surface_data(image_data) + + if not predictions: + print("❌ 예측 결과를 받을 수 없습니다.") + return + + # 결과 처리 + combined_result = predictions.get("combined") + if not combined_result: + print("❌ 조합된 예측 결과가 없습니다.") + return + + # 이상 감지 여부에 따른 로깅 + if combined_result.get("status") == "anomaly": + # 전체 예측 정보와 원본 데이터 함께 로깅 + anomaly_logger.log_anomaly( + "painting-surface", # 도장 표면 서비스로 수정 + combined_result, + { + "image_data": image_data, + "detailed_results": predictions, + "simulation_data": simulated_data + } + ) + print("🚨 이상 감지!") + else: + anomaly_logger.log_normal_processing( + "painting-surface", combined_result) # 도장 표면 서비스로 수정 + print("✅ 정상 상태") + + # 상세 결과 출력 + print(f"📋 {combined_result.get('combined_logic', 'N/A')}") + print("-" * 60) + + except Exception as e: + print(f"❌ 데이터 수집 중 오류 발생: {e}") + anomaly_logger.log_error("painting-surface-scheduler", str(e)) # 도장 표면 서비스로 수정 + + async def _send_batch_info_to_backend(self, total_images: int): + """백엔드에 배치 시작 정보 전송""" + try: + import httpx + from datetime import datetime + + batch_info = { + "totalImages": total_images, + "batchStartTime": datetime.now().isoformat() + } + + backend_url = settings.backend_service_url # 설정 파일에서 가져오기 + + async with httpx.AsyncClient(timeout=10.0) as client: + response = await client.post( + f"{backend_url}/api/painting-surface/defect-detection/batch-start", + json=batch_info + ) + + if response.status_code == 200: + print(f"✅ 배치 정보 전송 성공: {total_images}개 이미지") + else: + print(f"❌ 배치 정보 전송 실패: {response.status_code}") + + except Exception as e: + print(f"❌ 배치 정보 전송 중 오류: {e}") + + def get_status(self) -> dict: + """스케줄러 상태 정보""" + jobs = self.scheduler.get_jobs() + return { + "is_running": self.is_running, + "interval_minutes": settings.scheduler_interval_minutes, + "next_run": str(jobs[0].next_run_time) if jobs else None, + "target_service": "painting-surface-defect-detection" + } + + +# 글로벌 스케줄러 인스턴스 +simulator_scheduler = SimulatorScheduler() diff --git a/services/painting-surface-data-simulator-service/app/utils/logger.py b/services/painting-surface-data-simulator-service/app/utils/logger.py new file mode 100644 index 0000000..92ec3b1 --- /dev/null +++ b/services/painting-surface-data-simulator-service/app/utils/logger.py @@ -0,0 +1,59 @@ +import json +import os +from datetime import datetime +from typing import Dict, Any +from app.config.settings import settings + + +class AnomalyLogger: + def __init__(self): + # 로그 디렉토리 생성 + os.makedirs(settings.log_directory, exist_ok=True) + self.log_file_path = os.path.join( + settings.log_directory, settings.log_filename) + + def log_anomaly(self, service_name: str, prediction_result: Dict[str, Any], original_data: Dict[str, Any]): + """결함 탐지 결과를 로그 파일에 저장""" + log_entry = { + "timestamp": datetime.now().isoformat(), + "service_name": service_name, + "prediction": prediction_result, + "original_data": original_data + } + + # JSON 파일에 추가 + with open(self.log_file_path, "a", encoding="utf-8") as f: + f.write(json.dumps(log_entry, ensure_ascii=False) + "\n") + + # 콘솔 출력 + print(f"🚨 DEFECT DETECTED: {service_name}") + print(f" └─ Defect Count: {prediction_result.get('defect_count', 'N/A')}") + print(f" └─ Total Images: {prediction_result.get('total_count', 'N/A')}") + print(f" └─ Status: {prediction_result.get('status', 'N/A')}") + print(f" └─ Time: {log_entry['timestamp']}") + print("-" * 50) + + def log_normal_processing(self, service_name: str, prediction_result: Dict[str, Any]): + """정상 처리 결과를 콘솔에만 출력""" + print( + f"✅ NORMAL: {service_name} - 결함 없음 (총 {prediction_result.get('total_count', 'N/A')}개 이미지)") + + def log_error(self, service_name: str, error_message: str, original_data: Dict[str, Any] = None): + """에러 로그""" + log_entry = { + "timestamp": datetime.now().isoformat(), + "service_name": service_name, + "error": error_message, + "original_data": original_data + } + + error_log_path = os.path.join( + settings.log_directory, settings.error_log_filename) + with open(error_log_path, "a", encoding="utf-8") as f: + f.write(json.dumps(log_entry, ensure_ascii=False) + "\n") + + print(f"❌ ERROR: {service_name} - {error_message}") + + +# 글로벌 로거 인스턴스 +anomaly_logger = AnomalyLogger() diff --git a/services/painting-surface-data-simulator-service/pytest.ini b/services/painting-surface-data-simulator-service/pytest.ini new file mode 100644 index 0000000..2c9da40 --- /dev/null +++ b/services/painting-surface-data-simulator-service/pytest.ini @@ -0,0 +1,5 @@ +[pytest] +pythonpath = . +testpaths = tests +addopts = -v +asyncio_mode = auto \ No newline at end of file diff --git a/services/painting-surface-data-simulator-service/requirements.txt b/services/painting-surface-data-simulator-service/requirements.txt new file mode 100644 index 0000000..86a5379 --- /dev/null +++ b/services/painting-surface-data-simulator-service/requirements.txt @@ -0,0 +1,33 @@ +# Core serving dependencies +fastapi==0.110.0 +uvicorn==0.29.0 +python-multipart==0.0.9 +python-dotenv==1.0.1 +pydantic-settings==2.1.0 + +# Azure Storage dependencies +azure-storage-blob==12.19.0 +aiohttp==3.9.1 + +# Scheduler dependencies +APScheduler==3.10.4 + +# Testing dependencies +pytest==7.4.3 +pytest-asyncio==0.21.1 +pytest-cov>=4.1.0 +pytest-mock>=3.11.0 + + +httpx==0.27.2 + + +black>=23.0.0 +flake8>=6.0.0 +mypy>=1.5.0 + + +coverage>=7.2.0 + + +pre-commit>=3.3.0 diff --git a/services/painting-surface-data-simulator-service/tests/conftest.py b/services/painting-surface-data-simulator-service/tests/conftest.py new file mode 100644 index 0000000..7580f95 --- /dev/null +++ b/services/painting-surface-data-simulator-service/tests/conftest.py @@ -0,0 +1,109 @@ +import pytest +import asyncio +import os +import sys +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock + +# pytest 설정 (pytest.ini와 동일한 효과) +def pytest_configure(config): + """pytest 설정""" + config.addinivalue_line( + "markers", "asyncio: 비동기 테스트" + ) + config.addinivalue_line( + "markers", "unit: 단위 테스트" + ) + config.addinivalue_line( + "markers", "integration: 통합 테스트" + ) + +def pytest_collection_modifyitems(config, items): + """테스트 아이템 수정""" + for item in items: + # 비동기 테스트에 자동으로 asyncio 마커 추가 + if asyncio.iscoroutinefunction(item.function): + item.add_marker(pytest.mark.asyncio) + + +@pytest.fixture +def event_loop(): + """비동기 테스트를 위한 이벤트 루프 fixture""" + loop = asyncio.new_event_loop() + yield loop + loop.close() + + +@pytest.fixture +def mock_settings(): + """테스트용 설정 모킹""" + # MagicMock을 사용해서 Settings 객체를 모킹 + mock = MagicMock() + mock.azure_connection_string = "test_connection_string" + mock.azure_container_name = "test-container" + mock.painting_data_folder = "test-painting" + mock.scheduler_interval_minutes = 1 + mock.batch_size = 5 + mock.painting_model_url = "http://test-model:8002" + mock.model_service_url = "http://test-model:8002" # 프로퍼티 추가 + mock.log_directory = "test-logs" + mock.log_filename = "test_log.json" + mock.error_log_filename = "test_error.json" + mock.http_timeout = 5 + mock.max_retries = 2 + return mock + + +@pytest.fixture +def mock_azure_storage(): + """Azure Storage 서비스 모킹""" + mock_storage = AsyncMock() + mock_storage.connect = AsyncMock() + mock_storage.disconnect = AsyncMock() + mock_storage.list_data_files = AsyncMock(return_value=["test1.jpg", "test2.jpg"]) + mock_storage.read_image_data = AsyncMock(return_value=b"test_image_data") + mock_storage.simulate_painting_surface_data = AsyncMock(return_value={ + "images": ["test1.jpg", "test2.jpg"], + "metadata": {"source": "test"} + }) + return mock_storage + + +@pytest.fixture +def mock_model_client(): + """모델 클라이언트 모킹""" + mock_client = AsyncMock() + mock_client.predict_painting_surface_data = AsyncMock(return_value={ + "status": "normal", + "defect_count": 0, + "total_count": 2 + }) + mock_client.health_check = AsyncMock(return_value=True) + return mock_client + + +@pytest.fixture +def mock_scheduler(): + """스케줄러 서비스 모킹""" + mock_scheduler = AsyncMock() + mock_scheduler.is_running = False + mock_scheduler.start = AsyncMock() + mock_scheduler.stop = AsyncMock() + mock_scheduler.get_status = MagicMock(return_value={ + "running": False, + "started_at": None, + "jobs": [] + }) + return mock_scheduler + + +@pytest.fixture +def test_log_directory(): + """테스트용 로그 디렉토리""" + test_dir = "test-logs" + os.makedirs(test_dir, exist_ok=True) + yield test_dir + # 테스트 후 정리 + if os.path.exists(test_dir): + import shutil + shutil.rmtree(test_dir) diff --git a/services/painting-surface-data-simulator-service/tests/run_tests.py b/services/painting-surface-data-simulator-service/tests/run_tests.py new file mode 100644 index 0000000..bb18846 --- /dev/null +++ b/services/painting-surface-data-simulator-service/tests/run_tests.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +""" +테스트 실행 스크립트 +사용법: python run_tests.py [옵션] +""" + +import subprocess +import sys +import os + + +def run_command(command, description): + """명령어 실행 및 결과 출력""" + print(f"\n{'='*60}") + print(f"🚀 {description}") + print(f"{'='*60}") + + try: + result = subprocess.run(command, shell=True, check=True, capture_output=True, text=True) + print("✅ 성공!") + if result.stdout: + print(result.stdout) + return True + except subprocess.CalledProcessError as e: + print(f"❌ 실패: {e}") + if e.stdout: + print("STDOUT:", e.stdout) + if e.stderr: + print("STDERR:", e.stderr) + return False + + +def main(): + """메인 함수""" + print("🧪 Painting Surface Data Simulator Service 테스트 실행") + print("=" * 60) + + # 현재 디렉토리 확인 + if not os.path.exists("tests"): + print("❌ tests 폴더를 찾을 수 없습니다.") + print(" 이 스크립트는 프로젝트 루트에서 실행해야 합니다.") + sys.exit(1) + + # 테스트 의존성 설치 확인 + print("📦 테스트 의존성 확인 중...") + try: + import pytest + import httpx + print("✅ 필요한 패키지가 설치되어 있습니다.") + except ImportError as e: + print(f"❌ 필요한 패키지가 설치되지 않았습니다: {e}") + print(" 다음 명령어로 설치하세요:") + print(" pip install -r requirements-dev.txt") + sys.exit(1) + + # 테스트 실행 + success = True + + # 1. 기본 테스트 실행 + success &= run_command( + "py -3.10 -m pytest tests/ -v", + "기본 테스트 실행" + ) + + # 2. 커버리지 테스트 실행 + success &= run_command( + "py -3.10 -m pytest tests/ --cov=app --cov-report=term-missing", + "코드 커버리지 테스트 실행" + ) + + # 3. 특정 테스트 파일 실행 (예시) + if success: + print("\n📋 개별 테스트 파일 실행 예시:") + print(" py -3.10 -m pytest tests/test_settings.py -v") + print(" py -3.10 -m pytest tests/test_logger.py -v") + print(" py -3.10 -m pytest tests/test_azure_storage.py -v") + print(" py -3.10 -m pytest tests/test_model_client.py -v") + print(" py -3.10 -m pytest tests/test_scheduler_service.py -v") + print(" py -3.10 -m pytest tests/test_simulator_router.py -v") + print(" py -3.10 -m pytest tests/test_main.py -v") + + # 결과 요약 + print(f"\n{'='*60}") + if success: + print("🎉 모든 테스트가 성공적으로 실행되었습니다!") + else: + print("⚠️ 일부 테스트 실행에 실패했습니다.") + print(" 위의 오류 메시지를 확인하고 수정하세요.") + print(f"{'='*60}") + + +if __name__ == "__main__": + main() diff --git a/services/painting-surface-data-simulator-service/tests/test_azure_storage.py b/services/painting-surface-data-simulator-service/tests/test_azure_storage.py new file mode 100644 index 0000000..bdb56ad --- /dev/null +++ b/services/painting-surface-data-simulator-service/tests/test_azure_storage.py @@ -0,0 +1,293 @@ +import pytest +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch +from azure.core.exceptions import AzureError, ClientAuthenticationError +from app.services.azure_storage import AzureStorageService + + +class TestAzureStorageService: + """Azure Storage 서비스 테스트""" + + def setup_method(self): + """각 테스트 메서드 실행 전 설정""" + self.test_connection_string = "test_connection_string" + self.test_container_name = "test-container" + self.test_data_folder = "test-painting" + + @patch('app.services.azure_storage.settings') + def test_azure_storage_initialization(self, mock_settings): + """Azure Storage 서비스 초기화 테스트""" + mock_settings.azure_connection_string = self.test_connection_string + mock_settings.azure_container_name = self.test_container_name + mock_settings.painting_data_folder = self.test_data_folder + + storage_service = AzureStorageService() + + assert storage_service.connection_string == self.test_connection_string + assert storage_service.container_name == self.test_container_name + assert storage_service.client is None + assert storage_service.image_index == 0 + + @pytest.mark.asyncio + @patch('app.services.azure_storage.settings') + @patch('app.services.azure_storage.BlobServiceClient') + async def test_connect_success(self, mock_blob_service_client, mock_settings): + """연결 성공 테스트""" + mock_settings.azure_connection_string = self.test_connection_string + mock_settings.azure_container_name = self.test_container_name + + # Mock 클라이언트 설정 + mock_client = MagicMock() + mock_container_client = AsyncMock() + mock_properties = MagicMock() + mock_properties.name = self.test_container_name + mock_properties.created_on = "2024-01-01" + + mock_container_client.get_container_properties = AsyncMock(return_value=mock_properties) + mock_client.get_container_client.return_value = mock_container_client + mock_blob_service_client.from_connection_string.return_value = mock_client + + storage_service = AzureStorageService() + + # 연결 테스트 + await storage_service.connect() + + assert storage_service.client == mock_client + mock_blob_service_client.from_connection_string.assert_called_once_with( + self.test_connection_string + ) + + @pytest.mark.asyncio + @patch('app.services.azure_storage.settings') + @patch('app.services.azure_storage.BlobServiceClient') + async def test_connect_failure_no_connection_string(self, mock_blob_service_client, mock_settings): + """연결 문자열 없음 실패 테스트""" + mock_settings.azure_connection_string = None + + storage_service = AzureStorageService() + + with pytest.raises(ValueError, match="Azure connection string이 설정되지 않았습니다."): + await storage_service.connect() + + @pytest.mark.asyncio + @patch('app.services.azure_storage.settings') + @patch('app.services.azure_storage.BlobServiceClient') + async def test_connect_failure_authentication_error(self, mock_blob_service_client, mock_settings): + """인증 실패 테스트""" + mock_settings.azure_connection_string = self.test_connection_string + + # 인증 오류 시뮬레이션 + mock_blob_service_client.from_connection_string.side_effect = ClientAuthenticationError("Auth failed") + + storage_service = AzureStorageService() + + with pytest.raises(ClientAuthenticationError): + await storage_service.connect() + + @pytest.mark.asyncio + @patch('app.services.azure_storage.settings') + @patch('app.services.azure_storage.BlobServiceClient') + async def test_connect_failure_general_error(self, mock_blob_service_client, mock_settings): + """일반 오류 실패 테스트""" + mock_settings.azure_connection_string = self.test_connection_string + + # 일반 오류 시뮬레이션 + mock_blob_service_client.from_connection_string.side_effect = Exception("General error") + + storage_service = AzureStorageService() + + with pytest.raises(Exception, match="General error"): + await storage_service.connect() + + @pytest.mark.asyncio + @patch('app.services.azure_storage.settings') + async def test_disconnect(self, mock_settings): + """연결 종료 테스트""" + mock_settings.azure_connection_string = self.test_connection_string + + storage_service = AzureStorageService() + storage_service.client = AsyncMock() + + await storage_service.disconnect() + + storage_service.client.close.assert_called_once() + + @pytest.mark.asyncio + @patch('app.services.azure_storage.settings') + @patch('app.services.azure_storage.BlobServiceClient') + async def test_list_data_files_success(self, mock_blob_service_client, mock_settings): + """데이터 파일 목록 조회 성공 테스트""" + mock_settings.azure_connection_string = self.test_connection_string + mock_settings.azure_container_name = self.test_container_name + mock_settings.painting_data_folder = self.test_data_folder + + # Mock 클라이언트 설정 + mock_client = MagicMock() + mock_container_client = AsyncMock() + + # Mock blob 객체들 + mock_blob1 = MagicMock() + mock_blob1.name = f"{self.test_data_folder}/image1.jpg" + mock_blob2 = MagicMock() + mock_blob2.name = f"{self.test_data_folder}/image2.png" + mock_blob3 = MagicMock() + mock_blob3.name = f"{self.test_data_folder}/document.txt" # 이미지가 아닌 파일 + + # list_blobs는 비동기 이터레이터를 반환해야 함 + async def mock_list_blobs(*args, **kwargs): + for blob in [mock_blob1, mock_blob2, mock_blob3]: + yield blob + mock_container_client.list_blobs = mock_list_blobs + mock_client.get_container_client.return_value = mock_container_client + mock_blob_service_client.from_connection_string.return_value = mock_client + + storage_service = AzureStorageService() + + # 파일 목록 조회 + files = await storage_service.list_data_files() + + expected_files = [f"{self.test_data_folder}/image1.jpg", f"{self.test_data_folder}/image2.png"] + assert files == expected_files + + @pytest.mark.asyncio + @patch('app.services.azure_storage.settings') + @patch('app.services.azure_storage.BlobServiceClient') + async def test_list_data_files_authentication_error(self, mock_blob_service_client, mock_settings): + """인증 오류로 인한 파일 목록 조회 실패 테스트""" + mock_settings.azure_connection_string = self.test_connection_string + mock_settings.azure_container_name = self.test_container_name + + # Mock 클라이언트 설정 + mock_client = MagicMock() + mock_container_client = MagicMock() + mock_container_client.list_blobs.side_effect = ClientAuthenticationError("Auth failed") + + mock_client.get_container_client.return_value = mock_container_client + mock_blob_service_client.from_connection_string.return_value = mock_client + + storage_service = AzureStorageService() + + # 파일 목록 조회 (인증 오류) + files = await storage_service.list_data_files() + + assert files == [] + + @pytest.mark.asyncio + @patch('app.services.azure_storage.settings') + @patch('app.services.azure_storage.BlobServiceClient') + async def test_list_data_files_general_error(self, mock_blob_service_client, mock_settings): + """일반 오류로 인한 파일 목록 조회 실패 테스트""" + mock_settings.azure_connection_string = self.test_connection_string + mock_settings.azure_container_name = self.test_container_name + + # Mock 클라이언트 설정 + mock_client = MagicMock() + mock_container_client = MagicMock() + mock_container_client.list_blobs.side_effect = Exception("General error") + + mock_client.get_container_client.return_value = mock_container_client + mock_blob_service_client.from_connection_string.return_value = mock_client + + storage_service = AzureStorageService() + + # 파일 목록 조회 (일반 오류) + files = await storage_service.list_data_files() + + assert files == [] + + @pytest.mark.asyncio + @patch('app.services.azure_storage.settings') + @patch('app.services.azure_storage.BlobServiceClient') + async def test_read_image_data_success(self, mock_blob_service_client, mock_settings): + """이미지 데이터 읽기 성공 테스트""" + mock_settings.azure_connection_string = self.test_connection_string + mock_settings.azure_container_name = self.test_container_name + + # Mock 클라이언트 설정 + mock_client = MagicMock() + mock_blob_client = AsyncMock() + mock_blob_data = MagicMock() + # readall은 Azure SDK에서 비동기 함수임 + mock_blob_data.readall = AsyncMock(return_value=b"test_image_data") + + mock_blob_client.download_blob = AsyncMock(return_value=mock_blob_data) + mock_client.get_blob_client.return_value = mock_blob_client + mock_blob_service_client.from_connection_string.return_value = mock_client + + storage_service = AzureStorageService() + + # Azure Storage 연결 Mock 설정 + storage_service.client = mock_client + storage_service.image_index = 0 # image_index를 실제 정수로 설정 + + # 이미지 데이터 읽기 + image_data = await storage_service.read_image_data("test.jpg") + + assert image_data == b"test_image_data" + + @pytest.mark.asyncio + @patch('app.services.azure_storage.settings') + @patch('app.services.azure_storage.BlobServiceClient') + async def test_read_image_data_failure(self, mock_blob_service_client, mock_settings): + """이미지 데이터 읽기 실패 테스트""" + mock_settings.azure_connection_string = self.test_connection_string + mock_settings.azure_container_name = self.test_container_name + + # Mock 클라이언트 설정 + mock_client = MagicMock() + mock_blob_client = MagicMock() + mock_blob_client.download_blob.side_effect = Exception("Download failed") + + mock_client.get_blob_client.return_value = mock_blob_client + mock_blob_service_client.from_connection_string.return_value = mock_client + + storage_service = AzureStorageService() + + # 이미지 데이터 읽기 (실패) + image_data = await storage_service.read_image_data("test.jpg") + + assert image_data is None + + @pytest.mark.asyncio + @patch('app.services.azure_storage.settings') + @patch('app.services.azure_storage.BlobServiceClient') + async def test_simulate_painting_surface_data(self, mock_blob_service_client, mock_settings): + """도장 표면 데이터 시뮬레이션 테스트""" + mock_settings.azure_connection_string = self.test_connection_string + mock_settings.azure_container_name = self.test_container_name + mock_settings.painting_data_folder = self.test_data_folder + mock_settings.batch_size = 10 # batch_size를 실제 정수로 설정 + + # Mock 클라이언트 설정 + mock_client = MagicMock() + mock_container_client = AsyncMock() + + # Mock blob 객체들 + mock_blob1 = MagicMock() + mock_blob1.name = f"{self.test_data_folder}/image1.jpg" + mock_blob2 = MagicMock() + mock_blob2.name = f"{self.test_data_folder}/image2.png" + + # list_blobs는 비동기 이터레이터를 반환해야 함 + async def mock_list_blobs(*args, **kwargs): + for blob in [mock_blob1, mock_blob2]: + yield blob + mock_container_client.list_blobs = mock_list_blobs + mock_client.get_container_client.return_value = mock_container_client + mock_blob_service_client.from_connection_string.return_value = mock_client + + storage_service = AzureStorageService() + + # Azure Storage 연결 Mock 설정 + storage_service.client = mock_client + storage_service.image_index = 0 # image_index를 실제 정수로 설정 + + # 데이터 시뮬레이션 + simulated_data = await storage_service.simulate_painting_surface_data() + + assert "images" in simulated_data + assert "total_images" in simulated_data + assert "batch_size" in simulated_data + assert len(simulated_data["images"]) == 2 + assert simulated_data["images"][0] == f"{self.test_data_folder}/image1.jpg" + assert simulated_data["images"][1] == f"{self.test_data_folder}/image2.png" diff --git a/services/painting-surface-data-simulator-service/tests/test_logger.py b/services/painting-surface-data-simulator-service/tests/test_logger.py new file mode 100644 index 0000000..262379a --- /dev/null +++ b/services/painting-surface-data-simulator-service/tests/test_logger.py @@ -0,0 +1,194 @@ +import pytest +import os +import json +import tempfile +from unittest.mock import patch, MagicMock +from app.utils.logger import AnomalyLogger + + +class TestAnomalyLogger: + """AnomalyLogger 클래스 테스트""" + + def setup_method(self): + """각 테스트 메서드 실행 전 설정""" + self.test_log_dir = "test-logs" + self.test_log_file = os.path.join(self.test_log_dir, "test_log.json") + self.test_error_file = os.path.join(self.test_log_dir, "test_error.json") + + def teardown_method(self): + """각 테스트 메서드 실행 후 정리""" + # 테스트 로그 파일 정리 + if os.path.exists(self.test_log_dir): + import shutil + shutil.rmtree(self.test_log_dir) + + @patch('app.utils.logger.settings') + def test_logger_initialization(self, mock_settings): + """로거 초기화 테스트""" + mock_settings.log_directory = self.test_log_dir + mock_settings.log_filename = "test_log.json" + + logger = AnomalyLogger() + + assert logger.log_file_path == self.test_log_file + assert os.path.exists(self.test_log_dir) + + @patch('app.utils.logger.settings') + def test_log_anomaly(self, mock_settings): + """결함 탐지 로그 테스트""" + mock_settings.log_directory = self.test_log_dir + mock_settings.log_filename = "test_log.json" + + logger = AnomalyLogger() + + # 테스트 데이터 + service_name = "test-service" + prediction_result = { + "defect_count": 2, + "total_count": 5, + "status": "anomaly" + } + original_data = {"image_path": "test.jpg"} + + # 로그 기록 + logger.log_anomaly(service_name, prediction_result, original_data) + + # 로그 파일 확인 + assert os.path.exists(self.test_log_file) + + with open(self.test_log_file, 'r', encoding='utf-8') as f: + log_entry = json.loads(f.readline().strip()) + + assert log_entry["service_name"] == service_name + assert log_entry["prediction"] == prediction_result + assert log_entry["original_data"] == original_data + assert "timestamp" in log_entry + + @patch('app.utils.logger.settings') + def test_log_normal_processing(self, mock_settings): + """정상 처리 로그 테스트""" + mock_settings.log_directory = self.test_log_dir + mock_settings.log_filename = "test_log.json" + + logger = AnomalyLogger() + + # 테스트 데이터 + service_name = "test-service" + prediction_result = { + "defect_count": 0, + "total_count": 5, + "status": "normal" + } + + # 콘솔 출력 모킹 + with patch('builtins.print') as mock_print: + logger.log_normal_processing(service_name, prediction_result) + + # 콘솔 출력 확인 + mock_print.assert_called() + + @patch('app.utils.logger.settings') + def test_log_error(self, mock_settings): + """에러 로그 테스트""" + mock_settings.log_directory = self.test_log_dir + mock_settings.error_log_filename = "test_error.json" + + logger = AnomalyLogger() + + # 테스트 데이터 + service_name = "test-service" + error_message = "Test error message" + original_data = {"image_path": "test.jpg"} + + # 에러 로그 기록 + logger.log_error(service_name, error_message, original_data) + + # 에러 로그 파일 확인 + assert os.path.exists(self.test_error_file) + + with open(self.test_error_file, 'r', encoding='utf-8') as f: + log_entry = json.loads(f.readline().strip()) + + assert log_entry["service_name"] == service_name + assert log_entry["error"] == error_message + assert log_entry["original_data"] == original_data + assert "timestamp" in log_entry + + @patch('app.utils.logger.settings') + def test_log_error_without_original_data(self, mock_settings): + """원본 데이터 없이 에러 로그 테스트""" + mock_settings.log_directory = self.test_log_dir + mock_settings.error_log_filename = "test_error.json" + + logger = AnomalyLogger() + + # 테스트 데이터 + service_name = "test-service" + error_message = "Test error message" + + # 에러 로그 기록 (원본 데이터 없음) + logger.log_error(service_name, error_message) + + # 에러 로그 파일 확인 + assert os.path.exists(self.test_error_file) + + with open(self.test_error_file, 'r', encoding='utf-8') as f: + log_entry = json.loads(f.readline().strip()) + + assert log_entry["service_name"] == service_name + assert log_entry["error"] == error_message + assert log_entry["original_data"] is None + + @patch('app.utils.logger.settings') + def test_multiple_log_entries(self, mock_settings): + """여러 로그 항목 테스트""" + mock_settings.log_directory = self.test_log_dir + mock_settings.log_filename = "test_log.json" + + logger = AnomalyLogger() + + # 여러 로그 기록 + for i in range(3): + service_name = f"service-{i}" + prediction_result = { + "defect_count": i, + "total_count": 5, + "status": "anomaly" if i > 0 else "normal" + } + original_data = {"image_path": f"test_{i}.jpg"} + + logger.log_anomaly(service_name, prediction_result, original_data) + + # 로그 파일에서 모든 항목 확인 + with open(self.test_log_file, 'r', encoding='utf-8') as f: + lines = f.readlines() + + assert len(lines) == 3 + + for i, line in enumerate(lines): + log_entry = json.loads(line.strip()) + assert log_entry["service_name"] == f"service-{i}" + assert log_entry["prediction"]["defect_count"] == i + + @patch('app.utils.logger.settings') + def test_logger_with_existing_directory(self, mock_settings): + """기존 디렉토리가 있는 경우 테스트""" + mock_settings.log_directory = self.test_log_dir + mock_settings.log_filename = "test_log.json" + + # 디렉토리 미리 생성 + os.makedirs(self.test_log_dir, exist_ok=True) + + logger = AnomalyLogger() + + # 로거가 정상적으로 작동하는지 확인 + assert logger.log_file_path == self.test_log_file + + # 로그 기록 테스트 + service_name = "test-service" + prediction_result = {"defect_count": 1, "total_count": 5, "status": "anomaly"} + original_data = {"image_path": "test.jpg"} + + logger.log_anomaly(service_name, prediction_result, original_data) + + assert os.path.exists(self.test_log_file) diff --git a/services/painting-surface-data-simulator-service/tests/test_main.py b/services/painting-surface-data-simulator-service/tests/test_main.py new file mode 100644 index 0000000..557d6ce --- /dev/null +++ b/services/painting-surface-data-simulator-service/tests/test_main.py @@ -0,0 +1,268 @@ +import pytest +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch +from fastapi.testclient import TestClient +from app.main import app, lifespan + + +class TestMainApplication: + """메인 애플리케이션 테스트""" + + def setup_method(self): + """각 테스트 메서드 실행 전 설정""" + self.client = TestClient(app) + + def test_root_endpoint(self): + """루트 엔드포인트 테스트""" + response = self.client.get("/") + + assert response.status_code == 200 + data = response.json() + + assert data["service"] == "Painting Surface Data Simulator Service" + assert data["version"] == "1.0.0" + assert data["status"] == "running" + assert data["target_model"] == "painting-surface-defect-detection" + assert "scheduler_status" in data + assert "azure_storage" in data + + def test_health_check_endpoint(self): + """헬스 체크 엔드포인트 테스트""" + response = self.client.get("/health") + + assert response.status_code == 200 + data = response.json() + + assert data["status"] == "healthy" + + def test_simulator_router_included(self): + """시뮬레이터 라우터 포함 여부 테스트""" + response = self.client.get("/simulator/status") + + # 라우터가 포함되어 있다면 200 또는 적절한 응답을 받아야 함 + # 실제 구현에 따라 응답이 달라질 수 있음 + assert response.status_code in [200, 404, 500] + + def test_test_connection_router_included(self): + """테스트 연결 라우터 포함 여부 테스트""" + response = self.client.post("/test/azure-storage-connection") + + # 라우터가 포함되어 있다면 적절한 응답을 받아야 함 + # 실제 구현에 따라 응답이 달라질 수 있음 + assert response.status_code in [200, 404, 500] + + @pytest.mark.asyncio + @patch('app.main.settings') + @patch('app.main.azure_storage') + async def test_lifespan_startup_success(self, mock_azure_storage, mock_settings): + """애플리케이션 시작 성공 테스트""" + # Mock 설정 + mock_settings.azure_connection_string = "test_connection_string" + mock_settings.azure_container_name = "test-container" + mock_settings.painting_data_folder = "test-painting" + mock_settings.painting_model_url = "http://test-model:8002" + mock_settings.log_directory = "test-logs" + mock_settings.scheduler_interval_minutes = 1 + + # Mock Azure Storage 연결 성공 + mock_azure_storage.connect = AsyncMock() + + # 콘솔 출력 모킹 + with patch('builtins.print') as mock_print: + # 애플리케이션 시작 + async with lifespan(app): + pass + + # 시작 메시지 출력 확인 + mock_print.assert_any_call("🚀 Painting Surface Defect Simulator Service 시작 중...") + mock_print.assert_any_call("🔍 환경 변수 확인 중...") + mock_print.assert_any_call("✅ Azure Storage 연결 성공!") + + @pytest.mark.asyncio + @patch('app.main.settings') + @patch('app.main.azure_storage') + async def test_lifespan_startup_no_connection_string(self, mock_azure_storage, mock_settings): + """연결 문자열 없음 시작 테스트""" + # Mock 설정 (연결 문자열 없음) + mock_settings.azure_connection_string = None + mock_settings.azure_container_name = "test-container" + mock_settings.painting_data_folder = "test-painting" + mock_settings.painting_model_url = "http://test-model:8002" + mock_settings.log_directory = "test-logs" + mock_settings.scheduler_interval_minutes = 1 + + # 콘솔 출력 모킹 + with patch('builtins.print') as mock_print: + # 애플리케이션 시작 + async with lifespan(app): + pass + + # 경고 메시지 출력 확인 + mock_print.assert_any_call("⚠️ AZURE_CONNECTION_STRING 환경 변수가 설정되지 않았습니다.") + + @pytest.mark.asyncio + @patch('app.main.settings') + @patch('app.main.azure_storage') + async def test_lifespan_startup_azure_connection_failure(self, mock_azure_storage, mock_settings): + """Azure Storage 연결 실패 시작 테스트""" + # Mock 설정 + mock_settings.azure_connection_string = "test_connection_string" + mock_settings.azure_container_name = "test-container" + mock_settings.painting_data_folder = "test-painting" + mock_settings.painting_model_url = "http://test-model:8002" + mock_settings.log_directory = "test-logs" + mock_settings.scheduler_interval_minutes = 1 + + # Mock Azure Storage 연결 실패 + mock_azure_storage.connect = AsyncMock(side_effect=Exception("Connection failed")) + + # 콘솔 출력 모킹 + with patch('builtins.print') as mock_print: + # 애플리케이션 시작 + async with lifespan(app): + pass + + # 연결 실패 메시지 출력 확인 + mock_print.assert_any_call("❌ Azure Storage 연결 실패: Connection failed") + + @pytest.mark.asyncio + @patch('app.main.settings') + @patch('app.main.azure_storage') + @patch('app.main.simulator_scheduler') + async def test_lifespan_shutdown(self, mock_scheduler, mock_azure_storage, mock_settings): + """애플리케이션 종료 테스트""" + # Mock 설정 + mock_settings.azure_connection_string = "test_connection_string" + mock_settings.azure_container_name = "test-container" + mock_settings.painting_data_folder = "test-painting" + mock_settings.painting_model_url = "http://test-model:8002" + mock_settings.log_directory = "test-logs" + mock_settings.scheduler_interval_minutes = 1 + + # Mock Azure Storage 연결 성공 + mock_azure_storage.connect = AsyncMock() + + # Mock 스케줄러가 실행 중 + mock_scheduler.is_running = True + mock_scheduler.stop = AsyncMock() + + # 콘솔 출력 모킹 + with patch('builtins.print') as mock_print: + # 애플리케이션 시작 및 종료 + async with lifespan(app): + pass + + # 종료 메시지 출력 확인 + mock_print.assert_any_call("🛑 Painting Surface Defect Simulator Service 종료 중...") + + # 스케줄러 중지 호출 확인 + mock_scheduler.stop.assert_called_once() + + @pytest.mark.asyncio + @patch('app.main.settings') + @patch('app.main.azure_storage') + @patch('app.main.simulator_scheduler') + async def test_lifespan_shutdown_scheduler_not_running(self, mock_scheduler, mock_azure_storage, mock_settings): + """스케줄러가 실행 중이 아닌 경우 종료 테스트""" + # Mock 설정 + mock_settings.azure_connection_string = "test_connection_string" + mock_settings.azure_container_name = "test-container" + mock_settings.painting_data_folder = "test-painting" + mock_settings.painting_model_url = "http://test-model:8002" + mock_settings.log_directory = "test-logs" + mock_settings.scheduler_interval_minutes = 1 + + # Mock Azure Storage 연결 성공 + mock_azure_storage.connect = AsyncMock() + + # Mock 스케줄러가 실행 중이 아님 + mock_scheduler.is_running = False + mock_scheduler.stop = AsyncMock() + + # 콘솔 출력 모킹 + with patch('builtins.print') as mock_print: + # 애플리케이션 시작 및 종료 + async with lifespan(app): + pass + + # 종료 메시지 출력 확인 + mock_print.assert_any_call("🛑 Painting Surface Defect Simulator Service 종료 중...") + + # 스케줄러 중지가 호출되지 않음 + mock_scheduler.stop.assert_not_called() + + @pytest.mark.asyncio + @patch('app.main.settings') + @patch('app.main.azure_storage') + async def test_lifespan_log_directory_creation(self, mock_azure_storage, mock_settings): + """로그 디렉토리 생성 테스트""" + # Mock 설정 + mock_settings.azure_connection_string = "test_connection_string" + mock_settings.azure_container_name = "test-container" + mock_settings.painting_data_folder = "test-painting" + mock_settings.painting_model_url = "http://test-model:8002" + mock_settings.log_directory = "test-logs" + mock_settings.scheduler_interval_minutes = 1 + + # Mock Azure Storage 연결 성공 + mock_azure_storage.connect = AsyncMock() + + # Mock os.makedirs + with patch('os.makedirs') as mock_makedirs: + # 애플리케이션 시작 + async with lifespan(app): + pass + + # 로그 디렉토리 생성 호출 확인 + mock_makedirs.assert_called_once_with("test-logs", exist_ok=True) + + def test_app_metadata(self): + """애플리케이션 메타데이터 테스트""" + assert app.title == "Painting Surface Defect Simulator Service" + assert app.description == "도장 표면 결함 탐지 모델을 위한 실시간 데이터 시뮬레이터" + assert app.version == "1.0.0" + + def test_app_has_lifespan(self): + """애플리케이션에 lifespan이 설정되어 있는지 테스트""" + assert app.router.lifespan_context == lifespan + + def test_app_includes_routers(self): + """애플리케이션에 라우터가 포함되어 있는지 테스트""" + # 시뮬레이터 라우터 확인 (경로에 /simulator가 포함된 라우트) + simulator_routes = [route for route in app.routes if hasattr(route, 'path') and '/simulator' in route.path] + test_routes = [route for route in app.routes if hasattr(route, 'path') and '/test' in route.path] + + assert len(simulator_routes) > 0, "시뮬레이터 라우터가 포함되지 않았습니다" + assert len(test_routes) > 0, "테스트 연결 라우터가 포함되지 않았습니다" + + + + @pytest.mark.asyncio + @patch('app.main.settings') + @patch('app.main.azure_storage') + async def test_lifespan_environment_variables_display(self, mock_azure_storage, mock_settings): + """환경 변수 표시 테스트""" + # Mock 설정 + mock_settings.azure_connection_string = "test_connection_string" + mock_settings.azure_container_name = "test-container" + mock_settings.painting_data_folder = "test-painting" + mock_settings.painting_model_url = "http://test-model:8002" + mock_settings.log_directory = "test-logs" + mock_settings.scheduler_interval_minutes = 1 + + # Mock Azure Storage 연결 성공 + mock_azure_storage.connect = AsyncMock() + + # 콘솔 출력 모킹 + with patch('builtins.print') as mock_print: + # 애플리케이션 시작 + async with lifespan(app): + pass + + # 환경 변수 표시 확인 + mock_print.assert_any_call(" Azure Connection String: ✅ 설정됨") + mock_print.assert_any_call(" Azure Container: test-container") + mock_print.assert_any_call(" Painting Data Folder: test-painting") + mock_print.assert_any_call(" Model URL: http://test-model:8002") + mock_print.assert_any_call("🔧 스케줄러 간격: 1분") + mock_print.assert_any_call("🎯 대상 서비스: 도장 표면 결함탐지 모델") diff --git a/services/painting-surface-data-simulator-service/tests/test_model_client.py b/services/painting-surface-data-simulator-service/tests/test_model_client.py new file mode 100644 index 0000000..0dbbcbc --- /dev/null +++ b/services/painting-surface-data-simulator-service/tests/test_model_client.py @@ -0,0 +1,312 @@ +import pytest +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch +import httpx +from app.services.model_client import PaintingSurfaceModelClient + + +class TestPaintingSurfaceModelClient: + """도장 표면 모델 클라이언트 테스트""" + + def setup_method(self): + """각 테스트 메서드 실행 전 설정""" + self.test_service_url = "http://test-model:8002" + self.test_timeout = 5 + self.test_max_retries = 2 + + @patch('app.services.model_client.settings') + def test_model_client_initialization(self, mock_settings): + """모델 클라이언트 초기화 테스트""" + mock_settings.http_timeout = self.test_timeout + mock_settings.max_retries = self.test_max_retries + mock_settings.model_service_url = self.test_service_url + + client = PaintingSurfaceModelClient() + + assert client.timeout.connect == self.test_timeout + assert client.max_retries == self.test_max_retries + assert client.service_name == "painting-surface" + assert client.service_url == self.test_service_url + + @pytest.mark.asyncio + @patch('app.services.model_client.settings') + async def test_predict_painting_surface_data_success(self, mock_settings): + """도장 표면 데이터 예측 성공 테스트""" + mock_settings.model_service_url = self.test_service_url + + client = PaintingSurfaceModelClient() + + # Mock Azure Storage 다운로드 + with patch.object(client, '_download_image_from_azure') as mock_download: + mock_download.return_value = b"test_image_data" + + # Mock HTTP 요청 + with patch.object(client, '_predict_with_file_upload') as mock_predict: + mock_predict.return_value = { + "predictions": [], + "status": "normal", + "defect_count": 0, + "total_count": 1 + } + + # Mock 상세 로깅 + with patch.object(client, '_log_detailed_prediction_result') as mock_log: + result = await client.predict_painting_surface_data(["test1.jpg", "test2.jpg"]) + + assert result is not None + assert "images" in result + assert "combined" in result + assert len(result["images"]) == 2 + assert result["combined"]["status"] == "normal" + assert result["combined"]["defect_count"] == 0 + assert result["combined"]["total_count"] == 2 + + @pytest.mark.asyncio + @patch('app.services.model_client.settings') + async def test_predict_painting_surface_data_no_service_url(self, mock_settings): + """서비스 URL 없음 테스트""" + mock_settings.model_service_url = None + + client = PaintingSurfaceModelClient() + + result = await client.predict_painting_surface_data(["test.jpg"]) + + assert result is None + + @pytest.mark.asyncio + @patch('app.services.model_client.settings') + async def test_predict_painting_surface_data_download_failure(self, mock_settings): + """이미지 다운로드 실패 테스트""" + mock_settings.model_service_url = self.test_service_url + + client = PaintingSurfaceModelClient() + + # Mock Azure Storage 다운로드 실패 + with patch.object(client, '_download_image_from_azure') as mock_download: + mock_download.return_value = None + + result = await client.predict_painting_surface_data(["test.jpg"]) + + assert result is None + + @pytest.mark.asyncio + @patch('app.services.model_client.settings') + async def test_predict_painting_surface_data_prediction_failure(self, mock_settings): + """예측 요청 실패 테스트""" + mock_settings.model_service_url = self.test_service_url + + client = PaintingSurfaceModelClient() + + # Mock Azure Storage 다운로드 성공 + with patch.object(client, '_download_image_from_azure') as mock_download: + mock_download.return_value = b"test_image_data" + + # Mock HTTP 요청 실패 + with patch.object(client, '_predict_with_file_upload') as mock_predict: + mock_predict.return_value = None + + result = await client.predict_painting_surface_data(["test.jpg"]) + + assert result is None + + @pytest.mark.asyncio + @patch('app.services.model_client.settings') + async def test_predict_painting_surface_data_exception_handling(self, mock_settings): + """예외 처리 테스트""" + mock_settings.model_service_url = self.test_service_url + + client = PaintingSurfaceModelClient() + + # Mock Azure Storage 다운로드에서 예외 발생 + with patch.object(client, '_download_image_from_azure') as mock_download: + mock_download.side_effect = Exception("Download error") + + result = await client.predict_painting_surface_data(["test.jpg"]) + + assert result is None + + def test_combine_painting_results_normal(self): + """정상 결과 조합 테스트""" + client = PaintingSurfaceModelClient() + + image_results = [ + {"predictions": [], "status": "normal"}, + {"predictions": [], "status": "normal"} + ] + + combined = client._combine_painting_results(image_results) + + assert combined["status"] == "normal" + assert combined["defect_count"] == 0 + assert combined["total_count"] == 2 + assert combined["defect_ratio"] == 0.0 + + def test_combine_painting_results_anomaly(self): + """결함 결과 조합 테스트""" + client = PaintingSurfaceModelClient() + + image_results = [ + {"predictions": [{"defect": "scratch"}], "status": "anomaly"}, + {"predictions": [], "status": "normal"} + ] + + combined = client._combine_painting_results(image_results) + + assert combined["status"] == "anomaly" + assert combined["defect_count"] == 1 + assert combined["total_count"] == 2 + assert combined["defect_ratio"] == 0.5 + + def test_combine_painting_results_empty(self): + """빈 결과 조합 테스트""" + client = PaintingSurfaceModelClient() + + combined = client._combine_painting_results([]) + + assert combined["status"] == "error" + assert "message" in combined + + @pytest.mark.asyncio + @patch('app.services.model_client.settings') + async def test_health_check_success(self, mock_settings): + """헬스 체크 성공 테스트""" + mock_settings.model_service_url = self.test_service_url + + client = PaintingSurfaceModelClient() + + # Mock HTTP 응답 + with patch('httpx.AsyncClient') as mock_client_class: + mock_client = MagicMock() + mock_response = MagicMock() + mock_response.status_code = 200 + mock_client.get = AsyncMock(return_value=mock_response) + mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) + + result = await client.health_check() + + assert result is True + + @pytest.mark.asyncio + @patch('app.services.model_client.settings') + async def test_health_check_failure(self, mock_settings): + """헬스 체크 실패 테스트""" + mock_settings.model_service_url = self.test_service_url + + client = PaintingSurfaceModelClient() + + # Mock HTTP 응답 실패 + with patch('httpx.AsyncClient') as mock_client_class: + mock_client = MagicMock() + mock_response = MagicMock() + mock_response.status_code = 500 + mock_client.get.return_value = mock_response + mock_client_class.return_value.__aenter__.return_value = mock_client + + result = await client.health_check() + + assert result is False + + @pytest.mark.asyncio + @patch('app.services.model_client.settings') + async def test_health_check_exception(self, mock_settings): + """헬스 체크 예외 테스트""" + mock_settings.model_service_url = self.test_service_url + + client = PaintingSurfaceModelClient() + + # Mock HTTP 요청에서 예외 발생 + with patch('httpx.AsyncClient') as mock_client_class: + mock_client = MagicMock() + mock_client.get.side_effect = Exception("Connection error") + mock_client_class.return_value.__aenter__.return_value = mock_client + + result = await client.health_check() + + assert result is False + + @pytest.mark.asyncio + @patch('app.services.model_client.settings') + async def test_health_check_no_service_url(self, mock_settings): + """서비스 URL 없음 헬스 체크 테스트""" + mock_settings.model_service_url = None + + client = PaintingSurfaceModelClient() + + result = await client.health_check() + + assert result is False + + @patch('app.services.model_client.settings') + def test_log_detailed_prediction_result(self, mock_settings): + """상세 예측 결과 로깅 테스트""" + mock_settings.model_service_url = self.test_service_url + + client = PaintingSurfaceModelClient() + + # 콘솔 출력 모킹 + with patch('builtins.print') as mock_print: + image_file = "test.jpg" + result = { + "predictions": [{"defect": "scratch", "confidence": 0.95}], + "status": "anomaly" + } + + client._log_detailed_prediction_result(image_file, result) + + # 콘솔 출력 확인 + mock_print.assert_called() + + @pytest.mark.asyncio + @patch('app.services.model_client.settings') + async def test_download_image_from_azure(self, mock_settings): + """Azure에서 이미지 다운로드 테스트""" + mock_settings.model_service_url = self.test_service_url + + client = PaintingSurfaceModelClient() + + # Mock Azure Storage 서비스 (메서드 자체를 Mock) + with patch.object(client, '_download_image_from_azure', new_callable=AsyncMock) as mock_download: + mock_download.return_value = b"test_image_data" + + result = await client._download_image_from_azure("test.jpg") + + assert result == b"test_image_data" + mock_download.assert_called_once_with("test.jpg") + + @pytest.mark.asyncio + @patch('app.services.model_client.settings') + async def test_predict_with_file_upload(self, mock_settings): + """파일 업로드 방식 예측 요청 테스트""" + mock_settings.model_service_url = self.test_service_url + mock_settings.max_retries = 2 # max_retries를 실제 정수로 설정 + + client = PaintingSurfaceModelClient() + + # Mock HTTP 요청 + with patch('httpx.AsyncClient') as mock_client_class: + mock_client = MagicMock() + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "predictions": [], + "status": "normal" + } + mock_client.post = AsyncMock(return_value=mock_response) + mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) + + image_data = b"test_image_data" + image_file = "test.jpg" + confidence_threshold = 0.5 + + result = await client._predict_with_file_upload( + "http://test-model:8002/predict/file", + image_data, + image_file, + confidence_threshold + ) + + assert result is not None + assert result["status"] == "normal" + mock_client.post.assert_called_once() diff --git a/services/painting-surface-data-simulator-service/tests/test_scheduler_service.py b/services/painting-surface-data-simulator-service/tests/test_scheduler_service.py new file mode 100644 index 0000000..fbc1744 --- /dev/null +++ b/services/painting-surface-data-simulator-service/tests/test_scheduler_service.py @@ -0,0 +1,336 @@ +import pytest +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch +from datetime import datetime +from app.services.scheduler_service import SimulatorScheduler + + +class TestSimulatorScheduler: + """시뮬레이터 스케줄러 테스트""" + + def setup_method(self): + """각 테스트 메서드 실행 전 설정""" + self.scheduler = SimulatorScheduler() + + def teardown_method(self): + """각 테스트 메서드 실행 후 정리""" + if self.scheduler.is_running: + # 이벤트 루프가 실행 중일 때만 create_task 사용 + try: + loop = asyncio.get_running_loop() + asyncio.create_task(self.scheduler.stop()) + except RuntimeError: + # 이벤트 루프가 실행 중이 아니면 무시 + pass + + def test_scheduler_initialization(self): + """스케줄러 초기화 테스트""" + assert self.scheduler.is_running is False + assert self.scheduler.scheduler is not None + + @pytest.mark.asyncio + @patch('app.services.scheduler_service.settings') + @patch('app.services.scheduler_service.painting_surface_model_client') + @patch('app.services.scheduler_service.azure_storage') + async def test_start_success(self, mock_azure_storage, mock_model_client, mock_settings): + """스케줄러 시작 성공 테스트""" + mock_settings.scheduler_interval_minutes = 1 + + # Mock 헬스 체크 성공 + mock_model_client.health_check = AsyncMock(return_value=True) + + # Mock Azure Storage 연결 + mock_azure_storage.connect = AsyncMock() + + # Mock 스케줄러 + mock_scheduler_instance = MagicMock() + self.scheduler.scheduler = mock_scheduler_instance + + # 스케줄러 시작 + await self.scheduler.start() + + assert self.scheduler.is_running is True + mock_scheduler_instance.add_job.assert_called_once() + mock_scheduler_instance.start.assert_called_once() + + @pytest.mark.asyncio + @patch('app.services.scheduler_service.settings') + @patch('app.services.scheduler_service.painting_surface_model_client') + async def test_start_already_running(self, mock_model_client, mock_settings): + """이미 실행 중인 스케줄러 시작 테스트""" + mock_settings.scheduler_interval_minutes = 1 + + # 이미 실행 중으로 설정 + self.scheduler.is_running = True + + # Mock 헬스 체크 성공 + mock_model_client.health_check = AsyncMock(return_value=True) + + # Mock Azure Storage 연결 + with patch('app.services.scheduler_service.azure_storage') as mock_azure: + mock_azure.connect = AsyncMock() + + # Mock 스케줄러 + mock_scheduler_instance = MagicMock() + self.scheduler.scheduler = mock_scheduler_instance + + # 스케줄러 시작 시도 + await self.scheduler.start() + + # 이미 실행 중이므로 작업이 추가되지 않음 + mock_scheduler_instance.add_job.assert_not_called() + + @pytest.mark.asyncio + @patch('app.services.scheduler_service.settings') + @patch('app.services.scheduler_service.painting_surface_model_client') + async def test_start_health_check_failure(self, mock_model_client, mock_settings): + """헬스 체크 실패로 인한 시작 실패 테스트""" + mock_settings.scheduler_interval_minutes = 1 + + # Mock 헬스 체크 실패 + mock_model_client.health_check = AsyncMock(return_value=False) + + # 스케줄러 시작 시도 (헬스 체크 실패로 예외 발생) + with pytest.raises(Exception, match="도장 표면 결함탐지 서비스가 비활성 상태입니다."): + await self.scheduler.start() + + @pytest.mark.asyncio + @patch('app.services.scheduler_service.settings') + @patch('app.services.scheduler_service.painting_surface_model_client') + async def test_start_exception_handling(self, mock_model_client, mock_settings): + """시작 중 예외 처리 테스트""" + mock_settings.scheduler_interval_minutes = 1 + + # Mock 헬스 체크 성공 + mock_model_client.health_check = AsyncMock(return_value=True) + + # Mock 스케줄러에서 예외 발생 + mock_scheduler_instance = MagicMock() + mock_scheduler_instance.add_job.side_effect = Exception("Scheduler error") + self.scheduler.scheduler = mock_scheduler_instance + + # 스케줄러 시작 시도 (예외 발생) + with pytest.raises(Exception, match="Scheduler error"): + await self.scheduler.start() + + @pytest.mark.asyncio + async def test_stop_not_running(self): + """실행 중이 아닌 스케줄러 중지 테스트""" + # 실행 중이 아닌 상태 + self.scheduler.is_running = False + + # Mock 스케줄러 + mock_scheduler_instance = MagicMock() + self.scheduler.scheduler = mock_scheduler_instance + + # 스케줄러 중지 + await self.scheduler.stop() + + # 이미 중지된 상태이므로 shutdown이 호출되지 않음 + mock_scheduler_instance.shutdown.assert_not_called() + + @pytest.mark.asyncio + async def test_stop_running(self): + """실행 중인 스케줄러 중지 테스트""" + # 실행 중인 상태 + self.scheduler.is_running = True + + # Mock 스케줄러 + mock_scheduler_instance = MagicMock() + self.scheduler.scheduler = mock_scheduler_instance + + # Mock Azure Storage 연결 종료 + with patch('app.services.scheduler_service.azure_storage') as mock_azure: + mock_azure.disconnect = AsyncMock() + + # 스케줄러 중지 + await self.scheduler.stop() + + assert self.scheduler.is_running is False + mock_scheduler_instance.shutdown.assert_called_once() + mock_azure.disconnect.assert_called_once() + + def test_get_status(self): + """상태 조회 테스트""" + # Mock 스케줄러 + mock_scheduler_instance = MagicMock() + mock_scheduler_instance.get_jobs.return_value = [] + self.scheduler.scheduler = mock_scheduler_instance + + # 상태 조회 + status = self.scheduler.get_status() + + assert "is_running" in status + assert "interval_minutes" in status + assert "target_service" in status + assert status["is_running"] is False + + @pytest.mark.asyncio + @patch('app.services.scheduler_service.settings') + @patch('app.services.scheduler_service.painting_surface_model_client') + @patch('app.services.scheduler_service.azure_storage') + @patch('app.services.scheduler_service.anomaly_logger') + async def test_simulate_data_collection_success(self, mock_logger, mock_azure_storage, mock_model_client, mock_settings): + """데이터 수집 시뮬레이션 성공 테스트""" + mock_settings.scheduler_interval_minutes = 1 + + # Mock Azure Storage 데이터 시뮬레이션 + mock_azure_storage.simulate_painting_surface_data = AsyncMock(return_value={ + "images": ["image1.jpg", "image2.jpg"], + "metadata": {"source": "test"} + }) + + # Mock 모델 클라이언트 예측 (실제 코드 로직에 맞춤) + mock_model_client.predict_painting_surface_data = AsyncMock(return_value={ + "combined": { + "status": "normal", + "defect_count": 0, + "total_count": 2, + "combined_logic": "정상 처리 완료" + } + }) + + # Mock 로거 + mock_logger.log_normal_processing = MagicMock() + + # 데이터 수집 시뮬레이션 실행 + await self.scheduler._simulate_data_collection() + + # Azure Storage에서 데이터 시뮬레이션 호출 확인 + mock_azure_storage.simulate_painting_surface_data.assert_called_once() + + # 모델 클라이언트 예측 호출 확인 + mock_model_client.predict_painting_surface_data.assert_called_once_with(["image1.jpg", "image2.jpg"]) + + # 로거 호출 확인 + mock_logger.log_normal_processing.assert_called_once() + + @pytest.mark.asyncio + @patch('app.services.scheduler_service.settings') + @patch('app.services.scheduler_service.azure_storage') + async def test_simulate_data_collection_no_data(self, mock_azure_storage, mock_settings): + """데이터가 없는 경우 테스트""" + mock_settings.scheduler_interval_minutes = 1 + + # Mock Azure Storage에서 데이터 없음 + mock_azure_storage.simulate_painting_surface_data = AsyncMock(return_value=None) + + # 콘솔 출력 모킹 + with patch('builtins.print') as mock_print: + # 데이터 수집 시뮬레이션 실행 + await self.scheduler._simulate_data_collection() + + # 경고 메시지 출력 확인 + mock_print.assert_called_with("⚠️ 수집할 데이터가 없습니다.") + + @pytest.mark.asyncio + @patch('app.services.scheduler_service.settings') + @patch('app.services.scheduler_service.painting_surface_model_client') + @patch('app.services.scheduler_service.azure_storage') + async def test_simulate_data_collection_prediction_failure(self, mock_azure_storage, mock_model_client, mock_settings): + """예측 실패 테스트""" + mock_settings.scheduler_interval_minutes = 1 + + # Mock Azure Storage 데이터 시뮬레이션 + mock_azure_storage.simulate_painting_surface_data = AsyncMock(return_value={ + "images": ["image1.jpg", "image2.jpg"], + "metadata": {"source": "test"} + }) + + # Mock 모델 클라이언트 예측 실패 (실제 코드 로직에 맞춤) + mock_model_client.predict_painting_surface_data = AsyncMock(return_value=None) + + # 콘솔 출력 모킹 + with patch('builtins.print') as mock_print: + # 데이터 수집 시뮬레이션 실행 + await self.scheduler._simulate_data_collection() + + # 예측 실패 메시지 출력 확인 + mock_print.assert_called_with("❌ 예측 결과를 받을 수 없습니다.") + + @pytest.mark.asyncio + @patch('app.services.scheduler_service.settings') + @patch('app.services.scheduler_service.painting_surface_model_client') + @patch('app.services.scheduler_service.azure_storage') + @patch('app.services.scheduler_service.anomaly_logger') + async def test_simulate_data_collection_anomaly_detected(self, mock_logger, mock_azure_storage, mock_model_client, mock_settings): + """결함 탐지 테스트""" + mock_settings.scheduler_interval_minutes = 1 + + # Mock Azure Storage 데이터 시뮬레이션 + mock_azure_storage.simulate_painting_surface_data = AsyncMock(return_value={ + "images": ["image1.jpg", "image2.jpg"], + "metadata": {"source": "test"} + }) + + # Mock 모델 클라이언트 예측 (결함 탐지 - 실제 코드 로직에 맞춤) + mock_model_client.predict_painting_surface_data = AsyncMock(return_value={ + "combined": { + "status": "anomaly", + "defect_count": 1, + "total_count": 2, + "combined_logic": "결함 탐지됨" + } + }) + + # Mock 로거 + mock_logger.log_anomaly = MagicMock() + + # 데이터 수집 시뮬레이션 실행 + await self.scheduler._simulate_data_collection() + + # 결함 탐지 로그 호출 확인 + mock_logger.log_anomaly.assert_called_once() + + @pytest.mark.asyncio + @patch('app.services.scheduler_service.settings') + @patch('app.services.scheduler_service.painting_surface_model_client') + @patch('app.services.scheduler_service.azure_storage') + async def test_simulate_data_collection_exception_handling(self, mock_azure_storage, mock_model_client, mock_settings): + """예외 처리 테스트""" + mock_settings.scheduler_interval_minutes = 1 + + # Mock Azure Storage에서 예외 발생 + mock_azure_storage.simulate_painting_surface_data = AsyncMock(side_effect=Exception("Storage error")) + + # Mock 로거 + with patch('app.services.scheduler_service.anomaly_logger') as mock_logger: + mock_logger.log_error = MagicMock() + + # 데이터 수집 시뮬레이션 실행 + await self.scheduler._simulate_data_collection() + + # 로거 에러 호출 확인 + mock_logger.log_error.assert_called_once_with("painting-surface-scheduler", "Storage error") + + @pytest.mark.asyncio + @patch('app.services.scheduler_service.settings') + @patch('app.services.scheduler_service.painting_surface_model_client') + async def test_initial_health_check_success(self, mock_model_client, mock_settings): + """초기 헬스 체크 성공 테스트""" + mock_settings.scheduler_interval_minutes = 1 + + # Mock 헬스 체크 성공 + mock_model_client.health_check = AsyncMock(return_value=True) + + # 콘솔 출력 모킹 + with patch('builtins.print') as mock_print: + # 초기 헬스 체크 실행 + await self.scheduler._initial_health_check() + + # 성공 메시지 출력 확인 (실제 출력 메시지에 맞춤) + mock_print.assert_any_call(" ✅ 도장 표면 결함탐지 서비스") + + @pytest.mark.asyncio + @patch('app.services.scheduler_service.settings') + @patch('app.services.scheduler_service.painting_surface_model_client') + async def test_initial_health_check_failure(self, mock_model_client, mock_settings): + """초기 헬스 체크 실패 테스트""" + mock_settings.scheduler_interval_minutes = 1 + + # Mock 헬스 체크 실패 + mock_model_client.health_check = AsyncMock(return_value=False) + + # 초기 헬스 체크 실행 (예외 발생) + with pytest.raises(Exception, match="도장 표면 결함탐지 서비스가 비활성 상태입니다."): + await self.scheduler._initial_health_check() diff --git a/services/painting-surface-data-simulator-service/tests/test_settings.py b/services/painting-surface-data-simulator-service/tests/test_settings.py new file mode 100644 index 0000000..1f13ea3 --- /dev/null +++ b/services/painting-surface-data-simulator-service/tests/test_settings.py @@ -0,0 +1,99 @@ +import pytest +import os +from app.config.settings import Settings + + +class TestSettings: + """설정 클래스 테스트""" + + def test_settings_default_values(self): + """기본값 테스트""" + # 환경 변수가 설정되어 있으므로 기본값 확인 테스트 + settings = Settings() + # 기본값 확인 + assert settings.azure_container_name == "simulator-data" + assert settings.painting_data_folder == "painting-surface" + assert settings.scheduler_interval_minutes == 1 + assert settings.batch_size == 10 + assert settings.painting_model_url == "http://localhost:8002" + assert settings.log_directory == "logs" + assert settings.log_filename == "painting_defect_detections.json" + assert settings.error_log_filename == "painting_errors.json" + assert settings.http_timeout == 30 + assert settings.max_retries == 3 + + def test_settings_with_environment_variables(self): + """환경 변수를 통한 설정 테스트""" + # 테스트 환경 변수 설정 + os.environ['AZURE_CONNECTION_STRING'] = 'test_connection_string' + + try: + settings = Settings() + + # 기본값 확인 + assert settings.azure_container_name == "simulator-data" + assert settings.painting_data_folder == "painting-surface" + assert settings.scheduler_interval_minutes == 1 + assert settings.batch_size == 10 + assert settings.painting_model_url == "http://localhost:8002" + assert settings.log_directory == "logs" + assert settings.log_filename == "painting_defect_detections.json" + assert settings.error_log_filename == "painting_errors.json" + assert settings.http_timeout == 30 + assert settings.max_retries == 3 + + # 커스텀 값 확인 + assert settings.azure_connection_string == "test_connection_string" + + finally: + # 환경 변수 정리 + if 'AZURE_CONNECTION_STRING' in os.environ: + del os.environ['AZURE_CONNECTION_STRING'] + + def test_model_service_url_property(self): + """model_service_url 프로퍼티 테스트""" + os.environ['AZURE_CONNECTION_STRING'] = 'test_connection_string' + + try: + settings = Settings() + assert settings.model_service_url == "http://localhost:8002" + finally: + if 'AZURE_CONNECTION_STRING' in os.environ: + del os.environ['AZURE_CONNECTION_STRING'] + + def test_settings_custom_values(self): + """커스텀 값 설정 테스트""" + os.environ['AZURE_CONNECTION_STRING'] = 'test_connection_string' + os.environ['AZURE_CONTAINER_NAME'] = 'custom-container' + os.environ['PAINTING_DATA_FOLDER'] = 'custom-painting' + os.environ['SCHEDULER_INTERVAL_MINUTES'] = '5' + os.environ['BATCH_SIZE'] = '20' + + try: + settings = Settings() + + assert settings.azure_container_name == "custom-container" + assert settings.painting_data_folder == "custom-painting" + assert settings.scheduler_interval_minutes == 5 + assert settings.batch_size == 20 + + finally: + # 환경 변수 정리 + for key in ['AZURE_CONNECTION_STRING', 'AZURE_CONTAINER_NAME', + 'PAINTING_DATA_FOLDER', 'SCHEDULER_INTERVAL_MINUTES', 'BATCH_SIZE']: + if key in os.environ: + del os.environ[key] + + def test_settings_model_config(self): + """모델 설정 테스트""" + os.environ['AZURE_CONNECTION_STRING'] = 'test_connection_string' + + try: + settings = Settings() + + assert settings.model_config["env_file"] == ".env" + assert settings.model_config["env_file_encoding"] == "utf-8" + + finally: + if 'AZURE_CONNECTION_STRING' in os.environ: + del os.environ['AZURE_CONNECTION_STRING'] diff --git a/services/painting-surface-data-simulator-service/tests/test_simulator_router.py b/services/painting-surface-data-simulator-service/tests/test_simulator_router.py new file mode 100644 index 0000000..83793d3 --- /dev/null +++ b/services/painting-surface-data-simulator-service/tests/test_simulator_router.py @@ -0,0 +1,275 @@ +import pytest +import json +import os +from unittest.mock import AsyncMock, MagicMock, patch +from fastapi.testclient import TestClient +from fastapi import HTTPException +from app.routers.simulator_router import get_simulator_status, start_simulator, stop_simulator, get_recent_logs +from app.services.scheduler_service import simulator_scheduler +from app.services.model_client import painting_surface_model_client + + +class TestSimulatorRouter: + """시뮬레이터 라우터 테스트""" + + def setup_method(self): + """각 테스트 메서드 실행 전 설정""" + self.test_log_dir = "test-logs" + self.test_log_file = os.path.join(self.test_log_dir, "test_log.json") + + # 테스트 로그 디렉토리 생성 + os.makedirs(self.test_log_dir, exist_ok=True) + + def teardown_method(self): + """각 테스트 메서드 실행 후 정리""" + # 테스트 로그 파일 정리 + if os.path.exists(self.test_log_dir): + import shutil + shutil.rmtree(self.test_log_dir) + + @pytest.mark.asyncio + @patch('app.routers.simulator_router.simulator_scheduler') + @patch('app.routers.simulator_router.painting_surface_model_client') + async def test_get_simulator_status_success(self, mock_model_client, mock_scheduler): + """시뮬레이터 상태 조회 성공 테스트""" + # Mock 스케줄러 상태 + mock_scheduler.is_running = True + mock_scheduler.get_status.return_value = { + "running": True, + "started_at": "2024-01-01T00:00:00", + "jobs": ["data_simulation"] + } + + # Mock 모델 클라이언트 헬스 체크 + mock_model_client.health_check = AsyncMock(return_value=True) + + # 상태 조회 + response = await get_simulator_status() + + assert response["running"] is True + assert response["started_at"] == "2024-01-01T00:00:00" + assert response["jobs"] == ["data_simulation"] + assert response["painting_surface_service_health"] is True + + @pytest.mark.asyncio + @patch('app.routers.simulator_router.simulator_scheduler') + @patch('app.routers.simulator_router.painting_surface_model_client') + async def test_get_simulator_status_not_running(self, mock_model_client, mock_scheduler): + """시뮬레이터가 실행 중이 아닌 경우 테스트""" + # Mock 스케줄러 상태 + mock_scheduler.is_running = False + mock_scheduler.get_status.return_value = { + "running": False, + "started_at": None, + "jobs": [] + } + + # 상태 조회 + response = await get_simulator_status() + + assert response["running"] is False + assert response["started_at"] is None + assert response["jobs"] == [] + # 실행 중이 아니므로 헬스 체크가 호출되지 않음 + assert "painting_surface_service_health" not in response + + @pytest.mark.asyncio + @patch('app.routers.simulator_router.simulator_scheduler') + async def test_start_simulator_success(self, mock_scheduler): + """시뮬레이터 시작 성공 테스트""" + # Mock 스케줄러 시작 + mock_scheduler.start = AsyncMock() + mock_scheduler.get_status.return_value = { + "running": True, + "started_at": "2024-01-01T00:00:00", + "jobs": ["data_simulation"] + } + + # 시뮬레이터 시작 + response = await start_simulator() + + assert response["message"] == "시뮬레이터가 시작되었습니다." + assert response["status"]["running"] is True + mock_scheduler.start.assert_called_once() + + @pytest.mark.asyncio + @patch('app.routers.simulator_router.simulator_scheduler') + async def test_start_simulator_failure(self, mock_scheduler): + """시뮬레이터 시작 실패 테스트""" + # Mock 스케줄러 시작 실패 + mock_scheduler.start = AsyncMock(side_effect=Exception("Start failed")) + + # 시뮬레이터 시작 시도 (예외 발생) + with pytest.raises(HTTPException) as exc_info: + await start_simulator() + + assert exc_info.value.status_code == 500 + assert "시뮬레이터 시작 실패" in exc_info.value.detail + + @pytest.mark.asyncio + @patch('app.routers.simulator_router.simulator_scheduler') + async def test_stop_simulator_success(self, mock_scheduler): + """시뮬레이터 중지 성공 테스트""" + # Mock 스케줄러 중지 + mock_scheduler.stop = AsyncMock() + mock_scheduler.get_status.return_value = { + "running": False, + "started_at": None, + "jobs": [] + } + + # 시뮬레이터 중지 + response = await stop_simulator() + + assert response["message"] == "시뮬레이터가 중지되었습니다." + assert response["status"]["running"] is False + mock_scheduler.stop.assert_called_once() + + @pytest.mark.asyncio + @patch('app.routers.simulator_router.simulator_scheduler') + async def test_stop_simulator_failure(self, mock_scheduler): + """시뮬레이터 중지 실패 테스트""" + # Mock 스케줄러 중지 실패 + mock_scheduler.stop = AsyncMock(side_effect=Exception("Stop failed")) + + # 시뮬레이터 중지 시도 (예외 발생) + with pytest.raises(HTTPException) as exc_info: + await stop_simulator() + + assert exc_info.value.status_code == 500 + assert "시뮬레이터 중지 실패" in exc_info.value.detail + + @pytest.mark.asyncio + @patch('app.routers.simulator_router.settings') + async def test_get_recent_logs_success(self, mock_settings): + """최근 로그 조회 성공 테스트""" + mock_settings.log_directory = self.test_log_dir + mock_settings.log_filename = "test_log.json" + + # 테스트 로그 파일 생성 + test_logs = [ + {"timestamp": "2024-01-01T00:00:00", "service": "test1", "status": "normal"}, + {"timestamp": "2024-01-01T00:01:00", "service": "test2", "status": "anomaly"}, + {"timestamp": "2024-01-01T00:02:00", "service": "test3", "status": "normal"} + ] + + with open(self.test_log_file, 'w', encoding='utf-8') as f: + for log in test_logs: + f.write(json.dumps(log, ensure_ascii=False) + '\n') + + # 최근 로그 조회 + response = await get_recent_logs() + + assert "logs" in response + assert "total_count" in response + assert len(response["logs"]) == 3 + assert response["total_count"] == 3 + + # 로그 내용 확인 + assert response["logs"][0]["service"] == "test1" + assert response["logs"][1]["service"] == "test2" + assert response["logs"][2]["service"] == "test3" + + @pytest.mark.asyncio + @patch('app.routers.simulator_router.settings') + async def test_get_recent_logs_no_file(self, mock_settings): + """로그 파일이 없는 경우 테스트""" + mock_settings.log_directory = self.test_log_dir + mock_settings.log_filename = "nonexistent.json" + + # 최근 로그 조회 + response = await get_recent_logs() + + assert response["logs"] == [] + assert response["message"] == "로그 파일이 없습니다." + + @pytest.mark.asyncio + @patch('app.routers.simulator_router.settings') + async def test_get_recent_logs_invalid_json(self, mock_settings): + """잘못된 JSON 형식의 로그 파일 테스트""" + mock_settings.log_directory = self.test_log_dir + mock_settings.log_filename = "test_log.json" + + # 잘못된 JSON 형식의 로그 파일 생성 + with open(self.test_log_file, 'w', encoding='utf-8') as f: + f.write('{"valid": "json"}\n') + f.write('invalid json line\n') + f.write('{"another": "valid"}\n') + + # 최근 로그 조회 + response = await get_recent_logs() + + assert "logs" in response + assert "total_count" in response + # 유효한 JSON만 파싱됨 + assert len(response["logs"]) == 2 + assert response["total_count"] == 3 + + @pytest.mark.asyncio + @patch('app.routers.simulator_router.settings') + async def test_get_recent_logs_exception_handling(self, mock_settings): + """로그 조회 중 예외 처리 테스트""" + mock_settings.log_directory = self.test_log_dir + mock_settings.log_filename = "test_log.json" + + # 파일이 존재하지만 읽기 실패하는 경우 모킹 + with patch('os.path.exists', return_value=True): + with patch('builtins.open', side_effect=PermissionError("Permission denied")): + # 로그 조회 시도 (예외 발생) + with pytest.raises(HTTPException) as exc_info: + await get_recent_logs() + + assert exc_info.value.status_code == 500 + assert "로그 조회 실패" in exc_info.value.detail + + @pytest.mark.asyncio + @patch('app.routers.simulator_router.settings') + async def test_get_recent_logs_limit_to_10(self, mock_settings): + """로그 10개 제한 테스트""" + mock_settings.log_directory = self.test_log_dir + mock_settings.log_filename = "test_log.json" + + # 15개의 테스트 로그 생성 + test_logs = [] + for i in range(15): + test_logs.append({ + "timestamp": f"2024-01-01T00:{i:02d}:00", + "service": f"test{i}", + "status": "normal" + }) + + with open(self.test_log_file, 'w', encoding='utf-8') as f: + for log in test_logs: + f.write(json.dumps(log, ensure_ascii=False) + '\n') + + # 최근 로그 조회 + response = await get_recent_logs() + + assert len(response["logs"]) == 10 + assert response["total_count"] == 15 + + # 최근 10개만 반환되는지 확인 (마지막 10개) + assert response["logs"][0]["service"] == "test5" + assert response["logs"][-1]["service"] == "test14" + + @pytest.mark.asyncio + @patch('app.routers.simulator_router.simulator_scheduler') + @patch('app.routers.simulator_router.painting_surface_model_client') + async def test_get_simulator_status_with_health_check_failure(self, mock_model_client, mock_scheduler): + """헬스 체크 실패가 포함된 상태 조회 테스트""" + # Mock 스케줄러 상태 + mock_scheduler.is_running = True + mock_scheduler.get_status.return_value = { + "running": True, + "started_at": "2024-01-01T00:00:00", + "jobs": ["data_simulation"] + } + + # Mock 모델 클라이언트 헬스 체크 실패 + mock_model_client.health_check = AsyncMock(return_value=False) + + # 상태 조회 + response = await get_simulator_status() + + assert response["running"] is True + assert response["painting_surface_service_health"] is False diff --git a/services/painting-surface-defect-detection-model-service/app/main.py b/services/painting-surface-defect-detection-model-service/app/main.py index b7a8e53..e44d535 100644 --- a/services/painting-surface-defect-detection-model-service/app/main.py +++ b/services/painting-surface-defect-detection-model-service/app/main.py @@ -3,26 +3,37 @@ from datetime import datetime, timezone from contextlib import asynccontextmanager from app.services.inference import PaintingSurfaceDefectDetectionService +from typing import Annotated, Optional +from fastapi import Depends +import uvicorn +# Storage for the service instance +_detection_service: Optional[PaintingSurfaceDefectDetectionService] = None + +async def get_detection_service() -> PaintingSurfaceDefectDetectionService: + if _detection_service is None: + raise RuntimeError("Detection service not initialized") + return _detection_service @asynccontextmanager async def lifespan(app: FastAPI): # 앱 시작 시 실행 try: - global detection_service - detection_service = PaintingSurfaceDefectDetectionService() - await detection_service.load_model() + global _detection_service + _detection_service = PaintingSurfaceDefectDetectionService() + await _detection_service.load_model() - # 전역 변수로 라우터에서 접근할 수 있도록 설정 - predict_router.detection_service = detection_service + # predict.py의 전역 변수 설정 + import app.routers.predict as predict_module + predict_module.detection_service = _detection_service print(f"[{datetime.now()}] Painting Surface Defect Detection Model loaded successfully") except Exception as e: raise RuntimeError(f"Failed to initialize models: {str(e)}") from e yield # 앱 종료 시 실행 - if detection_service: - await detection_service.cleanup() + if _detection_service: + await _detection_service.cleanup() app = FastAPI( @@ -31,7 +42,7 @@ async def lifespan(app: FastAPI): lifespan=lifespan ) -app.include_router(predict_router.router, prefix="/api") +app.include_router(predict_router.router) @app.get("/health") @@ -47,3 +58,14 @@ async def ready(): @app.get("/startup") async def startup(): return {"status": "started", "initialization_complete": True} + + +if __name__ == "__main__": + # 시뮬레이터에서 접근할 수 있도록 포트 8002로 설정 + uvicorn.run( + "main:app", + host="0.0.0.0", # 모든 인터페이스에서 접근 가능 + port=8002, + reload=True + ) + diff --git a/services/painting-surface-defect-detection-model-service/app/models/yolo_model.py b/services/painting-surface-defect-detection-model-service/app/models/yolo_model.py index 5a9d3c3..f5482e7 100644 --- a/services/painting-surface-defect-detection-model-service/app/models/yolo_model.py +++ b/services/painting-surface-defect-detection-model-service/app/models/yolo_model.py @@ -69,10 +69,8 @@ def load_model_config(self) -> Dict[str, Any]: """모델 설정 파일 로드""" try: config_filename = "model_config.json" - config_path = hf_hub_download( - repo_id=self.model_name, - filename=config_filename - ) + # 로컬 파일 경로 사용 + config_path = os.path.join(os.path.dirname(__file__), config_filename) with open(config_path, "r", encoding="utf-8") as f: self.model_config = json.load(f) @@ -91,10 +89,8 @@ def load_class_mapping(self) -> Dict[int, str]: """클래스 매핑 파일 로드""" try: mapping_filename = "class_mapping.json" - mapping_path = hf_hub_download( - repo_id=self.model_name, - filename=mapping_filename - ) + # 로컬 파일 경로 사용 + mapping_path = os.path.join(os.path.dirname(__file__), mapping_filename) with open(mapping_path, "r", encoding="utf-8") as f: class_mapping = json.load(f) @@ -113,10 +109,8 @@ def load_threshold_config(self) -> Dict[str, float]: """임계값 설정 파일 로드""" try: threshold_filename = "thresholds.json" - threshold_path = hf_hub_download( - repo_id=self.model_name, - filename=threshold_filename - ) + # 로컬 파일 경로 사용 + threshold_path = os.path.join(os.path.dirname(__file__), threshold_filename) with open(threshold_path, "r", encoding="utf-8") as f: thresholds = json.load(f) @@ -124,9 +118,6 @@ def load_threshold_config(self) -> Dict[str, float]: logger.info("Threshold configuration loaded successfully") return thresholds - except FileNotFoundError: - logger.warning(f"Threshold file '{threshold_filename}' not found, using empty thresholds") - return {} except Exception as e: logger.error(f"Failed to load threshold config: {str(e)}") return {} diff --git a/services/painting-surface-defect-detection-model-service/app/routers/predict.py b/services/painting-surface-defect-detection-model-service/app/routers/predict.py index d156f20..a93b8f9 100644 --- a/services/painting-surface-defect-detection-model-service/app/routers/predict.py +++ b/services/painting-surface-defect-detection-model-service/app/routers/predict.py @@ -37,6 +37,13 @@ def validate_base64(cls, v): return v +class ImagePathRequest(BaseModel): + """이미지 경로 기반 예측 요청 스키마""" + image_path: str = Field(..., description="이미지 파일 경로") + confidence_threshold: Optional[float] = Field(0.5, ge=0.0, le=1.0, description="신뢰도 임계값 (0.0-1.0)") + timestamp: Optional[str] = Field(None, description="타임스탬프") + + class PredictionResponse(BaseModel): """예측 응답 스키마""" predictions: List[Dict[str, Any]] @@ -53,7 +60,7 @@ def get_detection_service() -> PaintingSurfaceDefectDetectionService: return detection_service -@router.post("/predict", response_model=PredictionResponse) +@router.post("/predict/file", response_model=PredictionResponse) async def predict( image: UploadFile = File(..., description="이미지 파일"), confidence_threshold: float = Form(0.5, ge=0.0, le=1.0, description="신뢰도 임계값 (0.0-1.0)") @@ -94,7 +101,7 @@ async def predict( raise HTTPException(status_code=500, detail=f"Prediction failed: {str(e)}") -@router.post("/predict/base64", response_model=PredictionResponse) +@router.post("/predict", response_model=PredictionResponse) def predict_base64(data: PredictionRequest): """ 도장 표면 결함 탐지 (Base64 방식) @@ -120,6 +127,9 @@ def predict_base64(data: PredictionRequest): raise HTTPException(status_code=500, detail=f"Base64 prediction failed: {str(e)}") + + + def predict_anomaly(image_data: bytes, confidence_threshold: Optional[float] = None): """도장 표면 결함 탐지 (동기 함수)""" if detection_service is None: diff --git a/services/painting-surface-defect-detection-model-service/requirements.txt b/services/painting-surface-defect-detection-model-service/requirements.txt index 62eb49f..1b4b2e4 100644 --- a/services/painting-surface-defect-detection-model-service/requirements.txt +++ b/services/painting-surface-defect-detection-model-service/requirements.txt @@ -10,6 +10,7 @@ python-dotenv==1.0.1 requests==2.31.0 torch==2.1.0 Pillow==10.2.0 +urllib3==2.5.0 # HF 관련 (필요시만) transformers==4.37.2 diff --git a/services/painting-surface-defect-detection-model-service/tests/test_model_loader.py b/services/painting-surface-defect-detection-model-service/tests/test_model_loader.py index 70249db..fbc0c95 100644 --- a/services/painting-surface-defect-detection-model-service/tests/test_model_loader.py +++ b/services/painting-surface-defect-detection-model-service/tests/test_model_loader.py @@ -365,9 +365,6 @@ def test_model_info_without_config(self): # model_type이 있을 수 있으므로 제거하지 않음 -if __name__ == "__main__": - pytest.main([__file__]) - def test_load_yolo_model_invalid_file_extension(self): """잘못된 파일 확장자로 모델 로딩 실패 테스트""" with patch('app.models.yolo_model.hf_hub_download') as mock_hf_download: @@ -785,4 +782,7 @@ def test_model_loader_attribute_isolation(self): assert second_loader.model is None assert second_loader.model_config is None assert second_loader.org == self.model_loader.org - assert second_loader.repo == self.model_loader.repo \ No newline at end of file + assert second_loader.repo == self.model_loader.repo + +if __name__ == "__main__": + pytest.main([__file__]) \ No newline at end of file