ca7f057e41
협의로 선정한 조기 신호 4종을 AlertManager 에 추가한다. - deadline: 최근 1h JobDeadlineExceeded 수 ≥ LPS_ALERT_DEADLINE_1H(5). 재시도로 살아나면 dead 룰엔 안 잡히는 크롤 행 반복 신호를 별도 집계. - cost: 최근 1h 완료 잡 검색원가 합 ≥ LPS_ALERT_COST_1H_USD(1.0). 비용의 87%가 프록시 대역폭 — 리소스차단 풀림·재시도 루프의 조용한 비용 폭주를 감시. job.result 의 metrics.cost.total_usd JSONB 합산. - proxy_ports_low: 가용 포트 비율 ≤ LPS_ALERT_PORTS_LOW_PCT(30%). 쿨다운 격리 누적 — blocks_1h(80건)보다 먼저 우는 대규모 차단 조기 신호. 워커별 프록시 중 가장 소진된 것 기준(min). - budget_leak: 최근 6h end_reason=block 세션 ≥ LPS_ALERT_BLOCK_SESSIONS_6H(1). 요청 예산(3회)을 지켰는데도 차단됨 = 예산 하향 검토 신호. - deadline_1h·cost_1h_usd 는 queue.ops() 에 편입 → /v1/lps/ops 로도 노출. 포트·세션 지표는 워커 웹훅 스냅샷에 포함(프록시 상태는 워커에만 있음). - 테스트 4건 추가(ops 집계 2·포트 스냅샷 2), 전체 145 passed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
97 lines
3.9 KiB
Python
97 lines
3.9 KiB
Python
"""LPS API 라우터 테스트 — 검색 적재/중복/상태조회/큐 통계 (ASGI 클라이언트 + 실 lps_db)."""
|
|
|
|
import pytest_asyncio
|
|
from sqlalchemy import text
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def clean_jobs(db_engine):
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(text("TRUNCATE job"))
|
|
|
|
|
|
async def test_search_enqueues_jobs(client, clean_jobs):
|
|
body = {"data": [
|
|
{"product_code": "P1", "product_name": "커피", "job_type": "new_product"},
|
|
{"product_code": "P2", "product_name": "무선마우스", "specification": "M170"},
|
|
]}
|
|
r = await client.post("/v1/lps/search", json=body)
|
|
assert r.status_code == 200
|
|
j = r.json()
|
|
assert j["accepted"] == 2
|
|
assert {i["product_code"] for i in j["items"]} == {"P1", "P2"}
|
|
assert all(i.get("job_id") for i in j["items"])
|
|
|
|
|
|
async def test_search_dedupes_active_product(client, clean_jobs):
|
|
body = {"data": [{"product_code": "P1", "product_name": "커피", "job_type": "new_product"}]}
|
|
await client.post("/v1/lps/search", json=body)
|
|
r2 = await client.post("/v1/lps/search", json=body) # 같은 product_code 재요청
|
|
item = r2.json()["items"][0]
|
|
assert item["duplicated"] is True
|
|
assert "job_id" not in item # None → RemoveNoneResponse 로 제거됨
|
|
assert r2.json()["accepted"] == 0
|
|
|
|
|
|
async def test_job_status_flow(client, clean_jobs):
|
|
jid = (await client.post("/v1/lps/search", json={"data": [{"product_code": "P9", "product_name": "커피"}]})).json()["items"][0]["job_id"]
|
|
r = await client.get(f"/v1/lps/jobs/{jid}")
|
|
body = r.json()
|
|
assert body["status"] == "PENDING" and body["attempts"] == 0
|
|
assert body["result"]["success"] is True
|
|
|
|
|
|
async def test_job_status_not_found(client, clean_jobs):
|
|
# 존재하지 않는(유효 UUID) 잡
|
|
r = await client.get("/v1/lps/jobs/00000000-0000-0000-0000-000000000000")
|
|
assert r.json()["result"]["success"] is False
|
|
assert r.json()["result"]["desc"] == "LPS_JOB_NOT_FOUND"
|
|
# 잘못된 형식의 id 도 not-found 처리
|
|
r2 = await client.get("/v1/lps/jobs/not-a-uuid")
|
|
assert r2.json()["result"]["desc"] == "LPS_JOB_NOT_FOUND"
|
|
|
|
|
|
async def test_price_history_endpoint(client, db_engine):
|
|
from crud.price_history import PriceHistory
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(text("TRUNCATE price_history"))
|
|
ph = PriceHistory()
|
|
await ph.record({"product_code": "GRAPH1", "outcome": "found", "final_lowest": 2500,
|
|
"final_source": "naver", "naver_lowest": 2500, "coupang_lowest": 2700, "matched_count": 2})
|
|
|
|
r = await client.get("/v1/lps/products/GRAPH1/history")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["product_code"] == "GRAPH1"
|
|
assert len(body["points"]) == 1
|
|
pt = body["points"][0]
|
|
assert pt["final"] == 2500 and pt["naver"] == 2500 and pt["coupang"] == 2700 and pt["final_source"] == "naver"
|
|
assert "triggered_at" in pt
|
|
|
|
|
|
async def test_queue_stats(client, clean_jobs):
|
|
await client.post("/v1/lps/search", json={"data": [
|
|
{"product_code": "A", "product_name": "x"},
|
|
{"product_code": "B", "product_name": "y"},
|
|
]})
|
|
r = await client.get("/v1/lps/queue/stats")
|
|
counts = r.json()["counts"]
|
|
assert counts["PENDING"] == 2 and counts["DONE"] == 0 and counts["DEAD"] == 0
|
|
|
|
|
|
async def test_readyz(client):
|
|
r = await client.get("/readyz") # DB 도달 → ready
|
|
assert r.status_code == 200 and r.json()["ready"] is True
|
|
|
|
|
|
async def test_ops_snapshot(client, clean_jobs):
|
|
await client.post("/v1/lps/search", json={"data": [{"product_code": "A", "product_name": "x"}]})
|
|
r = await client.get("/v1/lps/ops")
|
|
assert r.status_code == 200
|
|
j = r.json()
|
|
for k in ("pending", "running", "done", "dead", "dead_1h", "stuck_running", "oldest_pending_sec",
|
|
"blocks_1h", "deadline_1h", "pool_pct"):
|
|
assert k in j and isinstance(j[k], int)
|
|
assert isinstance(j["cost_1h_usd"], (int, float))
|
|
assert j["pending"] == 1
|