diff --git a/backend/common/database/db_session_manager.py b/backend/common/database/db_session_manager.py index 31d9d2f..b0c4a36 100644 --- a/backend/common/database/db_session_manager.py +++ b/backend/common/database/db_session_manager.py @@ -34,17 +34,21 @@ class DBSessionManager(Singleton): # 종료 시 dispose 하기 위해 생성한 엔진을 모아둔다. self.__engines = [] # 논리 DB -> config. DB 가 늘어나면 여기에 추가만 하면 된다. + # USER/PARTNER 는 물리적으로 같은 negosium_db 라 main_db_config 를 재사용한다(도메인별 논리 구분용). self.__db_type_map = { - DBType.MAIN.value: main_db_config, + DBType.USER.value: main_db_config, + DBType.PARTNER.value: main_db_config, } # Write 엔진 맵 self.__write_session = { - DBType.MAIN.value: self.create_engine(DBType.MAIN.value, DBWRType.DB_WRITE.value), + DBType.USER.value: self.create_engine(DBType.USER.value, DBWRType.DB_WRITE.value), + DBType.PARTNER.value: self.create_engine(DBType.PARTNER.value, DBWRType.DB_WRITE.value), } # Read 엔진 맵 self.__read_session = { - DBType.MAIN.value: self.create_engine(DBType.MAIN.value, DBWRType.DB_READ.value), + DBType.USER.value: self.create_engine(DBType.USER.value, DBWRType.DB_READ.value), + DBType.PARTNER.value: self.create_engine(DBType.PARTNER.value, DBWRType.DB_READ.value), } def create_engine(self, db_type: int, db_wr_type: int): diff --git a/backend/common/database/model/models.py b/backend/common/database/model/models.py index 0da66c6..edb1924 100644 --- a/backend/common/database/model/models.py +++ b/backend/common/database/model/models.py @@ -1,5 +1,6 @@ from sqlalchemy.orm import declarative_base -from sqlalchemy import Column, Integer, String, Boolean, DateTime +from sqlalchemy import Column, Integer, String, Boolean, DateTime, SmallInteger +from sqlalchemy.dialects.postgresql import UUID, JSONB from sqlalchemy.sql import text from common.enums import DBType @@ -8,19 +9,69 @@ from common.enums import DBType MAIN_BASE = declarative_base() -class tbl_account(MAIN_BASE): - # 모델이 자신이 속한 논리 DB 를 알려준다 (람다 실행 시 DBType 으로 세션 선택). +class supplier_users(MAIN_BASE): + # 이 프로젝트의 기본 유저. 실제 테이블은 negosium_db 의 supplier 스키마(supplier_users). @staticmethod def DBType(): - return DBType.MAIN.value + return DBType.USER.value - __tablename__ = "tbl_account" + __tablename__ = "supplier_users" + __table_args__ = {"schema": "supplier"} - uid = Column(Integer, primary_key=True, autoincrement=True) - id = Column(String(45), nullable=False, unique=True) # 로그인 ID. 중복 가입 방지 위해 unique. - pw = Column(String(255), nullable=False, default="") # bcrypt 해시 저장 - nickname = Column(String(45), nullable=False, default="") - is_blocked = Column(Boolean, nullable=False, default=False) - # PostgreSQL UTC now: now() 는 timestamptz 이므로 utc 로 변환해 timestamp 로 저장. - last_login_at = Column(DateTime, nullable=False, server_default=text("(now() AT TIME ZONE 'utc')")) - create_at = Column(DateTime, server_default=text("(now() AT TIME ZONE 'utc')")) + # gen_random_uuid() 는 pgcrypto 확장 기준. 코드값(status/role/type)은 SMALLINT 정수 코드(앱 enum 매핑). + su_id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()")) # 유저 식별자(PK) + supplier_id = Column(UUID(as_uuid=True), nullable=False) # 소속 공급사(partner.suppliers.supplier_id) + id = Column(String(20), nullable=False) # 로그인 ID + password = Column(String(255), nullable=False) # 해시된 비밀번호이어야 함 + name = Column(String(50), nullable=True) # 이름 + email = Column(String(255), nullable=True) # 이메일 + contact_number = Column(String(20), nullable=True) # 연락처 + last_accessed_at = Column(DateTime(timezone=True), nullable=False) # 마지막 접속 시각 + status = Column(SmallInteger, nullable=False, server_default=text("1")) # 상태: 1=active, 2=inactive + role = Column(SmallInteger, nullable=False, server_default=text("1")) # 권한: 1=user, 2=manager + created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("(now() AT TIME ZONE 'utc')")) # 생성 시각(UTC) + updated_at = Column(DateTime(timezone=True), nullable=False, server_default=text("(now() AT TIME ZONE 'utc')")) # 수정 시각(UTC, 앱에서 갱신) + deleted = Column(Boolean, nullable=False, server_default=text("false")) # 소프트 삭제 여부 + + +class suppliers(MAIN_BASE): + # partner.suppliers (공급사 회사). 공급사명(name) 조회용. partner 도메인이라 DBType 은 PARTNER. + @staticmethod + def DBType(): + return DBType.PARTNER.value + + __tablename__ = "suppliers" + __table_args__ = {"schema": "partner"} + + supplier_id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()")) # 공급사 식별자(PK) + company_id = Column(UUID(as_uuid=True), nullable=False) # 소속 회사(company.companies.company_id) + user_id = Column(UUID(as_uuid=True), nullable=False) # 등록 유저(company.users.user_id) + name = Column(String(100), nullable=False) # 공급사명 + code = Column(String(20), nullable=True) # 공급사 코드 + manager_name = Column(String(50), nullable=True) # 담당자명 + manager_email = Column(String(255), nullable=True) # 담당자 이메일 + manager_contact_number = Column(String(20), nullable=True) # 담당자 연락처 + priority = Column(String(10), nullable=True) # 우선순위 (고객사별 문자열 값 가능) + created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("(now() AT TIME ZONE 'utc')")) # 생성 시각(UTC) + updated_at = Column(DateTime(timezone=True), nullable=False, server_default=text("(now() AT TIME ZONE 'utc')")) # 수정 시각(UTC, 앱에서 갱신) + deleted = Column(Boolean, nullable=False, server_default=text("false")) # 소프트 삭제 여부 + + +class supplier_user_tokens(MAIN_BASE): + # 유저 인증 토큰. supplier_users 1 : N tokens. + @staticmethod + def DBType(): + return DBType.USER.value + + __tablename__ = "supplier_user_tokens" + __table_args__ = {"schema": "supplier"} + + sut_id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()")) # 토큰 식별자(PK) + su_id = Column(UUID(as_uuid=True), nullable=False) # 소유 유저(supplier.supplier_users.su_id) + type = Column(SmallInteger, nullable=False) # 토큰 종류 (코드, 앱 enum 매핑) + token = Column(JSONB, nullable=False) # 토큰 본문(JSON) + issued_at = Column(DateTime(timezone=True), nullable=False) # 발급 시각 + expired_at = Column(DateTime(timezone=True), nullable=False) # 만료 시각 + created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("(now() AT TIME ZONE 'utc')")) # 생성 시각(UTC) + updated_at = Column(DateTime(timezone=True), nullable=False, server_default=text("(now() AT TIME ZONE 'utc')")) # 수정 시각(UTC, 앱에서 갱신) + deleted = Column(Boolean, nullable=False, server_default=text("false")) # 소프트 삭제 여부 diff --git a/backend/common/enums.py b/backend/common/enums.py index 8793ea9..99c76ba 100644 --- a/backend/common/enums.py +++ b/backend/common/enums.py @@ -49,9 +49,11 @@ EXCEPTION_HTTP_INVALID_TOKEN_ACCESS = HTTPException(status_code=ErrorType.HTTP_I class DBType(Enum): """논리 DB 구분. 모델마다 DBType() 으로 자신이 속한 DB 를 반환한다. DB 가 늘어나면 여기에 추가하고 db_session_manager 의 맵에 등록만 하면 된다. + 물리적으로 같은 negosium_db 라도 도메인별 논리 구분으로 나눠 둘 수 있다(커넥션 config 는 재사용). """ - MAIN = 1 + USER = 1 # 기본 유저 (supplier_users 테이블) + PARTNER = 2 # partner 도메인 (partner.suppliers 등) class DBWRType(Enum): @@ -59,3 +61,28 @@ class DBWRType(Enum): DB_READ = 1 DB_WRITE = 2 + + +# ============================================================ +# 도메인 코드값. 스키마는 SMALLINT 정수 코드(1부터)로 두고, 의미 매핑은 여기 enum 으로 한다. +# (postgres-init/01-schema.sql: "코드값(status/role/type 등)은 SMALLINT 정수 코드로 둔다") +# ============================================================ +class AccountStatus(Enum): + """계정 상태 코드. company.users / supplier.supplier_users 의 status 컬럼.""" + + ACTIVE = 1 # 활성 + INACTIVE = 2 # 비활성 + + +class UserRole(Enum): + """유저 권한 코드. company.users / supplier.supplier_users 의 role 컬럼.""" + + USER = 1 # 일반 유저 + MANAGER = 2 # 매니저 + + +class TokenType(Enum): + """토큰 종류 코드. supplier.supplier_user_tokens 의 type 컬럼.""" + + ACCESS = 1 + REFRESH = 2 diff --git a/backend/common/models/gmodel.py b/backend/common/models/gmodel.py index b09f002..36d3ab9 100644 --- a/backend/common/models/gmodel.py +++ b/backend/common/models/gmodel.py @@ -46,11 +46,14 @@ class Res_WebPacketProtocol(WebPacketProtocol): class UserInfo(StructModel): - """JWT subject 로 인코딩되는 유저 식별 정보.""" + """JWT subject 로 인코딩되는 유저 식별 정보. su_id/supplier_id 는 uuid 문자열로 인코딩한다.""" - uid: int + su_id: str id: str - nickname: str + name: str + supplier_id: str + supplier_name: str + role: int def __init__(self, *args, **kwargs) -> None: super().__init__() diff --git a/backend/conftest.py b/backend/conftest.py index d5a3fe8..b1f138f 100644 --- a/backend/conftest.py +++ b/backend/conftest.py @@ -6,7 +6,6 @@ os.environ.setdefault("APP_ENV", "local") import pytest_asyncio from httpx import ASGITransport, AsyncClient -from sqlalchemy import text from sqlalchemy.ext.asyncio import create_async_engine from common.database.model.models import MAIN_BASE @@ -20,15 +19,13 @@ def _write_url(cfg) -> str: @pytest_asyncio.fixture async def db_engine(): - """테스트용 스키마를 보장하고, 매 테스트 시작 시 테이블을 비워 격리한다. + """테스트용 스키마를 보장한다. 격리는 각 테스트가 전용 행만 시드/정리하는 방식으로 한다. - 앱(DB_SESSION_MNG)은 자체 엔진으로 같은 DB(config.test.toml)에 접속하므로, - 여기서 만든 스키마를 그대로 공유한다. + 앱(DB_SESSION_MNG)은 자체 엔진으로 같은 DB 에 접속하므로 여기서 만든 스키마를 그대로 공유한다. """ engine = create_async_engine(_write_url(main_db_config)) async with engine.begin() as conn: await conn.run_sync(MAIN_BASE.metadata.create_all) # 이미 있으면 skip - await conn.execute(text("TRUNCATE TABLE tbl_account")) yield engine await engine.dispose() diff --git a/backend/crud/user_crud.py b/backend/crud/user_crud.py index b2a7ac6..ee70bfd 100644 --- a/backend/crud/user_crud.py +++ b/backend/crud/user_crud.py @@ -1,12 +1,12 @@ from abc import ABC, abstractmethod from typing import Tuple -from sqlalchemy import select, update +from sqlalchemy import delete, select, update from sqlalchemy.ext.asyncio import AsyncSession from common.database.db_session_manager import DB_SESSION_MNG -from common.database.model.models import tbl_account -from common.enums import ErrorType +from common.database.model.models import supplier_user_tokens, supplier_users, suppliers +from common.enums import ErrorType, TokenType from common.logger import LOG from common.utils.gtime import GTime @@ -14,29 +14,54 @@ from common.utils.gtime import GTime # CRUD 는 인터페이스(I*) 와 구현(*) 으로 분리한다. # - service 는 인터페이스 타입에 의존하고 Depends 로 구현을 주입받는다 (테스트/교체 용이). # - 모든 메서드는 (session, ...) 을 받는다. session 은 람다 호출 시 매니저가 넘겨준다. +# - 유저는 supplier_users 테이블, 공급사명은 partner.suppliers 에서 조회한다(no-FK). class IUserCRUD(ABC): @abstractmethod - async def get_account_by_id(self, cdb: AsyncSession, user_id: str) -> Tuple[ErrorType, tbl_account]: + async def get_account_by_id(self, cdb: AsyncSession, login_id: str) -> Tuple[ErrorType, supplier_users]: pass @abstractmethod - async def is_account(self, cdb: AsyncSession, user_id: str) -> ErrorType: + async def get_account_by_su_id(self, cdb: AsyncSession, su_id) -> Tuple[ErrorType, supplier_users]: pass @abstractmethod - async def add_account(self, cdb: AsyncSession, account: tbl_account) -> ErrorType: + async def get_supplier_name(self, cdb: AsyncSession, supplier_id) -> Tuple[ErrorType, str]: pass @abstractmethod - async def update_last_login(self, cdb: AsyncSession, user_uid: int) -> ErrorType: + async def is_account(self, cdb: AsyncSession, login_id: str) -> ErrorType: + pass + + @abstractmethod + async def add_account(self, cdb: AsyncSession, account: supplier_users) -> ErrorType: + pass + + @abstractmethod + async def add_token(self, cdb: AsyncSession, token: supplier_user_tokens) -> ErrorType: + pass + + @abstractmethod + async def delete_tokens_by_su_id(self, cdb: AsyncSession, su_id) -> ErrorType: + pass + + @abstractmethod + async def update_access_token(self, cdb: AsyncSession, su_id, token, issued_at, expired_at) -> ErrorType: + pass + + @abstractmethod + async def update_last_accessed(self, cdb: AsyncSession, su_id) -> ErrorType: pass class UserCRUD(IUserCRUD): - async def get_account_by_id(self, cdb: AsyncSession, user_id: str) -> Tuple[ErrorType, tbl_account]: + async def get_account_by_id(self, cdb: AsyncSession, login_id: str) -> Tuple[ErrorType, supplier_users]: try: - query = select(tbl_account).where(tbl_account.id == user_id).limit(1) - err_type, row_list = await DB_SESSION_MNG.execute(cdb, query, f"get_account_by_id(ID:{user_id}) failed.") + query = ( + select(supplier_users) + .where(supplier_users.id == login_id, supplier_users.deleted == False) # noqa: E712 + .limit(1) + ) + err_type, row_list = await DB_SESSION_MNG.execute(cdb, query, f"get_account_by_id(ID:{login_id}) failed.") if err_type != ErrorType.SUCCESS: return err_type, None if len(row_list) != 1: @@ -46,9 +71,47 @@ class UserCRUD(IUserCRUD): LOG.e_no_callstack(ex) return ErrorType.DB_RUN_FAILED, None - async def is_account(self, cdb: AsyncSession, user_id: str) -> ErrorType: + async def get_account_by_su_id(self, cdb: AsyncSession, su_id) -> Tuple[ErrorType, supplier_users]: try: - query = select(tbl_account).where(tbl_account.id == user_id).limit(1) + query = ( + select(supplier_users) + .where(supplier_users.su_id == su_id, supplier_users.deleted == False) # noqa: E712 + .limit(1) + ) + err_type, row_list = await DB_SESSION_MNG.execute(cdb, query, f"get_account_by_su_id(su_id:{su_id}) failed.") + if err_type != ErrorType.SUCCESS: + return err_type, None + if len(row_list) != 1: + return ErrorType.DB_INVALID_KEY, None + return ErrorType.SUCCESS, row_list[0] + except Exception as ex: + LOG.e_no_callstack(ex) + return ErrorType.DB_RUN_FAILED, None + + async def get_supplier_name(self, cdb: AsyncSession, supplier_id) -> Tuple[ErrorType, str]: + try: + query = ( + select(suppliers.name) + .where(suppliers.supplier_id == supplier_id, suppliers.deleted == False) # noqa: E712 + .limit(1) + ) + err_type, row_list = await DB_SESSION_MNG.execute(cdb, query, f"get_supplier_name(supplier_id:{supplier_id}) failed.") + if err_type != ErrorType.SUCCESS: + return err_type, None + if len(row_list) != 1: + return ErrorType.DB_INVALID_KEY, None + return ErrorType.SUCCESS, row_list[0] + except Exception as ex: + LOG.e_no_callstack(ex) + return ErrorType.DB_RUN_FAILED, None + + async def is_account(self, cdb: AsyncSession, login_id: str) -> ErrorType: + try: + query = ( + select(supplier_users) + .where(supplier_users.id == login_id, supplier_users.deleted == False) # noqa: E712 + .limit(1) + ) err_type, row_list = await DB_SESSION_MNG.execute(cdb, query) if err_type != ErrorType.SUCCESS: return err_type @@ -59,16 +122,49 @@ class UserCRUD(IUserCRUD): LOG.e_no_callstack(ex) return ErrorType.DB_RUN_FAILED - async def add_account(self, cdb: AsyncSession, account: tbl_account) -> ErrorType: + async def add_account(self, cdb: AsyncSession, account: supplier_users) -> ErrorType: try: return await DB_SESSION_MNG.insert(cdb, account) except Exception as ex: LOG.e_no_callstack(ex) return ErrorType.DB_RUN_FAILED - async def update_last_login(self, cdb: AsyncSession, user_uid: int) -> ErrorType: + async def add_token(self, cdb: AsyncSession, token: supplier_user_tokens) -> ErrorType: try: - query = update(tbl_account).where(tbl_account.uid == user_uid).values(last_login_at=GTime.UTC()) + return await DB_SESSION_MNG.insert(cdb, token) + except Exception as ex: + LOG.e_no_callstack(ex) + return ErrorType.DB_RUN_FAILED + + async def delete_tokens_by_su_id(self, cdb: AsyncSession, su_id) -> ErrorType: + # 단일 세션: 로그인/로그아웃 시 해당 유저의 토큰 행을 모두 제거한다(하드 삭제, 누적 방지). + try: + query = delete(supplier_user_tokens).where(supplier_user_tokens.su_id == su_id) + return await DB_SESSION_MNG.add(cdb, query) + except Exception as ex: + LOG.e_no_callstack(ex) + return ErrorType.DB_RUN_FAILED + + async def update_access_token(self, cdb: AsyncSession, su_id, token, issued_at, expired_at) -> ErrorType: + # 재발급 시 저장된 access 행만 새 토큰으로 갱신한다. + try: + query = ( + update(supplier_user_tokens) + .where( + supplier_user_tokens.su_id == su_id, + supplier_user_tokens.type == TokenType.ACCESS.value, + supplier_user_tokens.deleted == False, # noqa: E712 + ) + .values(token=token, issued_at=issued_at, expired_at=expired_at) + ) + return await DB_SESSION_MNG.add(cdb, query) + except Exception as ex: + LOG.e_no_callstack(ex) + return ErrorType.DB_RUN_FAILED + + async def update_last_accessed(self, cdb: AsyncSession, su_id) -> ErrorType: + try: + query = update(supplier_users).where(supplier_users.su_id == su_id).values(last_accessed_at=GTime.UTC()) return await DB_SESSION_MNG.add(cdb, query) except Exception as ex: LOG.e_no_callstack(ex) diff --git a/backend/requirements.txt b/backend/requirements.txt index 2f928b6..332cd3b 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -1,6 +1,7 @@ fastapi uvicorn[standard] sqlalchemy>=2.0 +greenlet # SQLAlchemy async 의 sync/async 브리지에 필수 (일부 환경에서 자동 설치 누락됨) asyncpg python-jose[cryptography] bcrypt diff --git a/backend/router/v1/auth/account.py b/backend/router/v1/auth/account.py index b5d4648..df3de34 100644 --- a/backend/router/v1/auth/account.py +++ b/backend/router/v1/auth/account.py @@ -1,12 +1,9 @@ from fastapi import APIRouter, Depends, Request -from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from common.models.gmodel import UserInfo from router.v1.validator.dependencies import IsValidAccessToken, IsValidRefreshToken, RemoveNoneResponse from services.auth_service import AuthService -from .protocol import Req_CreateAccount, Req_Login, Res_CreateAccount, Res_Login, Res_RefreshToken - -security = HTTPBearer() +from .protocol import Req_CreateAccount, Req_Login, Res_CreateAccount, Res_Login, Res_Me, Res_RefreshToken # 라우터(MVC 의 컨트롤러). 요청 검증 -> service 호출 -> RemoveNoneResponse 반환만 담당. router = APIRouter(prefix="/v1/auth", tags=["Auth"], responses={404: {"description": "Not found"}}) @@ -17,26 +14,31 @@ async def login(request: Request, req: Req_Login, service: AuthService = Depends return RemoveNoneResponse(await service.attempt_login(req.id, req.pw, request.client.host)) -@router.post(path="/create", response_model=Res_CreateAccount, summary="계정 생성", description="새 계정을 생성한다.") +# TODO: 계정 생성은 관리자/매니저 권한으로 제한할 가능성이 있음(현재는 비보호). +@router.post(path="/create", response_model=Res_CreateAccount, summary="계정 생성", description="supplier_id 소속의 유저를 생성한다(공급사 존재를 앱에서 검증).") async def create_account(request: Request, req: Req_CreateAccount, service: AuthService = Depends()): - return RemoveNoneResponse(await service.create_account(req.id, req.pw, req.nickname, request.client.host)) + return RemoveNoneResponse( + await service.create_account( + req.supplier_id, req.id, req.pw, req.name, req.email, req.contact_number, req.role, request.client.host + ) + ) @router.post( path="/refresh_token", - dependencies=[Depends(IsValidRefreshToken)], response_model=Res_RefreshToken, summary="액세스 토큰 갱신", - description="refresh 토큰으로 access 토큰을 재발급한다.", + description="refresh 토큰으로 access 토큰을 재발급한다. su_id DB 존재/활성은 service 에서 확인한다.", ) -async def refresh_token(service: AuthService = Depends(), credentials: HTTPAuthorizationCredentials = Depends(security)): - return RemoveNoneResponse(await service.refresh_token(credentials.credentials)) +async def refresh_token(user_info: UserInfo = Depends(IsValidRefreshToken), service: AuthService = Depends()): + return RemoveNoneResponse(await service.refresh_token(user_info)) @router.get( path="/me", - summary="내 정보 (보호된 엔드포인트 예시)", - description="유효한 access 토큰이 있어야 호출 가능. 토큰 검증 결과 UserInfo 를 주입받는다.", + response_model=Res_Me, + summary="내 정보 (보호된 엔드포인트)", + description="access 토큰 검증(validator) 후 su_id DB 존재/활성을 service 에서 확인해 반환한다.", ) -async def me(user_info: UserInfo = Depends(IsValidAccessToken)): - return {"uid": user_info.uid, "id": user_info.id, "nickname": user_info.nickname} +async def me(user_info: UserInfo = Depends(IsValidAccessToken), service: AuthService = Depends()): + return RemoveNoneResponse(await service.get_me(user_info)) diff --git a/backend/router/v1/auth/protocol.py b/backend/router/v1/auth/protocol.py index 2ef226e..c3cda59 100644 --- a/backend/router/v1/auth/protocol.py +++ b/backend/router/v1/auth/protocol.py @@ -1,5 +1,3 @@ -from pydantic import Field - from common.models.gmodel import Res_WebPacketProtocol, WebPacketProtocol @@ -14,21 +12,37 @@ class Req_Login(AuthProtocol): class Res_Login(Res_WebPacketProtocol): - uid: int = Field(0, description="user uid", json_schema_extra={"format": "int64"}) - nickname: str = "" + su_id: str = "" + name: str = "" # 유저 개인 이름 + supplier_id: str = "" # 소속 공급사(partner.suppliers) + supplier_name: str = "" # 공급사명 + role: int = 0 access_token: str = "" refresh_token: str = "" class Req_CreateAccount(AuthProtocol): - id: str = "" + supplier_id: str = "" # 소속 공급사(partner.suppliers.supplier_id) + id: str = "" # 로그인 ID pw: str = "" - nickname: str = "" + name: str = "" + email: str = "" + contact_number: str = "" + role: int = 1 # 1=user, 2=manager (UserRole) class Res_CreateAccount(Res_WebPacketProtocol): - uid: int = Field(0, description="생성된 user uid", json_schema_extra={"format": "int64"}) + su_id: str = "" class Res_RefreshToken(Res_WebPacketProtocol): access_token: str = "" + + +class Res_Me(Res_WebPacketProtocol): + su_id: str = "" + id: str = "" + name: str = "" + supplier_id: str = "" + supplier_name: str = "" + role: int = 0 diff --git a/backend/router/v1/validator/dependencies.py b/backend/router/v1/validator/dependencies.py index 73ad74e..08d676f 100644 --- a/backend/router/v1/validator/dependencies.py +++ b/backend/router/v1/validator/dependencies.py @@ -88,8 +88,7 @@ def DecodeRefreshToken(jwt_token: str) -> UserInfo: return __decode_token(jwt_token, JWT_REFRESH_SECRET, EXCEPTION_REFRESH_TOKEN_EXPIRED) -# ---- Depends 용 토큰 검증기 ------------------------------------------------ -# 보호된 엔드포인트에서 dependencies=[Depends(IsValidAccessToken)] 로 사용. +# ---- Depends 용 토큰 검증기 (디코드만; DB 존재/활성 검증은 service 가 담당) ----- async def IsValidAccessToken(credentials: HTTPAuthorizationCredentials = Depends(security)) -> UserInfo: return DecodeAccessToken(credentials.credentials) diff --git a/backend/services/auth_service.py b/backend/services/auth_service.py index 1d41675..e91c48c 100644 --- a/backend/services/auth_service.py +++ b/backend/services/auth_service.py @@ -1,24 +1,23 @@ +import uuid + from fastapi import Depends from common.database.db_session_manager import DB_SESSION_MNG -from common.database.model.models import tbl_account -from common.enums import DBWRType, ErrorType +from common.database.model.models import supplier_user_tokens, supplier_users, suppliers +from common.enums import AccountStatus, DBWRType, ErrorType, TokenType from common.logger import LOG from common.models.gmodel import UserInfo +from common.utils.gtime import GTime +from config.server_configs import jwt_token_config from crud.user_crud import IUserCRUD, UserCRUD -from router.v1.auth.protocol import Res_CreateAccount, Res_Login, Res_RefreshToken -from router.v1.validator.dependencies import ( - CreateAccessToken, - CreateRefreshToken, - DecodeRefreshToken, - GetHashedPW, - VerifyPW, -) +from router.v1.auth.protocol import Res_CreateAccount, Res_Login, Res_Me, Res_RefreshToken +from router.v1.validator.dependencies import CreateAccessToken, CreateRefreshToken, GetHashedPW, VerifyPW class AuthService: """비즈니스 로직 계층 (MVC 의 컨트롤러-서비스 분리에서 서비스). + - 유저는 supplier_users 테이블(JWT subject = UserInfo). uuid 는 문자열로 인코딩. - CRUD 는 Depends 로 인터페이스 타입으로 주입받는다. - DB 접근은 DB_SESSION_MNG 의 람다 실행으로만 한다. 조회 = execute_lambda(..., DB_READ, lambda s: crud.xxx(s, ...)) @@ -29,57 +28,33 @@ class AuthService: def __init__(self, user_crud: IUserCRUD = Depends(UserCRUD)): self.user_crud = user_crud - async def attempt_login(self, id: str, pw: str, connect_ip: str) -> Res_Login: - LOG.i(f"LOGIN : {id=}") - res = Res_Login() - - # 1) 계정 조회 (Read DB) - err_type, account = await DB_SESSION_MNG.execute_lambda( - tbl_account.DBType(), - DBWRType.DB_READ.value, - lambda s: self.user_crud.get_account_by_id(s, id), - ) - if err_type != ErrorType.SUCCESS: - # 계정 없음/조회 실패 모두 로그인 실패로 일반화 - res.result.SetResult(ErrorType.ACCOUNT_INVALID_INFO) - return res - account: tbl_account - - # 2) 비밀번호 검증 - if not await VerifyPW(pw, account.pw): - res.result.SetResult(ErrorType.ACCOUNT_INVALID_INFO) - return res - - # 3) 차단 여부 - if account.is_blocked: - res.result.SetResult(ErrorType.ACCOUNT_BLOCKED_USER) - return res - - # 4) 토큰 발급 - user_info = UserInfo(uid=account.uid, id=account.id, nickname=account.nickname) - res.access_token = CreateAccessToken(user_info) - res.refresh_token = CreateRefreshToken(user_info) - - # 5) 마지막 로그인 시간 갱신 (Write DB, 트랜잭션) - err_type = await DB_SESSION_MNG.execute_lambda_run( - [tbl_account.DBType()], - [lambda s: self.user_crud.update_last_login(s, account.uid)], - ) - if err_type != ErrorType.SUCCESS: - res.result.SetResult(err_type) - return res - - res.uid = account.uid - res.nickname = account.nickname - return res - - async def create_account(self, id: str, pw: str, nickname: str, connect_ip: str) -> Res_CreateAccount: - LOG.i(f"CREATE : {id=}, {nickname=}") + async def create_account( + self, supplier_id: str, id: str, pw: str, name: str, email: str, contact_number: str, role: int, connect_ip: str + ) -> Res_CreateAccount: + LOG.i(f"CREATE : {id=}, {supplier_id=}") res = Res_CreateAccount() - # 1) 중복 ID 확인 (Read DB) + # 0) supplier_id 형식 검증 + try: + sid = uuid.UUID(supplier_id) + except (ValueError, TypeError): + res.result.SetResult(ErrorType.INVALID_REQUEST_DATA) + return res + + # 1) 공급사 존재 확인 (no-FK 라 앱에서 무결성 검증, PARTNER Read) + err_type, _ = await DB_SESSION_MNG.execute_lambda( + suppliers.DBType(), + DBWRType.DB_READ.value, + lambda s: self.user_crud.get_supplier_name(s, sid), + ) + if err_type != ErrorType.SUCCESS: + # 존재하지 않는 supplier_id + res.result.SetResult(ErrorType.INVALID_REQUEST_DATA) + return res + + # 2) 중복 로그인 ID 확인 (USER Read) err_type = await DB_SESSION_MNG.execute_lambda( - tbl_account.DBType(), + supplier_users.DBType(), DBWRType.DB_READ.value, lambda s: self.user_crud.is_account(s, id), ) @@ -90,26 +65,188 @@ class AuthService: res.result.SetResult(err_type) return res - # 2) 계정 생성 (비밀번호는 bcrypt 해시로 저장) - account = tbl_account(id=id, pw=await GetHashedPW(pw), nickname=nickname or id) + # 3) 생성 (pw bcrypt 해시. last_accessed_at 은 NOT NULL/무기본값이라 생성 시각으로 둔다) + account = supplier_users( + supplier_id=sid, + id=id, + password=await GetHashedPW(pw), + name=name or None, + email=email or None, + contact_number=contact_number or None, + last_accessed_at=GTime.UTC(), + status=AccountStatus.ACTIVE.value, + role=role, + ) err_type = await DB_SESSION_MNG.execute_lambda_run( - [tbl_account.DBType()], + [supplier_users.DBType()], [lambda s: self.user_crud.add_account(s, account)], ) if err_type != ErrorType.SUCCESS: - # 사전 검사와 INSERT 사이의 경쟁 조건에서 unique 위반이 나면 동일 코드로 매핑. + # 사전 검사와 INSERT 사이 경쟁 조건의 unique 위반은 동일 코드로 매핑. if err_type == ErrorType.DB_ALREADY_SAME_KEY: res.result.SetResult(ErrorType.ACCOUNT_ALREADY_EXIST) else: res.result.SetResult(err_type) return res - res.uid = account.uid + res.su_id = str(account.su_id) return res - async def refresh_token(self, refresh_token: str) -> Res_RefreshToken: - res = Res_RefreshToken() - # refresh 토큰 검증은 라우터 Depends(IsValidRefreshToken) 에서 1차 수행됨. - user_info = DecodeRefreshToken(refresh_token) + async def attempt_login(self, id: str, pw: str, connect_ip: str) -> Res_Login: + LOG.i(f"LOGIN : {id=}") + res = Res_Login() + + # 1) 계정 조회 (USER Read 세션) + err_type, account = await DB_SESSION_MNG.execute_lambda( + supplier_users.DBType(), + DBWRType.DB_READ.value, + lambda s: self.user_crud.get_account_by_id(s, id), + ) + if err_type != ErrorType.SUCCESS: + # 계정 없음/조회 실패 모두 로그인 실패로 일반화 + res.result.SetResult(ErrorType.ACCOUNT_INVALID_INFO) + return res + account: supplier_users + + # 2) 비밀번호 검증 + if not await VerifyPW(pw, account.password): + res.result.SetResult(ErrorType.ACCOUNT_INVALID_INFO) + return res + + # 3) 상태 확인 (active 만 허용) + if account.status != AccountStatus.ACTIVE.value: + res.result.SetResult(ErrorType.ACCOUNT_BLOCKED_USER) + return res + + # 3-1) 공급사명 조회 (PARTNER Read 세션). 부가 정보라 실패해도 로그인은 막지 않고 빈 값. + _, sname = await DB_SESSION_MNG.execute_lambda( + suppliers.DBType(), + DBWRType.DB_READ.value, + lambda s: self.user_crud.get_supplier_name(s, account.supplier_id), + ) + supplier_name = sname or "" + + # 4) 토큰 발급 + user_info = UserInfo( + su_id=str(account.su_id), + id=account.id, + name=account.name or "", + supplier_id=str(account.supplier_id), + supplier_name=supplier_name, + role=account.role, + ) res.access_token = CreateAccessToken(user_info) + res.refresh_token = CreateRefreshToken(user_info) + + # 5) 마지막 접속 시간 갱신 + 토큰 교체 (Write DB, 한 트랜잭션) + # 단일 세션: 이전 토큰 행을 모두 지우고 access/refresh 2행을 새로 넣어 이전 세션을 무효화한다. + # 저장된 토큰은 추후 로그아웃/검증(토큰 대조)에서 사용한다. + now = GTime.UTC() + access_row = supplier_user_tokens( + su_id=account.su_id, + type=TokenType.ACCESS.value, + token={"jwt": res.access_token}, + issued_at=now, + expired_at=GTime.AddMinutes(jwt_token_config.access_expire_min), + ) + refresh_row = supplier_user_tokens( + su_id=account.su_id, + type=TokenType.REFRESH.value, + token={"jwt": res.refresh_token}, + issued_at=now, + expired_at=GTime.AddDays(jwt_token_config.refresh_expire_day), + ) + err_type = await DB_SESSION_MNG.execute_lambda_run( + [supplier_users.DBType()], + [ + lambda s: self.user_crud.update_last_accessed(s, account.su_id), + lambda s: self.user_crud.delete_tokens_by_su_id(s, account.su_id), # 단일 세션: 이전 토큰 제거 + lambda s: self.user_crud.add_token(s, access_row), + lambda s: self.user_crud.add_token(s, refresh_row), + ], + ) + if err_type != ErrorType.SUCCESS: + res.result.SetResult(err_type) + return res + + res.su_id = str(account.su_id) + res.name = account.name or "" + res.supplier_id = str(account.supplier_id) + res.supplier_name = supplier_name + res.role = account.role + return res + + async def __load_active_account(self, su_id_str: str) -> tuple[ErrorType, UserInfo]: + """su_id 로 유저를 조회해 존재 + status=active 확인 후, 공급사명까지 채운 DB 최신값 UserInfo 를 + 반환한다. (토큰 발급 후 삭제/비활성된 계정 차단용) + 실패 시 (에러코드, None) 을 반환하며, HTTP 변환은 라우터가 result 로 내려보낸다. + """ + # 1) 계정 조회 (USER Read 세션) + err_type, account = await DB_SESSION_MNG.execute_lambda( + supplier_users.DBType(), + DBWRType.DB_READ.value, + lambda s: self.user_crud.get_account_by_su_id(s, uuid.UUID(su_id_str)), + ) + if err_type != ErrorType.SUCCESS or account is None: + return ErrorType.ACCOUNT_INVALID_INFO, None + if account.status != AccountStatus.ACTIVE.value: # active 만 허용 + return ErrorType.ACCOUNT_BLOCKED_USER, None + + # 2) 공급사명 조회 (PARTNER Read 세션). 부가 정보라 실패해도 빈 값. + _, sname = await DB_SESSION_MNG.execute_lambda( + suppliers.DBType(), + DBWRType.DB_READ.value, + lambda s: self.user_crud.get_supplier_name(s, account.supplier_id), + ) + return ErrorType.SUCCESS, UserInfo( + su_id=str(account.su_id), + id=account.id, + name=account.name or "", + supplier_id=str(account.supplier_id), + supplier_name=sname or "", + role=account.role, + ) + + async def get_me(self, user_info: UserInfo) -> Res_Me: + # 토큰 디코드는 라우터 Depends(IsValidAccessToken) 에서 수행됨. 여기선 su_id DB 검증. + res = Res_Me() + err_type, info = await self.__load_active_account(user_info.su_id) + if err_type != ErrorType.SUCCESS: + res.result.SetResult(err_type) + return res + res.su_id = info.su_id + res.id = info.id + res.name = info.name + res.supplier_id = info.supplier_id + res.supplier_name = info.supplier_name + res.role = info.role + return res + + async def refresh_token(self, user_info: UserInfo) -> Res_RefreshToken: + # 토큰 디코드는 라우터 Depends(IsValidRefreshToken) 에서 수행됨. 여기선 su_id DB 검증 후 재발급. + res = Res_RefreshToken() + err_type, info = await self.__load_active_account(user_info.su_id) + if err_type != ErrorType.SUCCESS: + res.result.SetResult(err_type) + return res + + new_access = CreateAccessToken(info) # DB 최신값으로 재구성한 토큰 + # 단일 세션: 저장된 access 행을 새 토큰으로 갱신한다(refresh 행은 유지). + err_type = await DB_SESSION_MNG.execute_lambda_run( + [supplier_users.DBType()], + [ + lambda s: self.user_crud.update_access_token( + s, + uuid.UUID(info.su_id), + {"jwt": new_access}, + GTime.UTC(), + GTime.AddMinutes(jwt_token_config.access_expire_min), + ) + ], + ) + if err_type != ErrorType.SUCCESS: + res.result.SetResult(err_type) + return res + + res.access_token = new_access return res diff --git a/backend/tests/test_auth.py b/backend/tests/test_auth.py index 47963c4..c59f866 100644 --- a/backend/tests/test_auth.py +++ b/backend/tests/test_auth.py @@ -1,63 +1,272 @@ -"""auth 도메인 e2e 테스트. +"""인증 e2e 테스트 (supplier_users 기반 유저). -실행 전제: docker-compose 로 PostgreSQL 이 떠 있어야 한다 (negosium_db 사용). - docker compose up -d # 또는 로컬 postgres +실행 전제: PostgreSQL 이 떠 있어야 한다 (negosium_db, supplier/partner 스키마 적용). cd backend && python -m pytest + +테스트는 dev negosium_db 를 그대로 쓰므로, 다른 데이터를 건드리지 않도록 +TRUNCATE 대신 전용 테스트 행(pytest_user)만 시드/정리한다. """ +import uuid -async def test_create_and_login_flow(client): - # 1) 계정 생성 - r = await client.post("/v1/auth/create", json={"id": "user1", "pw": "pw1234", "nickname": "닉네임"}) +import bcrypt +import pytest_asyncio +from sqlalchemy import text + +TEST_LOGIN_ID = "pytest_user" +TEST_PW = "pytest1234" +TEST_USER_NAME = "테스트담당자" # supplier_users.name (유저 개인 이름) +TEST_SUPPLIER_NAME = "파이테스트공급사" # partner.suppliers.name (공급사명) + + +@pytest_asyncio.fixture +async def account_seed(db_engine): + """partner.suppliers(공급사) + supplier.supplier_users(유저) 테스트 행을 시드하고, 끝나면 정리한다.""" + supplier_id = uuid.uuid4() + pw_hash = bcrypt.hashpw(TEST_PW.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") + + async def _cleanup(conn): + await conn.execute( + text( + "DELETE FROM supplier.supplier_user_tokens WHERE su_id IN " + "(SELECT su_id FROM supplier.supplier_users WHERE id = :id)" + ), + {"id": TEST_LOGIN_ID}, + ) + await conn.execute(text("DELETE FROM supplier.supplier_users WHERE id = :id"), {"id": TEST_LOGIN_ID}) + await conn.execute(text("DELETE FROM partner.suppliers WHERE name = :n"), {"n": TEST_SUPPLIER_NAME}) + + async with db_engine.begin() as conn: + await _cleanup(conn) # 이전 실행 잔재 제거 + await conn.execute( + text( + "INSERT INTO partner.suppliers (supplier_id, company_id, user_id, name) " + "VALUES (:sid, gen_random_uuid(), gen_random_uuid(), :name)" + ), + {"sid": supplier_id, "name": TEST_SUPPLIER_NAME}, + ) + await conn.execute( + text( + "INSERT INTO supplier.supplier_users " + "(supplier_id, id, password, name, last_accessed_at, status, role) " + "VALUES (:sid, :id, :pw, :uname, now(), 1, 1)" + ), + {"sid": supplier_id, "id": TEST_LOGIN_ID, "pw": pw_hash, "uname": TEST_USER_NAME}, + ) + + yield {"supplier_id": supplier_id} + + # 테스트 중 생성된 유저/토큰까지 정리하기 위해 supplier_id 기준으로 지운다. + async with db_engine.begin() as conn: + await conn.execute( + text( + "DELETE FROM supplier.supplier_user_tokens WHERE su_id IN " + "(SELECT su_id FROM supplier.supplier_users WHERE supplier_id = :sid)" + ), + {"sid": supplier_id}, + ) + await conn.execute(text("DELETE FROM supplier.supplier_users WHERE supplier_id = :sid"), {"sid": supplier_id}) + await conn.execute(text("DELETE FROM partner.suppliers WHERE supplier_id = :sid"), {"sid": supplier_id}) + + +async def _set_account(db_engine, **values): + """테스트용 유저 행의 컬럼을 갱신한다 (status/deleted 등).""" + sets = ", ".join(f"{k} = :{k}" for k in values) + params = {**values, "id": TEST_LOGIN_ID} + async with db_engine.begin() as conn: + await conn.execute(text(f"UPDATE supplier.supplier_users SET {sets} WHERE id = :id"), params) + + +async def _login(client): + return await client.post("/v1/auth/login", json={"id": TEST_LOGIN_ID, "pw": TEST_PW}) + + +# ---- 생성 ------------------------------------------------------------------- +async def test_create_success(client, account_seed): + sid = str(account_seed["supplier_id"]) + r = await client.post( + "/v1/auth/create", + json={"supplier_id": sid, "id": "pytest_new", "pw": "newpw1234", "name": "새담당자", "role": 2}, + ) assert r.status_code == 200 body = r.json() assert body["result"]["success"] is True - assert body["uid"] > 0 + assert body["su_id"] + # 생성된 계정으로 즉시 로그인 가능 + r2 = await client.post("/v1/auth/login", json={"id": "pytest_new", "pw": "newpw1234"}) + lb = r2.json() + assert lb["result"]["success"] is True + assert lb["role"] == 2 # 매니저로 생성됨 - # 2) 로그인 -> 토큰 발급 - r = await client.post("/v1/auth/login", json={"id": "user1", "pw": "pw1234"}) + +async def test_create_duplicate(client, account_seed): + sid = str(account_seed["supplier_id"]) + payload = {"supplier_id": sid, "id": "pytest_dup", "pw": "x12345"} + r1 = await client.post("/v1/auth/create", json=payload) + assert r1.json()["result"]["success"] is True + r2 = await client.post("/v1/auth/create", json=payload) + assert r2.json()["result"]["code"] == 1201 # ACCOUNT_ALREADY_EXIST + + +async def test_create_invalid_supplier(client, account_seed): + # 존재하지 않는 supplier_id (no-FK 라 앱에서 검증) + r = await client.post( + "/v1/auth/create", + json={"supplier_id": str(uuid.uuid4()), "id": "pytest_orphan", "pw": "x12345"}, + ) + body = r.json() + assert body["result"]["success"] is False + assert body["result"]["code"] == 101 # INVALID_REQUEST_DATA + + +async def test_create_malformed_supplier_id(client, account_seed): + r = await client.post( + "/v1/auth/create", + json={"supplier_id": "not-a-uuid", "id": "pytest_bad", "pw": "x12345"}, + ) + assert r.json()["result"]["code"] == 101 # INVALID_REQUEST_DATA + + +# ---- 로그인 ----------------------------------------------------------------- +async def test_login_success(client, account_seed): + r = await _login(client) assert r.status_code == 200 body = r.json() assert body["result"]["success"] is True assert body["access_token"] assert body["refresh_token"] - assert body["nickname"] == "닉네임" - access_token = body["access_token"] - - # 3) 보호된 엔드포인트 호출 - r = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {access_token}"}) - assert r.status_code == 200 - assert r.json()["id"] == "user1" + assert body["name"] == TEST_USER_NAME # 유저 개인 이름 + assert body["supplier_name"] == TEST_SUPPLIER_NAME # 공급사명(partner.suppliers) + assert body["role"] == 1 + assert body["su_id"] + assert body["supplier_id"] == str(account_seed["supplier_id"]) -async def test_login_with_wrong_password(client): - await client.post("/v1/auth/create", json={"id": "user2", "pw": "correct", "nickname": "n"}) - - r = await client.post("/v1/auth/login", json={"id": "user2", "pw": "wrong"}) +async def test_login_wrong_password(client, account_seed): + r = await client.post("/v1/auth/login", json={"id": TEST_LOGIN_ID, "pw": "wrong"}) assert r.status_code == 200 body = r.json() assert body["result"]["success"] is False - # 자격증명 오류는 ACCOUNT_INVALID_INFO(1200) - assert body["result"]["code"] == 1200 - assert body.get("access_token", "") == "" # 실패 시 토큰은 빈 문자열 + assert body["result"]["code"] == 1200 # ACCOUNT_INVALID_INFO + assert body.get("access_token", "") == "" -async def test_login_nonexistent_account(client): - r = await client.post("/v1/auth/login", json={"id": "ghost", "pw": "whatever"}) - assert r.json()["result"]["success"] is False - - -async def test_duplicate_account_create(client): - r1 = await client.post("/v1/auth/create", json={"id": "dup", "pw": "pw1234", "nickname": "n"}) - assert r1.json()["result"]["success"] is True - - r2 = await client.post("/v1/auth/create", json={"id": "dup", "pw": "pw5678", "nickname": "n2"}) - body = r2.json() +async def test_login_nonexistent(client): + r = await client.post("/v1/auth/login", json={"id": "ghost_user", "pw": "whatever"}) + assert r.status_code == 200 + body = r.json() assert body["result"]["success"] is False - # ACCOUNT_ALREADY_EXIST(1201) - assert body["result"]["code"] == 1201 + assert body["result"]["code"] == 1200 -async def test_me_without_token_is_rejected(client): +async def _stored_tokens(db_engine, su_id): + """su_id 의 저장된 토큰을 {type: jwt} dict 로 반환.""" + import json + + async with db_engine.begin() as conn: + rows = ( + await conn.execute( + text("SELECT type, token FROM supplier.supplier_user_tokens WHERE su_id = :sid AND deleted = false"), + {"sid": uuid.UUID(su_id)}, + ) + ).fetchall() + out = {} + for t, tok in rows: + out[t] = (json.loads(tok) if isinstance(tok, str) else tok)["jwt"] + return out + + +async def test_login_stores_access_and_refresh(client, account_seed, db_engine): + # 로그인 시 access(type=1) + refresh(type=2) 2행이 저장되고, 응답 토큰과 일치한다. + body = (await _login(client)).json() + assert body["result"]["success"] is True + stored = await _stored_tokens(db_engine, body["su_id"]) + assert set(stored.keys()) == {1, 2} # ACCESS, REFRESH + assert stored[1] == body["access_token"] + assert stored[2] == body["refresh_token"] + + +async def test_relogin_replaces_tokens_single_session(client, account_seed, db_engine): + # 단일 세션: 재로그인해도 토큰 행이 누적되지 않고 항상 정확히 2행(access/refresh)만 유지된다. + await _login(client) + second = (await _login(client)).json() + su_id = second["su_id"] + async with db_engine.begin() as conn: + count = ( + await conn.execute( + text("SELECT count(*) FROM supplier.supplier_user_tokens WHERE su_id = :sid AND deleted = false"), + {"sid": uuid.UUID(su_id)}, + ) + ).scalar() + assert count == 2 # 누적되지 않음 (2번 로그인해도 2행) + stored = await _stored_tokens(db_engine, su_id) + assert stored[2] == second["refresh_token"] # 최신 로그인 토큰으로 교체됨 + + +async def test_login_inactive(client, account_seed, db_engine): + await _set_account(db_engine, status=2) # 비활성 + r = await _login(client) + body = r.json() + assert body["result"]["success"] is False + assert body["result"]["code"] == 1202 # ACCOUNT_BLOCKED_USER + + +# ---- /me (보호된 엔드포인트) ------------------------------------------------- +async def test_me_active(client, account_seed): + access = (await _login(client)).json()["access_token"] + r = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {access}"}) + assert r.status_code == 200 + body = r.json() + assert body["result"]["success"] is True + assert body["id"] == TEST_LOGIN_ID + assert body["supplier_name"] == TEST_SUPPLIER_NAME + + +async def test_me_inactive_after_token(client, account_seed, db_engine): + # 토큰 발급 후 계정이 비활성(status=2)되면 만료 전이라도 차단된다 (200 + result code). + access = (await _login(client)).json()["access_token"] + await _set_account(db_engine, status=2) + r = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {access}"}) + assert r.status_code == 200 + assert r.json()["result"]["code"] == 1202 # ACCOUNT_BLOCKED_USER + + +async def test_me_deleted_after_token(client, account_seed, db_engine): + # 소프트 삭제(deleted=TRUE)된 계정도 차단된다. + access = (await _login(client)).json()["access_token"] + await _set_account(db_engine, deleted=True) + r = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {access}"}) + assert r.status_code == 200 + assert r.json()["result"]["code"] == 1200 # ACCOUNT_INVALID_INFO (조회 안 됨) + + +async def test_me_without_token(client): r = await client.get("/v1/auth/me") assert r.status_code in (401, 403) # HTTPBearer 가 자격증명 없음을 거부 + + +async def test_me_invalid_token(client): + r = await client.get("/v1/auth/me", headers={"Authorization": "Bearer garbage.token.value"}) + assert r.status_code == 433 # HTTP_INVALID_CLIENT_ACCESS (validator 가 raise) + + +# ---- refresh ---------------------------------------------------------------- +async def test_refresh_success(client, account_seed): + refresh = (await _login(client)).json()["refresh_token"] + r = await client.post("/v1/auth/refresh_token", headers={"Authorization": f"Bearer {refresh}"}) + assert r.status_code == 200 + body = r.json() + assert body["result"]["success"] is True + assert body["access_token"] + + +async def test_refresh_inactive_after_token(client, account_seed, db_engine): + # 삭제/비활성 계정에는 토큰을 재발급하지 않는다. + refresh = (await _login(client)).json()["refresh_token"] + await _set_account(db_engine, status=2) + r = await client.post("/v1/auth/refresh_token", headers={"Authorization": f"Bearer {refresh}"}) + assert r.status_code == 200 + body = r.json() + assert body["result"]["success"] is False + assert body["result"]["code"] == 1202 + assert body.get("access_token", "") == ""