d1d1ed6ec5
2단계에서 저장한 sources/partial 을 운영자가 볼 수 있게 한다. 운영자 목적은 **진단**이라
상태를 접지 않는다 — blocked(IP 회전으로 자동 회복)와 env_blocked(사람이 환경·설정을 고쳐야 함)를
뭉뚱그리면 회복될 일에 매달리거나 손봐야 할 설정을 방치하게 된다.
API
- /v1/lps/products, /v1/lps/products/{code}/history 둘 다 sources·partial 을 싣는다.
- 이력은 **시점마다** 싣는다. 최신 상태를 과거 시점의 몰별 표 옆에 붙이면 '그때도 막혔던 것처럼'
보여 오해를 부른다 — 그래서 ProductItem 이 아니라 PricePoint 에 담았다.
화면
- lib/sourceState.ts: 상태별 라벨·색·설명·confirmed 를 한곳에. 미지의 상태가 와도 화면이 깨지지
않는다(값 그대로 표시 + '모름' 취급). 색은 전부 @theme 토큰 참조(raw hex 금지).
- 상품 목록: partial 이면 '일부 확인 못함' 배지 + 툴팁에 어느 몰인지.
- 몰별 비교 카드 위: 몰별 상태·수집 건수·실패 사유 원문(툴팁). 가격표에 없는 몰이 **왜** 없는지를
여기서 답한다 — by_mall 은 가격이 있는 몰만 담으므로 그 답이 여기밖에 없다.
검증: ASGI 직접 호출로 두 엔드포인트 응답 확인(한글 사유 포함), tsc 오류 없음.
테스트 3건 추가(목록 노출 / 시점별 상태가 각각 다르게 / 컬럼 추가 이전 옛 행 호환).
전체 292 passed. 진행 상황은 docs/result-states.md 4절.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
186 lines
8.4 KiB
Python
186 lines
8.4 KiB
Python
"""관리자 FE API 테스트 — 잡 목록/재큐·상품 목록·IP세션/차단/비용 통계 (실 lps_db)."""
|
|
|
|
import pytest_asyncio
|
|
from sqlalchemy import text
|
|
|
|
from common.enums import JobType
|
|
from crud.job_crud import JobQueue
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def clean_all(db_engine):
|
|
async with db_engine.begin() as conn:
|
|
for t in ("job", "price_history", "ip_session", "bot_detection"):
|
|
await conn.execute(text(f"TRUNCATE {t}"))
|
|
return db_engine
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def q(clean_all):
|
|
return JobQueue()
|
|
|
|
|
|
# ---- 잡 목록 --------------------------------------------------------------
|
|
|
|
async def test_jobs_list_with_filters(client, q):
|
|
await q.enqueue(JobType.SEARCH.value, {"product_code": "A1", "product_name": "맥심 커피"})
|
|
await q.enqueue(JobType.SEARCH.value, {"product_code": "B2", "product_name": "생수"})
|
|
job = await q.claim("w1")
|
|
await q.complete(job["job_id"], "w1", {"outcome": "found", "lowest": {"price": 12000},
|
|
"metrics": {"cost": {"total_usd": 0.01}}})
|
|
r = await client.get("/v1/lps/jobs")
|
|
j = r.json()
|
|
assert r.status_code == 200 and j["total"] == 2
|
|
|
|
r = await client.get("/v1/lps/jobs", params={"status": "DONE"})
|
|
j = r.json()
|
|
assert j["total"] == 1
|
|
assert j["items"][0]["outcome"] == "found" and j["items"][0]["final_lowest"] == 12000
|
|
assert j["items"][0]["cost_usd"] == 0.01
|
|
|
|
r = await client.get("/v1/lps/jobs", params={"q": "맥심"})
|
|
assert r.json()["total"] == 1
|
|
|
|
|
|
async def test_jobs_list_unknown_status(client, q):
|
|
r = await client.get("/v1/lps/jobs", params={"status": "NOPE"})
|
|
assert r.json()["result"]["success"] is False
|
|
|
|
|
|
# ---- 재큐 ------------------------------------------------------------------
|
|
|
|
async def test_requeue_dead_job(client, q):
|
|
jid = await q.enqueue(JobType.SEARCH.value, {"product_code": "A1"}, max_attempts=1, dedupe_key="search-A1")
|
|
await q.claim("w1")
|
|
await q.fail(jid, "w1", "boom", backoff_sec=0) # 1/1 → DEAD
|
|
r = await client.post(f"/v1/lps/jobs/{jid}/requeue")
|
|
assert r.json()["requeued"] is True
|
|
assert (await q.counts())["PENDING"] == 1 # DEAD → PENDING
|
|
|
|
|
|
async def test_requeue_rejects_non_dead_and_missing(client, q):
|
|
jid = await q.enqueue(JobType.SEARCH.value, {"product_code": "A1"})
|
|
r = await client.post(f"/v1/lps/jobs/{jid}/requeue") # PENDING — 재큐 대상 아님
|
|
assert r.json()["result"]["success"] is False
|
|
r = await client.post("/v1/lps/jobs/not-a-uuid/requeue")
|
|
assert r.json()["result"]["success"] is False
|
|
|
|
|
|
async def test_requeue_blocked_by_active_duplicate(client, q):
|
|
dead = await q.enqueue(JobType.SEARCH.value, {"product_code": "A1"}, max_attempts=1, dedupe_key="search-A1")
|
|
await q.claim("w1")
|
|
await q.fail(dead, "w1", "boom", backoff_sec=0)
|
|
await q.enqueue(JobType.SEARCH.value, {"product_code": "A1"}, dedupe_key="search-A1") # 활성 중복 생성
|
|
r = await client.post(f"/v1/lps/jobs/{dead}/requeue")
|
|
assert r.json()["requeued"] is False
|
|
assert r.json()["result"]["success"] is False # DB_ALREADY_SAME_KEY
|
|
|
|
|
|
# ---- 상품 목록 --------------------------------------------------------------
|
|
|
|
async def test_products_list_latest_snapshot(client, clean_all):
|
|
async with clean_all.begin() as conn:
|
|
await conn.execute(text("""
|
|
INSERT INTO price_history (product_code, outcome, naver_lowest, final_lowest, naver_name, triggered_at)
|
|
VALUES ('P1', 'found', 1000, 900, '커피 320개입', now() - interval '2 hour'),
|
|
('P1', 'found', 1100, 950, '커피 320개입', now() - interval '1 hour'),
|
|
('P2', 'not_found', NULL, NULL, NULL, now())
|
|
"""))
|
|
r = await client.get("/v1/lps/products")
|
|
items = r.json()["items"]
|
|
assert [i["product_code"] for i in items] == ["P2", "P1"] # 최근 검색순
|
|
p1 = items[1]
|
|
assert p1["searches"] == 2 and p1["final_lowest"] == 950 # 최신 스냅샷 + 누적 횟수
|
|
r = await client.get("/v1/lps/products", params={"q": "P1"})
|
|
assert len(r.json()["items"]) == 1
|
|
|
|
|
|
# ---- 몰별 확인 상태 (2026-08-07, 3단계) -------------------------------------
|
|
# 운영 화면은 '왜 그 몰 값이 없나'에 답할 수 있어야 한다 — by_mall 에 없는 몰이
|
|
# '거기엔 없더라'인지 '거기를 못 봤다'인지는 sources 에만 있다.
|
|
|
|
async def test_products_list_exposes_source_states(client, clean_all):
|
|
async with clean_all.begin() as conn:
|
|
await conn.execute(text("""
|
|
INSERT INTO price_history (product_code, outcome, final_lowest, partial, sources, triggered_at)
|
|
VALUES ('S1', 'found', 9000, true,
|
|
'{"naver": {"state": "matched", "count": 40},
|
|
"coupang": {"state": "env_blocked", "error": "사용권한이 제한된"}}'::jsonb,
|
|
now())
|
|
"""))
|
|
item = (await client.get("/v1/lps/products")).json()["items"][0]
|
|
assert item["partial"] is True
|
|
assert item["sources"]["coupang"]["state"] == "env_blocked"
|
|
assert "사용권한이 제한된" in item["sources"]["coupang"]["error"] # 원인 원문이 운영자에게 간다
|
|
|
|
|
|
async def test_history_points_carry_state_per_point(client, clean_all):
|
|
"""상태는 시점마다 다르다 — 최신 상태를 과거 시점 옆에 붙이면 오해를 부른다."""
|
|
async with clean_all.begin() as conn:
|
|
await conn.execute(text("""
|
|
INSERT INTO price_history (product_code, outcome, final_lowest, partial, sources, triggered_at)
|
|
VALUES ('S2', 'found', 9000, true,
|
|
'{"coupang": {"state": "blocked"}}'::jsonb, now() - interval '1 hour'),
|
|
('S2', 'found', 8500, false,
|
|
'{"coupang": {"state": "matched", "count": 60}}'::jsonb, now())
|
|
"""))
|
|
pts = (await client.get("/v1/lps/products/S2/history")).json()["points"]
|
|
assert [p["partial"] for p in pts] == [True, False]
|
|
assert pts[0]["sources"]["coupang"]["state"] == "blocked"
|
|
assert pts[1]["sources"]["coupang"]["state"] == "matched"
|
|
|
|
|
|
async def test_old_rows_without_state_still_work(client, clean_all):
|
|
"""이 컬럼 추가 이전 이력도 그대로 읽혀야 한다(마이그레이션 전 데이터)."""
|
|
async with clean_all.begin() as conn:
|
|
await conn.execute(text("""
|
|
INSERT INTO price_history (product_code, outcome, final_lowest, triggered_at)
|
|
VALUES ('S3', 'found', 7000, now())
|
|
"""))
|
|
item = (await client.get("/v1/lps/products")).json()["items"][0]
|
|
assert item["partial"] is False and item.get("sources") is None
|
|
|
|
|
|
# ---- 통계 3종 ---------------------------------------------------------------
|
|
|
|
async def test_ip_session_stats(client, clean_all):
|
|
async with clean_all.begin() as conn:
|
|
await conn.execute(text("""
|
|
INSERT INTO ip_session (source, proxy_port, requests, ok_count, blocked_count, end_reason)
|
|
VALUES ('coupang', 10001, 3, 3, 0, 'budget'),
|
|
('coupang', 10002, 3, 3, 0, 'budget'),
|
|
('coupang', 10003, 5, 4, 1, 'block')
|
|
"""))
|
|
r = await client.get("/v1/lps/stats/ip-sessions")
|
|
j = r.json()
|
|
assert j["by_reason"] == {"budget": 2, "block": 1}
|
|
assert j["block_min_requests"] == 5 # 예산 튜닝 기준선
|
|
assert {"requests": 3, "count": 2} in j["histogram"]
|
|
assert len(j["sessions"]) == 3
|
|
|
|
|
|
async def test_bot_stats(client, clean_all):
|
|
async with clean_all.begin() as conn:
|
|
await conn.execute(text("""
|
|
INSERT INTO bot_detection (source, query, ip_request_no, proxy_port, marker)
|
|
VALUES ('coupang', '생수', 4, 10001, '/akam/')
|
|
"""))
|
|
r = await client.get("/v1/lps/stats/bot")
|
|
j = r.json()
|
|
assert len(j["items"]) == 1 and j["items"][0]["marker"] == "/akam/"
|
|
assert sum(h["count"] for h in j["hourly"]) == 1
|
|
|
|
|
|
async def test_cost_stats(client, q):
|
|
for cost in (0.01, 0.02):
|
|
jid = await q.enqueue(JobType.SEARCH.value, {"product_code": f"C{cost}"})
|
|
job = await q.claim("w1")
|
|
await q.complete(job["job_id"], "w1",
|
|
{"metrics": {"cost": {"ai_usd": cost / 2, "proxy_usd": cost / 2, "total_usd": cost},
|
|
"duration_ms": 15000}})
|
|
r = await client.get("/v1/lps/stats/cost")
|
|
buckets = r.json()["buckets"]
|
|
assert sum(b["total_usd"] for b in buckets) == 0.03
|
|
assert sum(b["jobs"] for b in buckets) == 2
|
|
assert buckets[0]["avg_ms"] == 15000
|