Coverage for src / mysingle / auth / init_data.py: 0%

1import asyncio 

2 

3from ..core.config import settings 

4from ..core.logging import get_structured_logger 

5from .models import User 

6from .security.password import PasswordHelper 

7 

8password_helper = PasswordHelper() 

9 

10logger = get_structured_logger(__name__) 

11 

12 

13async def _try_create_with_retry( 

14 create_func, check_func, entity_name: str, max_retries: int = 3 

15) -> bool: 

16 """재시도 로직을 포함한 생성 함수. 

17 

18 여러 워커가 동시에 실행될 때 경쟁 조건(race condition)을 처리합니다. 

19 

20 Args: 

21 create_func: 생성을 시도할 비동기 함수 

22 check_func: 이미 존재하는지 확인할 비동기 함수 

23 entity_name: 엔티티 이름 (로깅용) 

24 max_retries: 최대 재시도 횟수 

25 

26 Returns: 

27 bool: 생성 성공 여부 (이미 존재하면 True) 

28 """ 

29 for attempt in range(max_retries): 

30 try: 

31 # 이미 존재하는지 확인 

32 existing = await check_func() 

33 if existing: 

34 logger.info(f"✅ {entity_name} already exists (attempt {attempt + 1})") 

35 return True 

36 

37 # 생성 시도 

38 await create_func() 

39 logger.info( 

40 f"✅ {entity_name} created successfully (attempt {attempt + 1})" 

41 ) 

42 return True 

43 

44 except Exception as e: 

45 error_msg = str(e).lower() 

46 

47 # 중복 키 에러인 경우 (다른 워커가 이미 생성함) 

48 if "duplicate" in error_msg or "e11000" in error_msg: 

49 logger.info( 

50 f"ℹ️ {entity_name} was created by another worker (attempt {attempt + 1})" 

51 ) 

52 # 잠시 대기 후 다시 확인 

53 await asyncio.sleep(0.5) 

54 existing = await check_func() 

55 if existing: 

56 logger.info(f"✅ {entity_name} verified after duplicate error") 

57 return True 

58 

59 # 마지막 시도가 아니면 재시도 

60 if attempt < max_retries - 1: 

61 logger.warning( 

62 f"⚠️ Failed to create {entity_name} (attempt {attempt + 1}/{max_retries}): {e}" 

63 ) 

64 await asyncio.sleep(1.0) # 대기 후 재시도 

65 else: 

66 logger.error( 

67 f"❌ Failed to create {entity_name} after {max_retries} attempts: {e}" 

68 ) 

69 return False 

70 

71 return False 

72 

73 

74async def create_first_super_admin() -> None: 

75 """첫 번째 Super Admin 사용자 생성 (멀티 워커 환경 지원)""" 

76 try: 

77 logger.info("🔍 Checking for existing Super Admin user...") 

78 

79 # 설정 값 확인 

80 if ( 

81 settings.SUPERUSER_EMAIL == "your_email@example.com" 

82 or settings.SUPERUSER_PASSWORD == "change-this-admin-password" 

83 ): 

84 logger.warning( 

85 "⏭️ Super Admin creation skipped: Default email/password values detected. " 

86 "Please set SUPERUSER_EMAIL and SUPERUSER_PASSWORD environment variables." 

87 ) 

88 return 

89 

90 logger.info(f"👤 Creating Super Admin with email: {settings.SUPERUSER_EMAIL}") 

91 

92 # 생성 함수 

93 async def create_user(): 

94 user = User( 

95 email=settings.SUPERUSER_EMAIL, 

96 hashed_password=password_helper.hash(settings.SUPERUSER_PASSWORD), 

97 full_name=settings.SUPERUSER_FULLNAME, 

98 is_active=True, 

99 is_superuser=True, 

100 is_verified=True, 

101 ) 

102 await user.save() 

103 logger.info( 

104 f"✅ Super Admin user created: {user.full_name} " 

105 f"({user.email}, ID: {user.id})" 

106 ) 

107 

108 # 확인 함수 

109 async def check_existing(): 

110 existing = await User.find_one( 

111 {"email": settings.SUPERUSER_EMAIL, "is_superuser": True} 

112 ) 

113 if existing: 

114 logger.info( 

115 f"✅ Super Admin already exists: {existing.email} (ID: {existing.id})" 

116 ) 

117 return existing 

118 

119 # 재시도 로직으로 생성 

120 success = await _try_create_with_retry( 

121 create_func=create_user, 

122 check_func=check_existing, 

123 entity_name=f"Super Admin ({settings.SUPERUSER_EMAIL})", 

124 max_retries=3, 

125 ) 

126 

127 if success: 

128 logger.info(f"✅ Super Admin setup completed: {settings.SUPERUSER_EMAIL}") 

129 else: 

130 logger.warning(f"⚠️ Super Admin setup failed: {settings.SUPERUSER_EMAIL}") 

131 

132 except Exception as e: 

133 logger.error( 

134 f"❌ Failed to create first Super Admin: {type(e).__name__}: {str(e)}" 

135 ) 

136 logger.error(f"Settings - Email: {settings.SUPERUSER_EMAIL}") 

137 logger.error(f"Settings - Fullname: {settings.SUPERUSER_FULLNAME}") 

138 # Super Admin 생성 실패가 애플리케이션 시작을 막지 않도록 함 

139 

140 

141async def create_test_users() -> None: 

142 """ 

143 테스트용 유저 생성 (development/local 환경에서만) 

144 

145 생성되는 테스트 유저: 

146 1. test_user: 일반 유저 (verified, not superuser) 

147 - email: "test_user" 

148 - password: "1234" 

149 - full_name: "Test User" 

150 

151 2. test_admin: 관리자 유저 (verified, superuser) 

152 - email: "test_admin" 

153 - password: "1234" 

154 - full_name: "Test Admin" 

155 

156 ⚠️ WARNING: production 환경에서는 절대 호출되지 않습니다! 

157 """ 

158 # production 환경에서는 실행 안 함 

159 if settings.ENVIRONMENT.lower() not in ["development", "local", "dev"]: 

160 logger.info( 

161 "⏭️ Test user creation skipped: Not in development/local environment" 

162 ) 

163 return 

164 

165 try: 

166 logger.info("🧪 Creating test users for development/local environment...") 

167 

168 # 1. 일반 테스트 유저 생성 (같은 이메일 + is_superuser=False) 

169 existing_test_user = await User.find_one( 

170 {"email": settings.TEST_USER_EMAIL, "is_superuser": False} 

171 ) 

172 if existing_test_user: 

173 logger.info( 

174 f"✅ Test user already exists: {settings.TEST_USER_EMAIL} " 

175 f"(ID: {existing_test_user.id})" 

176 ) 

177 else: 

178 test_user = User( 

179 email=settings.TEST_USER_EMAIL, 

180 hashed_password=password_helper.hash(settings.TEST_USER_PASSWORD), 

181 full_name=settings.TEST_USER_FULLNAME, 

182 is_active=True, 

183 is_superuser=False, 

184 is_verified=True, 

185 ) 

186 await test_user.save() 

187 logger.info( 

188 f"✅ Test user created: {settings.TEST_USER_FULLNAME} " 

189 f"({settings.TEST_USER_EMAIL}, ID: {test_user.id})" 

190 ) 

191 

192 # 2. 관리자 테스트 유저 생성 (같은 이메일 + is_superuser=True) 

193 existing_test_admin = await User.find_one( 

194 {"email": settings.TEST_ADMIN_EMAIL, "is_superuser": True} 

195 ) 

196 if existing_test_admin: 

197 logger.info( 

198 f"✅ Test admin already exists: {settings.TEST_ADMIN_EMAIL} " 

199 f"(ID: {existing_test_admin.id})" 

200 ) 

201 else: 

202 test_admin = User( 

203 email=settings.TEST_ADMIN_EMAIL, 

204 hashed_password=password_helper.hash(settings.TEST_ADMIN_PASSWORD), 

205 full_name=settings.TEST_ADMIN_FULLNAME, 

206 is_active=True, 

207 is_superuser=True, 

208 is_verified=True, 

209 ) 

210 await test_admin.save() 

211 logger.info( 

212 f"✅ Test admin created: {settings.TEST_ADMIN_FULLNAME} " 

213 f"({settings.TEST_ADMIN_EMAIL}, ID: {test_admin.id})" 

214 ) 

215 

216 logger.info("✅ Test users setup completed successfully") 

217 

218 except Exception as e: 

219 logger.error(f"❌ Failed to create test users: {type(e).__name__}: {str(e)}") 

220 # 테스트 유저 생성 실패가 애플리케이션 시작을 막지 않도록 함