Mina Choi
1f778ab975
- item/supplier/quotation/quotation_setting CRUD·service·router 추가 - item protocol delivery_type str→int (ERD/스키마 SMALLINT 일치) - DeliveryType enum + 한글 라벨, 공용 GET /v1/enums (도메인 코드 메타데이터) - CompanyBrief → CompanyData 로 *Data 네이밍 통일 - CORS: WebServerConfig.client_url(단일) 도입 (config_models/router) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
"""item 도메인 e2e — CRUD + company 멀티테넌트 스코프 검증.
|
|
|
|
실행 전제: PostgreSQL(negodata_db). docker compose up -d 후 python -m pytest.
|
|
"""
|
|
import uuid
|
|
|
|
import pytest_asyncio
|
|
from sqlalchemy import text
|
|
|
|
|
|
async def _headers(client, company_id, login_id="itemuser", pw="pw1234"):
|
|
await client.post(
|
|
"/v1/auth/create",
|
|
json={"id": login_id, "password": pw, "company_id": company_id, "name": "n"},
|
|
)
|
|
r = await client.post("/v1/auth/login", json={"id": login_id, "password": pw})
|
|
return {"Authorization": f"Bearer {r.json()['access_token']}"}
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def other_company_id(db_engine) -> str:
|
|
cid = uuid.uuid4()
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO companies (company_id, name) VALUES (:cid, :name)"),
|
|
{"cid": cid, "name": "다른회사"},
|
|
)
|
|
return str(cid)
|
|
|
|
|
|
async def test_item_crud_flow(client, company_id):
|
|
h = await _headers(client, company_id)
|
|
|
|
# 등록
|
|
r = await client.post("/v1/item/create", json={"name": "상품A", "price": 1000, "code": "C1"}, headers=h)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["result"]["success"] is True
|
|
item_id = body["item"]["item_id"]
|
|
assert body["item"]["name"] == "상품A"
|
|
|
|
# 목록
|
|
r = await client.get("/v1/item/list", headers=h)
|
|
body = r.json()
|
|
assert body["total"] == 1 and len(body["items"]) == 1
|
|
|
|
# 단건 조회
|
|
r = await client.get(f"/v1/item/{item_id}", headers=h)
|
|
assert r.json()["item"]["item_id"] == item_id
|
|
|
|
# 수정 (부분)
|
|
r = await client.patch(f"/v1/item/update/{item_id}", json={"price": 2000}, headers=h)
|
|
assert r.json()["item"]["price"] == 2000
|
|
assert r.json()["item"]["name"] == "상품A" # 미지정 필드 유지
|
|
|
|
# 삭제 (soft)
|
|
r = await client.delete(f"/v1/item/delete/{item_id}", headers=h)
|
|
assert r.json()["result"]["success"] is True
|
|
|
|
# 삭제 후 목록 0
|
|
r = await client.get("/v1/item/list", headers=h)
|
|
assert r.json()["total"] == 0
|
|
|
|
|
|
async def test_item_not_found(client, company_id):
|
|
h = await _headers(client, company_id)
|
|
r = await client.get(f"/v1/item/{uuid.uuid4()}", headers=h)
|
|
body = r.json()
|
|
assert body["result"]["success"] is False
|
|
assert body["result"]["code"] == 1300 # ITEM_NOT_FOUND
|
|
|
|
|
|
async def test_item_company_scope(client, company_id, other_company_id):
|
|
# 회사 A 가 상품 등록
|
|
ha = await _headers(client, company_id, login_id="userA")
|
|
r = await client.post("/v1/item/create", json={"name": "A상품"}, headers=ha)
|
|
a_item_id = r.json()["item"]["item_id"]
|
|
|
|
# 회사 B 유저는 A 의 상품을 목록/단건에서 볼 수 없다
|
|
hb = await _headers(client, other_company_id, login_id="userB")
|
|
r = await client.get("/v1/item/list", headers=hb)
|
|
assert r.json()["total"] == 0
|
|
|
|
r = await client.get(f"/v1/item/{a_item_id}", headers=hb)
|
|
assert r.json()["result"]["code"] == 1300 # 타사 자원은 ITEM_NOT_FOUND
|