|
1 | 1 | from unittest.mock import AsyncMock, MagicMock, patch |
| 2 | + |
| 3 | +import pytest |
| 4 | + |
2 | 5 | from app.auth.security import get_password_hash |
3 | | -from app.models.trading import User |
| 6 | +from app.auth.web_router import MAX_SESSIONS_PER_USER |
| 7 | +from app.models.trading import User, UserRole |
4 | 8 |
|
5 | 9 |
|
6 | 10 | def test_login_page_render(auth_test_client): |
@@ -115,3 +119,181 @@ def test_web_logout(auth_test_client, auth_mock_session, mock_auth_middleware_db |
115 | 119 | assert response.status_code == 303 |
116 | 120 | # Check if session cookie is deleted (expired) |
117 | 121 | assert "session" not in response.cookies or response.cookies["session"] == "" |
| 122 | + |
| 123 | + |
| 124 | +# ==================== 다중 세션 로그인 테스트 ==================== |
| 125 | + |
| 126 | + |
| 127 | +@pytest.fixture |
| 128 | +def mock_limiter(): |
| 129 | + """Rate limiter를 비활성화하는 fixture""" |
| 130 | + from app.auth import web_router |
| 131 | + |
| 132 | + # limiter의 enabled 속성을 False로 설정 |
| 133 | + original_enabled = web_router.limiter.enabled |
| 134 | + web_router.limiter.enabled = False |
| 135 | + yield |
| 136 | + web_router.limiter.enabled = original_enabled |
| 137 | + |
| 138 | + |
| 139 | +class TestMultipleSessionLogin: |
| 140 | + """다중 디바이스 로그인 테스트""" |
| 141 | + |
| 142 | + @pytest.fixture(autouse=True) |
| 143 | + def disable_rate_limit(self, mock_limiter): |
| 144 | + """이 클래스의 모든 테스트에서 rate limiter 비활성화""" |
| 145 | + pass |
| 146 | + |
| 147 | + @pytest.fixture |
| 148 | + def mock_redis_client(self): |
| 149 | + """Redis 클라이언트 mock""" |
| 150 | + mock_client = AsyncMock() |
| 151 | + mock_client.sismember = AsyncMock(return_value=True) |
| 152 | + mock_client.sadd = AsyncMock(return_value=1) |
| 153 | + mock_client.srem = AsyncMock(return_value=1) |
| 154 | + mock_client.scard = AsyncMock(return_value=0) |
| 155 | + mock_client.spop = AsyncMock(return_value=None) |
| 156 | + mock_client.expire = AsyncMock(return_value=True) |
| 157 | + mock_client.get = AsyncMock(return_value=None) |
| 158 | + mock_client.set = AsyncMock(return_value=True) |
| 159 | + mock_client.aclose = AsyncMock() |
| 160 | + return mock_client |
| 161 | + |
| 162 | + @pytest.fixture |
| 163 | + def test_user(self): |
| 164 | + """테스트용 사용자""" |
| 165 | + return User( |
| 166 | + id=1, |
| 167 | + username="testuser", |
| 168 | + |
| 169 | + hashed_password=get_password_hash("password123"), |
| 170 | + role=UserRole.viewer, |
| 171 | + is_active=True |
| 172 | + ) |
| 173 | + |
| 174 | + def test_multiple_logins_use_redis_set(self, auth_test_client, auth_mock_session, test_user, mock_redis_client): |
| 175 | + """여러 번 로그인해도 Redis Set에 세션 추가""" |
| 176 | + mock_result = MagicMock() |
| 177 | + mock_result.scalar_one_or_none.return_value = test_user |
| 178 | + auth_mock_session.execute.return_value = mock_result |
| 179 | + |
| 180 | + with patch("app.auth.web_router.redis.from_url", return_value=mock_redis_client): |
| 181 | + # 첫 번째 로그인 |
| 182 | + response1 = auth_test_client.post( |
| 183 | + "/web-auth/login", |
| 184 | + data={"username": "testuser", "password": "password123"}, |
| 185 | + follow_redirects=False |
| 186 | + ) |
| 187 | + assert response1.status_code == 303 |
| 188 | + assert "session" in response1.cookies |
| 189 | + |
| 190 | + # Redis sadd가 호출되었는지 확인 (Set에 추가) |
| 191 | + assert mock_redis_client.sadd.call_count >= 1 |
| 192 | + |
| 193 | + def test_logout_removes_only_current_session(self, auth_test_client, auth_mock_session, test_user, mock_redis_client): |
| 194 | + """로그아웃 시 현재 세션만 Set에서 제거""" |
| 195 | + mock_result = MagicMock() |
| 196 | + mock_result.scalar_one_or_none.return_value = test_user |
| 197 | + auth_mock_session.execute.return_value = mock_result |
| 198 | + |
| 199 | + # logout 함수 내에서 redis를 import하므로 redis.asyncio 모듈 자체를 mock |
| 200 | + with patch("app.auth.web_router.redis.from_url", return_value=mock_redis_client), \ |
| 201 | + patch("redis.asyncio.from_url", return_value=mock_redis_client): |
| 202 | + # 로그인 |
| 203 | + login_response = auth_test_client.post( |
| 204 | + "/web-auth/login", |
| 205 | + data={"username": "testuser", "password": "password123"}, |
| 206 | + follow_redirects=False |
| 207 | + ) |
| 208 | + session_cookie = login_response.cookies.get("session") |
| 209 | + |
| 210 | + # 로그아웃 |
| 211 | + auth_test_client.cookies.set("session", session_cookie) |
| 212 | + auth_test_client.get("/web-auth/logout", follow_redirects=False) |
| 213 | + |
| 214 | + # Redis srem이 호출되었는지 확인 (Set에서 제거) |
| 215 | + assert mock_redis_client.srem.called |
| 216 | + |
| 217 | + def test_session_limit_removes_oldest_when_exceeded(self, auth_test_client, auth_mock_session, test_user): |
| 218 | + """세션 개수가 MAX_SESSIONS_PER_USER를 초과하면 기존 세션 제거""" |
| 219 | + mock_result = MagicMock() |
| 220 | + mock_result.scalar_one_or_none.return_value = test_user |
| 221 | + auth_mock_session.execute.return_value = mock_result |
| 222 | + |
| 223 | + # 별도의 mock_redis_client 생성 (scard가 MAX_SESSIONS_PER_USER 반환) |
| 224 | + mock_redis = AsyncMock() |
| 225 | + mock_redis.scard = AsyncMock(return_value=MAX_SESSIONS_PER_USER) |
| 226 | + mock_redis.spop = AsyncMock(return_value="old_session_hash") |
| 227 | + mock_redis.sadd = AsyncMock(return_value=1) |
| 228 | + mock_redis.expire = AsyncMock(return_value=True) |
| 229 | + mock_redis.set = AsyncMock(return_value=True) |
| 230 | + mock_redis.aclose = AsyncMock() |
| 231 | + |
| 232 | + with patch("app.auth.web_router.redis.from_url", return_value=mock_redis): |
| 233 | + # 새 로그인 시도 |
| 234 | + response = auth_test_client.post( |
| 235 | + "/web-auth/login", |
| 236 | + data={"username": "testuser", "password": "password123"}, |
| 237 | + follow_redirects=False |
| 238 | + ) |
| 239 | + assert response.status_code == 303 |
| 240 | + |
| 241 | + # spop이 호출되어 기존 세션 하나 제거 |
| 242 | + assert mock_redis.spop.called |
| 243 | + |
| 244 | + def test_session_validation_uses_sismember(self, auth_test_client, auth_mock_session, mock_auth_middleware_db, test_user, mock_redis_client): |
| 245 | + """세션 검증 시 sismember로 Set 멤버십 확인""" |
| 246 | + mock_result = MagicMock() |
| 247 | + mock_result.scalar_one_or_none.return_value = test_user |
| 248 | + auth_mock_session.execute.return_value = mock_result |
| 249 | + mock_auth_middleware_db.execute.return_value = mock_result |
| 250 | + |
| 251 | + # sismember가 True를 반환하도록 설정 |
| 252 | + mock_redis_client.sismember = AsyncMock(return_value=True) |
| 253 | + |
| 254 | + with patch("app.auth.web_router.redis.from_url", return_value=mock_redis_client), \ |
| 255 | + patch("app.auth.web_router.get_session_blacklist") as mock_blacklist: |
| 256 | + mock_blacklist.return_value.is_blacklisted = AsyncMock(return_value=False) |
| 257 | + |
| 258 | + # 로그인 |
| 259 | + login_response = auth_test_client.post( |
| 260 | + "/web-auth/login", |
| 261 | + data={"username": "testuser", "password": "password123"}, |
| 262 | + follow_redirects=False |
| 263 | + ) |
| 264 | + session_cookie = login_response.cookies.get("session") |
| 265 | + |
| 266 | + # 세션으로 인증 필요한 페이지 접근 |
| 267 | + auth_test_client.cookies.set("session", session_cookie) |
| 268 | + # 로그인 페이지는 이미 로그인 상태면 리다이렉트 |
| 269 | + response = auth_test_client.get("/web-auth/login", follow_redirects=False) |
| 270 | + |
| 271 | + # sismember가 호출되었는지 확인 |
| 272 | + assert mock_redis_client.sismember.called |
| 273 | + |
| 274 | + def test_invalid_session_returns_none(self, auth_test_client, auth_mock_session, mock_auth_middleware_db, test_user, mock_redis_client): |
| 275 | + """Set에 없는 세션은 인증 실패""" |
| 276 | + mock_result = MagicMock() |
| 277 | + mock_result.scalar_one_or_none.return_value = test_user |
| 278 | + auth_mock_session.execute.return_value = mock_result |
| 279 | + mock_auth_middleware_db.execute.return_value = mock_result |
| 280 | + |
| 281 | + # sismember가 False를 반환 (세션이 Set에 없음) |
| 282 | + mock_redis_client.sismember = AsyncMock(return_value=False) |
| 283 | + |
| 284 | + with patch("app.auth.web_router.redis.from_url", return_value=mock_redis_client), \ |
| 285 | + patch("app.auth.web_router.get_session_blacklist") as mock_blacklist: |
| 286 | + mock_blacklist.return_value.is_blacklisted = AsyncMock(return_value=False) |
| 287 | + |
| 288 | + # 유효하지 않은 세션으로 접근 |
| 289 | + auth_test_client.cookies.set("session", "invalid_session_token") |
| 290 | + response = auth_test_client.get("/web-auth/login", follow_redirects=False) |
| 291 | + |
| 292 | + # 로그인 페이지가 표시됨 (세션 무효) |
| 293 | + assert response.status_code == 200 |
| 294 | + assert "로그인" in response.text |
| 295 | + |
| 296 | + |
| 297 | +def test_max_sessions_per_user_constant(): |
| 298 | + """MAX_SESSIONS_PER_USER 상수가 올바르게 설정되어 있는지 확인""" |
| 299 | + assert MAX_SESSIONS_PER_USER == 5 |
0 commit comments