[HCD-8] 상태 조회 시 포트 번호 변환

app/api/challenge.py

@@ -69,7 +69,7 @@ def get_userchallenge_status():
     try:
         # Challenge 관련 정보 가져오기
         res = request.get_json()
-        print(f"/status : {res}",file=sys.stderr)
+        # print(f"/status : {res}",file=sys.stderr)
         if not res:
             raise InvalidRequest(error_msg="Request body is empty or not valid JSON")
 
@@ -89,9 +89,9 @@ def get_userchallenge_status():
 
         # 사용자 챌린지 상태 조회
         repo = UserChallengesRepository()
-        status = repo.get_status(challenge_id, username)
-        if status is None:
+        status_dict = repo.get_status(challenge_id, username)
+        if status_dict is None:
             raise UserChallengeNotFoundError(error_msg=f"User challenge not found for {username} and {challenge_id}")
-        return jsonify({'data': {'status': status}}), 200
+        return jsonify({'data': status_dict}), 200
     except Exception as e:
         raise UserChallengeNotFoundError(error_msg=str(e))

app/extensions/db/repository.py

@@ -1,12 +1,14 @@
-
 import logging
+from sqlite3 import OperationalError
 from typing import List, Optional
 from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session, load_only
+from sqlalchemy.sql import text
+from contextlib import contextmanager
 from app.exceptions.api import InternalServerError
 from app.extensions_manager import db
 from app.extensions.db.models import Challenges, UserChallenges
 
-
 class UserChallengesRepository:
     def __init__(self, session=None):
         self.session = session or db.session
@@ -70,7 +72,10 @@ def update_status(self, challenge: UserChallenges, new_status: str) -> bool:
             bool: 업데이트 성공 여부
         """
         try:
-            challenge.status = new_status
+            db.session.execute(text("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED"))
+            fresh_challenge = self.session.merge(challenge)
+            self.session.refresh(fresh_challenge) 
+            fresh_challenge.status = new_status
             # self.session.add(challenge)  # Add this line to track the object
             # self.session.flush()  
             self.session.commit()
@@ -94,20 +99,15 @@ def update_port(self, challenge: UserChallenges, port: int) -> bool:
             bool: 업데이트 성공 여부
         """
         try:
-            # 1) 먼저 challenge 객체를 세션에 맞게 merge
+            db.session.execute(text("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED"))
             fresh_challenge = self.session.merge(challenge)
-
-            fresh_challenge = self.session.merge(challenge)
-            self.session.refresh(fresh_challenge)  # Ensure fresh data
+            self.session.refresh(fresh_challenge) 
             fresh_challenge.port = port
-
             self.session.commit()
             return True
         except SQLAlchemyError as e:
-            self.session.rollback()
+            db.session.rollback()
             raise InternalServerError(error_msg=f"Error updating challenge port: {e}") from e
-
-
     def is_running(self, challenge: UserChallenges) -> bool:
         """
         챌린지 실행 여부 확인
@@ -120,7 +120,7 @@ def is_running(self, challenge: UserChallenges) -> bool:
         """
         return challenge.status == 'Running'
 
-    def get_status(self, challenge_id, username) -> Optional[str]:
+    def get_status(self, challenge_id, username)  -> Optional[dict]:
         """
         챌린지 상태 조회
 
@@ -132,8 +132,24 @@ def get_status(self, challenge_id, username) -> Optional[str]:
             str: 챌린지 상태
         """
         challenge = UserChallenges.query.filter_by(C_idx=challenge_id, username=username).first()
-        return challenge.status if challenge else None
+        if not challenge:
+            return None
+
+        if challenge.status == 'Running':
+            return {'status': challenge.status, 'port': int(challenge.port)}
+        return {'status': challenge.status}
+
+# class ChallengeRepository:
+#     def __init__(self):
+#         self.db_session = db.session
+
+#     def get_challenge_name(self, challenge_id: int) -> Optional[str]:
+#         """챌린지 ID로 챌린지 조회"""
+#         with self.get_session() as session:
+#             challenge = session.query(Challenges).get(challenge_id)
+#             return challenge.title if challenge else None
 
+
 class ChallengeRepository:
     @staticmethod
     def get_challenge_name(challenge_id: int) -> Optional[str]:

app/extensions/db/repository_test.py

@@ -0,0 +1,32 @@
+import pytest
+from unittest.mock import MagicMock
+from app.repositories.user_challenges_repository import UserChallengesRepository
+from app.extensions.db.models import UserChallenges
+
+@pytest.fixture
+def mock_session():
+    return MagicMock()
+
+@pytest.fixture
+def user_challenges_repo(mock_session):
+    return UserChallengesRepository(session=mock_session)
+
+@pytest.fixture
+def mock_challenge():
+    return UserChallenges(C_idx=1, username="test_user", status="Running", port=8080)
+
+def test_get_status_running(user_challenges_repo, mock_session, mock_challenge):
+    mock_session.query.return_value.filter_by.return_value.first.return_value = mock_challenge
+    result = user_challenges_repo.get_status(challenge_id=1, username="test_user")
+    assert result == {'status': 'Running', 'port': 8080}
+
+def test_get_status_not_running(user_challenges_repo, mock_session):
+    mock_challenge = UserChallenges(C_idx=1, username="test_user", status="Stopped", port=8080)
+    mock_session.query.return_value.filter_by.return_value.first.return_value = mock_challenge
+    result = user_challenges_repo.get_status(challenge_id=1, username="test_user")
+    assert result == {'status': 'Stopped'}
+
+def test_get_status_not_found(user_challenges_repo, mock_session):
+    mock_session.query.return_value.filter_by.return_value.first.return_value = None
+    result = user_challenges_repo.get_status(challenge_id=1, username="nonexistent_user")
+    assert result is None