61 lines
2.4 KiB
Python
61 lines
2.4 KiB
Python
"""/v1/negotiation/step (P2 루프 HTTP 프리뷰) 회귀 스모크.
|
|
|
|
서버 기동 후 실제로 칠 수 있는 유일한 업무 엔드포인트 — 테넌트 라우팅 + config 주입 + 응답 형태를 검증.
|
|
DB 로깅은 db_engine 픽스처 유무와 무관하게 log=false 로 끄고 검증(순수 응답 형태).
|
|
"""
|
|
|
|
import pytest
|
|
|
|
_BODY = {
|
|
"revenue_amount": 20_000_000,
|
|
"distribution_code": "A",
|
|
"partner_count": 1,
|
|
"acceptance_ratio": 0.11,
|
|
"input_price": 9950,
|
|
"anchor_price": 9900, # KT 앵커링가 (anchor < target)
|
|
"target_price": 10000, # KT 목표 매입가
|
|
"round_number": 3,
|
|
"outcome": "success",
|
|
"log": False,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_step_requires_tenant_header(client):
|
|
r = await client.post("/v1/negotiation/step", json=_BODY)
|
|
assert r.status_code == 400
|
|
assert r.json()["result"]["desc"] == "TENANT_HEADER_MISSING"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_step_tenant_divergence(client):
|
|
rk = await client.post("/v1/negotiation/step", headers={"X-Tenant-ID": "ktcommerce"}, json=_BODY)
|
|
ri = await client.post("/v1/negotiation/step", headers={"X-Tenant-ID": "imarketkorea"}, json=_BODY)
|
|
assert rk.status_code == 200 and ri.status_code == 200
|
|
dk, di = rk.json(), ri.json()
|
|
# 같은 입력이 테넌트 config 에 따라 다른 상태/카드로 갈린다
|
|
assert dk["card_id"].startswith("NGC-A")
|
|
assert di["card_id"].startswith("NGC-B")
|
|
assert dk["state_index"] != di["state_index"]
|
|
# 응답 형태
|
|
assert dk["result"]["success"] is True
|
|
# input 9950: price_reward=(10000-9950)/(10000-9900)=0.5, round3 weight=0.2,
|
|
# penalty=0.06, end(success)=1.0 → 0.2*0.5+1.0-0.06 = 1.04
|
|
assert dk["reward"]["total"] == pytest.approx(1.04, abs=1e-6)
|
|
assert dk["logged"] is False # log=false
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_step_invalid_distribution_code_is_domain_error(client):
|
|
body = dict(_BODY, distribution_code="Z")
|
|
r = await client.post("/v1/negotiation/step", headers={"X-Tenant-ID": "ktcommerce"}, json=body)
|
|
assert r.status_code == 200 # HTTP 는 200, 결과코드로 에러 전달(backend 규약)
|
|
assert r.json()["result"]["desc"] == "NEGO_INVALID_STEP"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_step_invalid_outcome(client):
|
|
body = dict(_BODY, outcome="maybe")
|
|
r = await client.post("/v1/negotiation/step", headers={"X-Tenant-ID": "ktcommerce"}, json=body)
|
|
assert r.json()["result"]["desc"] == "INVALID_REQUEST_DATA"
|