Mina Choi
c62b66e35b
- 소유권 게이팅(common/authz): 변경 액션 본인∪OWNER, 협력사 삭제 OWNER 전용 - 견적 수동 낙찰(award) + 작성자명(creatorName) 표시 + 전화번호 입력 컴포넌트 + 카드 엑셀 업로드 - supplier_type 은 이번 커밋 미변경(다음 커밋에서 코드부터 정리 예정) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
123 lines
5.5 KiB
Python
123 lines
5.5 KiB
Python
"""소유자 게이팅 — 변경 액션은 '본인 소유' 또는 '최고관리자(OWNER)'만. 프론트 버튼 차단과 같은 규칙을 백엔드가 강제한다.
|
|
|
|
- item(상품): 같은 회사의 다른 일반유저는 남의 상품을 수정·삭제 못 한다(OWNER는 가능).
|
|
- quotation(견적): 남의 견적 삭제는 비소유 일반유저 차단, OWNER 허용, user_id 미지정(내부 호출)은 스킵.
|
|
공용 판정은 common.authz.is_owner_or_admin — 여기 통과하면 다른 변경 액션(마감·재생성·초청메일)도 같은 규칙을 탄다.
|
|
"""
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import text
|
|
|
|
from common.enums import ErrorType, QuotationStatus, QuotationType, UserRole
|
|
from crud.quotation_crud import QuotationCRUD
|
|
from services.quotation_service import QuotationService
|
|
|
|
PAST = datetime(2020, 1, 1)
|
|
|
|
|
|
# ===== item(상품) — HTTP e2e: 같은 회사, 다른 유저 =====
|
|
async def test_item_update_delete_blocked_for_non_owner(client, auth_headers):
|
|
"""검증: A가 등록한 상품을 같은 회사의 다른 일반유저 B가 수정·삭제 시도.
|
|
기대결과: 둘 다 거부(ACCOUNT_FORBIDDEN), 상품은 원값 그대로 남는다."""
|
|
ha = await auth_headers("item_owner") # 소유자
|
|
hb = await auth_headers("item_other") # 같은 회사, USER
|
|
item_id = (await client.post(
|
|
"/v1/item/create", json={"name": "상품A", "price": 1000, "code": "OWN1"}, headers=ha
|
|
)).json()["item"]["item_id"]
|
|
|
|
r_upd = await client.patch(f"/v1/item/update/{item_id}", json={"price": 9999}, headers=hb)
|
|
assert r_upd.json()["result"]["success"] is False
|
|
assert r_upd.json()["result"]["code"] == ErrorType.ACCOUNT_FORBIDDEN.value
|
|
|
|
r_del = await client.delete(f"/v1/item/delete/{item_id}", headers=hb)
|
|
assert r_del.json()["result"]["success"] is False
|
|
assert r_del.json()["result"]["code"] == ErrorType.ACCOUNT_FORBIDDEN.value
|
|
|
|
# 수정·삭제 모두 무산 — 원값으로 조회된다
|
|
assert (await client.get(f"/v1/item/{item_id}", headers=ha)).json()["item"]["price"] == 1000
|
|
|
|
|
|
async def test_item_delete_allowed_for_owner_role(client, auth_headers):
|
|
"""검증: A가 등록한 상품을 같은 회사 최고관리자(OWNER)가 삭제.
|
|
기대결과: 성공 — 소유자가 아니어도 OWNER는 허용."""
|
|
ha = await auth_headers("item_owner2")
|
|
hadmin = await auth_headers("item_admin", role=UserRole.OWNER.value)
|
|
item_id = (await client.post(
|
|
"/v1/item/create", json={"name": "상품B", "price": 500, "code": "OWN2"}, headers=ha
|
|
)).json()["item"]["item_id"]
|
|
|
|
r = await client.delete(f"/v1/item/delete/{item_id}", headers=hadmin)
|
|
assert r.json()["result"]["success"] is True
|
|
|
|
|
|
# ===== quotation(견적) — 서비스 직접: 삭제 게이팅 =====
|
|
async def test_quotation_delete_blocked_for_non_owner(db_engine):
|
|
"""검증: 남의 견적을 비소유 일반유저(USER)가 삭제 시도.
|
|
기대결과: 거부(ACCOUNT_FORBIDDEN) + soft-delete 안 됨(deleted=false)."""
|
|
owner, other = uuid.uuid4(), uuid.uuid4()
|
|
qt = await _seed_quotation(db_engine, user_id=owner, number="DEL-NONOWNER")
|
|
|
|
res = await _service().delete_quotation(str(qt), None, other, UserRole.USER.value)
|
|
|
|
assert res.result.success is False
|
|
assert res.result.code == ErrorType.ACCOUNT_FORBIDDEN.value
|
|
assert await _deleted(db_engine, qt) is False
|
|
|
|
|
|
async def test_quotation_delete_allowed_for_owner_role(db_engine):
|
|
"""검증: 남의 견적을 최고관리자(OWNER)가 삭제.
|
|
기대결과: 성공 + deleted=true."""
|
|
creator, admin = uuid.uuid4(), uuid.uuid4()
|
|
qt = await _seed_quotation(db_engine, user_id=creator, number="DEL-OWNER")
|
|
|
|
res = await _service().delete_quotation(str(qt), None, admin, UserRole.OWNER.value)
|
|
|
|
assert res.result.success is True
|
|
assert await _deleted(db_engine, qt) is True
|
|
|
|
|
|
async def test_quotation_delete_skips_gate_for_internal_call(db_engine):
|
|
"""검증: user_id 미지정(내부/스케줄러 호출)로 삭제.
|
|
기대결과: 소유권 검사 스킵 → 성공(회사 스코프만 적용)."""
|
|
creator = uuid.uuid4()
|
|
qt = await _seed_quotation(db_engine, user_id=creator, number="DEL-INTERNAL")
|
|
|
|
res = await _service().delete_quotation(str(qt), None)
|
|
|
|
assert res.result.success is True
|
|
assert await _deleted(db_engine, qt) is True
|
|
|
|
|
|
# ===== 헬퍼 =====
|
|
async def _seed_quotation(engine, *, user_id, number):
|
|
"""견적 1건 시드(삭제 게이팅 확인용 — 상태는 무관하므로 CLOSED로 고정)."""
|
|
qt_id = uuid.uuid4()
|
|
async with engine.begin() as conn:
|
|
await conn.execute(
|
|
text(
|
|
"INSERT INTO quotations "
|
|
"(qt_id, user_id, qt_setting_id, version_id, name, number, type, status, "
|
|
" round, iteration, start_time, end_time, deleted) VALUES "
|
|
"(:qt_id, :user_id, :qt_setting_id, :version_id, '견적', :number, :type, :status, "
|
|
" 1, 0, :t, :t, false)"
|
|
),
|
|
{
|
|
"qt_id": qt_id, "user_id": user_id, "qt_setting_id": uuid.uuid4(),
|
|
"version_id": uuid.uuid4(), "number": number, "type": QuotationType.REQUOTE.value,
|
|
"status": QuotationStatus.CLOSED.value, "t": PAST,
|
|
},
|
|
)
|
|
return qt_id
|
|
|
|
|
|
async def _deleted(engine, qt_id) -> bool:
|
|
async with engine.begin() as conn:
|
|
return (await conn.execute(
|
|
text("SELECT deleted FROM quotations WHERE qt_id = :q"), {"q": qt_id}
|
|
)).scalar_one()
|
|
|
|
|
|
def _service():
|
|
return QuotationService(QuotationCRUD())
|