3c8923c075
KT-NEGOWIZ fe_v2 의 서비스 안내 팝업을 우리 구조로 이식.
- frontend: GuideContent(공용 본문, VAT/배송비는 chat init 메타로 동적 표시)
+ ServiceInfoPopup(채팅 진입 시 자동 — 안내 보지 않기/오늘 하루 보지 않기)
+ ServiceGuidePopup(메뉴 유의사항·이용 가이드 버튼 수동 — X/배경/ESC 닫기)
+ useServiceInfoPopup 훅(서버 상태 + localStorage 오늘 하루)
+ apis/auth 에 popupStatus 쿼리·hidePopup 뮤테이션, 헬프데스크 연락처 반영,
한국어 단어 잘림 방지(break-keep)
- backend: GET /v1/auth/popup/status, POST /v1/auth/popup/hide
(authenticate 공통 인증 + supplier_users.hide_service_info 영구 저장), 테스트 4건
- db: supplier_users.hide_service_info 컬럼 — 00-init 테이블 정의 갱신,
기존 DB 보정은 alters/2026-07-07-supplier-users-hide-service-info.sql 별도 파일
(기준선 이후 보정 ALTER 는 alters/ 파일로 관리하도록 규칙 변경)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
374 lines
15 KiB
Python
374 lines
15 KiB
Python
"""인증 e2e 테스트 (supplier_users 기반 유저).
|
|
|
|
실행 전제: PostgreSQL 이 떠 있어야 한다 (negosium_db, supplier/partner 스키마 적용).
|
|
cd backend && python -m pytest
|
|
|
|
테스트는 dev negosium_db 를 그대로 쓰므로, 다른 데이터를 건드리지 않도록
|
|
TRUNCATE 대신 전용 테스트 행(pytest_user)만 시드/정리한다.
|
|
"""
|
|
|
|
import uuid
|
|
|
|
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["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 # 매니저로 생성됨
|
|
|
|
|
|
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["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_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
|
|
assert body["result"]["code"] == 1200 # ACCOUNT_INVALID_INFO
|
|
assert body.get("access_token", "") == ""
|
|
|
|
|
|
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
|
|
assert body["result"]["code"] == 1200
|
|
|
|
|
|
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", "") == ""
|
|
|
|
|
|
# ---- 로그아웃 / stateful 토큰 검증 -------------------------------------------
|
|
async def test_logout_revokes_tokens(client, account_seed, db_engine):
|
|
body = (await _login(client)).json()
|
|
su_id, access, refresh = body["su_id"], body["access_token"], body["refresh_token"]
|
|
|
|
# 로그아웃 성공
|
|
r = await client.post("/v1/auth/logout", headers={"Authorization": f"Bearer {access}"})
|
|
assert r.status_code == 200
|
|
assert r.json()["result"]["success"] is True
|
|
|
|
# 저장 토큰이 모두 삭제됨
|
|
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 == 0
|
|
|
|
# 로그아웃 후 같은 access 로 /me → TOKEN_REVOKED(1203)
|
|
r2 = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {access}"})
|
|
assert r2.status_code == 200
|
|
assert r2.json()["result"]["code"] == 1203
|
|
|
|
# 로그아웃 후 같은 refresh 로 재발급 → TOKEN_REVOKED(1203)
|
|
r3 = await client.post("/v1/auth/refresh_token", headers={"Authorization": f"Bearer {refresh}"})
|
|
assert r3.json()["result"]["code"] == 1203
|
|
|
|
|
|
async def test_logout_without_token(client):
|
|
r = await client.post("/v1/auth/logout")
|
|
assert r.status_code in (401, 403)
|
|
|
|
|
|
async def test_relogin_invalidates_previous_access(client, account_seed):
|
|
# 단일 세션: 재로그인하면 이전 세션의 access 가 무효화된다(저장 토큰이 교체됨).
|
|
import asyncio
|
|
|
|
first = (await _login(client)).json()
|
|
await asyncio.sleep(1.1) # exp(초 단위)가 달라져 토큰이 실제로 바뀌도록
|
|
second = (await _login(client)).json()
|
|
assert first["access_token"] != second["access_token"]
|
|
|
|
# 이전 access → 무효(TOKEN_REVOKED)
|
|
r_old = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {first['access_token']}"})
|
|
assert r_old.json()["result"]["code"] == 1203
|
|
# 새 access → 정상
|
|
r_new = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {second['access_token']}"})
|
|
assert r_new.json()["result"]["success"] is True
|
|
|
|
|
|
# ---- 팝업 숨김 (popup/status, popup/hide) --------------------------------------
|
|
async def test_popup_status_default_false(client, account_seed):
|
|
# 신규 유저는 숨김 상태가 아니다.
|
|
access = (await _login(client)).json()["access_token"]
|
|
r = await client.get("/v1/auth/popup/status", headers={"Authorization": f"Bearer {access}"})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["result"]["success"] is True
|
|
assert body["service_info"] is False
|
|
|
|
|
|
async def test_popup_hide_persists(client, account_seed):
|
|
# hide 후 status 가 True 로 바뀌고, 재로그인해도 유지된다(유저 행에 영구 저장).
|
|
access = (await _login(client)).json()["access_token"]
|
|
r = await client.post(
|
|
"/v1/auth/popup/hide",
|
|
json={"popup_type": "service_info"},
|
|
headers={"Authorization": f"Bearer {access}"},
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["result"]["success"] is True
|
|
|
|
r2 = await client.get("/v1/auth/popup/status", headers={"Authorization": f"Bearer {access}"})
|
|
assert r2.json()["service_info"] is True
|
|
|
|
# 재로그인(토큰 교체) 후에도 유지
|
|
new_access = (await _login(client)).json()["access_token"]
|
|
r3 = await client.get("/v1/auth/popup/status", headers={"Authorization": f"Bearer {new_access}"})
|
|
assert r3.json()["service_info"] is True
|
|
|
|
|
|
async def test_popup_hide_invalid_type(client, account_seed):
|
|
access = (await _login(client)).json()["access_token"]
|
|
r = await client.post(
|
|
"/v1/auth/popup/hide",
|
|
json={"popup_type": "unknown_popup"},
|
|
headers={"Authorization": f"Bearer {access}"},
|
|
)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["result"]["success"] is False
|
|
assert body["result"]["code"] == 101 # INVALID_REQUEST_DATA
|
|
|
|
|
|
async def test_popup_status_without_token(client):
|
|
r = await client.get("/v1/auth/popup/status")
|
|
assert r.status_code in (401, 403)
|