Mina Choi
82076e138e
- test DB 세션마다 자동 create/drop (팀원은 Postgres만 있으면 pytest 한 방) - auth_headers 시드 픽스처(무인증 /auth/create 제거 대응) + other_company_id - 커버: 회사 스코프(견적·상품·협력사·대시보드·세팅), 견적 마감 재견적 O/X + 알림, 견적 생성·목표가, 알림함 읽기, 회사유저 OWNER 게이팅, 기존 파일 검증/기대결과 주석 정비 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
100 lines
4.3 KiB
Python
100 lines
4.3 KiB
Python
"""알림함 '읽는' 쪽 테스트 — 목록 조회, 안 읽은 개수, 읽음 처리(하나/전체), 그리고 남의 알림은 안 보이는지.
|
|
|
|
'마감하면 알림이 쌓이는지'(쓰는 쪽)는 test_quotation_close_notify 가 본다. 여기선 겹치지 않게 '읽는' 동작만 본다.
|
|
알림은 원래 견적 마감 때 생기지만, 여기선 테스트를 위해 알림 행을 DB 에 직접 넣는다.
|
|
"""
|
|
import json
|
|
import uuid
|
|
|
|
from sqlalchemy import text
|
|
|
|
from common.enums import NotificationType
|
|
|
|
|
|
async def test_list_and_unread(client, auth_headers, db_engine):
|
|
"""검증: 내 알림 2건을 시드하고 인박스 목록 조회.
|
|
기대결과: total=2, unread=2, 안읽음이라 read_at 없음(None)."""
|
|
h = await auth_headers("notilist")
|
|
uid = await _user_id(db_engine, "notilist")
|
|
await _seed_notification(db_engine, uid)
|
|
await _seed_notification(db_engine, uid, ntype=NotificationType.REGENERATED.value)
|
|
|
|
r = await client.get("/v1/notification/list", headers=h)
|
|
body = r.json()
|
|
assert body["result"]["success"] is True
|
|
assert body["total"] == 2
|
|
assert body["unread"] == 2
|
|
assert len(body["notifications"]) == 2
|
|
# 안읽음은 read_at=None → RemoveNoneResponse 가 키를 제거하므로 .get() 으로 확인
|
|
assert all(n.get("read_at") is None for n in body["notifications"])
|
|
|
|
|
|
async def test_inbox_is_user_scoped(client, auth_headers, db_engine):
|
|
"""검증: 내 알림 1건 + 남의 알림 1건을 시드하고 내 인박스 조회.
|
|
기대결과: total=1, unread=1 — 내 것만 보인다(남의 알림 제외)."""
|
|
h = await auth_headers("notiscope")
|
|
me = await _user_id(db_engine, "notiscope")
|
|
await _seed_notification(db_engine, me) # 내 알림
|
|
await _seed_notification(db_engine, uuid.uuid4()) # 남의 알림(안 보여야 함)
|
|
|
|
r = await client.get("/v1/notification/list", headers=h)
|
|
body = r.json()
|
|
assert body["total"] == 1 and body["unread"] == 1
|
|
|
|
|
|
async def test_read_all_clears_unread(client, auth_headers, db_engine):
|
|
"""검증: 안읽음 2건 상태에서 read-all 호출 후 다시 목록 조회.
|
|
기대결과: unread=0, 목록엔 그대로 남고(total=2) 모든 read_at 채워짐."""
|
|
h = await auth_headers("notireadall")
|
|
uid = await _user_id(db_engine, "notireadall")
|
|
await _seed_notification(db_engine, uid)
|
|
await _seed_notification(db_engine, uid)
|
|
|
|
r = await client.post("/v1/notification/read-all", headers=h)
|
|
assert r.json()["result"]["success"] is True
|
|
|
|
r = await client.get("/v1/notification/list", headers=h)
|
|
body = r.json()
|
|
assert body["total"] == 2 and body["unread"] == 0
|
|
assert all(n["read_at"] is not None for n in body["notifications"])
|
|
|
|
|
|
async def test_read_one_decrements_unread(client, auth_headers, db_engine):
|
|
"""검증: 안읽음 2건 중 1건만 읽음 처리.
|
|
기대결과: unread 2 → 1."""
|
|
h = await auth_headers("notireadone")
|
|
uid = await _user_id(db_engine, "notireadone")
|
|
await _seed_notification(db_engine, uid)
|
|
await _seed_notification(db_engine, uid)
|
|
|
|
r = await client.get("/v1/notification/list", headers=h)
|
|
target_id = r.json()["notifications"][0]["notification_id"]
|
|
|
|
r = await client.post(f"/v1/notification/{target_id}/read", headers=h)
|
|
assert r.json()["result"]["success"] is True
|
|
|
|
r = await client.get("/v1/notification/list", headers=h)
|
|
assert r.json()["unread"] == 1
|
|
|
|
|
|
# ===== 헬퍼 (위 테스트들이 쓰는 도우미) =====
|
|
async def _user_id(engine, login_id):
|
|
"""auth_headers 로 시드된 유저의 user_id(알림 시드/스코프 확인용)."""
|
|
async with engine.begin() as conn:
|
|
return (await conn.execute(
|
|
text("SELECT user_id FROM users WHERE id = :id"), {"id": login_id}
|
|
)).scalar_one()
|
|
|
|
|
|
async def _seed_notification(engine, user_id, *, ntype=NotificationType.SUCCESS.value, data=None):
|
|
"""안읽음(read_at NULL) 알림 1건 시드."""
|
|
async with engine.begin() as conn:
|
|
await conn.execute(
|
|
text(
|
|
"INSERT INTO notifications (notification_id, user_id, type, data, read_at) "
|
|
"VALUES (:nid, :uid, :type, CAST(:data AS JSONB), NULL)"
|
|
),
|
|
{"nid": uuid.uuid4(), "uid": user_id, "type": ntype,
|
|
"data": json.dumps(data or {"qt_name": "견적A"})},
|
|
)
|