o2o-site-AEO/solution/backend/tests/test_collect_pipeline.py

"""수집 파이프라인 e2e — 잡이 실제로 돌아 fact·사진이 후보로 쌓이는지.

    상호명 → (URL 발견) → 확정 → 크롤링 → fact/사진 적재

★ 이 파이프라인이 절대 하면 안 되는 것:
   - 검증 안 된 사업장을 긁는 것
   - 수집값을 바로 사이트에 노출시키는 것 (전부 후보로 들어가야 한다)
   - 사장님 정정본을 덮어쓰는 것
"""
import uuid

from common.enums import FactStatus, JobStatus, JobType, LinkChannel, MediaStatus, PlaceCategory, SourceType
from crud.job_crud import JobQueue
from services.collector import MockAdapter
from worker.handlers import build_handler
from worker.runner import Worker


async def _ready_place(client, h, category=PlaceCategory.LODGING, kakao="k1"):
    """검증까지 끝나고 MockAdapter 가 처리할 수 있는 링크가 확정된 사업장."""
    pid = (await client.post("/v1/place", headers=h, json={"name": "하조대펜션", "category": category.value})).json()["place"]["place_id"]
    await client.post(f"/v1/place/{pid}/verify", headers=h, json={
        "external_place_id": kakao, "road_address": "강원 양양군 현북면 하조대해안길 3", "region_code": "4283025"})
    url = MockAdapter.url_for(category, pid, channel="yanolja")
    lid = (await client.post(f"/v1/place/{pid}/link", headers=h, json={
        "channel": LinkChannel.YANOLJA.value, "url": url, "discovered_by": SourceType.API.value})).json()["link"]["link_id"]
    await client.post(f"/v1/place/{pid}/link/{lid}/confirm", headers=h)
    return pid


async def _run_worker(job_id=None):
    """워커를 돌려 큐를 비운다. `job_id` 를 주면 그 잡이 끝날 때까지 돈다.

    ★ 1틱만 돌리면 안 된다. 수집이 끝나면 **지역 이야기 잡(LOCAL_SYNC)이 뒤따라 들어온다** —
      한 틱은 그걸 집어 가고, 정작 기다리던 수집 잡은 PENDING 인 채로 남는다. 테스트는
      `job["result"]` 를 읽다가 KeyError 로 죽는데, 화면에는 "그냥 안 끝난 것" 으로 보인다.
      큐에 뒤따르는 잡이 생길 때마다 이 헬퍼가 조용히 어긋나므로 개수를 세지 않고 비운다.
    """
    worker = Worker("test-worker", JobQueue(), build_handler(), job_deadline_sec=60)
    ran = 0
    for _ in range(10):
        if not await worker.process_one():
            break
        ran += 1
    assert ran, "워커가 집을 잡이 없다"


async def test_pipeline_publishes_collected_facts(auth_headers, client):
    """검증: 수집 잡을 끝까지 돌린다.
    기대결과: 빈 자리에 들어온 수집값이 **바로 노출값**이 된다(2026-09-14 결정).

    ★ 예전에는 전부 UNVERIFIED 후보였다. 그러면 수집 직후 발행이 '확인된 사실 0건' 으로
      막혀, 사장님이 한 건씩 승인하기 전에는 사이트가 만들어지지 않았다.
    ★ 사람이 넣은 값·정정본을 덮지 않는다는 보호는 그대로다 —
      test_recollect_cannot_overwrite_corrected_value 가 그 자리를 지킨다."""
    h = await auth_headers("u1")
    pid = await _ready_place(client, h)

    job_id = (await client.post(f"/v1/place/{pid}/collect", headers=h, json={})).json()["job_id"]
    await _run_worker()

    job = (await client.get(f"/v1/job/{job_id}", headers=h)).json()["job"]
    assert job["status"] == JobStatus.DONE.value, job.get("last_error")
    assert job["result"]["facts"]["stored"] > 0

    listed = (await client.get(f"/v1/place/{pid}/fact/list", headers=h)).json()
    assert len(listed["facts"]) > 0
    assert listed["publishable"] > 0, "수집값이 노출값 자리에 들어가지 않았다"
    assert all(
        f["status"] in (FactStatus.VERIFIED.value, FactStatus.UNVERIFIED.value)
        for f in listed["facts"]
    )


async def test_pipeline_records_source_on_every_fact(auth_headers, client):
    """검증: 수집된 fact 의 출처.
    기대결과: 전부 source_type=crawl + source_url 이 붙어 있다 — 출처 없는 사실은 없다."""
    h = await auth_headers("u1")
    pid = await _ready_place(client, h, kakao="k2")
    await client.post(f"/v1/place/{pid}/collect", headers=h, json={})
    await _run_worker()

    facts = (await client.get(f"/v1/place/{pid}/fact/list", headers=h)).json()["facts"]
    for f in facts:
        assert f["source_type"] == SourceType.CRAWL.value
        assert f["source_url"], f"출처 없는 fact: {f['key']}"


async def test_pipeline_creates_units_and_unit_scoped_facts(auth_headers, client):
    """검증: 숙박 수집 결과의 객실 단위 fact.
    기대결과: units 가 생기고 객실별 fact 가 각자 붙는다(A동·B동이 각자 기준인원을 갖는다)."""
    h = await auth_headers("u1")
    pid = await _ready_place(client, h, kakao="k3")
    await client.post(f"/v1/place/{pid}/collect", headers=h, json={})
    await _run_worker()

    units = (await client.get(f"/v1/place/{pid}/unit/list", headers=h)).json()["units"]
    assert len(units) >= 2

    facts = (await client.get(f"/v1/place/{pid}/fact/list", headers=h)).json()["facts"]
    unit_facts = [f for f in facts if f.get("unit_id")]
    assert unit_facts, "객실 단위 fact 가 하나도 없다"
    capacities = [f for f in unit_facts if f["key"] == "standard_capacity"]
    assert len({f["unit_id"] for f in capacities}) >= 2, "객실별로 따로 붙어야 한다"


async def test_pipeline_stores_media_with_origin_and_pending_review(auth_headers, client):
    """검증: 수집된 사진.
    기대결과: origin_url·source_type=crawl 이 남고 PENDING_REVIEW 다 —
    ★ 재게시 권리 결론에 따라 통째로 걸러낼 수 있어야 하고, Vision 전이라 사람 확인 큐다."""
    from sqlalchemy import text

    h = await auth_headers("u1")
    pid = await _ready_place(client, h, kakao="k4")
    job_id = (await client.post(f"/v1/place/{pid}/collect", headers=h, json={})).json()["job_id"]
    await _run_worker()

    job = (await client.get(f"/v1/job/{job_id}", headers=h)).json()["job"]
    assert job["result"]["media"]["stored"] > 0


async def test_recollect_skips_crawl_when_already_enough(auth_headers, client):
    """검증: 필수 항목이 이미 다 찬 사업장에 다시 수집을 건다.
    기대결과: ★ 크롤링을 아예 하지 않는다 — 사이트를 만들 정보가 충분하면 여분의 크롤링은 낭비다."""
    h = await auth_headers("u1")
    pid = await _ready_place(client, h, kakao="k5")

    await client.post(f"/v1/place/{pid}/collect", headers=h, json={})
    await _run_worker()
    first = (await client.get(f"/v1/place/{pid}/fact/list", headers=h)).json()

    job_id = (await client.post(f"/v1/place/{pid}/collect", headers=h, json={})).json()["job_id"]
    await _run_worker()
    job = (await client.get(f"/v1/job/{job_id}", headers=h)).json()["job"]

    assert job["result"]["coverage"]["enough"] is True
    assert "크롤링 생략" in job["result"].get("note", "")
    assert "fetch" not in job["result"], "충분한데 크롤링을 시도했다"

    second = (await client.get(f"/v1/place/{pid}/fact/list", headers=h)).json()
    assert len(second["facts"]) == len(first["facts"]), "재수집이 fact 를 중복 생성했다"


async def test_forced_recollect_is_idempotent(auth_headers, client):
    """검증: force=true 로 강제 재수집한다(항목이 이미 차 있어도).
    기대결과: 다시 긁되 fact 는 REFRESHED, 사진은 중복 스킵 — 데이터가 부풀지 않는다."""
    h = await auth_headers("u1")
    pid = await _ready_place(client, h, kakao="k5f")

    await client.post(f"/v1/place/{pid}/collect", headers=h, json={})
    await _run_worker()
    first = (await client.get(f"/v1/place/{pid}/fact/list", headers=h)).json()

    job_id = (await client.post(f"/v1/place/{pid}/collect", headers=h, json={"force": True})).json()["job_id"]
    await _run_worker()
    job = (await client.get(f"/v1/job/{job_id}", headers=h)).json()["job"]

    assert job["result"]["fetch"]["fetched"] == 1, "force 인데 크롤링을 안 했다"
    assert job["result"]["media"]["stored"] == 0
    assert job["result"]["media"]["skipped_duplicate"] > 0

    second = (await client.get(f"/v1/place/{pid}/fact/list", headers=h)).json()
    assert len(second["facts"]) == len(first["facts"]), "재수집이 fact 를 중복 생성했다"


async def test_coverage_reports_missing_required_fields(auth_headers, client):
    """검증: 수집 후 필수 항목 충족도.
    기대결과: coverage 에 required/covered/missing 이 담긴다 — UI 가 '뭐가 비었나'를 보여줄 수 있다."""
    h = await auth_headers("u1")
    pid = await _ready_place(client, h, kakao="k5c")

    job_id = (await client.post(f"/v1/place/{pid}/collect", headers=h, json={})).json()["job_id"]
    await _run_worker()
    cov = (await client.get(f"/v1/job/{job_id}", headers=h)).json()["job"]["result"]["coverage"]

    assert cov["total"] > 0
    assert cov["covered"] == cov["total"]
    assert cov["missing"] == []
    assert "check_in_time" in cov["required"]


async def test_recollect_does_not_touch_verified_value(auth_headers, client):
    """검증: 수집을 두 번 돌린다(같은 목데이터라 값이 같다).
    기대결과: 값이 같으므로 REFRESHED — ★ 사이트에 나가던 사실이 사라지지 않고 늘지도 않는다.

    ★ 재수집이 같은 값을 후보로 또 쌓으면 확인 큐가 중복으로 넘치고, 노출값을 지웠다
      다시 넣으면 그 사이에 사이트에서 사실이 사라진다. 둘 다 안 일어나야 한다."""
    h = await auth_headers("u1")
    pid = await _ready_place(client, h, kakao="k6")
    await client.post(f"/v1/place/{pid}/collect", headers=h, json={})
    await _run_worker()

    before = (await client.get(f"/v1/place/{pid}/fact/list", headers=h)).json()
    assert before["publishable"] > 0
    check_in_before = [f for f in before["facts"] if f["key"] == "check_in_time"]
    assert len(check_in_before) == 1 and check_in_before[0]["status"] == FactStatus.VERIFIED.value

    await client.post(f"/v1/place/{pid}/collect", headers=h, json={})
    await _run_worker()

    after = (await client.get(f"/v1/place/{pid}/fact/list", headers=h)).json()
    assert after["publishable"] == before["publishable"], "★ 재수집이 노출값 수를 바꿨다"
    check_in = [f for f in after["facts"] if f["key"] == "check_in_time"]
    assert len(check_in) == 1 and check_in[0]["status"] == FactStatus.VERIFIED.value


async def test_recollect_cannot_overwrite_corrected_value(auth_headers, client):
    """검증: 사장님이 정정한 값에 재수집이 다른 값을 들고 온다.
    기대결과: 노출값은 정정본 그대로, 크롤링 값은 후보로만 남는다 — ★ 절대규칙 6."""
    h = await auth_headers("u1")
    pid = await _ready_place(client, h, kakao="k7")
    await client.post(f"/v1/place/{pid}/collect", headers=h, json={})
    await _run_worker()

    facts = (await client.get(f"/v1/place/{pid}/fact/list", headers=h)).json()["facts"]
    target = next(f for f in facts if f["key"] == "check_in_time")
    fid = target["fact_id"]
    await client.post(f"/v1/place/{pid}/fact/{fid}/transition", headers=h, json={"status": FactStatus.VERIFIED.value})
    await client.post(f"/v1/place/{pid}/fact/{fid}/transition", headers=h,
                      json={"status": FactStatus.CORRECTED.value, "value": "16:30"})

    await client.post(f"/v1/place/{pid}/collect", headers=h, json={})
    await _run_worker()

    published = (await client.get(f"/v1/place/{pid}/fact/list", headers=h, params={"publishable_only": True})).json()["facts"]
    check_in = [f for f in published if f["key"] == "check_in_time"]
    assert len(check_in) == 1
    assert check_in[0]["value"] == "16:30", "★ 자동 수집이 사장님 정정본을 덮어썼다"
    assert check_in[0]["status"] == FactStatus.CORRECTED.value


async def test_pipeline_refuses_unverified_place(db_engine, owner_id):
    """검증: 검증 안 된 사업장의 수집 잡이 큐에 직접 들어간 경우(잡 적재 후 검증이 취소된 상황).
    기대결과: 잡이 실패한다 — ★ 잡 실행 시점에도 게이트를 다시 확인한다."""
    from sqlalchemy import text

    pid = uuid.uuid4()
    async with db_engine.begin() as conn:
        await conn.execute(
            text("INSERT INTO places (place_id, owner_user_id, name, category, status) "
                 "VALUES (:pid, :cid, :n, 1, 1)"),
            {"pid": pid, "cid": uuid.UUID(owner_id), "n": "미검증펜션"},
        )

    q = JobQueue()
    job_id = await q.enqueue(JobType.COLLECT.value, {"place_id": str(pid), "owner_user_id": owner_id}, max_attempts=1)
    worker = Worker("test-worker", q, build_handler(), backoff_fn=lambda _a: 0)
    await worker.process_one()

    row = await q.get(job_id)
    assert row["status"] == JobStatus.DEAD.value
    assert "동일 업소 검증" in row["last_error"]