feat(backend): supplier_users 기반 인증 재구축 및 단일세션 토큰 저장

- tbl_account 예시 인증 제거, supplier_users 기반으로 일원화(generic 네이밍 재사용)
- DBType USER/PARTNER 분리, AccountStatus/UserRole/TokenType enum 추가
- 보호 요청 시 su_id DB 존재/활성 검증(stateless JWT 빈틈 보완), 공급사명(partner.suppliers) 응답 포함
- 로그인 시 단일 세션 access/refresh 토큰을 supplier_user_tokens 에 저장(재로그인 시 교체)
- greenlet 의존성 추가, 인증 e2e 테스트(test_auth.py) 재작성

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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):

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):
+class supplier_users(MAIN_BASE):
-    # 모델이 자신이 속한 논리 DB 를 알려준다 (람다 실행 시 DBType 으로 세션 선택).
+    # 이 프로젝트의 기본 유저. 실제 테이블은 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)
+    # gen_random_uuid() 는 pgcrypto 확장 기준. 코드값(status/role/type)은 SMALLINT 정수 코드(앱 enum 매핑).
-    id = Column(String(45), nullable=False, unique=True)  # 로그인 ID. 중복 가입 방지 위해 unique.
+    su_id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()"))                   # 유저 식별자(PK)
-    pw = Column(String(255), nullable=False, default="")  # bcrypt 해시 저장
+    supplier_id = Column(UUID(as_uuid=True), nullable=False)                                                         # 소속 공급사(partner.suppliers.supplier_id)
-    nickname = Column(String(45), nullable=False, default="")
+    id = Column(String(20), nullable=False)                                                                          # 로그인 ID
-    is_blocked = Column(Boolean, nullable=False, default=False)
+    password = Column(String(255), nullable=False)                                                                   # 해시된 비밀번호이어야 함
-    # PostgreSQL UTC now: now() 는 timestamptz 이므로 utc 로 변환해 timestamp 로 저장.
+    name = Column(String(50), nullable=True)                                                                         # 이름
-    last_login_at = Column(DateTime, nullable=False, server_default=text("(now() AT TIME ZONE 'utc')"))
+    email = Column(String(255), nullable=True)                                                                       # 이메일
-    create_at = Column(DateTime, server_default=text("(now() AT TIME ZONE 'utc')"))
+    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"))                                          # 소프트 삭제 여부

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

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__()

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()
 

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.database.model.models import supplier_user_tokens, supplier_users, suppliers
-from common.enums import ErrorType
+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)
+            query = (
-            err_type, row_list = await DB_SESSION_MNG.execute(cdb, query, f"get_account_by_id(ID:{user_id}) failed.")
+                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)

backend/requirements.txt

@@ -1,6 +1,7 @@
 fastapi
 uvicorn[standard]
 sqlalchemy>=2.0
+greenlet  # SQLAlchemy async 의 sync/async 브리지에 필수 (일부 환경에서 자동 설치 누락됨)
 asyncpg
 python-jose[cryptography]
 bcrypt

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
+from .protocol import Req_CreateAccount, Req_Login, Res_CreateAccount, Res_Login, Res_Me, Res_RefreshToken
-
-security = HTTPBearer()
 
 # 라우터(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)):
+async def refresh_token(user_info: UserInfo = Depends(IsValidRefreshToken), service: AuthService = Depends()):
-    return RemoveNoneResponse(await service.refresh_token(credentials.credentials))
+    return RemoveNoneResponse(await service.refresh_token(user_info))
 
 
 @router.get(
     path="/me",
-    summary="내 정보 (보호된 엔드포인트 예시)",
+    response_model=Res_Me,
-    description="유효한 access 토큰이 있어야 호출 가능. 토큰 검증 결과 UserInfo 를 주입받는다.",
+    summary="내 정보 (보호된 엔드포인트)",
+    description="access 토큰 검증(validator) 후 su_id DB 존재/활성을 service 에서 확인해 반환한다.",
 )
-async def me(user_info: UserInfo = Depends(IsValidAccessToken)):
+async def me(user_info: UserInfo = Depends(IsValidAccessToken), service: AuthService = Depends()):
-    return {"uid": user_info.uid, "id": user_info.id, "nickname": user_info.nickname}
+    return RemoveNoneResponse(await service.get_me(user_info))

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"})
+    su_id: str = ""
-    nickname: 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

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 용 토큰 검증기 ------------------------------------------------
+# ---- Depends 용 토큰 검증기 (디코드만; DB 존재/활성 검증은 service 가 담당) -----
-# 보호된 엔드포인트에서 dependencies=[Depends(IsValidAccessToken)] 로 사용.
 async def IsValidAccessToken(credentials: HTTPAuthorizationCredentials = Depends(security)) -> UserInfo:
     return DecodeAccessToken(credentials.credentials)
 

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.database.model.models import supplier_user_tokens, supplier_users, suppliers
-from common.enums import DBWRType, ErrorType
+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.auth.protocol import Res_CreateAccount, Res_Login, Res_Me, Res_RefreshToken
-from router.v1.validator.dependencies import (
+from router.v1.validator.dependencies import CreateAccessToken, CreateRefreshToken, GetHashedPW, VerifyPW
-    CreateAccessToken,
-    CreateRefreshToken,
-    DecodeRefreshToken,
-    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:
+    async def create_account(
-        LOG.i(f"LOGIN : {id=}")
+        self, supplier_id: str, id: str, pw: str, name: str, email: str, contact_number: str, role: int, connect_ip: str
-        res = Res_Login()
+    ) -> Res_CreateAccount:
-
+        LOG.i(f"CREATE : {id=}, {supplier_id=}")
-        # 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=}")
         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 해시로 저장)
+        # 3) 생성 (pw bcrypt 해시. last_accessed_at 은 NOT NULL/무기본값이라 생성 시각으로 둔다)
-        account = tbl_account(id=id, pw=await GetHashedPW(pw), nickname=nickname or id)
+        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:
+    async def attempt_login(self, id: str, pw: str, connect_ip: str) -> Res_Login:
-        res = Res_RefreshToken()
+        LOG.i(f"LOGIN : {id=}")
-        # refresh 토큰 검증은 라우터 Depends(IsValidRefreshToken) 에서 1차 수행됨.
+        res = Res_Login()
-        user_info = DecodeRefreshToken(refresh_token)
+
+        # 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

backend/tests/test_auth.py

@@ -1,63 +1,272 @@
-"""auth 도메인 e2e 테스트.
+"""인증 e2e 테스트 (supplier_users 기반 유저).
 
-실행 전제: docker-compose 로 PostgreSQL 이 떠 있어야 한다 (negosium_db 사용).
+실행 전제: PostgreSQL 이 떠 있어야 한다 (negosium_db, supplier/partner 스키마 적용).
-    docker compose up -d  # 또는 로컬 postgres
     cd backend && python -m pytest
+
+테스트는 dev negosium_db 를 그대로 쓰므로, 다른 데이터를 건드리지 않도록
+TRUNCATE 대신 전용 테스트 행(pytest_user)만 시드/정리한다.
 """
 
+import uuid
 
-async def test_create_and_login_flow(client):
+import bcrypt
-    # 1) 계정 생성
+import pytest_asyncio
-    r = await client.post("/v1/auth/create", json={"id": "user1", "pw": "pw1234", "nickname": "닉네임"})
+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"] == "닉네임"
+    assert body["name"] == TEST_USER_NAME            # 유저 개인 이름
-    access_token = body["access_token"]
+    assert body["supplier_name"] == TEST_SUPPLIER_NAME  # 공급사명(partner.suppliers)
-
+    assert body["role"] == 1
-    # 3) 보호된 엔드포인트 호출
+    assert body["su_id"]
-    r = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {access_token}"})
+    assert body["supplier_id"] == str(account_seed["supplier_id"])
-    assert r.status_code == 200
-    assert r.json()["id"] == "user1"
 
 
-async def test_login_with_wrong_password(client):
+async def test_login_wrong_password(client, account_seed):
-    await client.post("/v1/auth/create", json={"id": "user2", "pw": "correct", "nickname": "n"})
+    r = await client.post("/v1/auth/login", json={"id": TEST_LOGIN_ID, "pw": "wrong"})
-
-    r = await client.post("/v1/auth/login", json={"id": "user2", "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  # ACCOUNT_INVALID_INFO
-    assert body["result"]["code"] == 1200
+    assert body.get("access_token", "") == ""
-    assert body.get("access_token", "") == ""  # 실패 시 토큰은 빈 문자열
 
 
-async def test_login_nonexistent_account(client):
+async def test_login_nonexistent(client):
-    r = await client.post("/v1/auth/login", json={"id": "ghost", "pw": "whatever"})
+    r = await client.post("/v1/auth/login", json={"id": "ghost_user", "pw": "whatever"})
-    assert r.json()["result"]["success"] is False
+    assert r.status_code == 200
-
+    body = r.json()
-
-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()
     assert body["result"]["success"] is False
-    # ACCOUNT_ALREADY_EXIST(1201)
+    assert body["result"]["code"] == 1200
-    assert body["result"]["code"] == 1201
 
 
-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", "") == ""