Web4ai/o2o-site-AEO
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Docs
o2o-site-AEO/solution/backend/tests/test_fact_api.py
Mina Choi 9d25ed613e 구조: 사장님(solution)과 내부 운영(admin)을 두 앱으로 가른다 
최상단을 프로젝트 단위로 평평하게 둔다 — o2o-negosium 과 같은 규약이고, 이 레포만
다르게 갈 이유가 없다. negodata/{backend,front} 가 프로젝트 안에서 f/b 를 가르는 선례,
lps-admin/ 이 백엔드 없이 프론트만 가진 최상단 폴더의 선례다.

  backend/ frontend/{admin,site,shared}  →  solution/{backend,front,site,shared} + admin/

## 왜

내부 라우트(/local-content, /places/:id/seo)의 이름과 화면 코드가 사장님 번들에
그대로 실려 나가고 있었다. UserRole.DEVELOPER 주석의 "고객사에 존재를 노출하지 않는다"를
번들이 깨고 있었다 — 라우트 가드는 화면을 가리지 번들은 못 가린다.
번들을 갈라 확인했다: 사장님 dist 에서 local-content · /places · SeoAudit 이 전부 0건이다.

그 과정에서 두 곳이 더 새고 있었다.
- AppShell 의 NAV 배열이 내부 메뉴를 하드코딩하고 있었다. 앱을 가른 뒤에도 dist 에
  local-content 가 남아서 찾았다. 메뉴는 이제 앱이 prop 으로 들고 온다.
- EditorHeader·BuilderPage·LoginPage 가 /places 로 링크하고 있었다. 그 화면이 admin 으로
  나갔으니 사장님 앱에서는 404 다. 링크를 걷어내고 LoginPage 기본 도착지는 '/' 로 바꿨다
  (앱마다 홈이 다르고 각 라우터의 '/' 가 이미 그걸 안다).

## admin 에 백엔드를 두지 않았다

내부 화면이 부르는 훅이 전부 router/v1/{place,fact,local,validator} 에 이미 있다.
자체 백엔드를 두면 place·fact·link 를 같은 DB 에 대고 두 번 구현하게 된다.
대가는 solution/backend 가 죽으면 admin 도 멈추는 것 — 내부 도구라 감수한다.

## admin 의 `@` 는 solution/front/src 를 가리킨다

내부 화면이 쓰는 API 클라이언트·UI·수집 배선이 solution 에 한 벌만 있고 그 파일들끼리도
`@/...` 로 서로를 부른다. admin 에서 `@` 를 자기 src 로 잡으면 그 참조가 전부 깨진다
(실측 TS2307 14건). 복제하는 길도 있지만 RecollectPanel 주석이 금지한다 —
"수집 경로를 두 벌 만들면 확정 게이트"가 갈라진다.
admin 자기 파일만 `@admin` 이고, 의존 방향은 admin → solution 한 쪽뿐이다.

admin 이 여는 빌더는 다른 오리진이라 절대 URL + 새 탭이다(admin/src/lib/solutionUrl.ts).
react-router Link 로 두면 admin 안에서 라우트를 찾다 404 다.

## 그 밖

- npm 워크스페이스 루트를 레포 루트로 올렸다(admin 이 solution 밖이라).
- docker-compose 를 255→174줄로 줄이고 admin(:3002) 서비스를 넣었다. ADMIN_BIND 기본값은
  127.0.0.1 — 0.0.0.0 으로 열면 앱을 가른 의미가 없다.
- 발행 호스트를 프론트 .env 에 따로 적지 않는다. compose 가 루트의 SITE_PUBLIC_HOST 를
  VITE_PUBLISH_HOST 로 흘려보낸다 — 두 곳에 적으면 canonical 과 화면 주소가 조용히 갈라진다.
- nginx/site.conf 를 git 에서 빼고 .example 만 남겼다(.env·*.toml 과 같은 규약).
  compose 가 bind mount 하므로 클론 직후 복사해야 한다 — 없으면 Docker 가 그 자리에
  디렉토리를 만들어 nginx 가 설정 없이 뜬다.
- config.test.toml.example 을 추가했다. 없으면 클론한 사람이 pytest 를 아예 못 돌린다
  (conftest import 단계에서 죽는다). 외부 API 키는 전부 빈값이다 —
  APP_ENV=test 가 .env 를 안 읽는 이유를 여기서 우회하면 안 된다.
- 경로가 한 칸 깊어져 test_schema_ddl(parents[2]→[3]) 과 test_site_theme 을 고쳤다.

검증: front·admin·site 전부 lint 0 / build 0. 백엔드 514 passed.
남은 4건(test_build_publish 3 · test_snapshot 1)은 이 변경 전부터 실패하던 것으로,
손대지 않은 메인 체크아웃에서 같은 4건이 같게 실패하는 것을 확인했다.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019uYhHQdssRubirPirrdJJC
2026-08-31 15:12:09 +09:00

281 lines
15 KiB
Python
Raw Permalink Blame History

"""facts 도메인 e2e — 생성 / 업데이트(재수집) / 수정 세 프로세스.
이 도메인이 지켜야 하는 것:
생성 : 자동 수집 → 후보 → 사람 승인 → 노출. 미검증은 절대 사이트에 안 나간다
업데이트: ★ 재수집이 노출 중인 사실을 밀어내지 않는다
값이 같으면 검증 유지, 다르면 후보로 쌓이고 사람이 승인해야 교체된다
수정 : 사람이 직접 넣으면 즉시 노출값 교체 + 재빌드 대상 표시
"""
from common.enums import ErrorType, FactStatus, FactWriteOutcome, PlaceCategory, SourceType
OTA = "https://ota.test/room/1"
async def _verified_place(client, headers, category=PlaceCategory.LODGING, kakao="999"):
r = await client.post("/v1/place", headers=headers, json={"name": "테스트업소", "category": category.value})
pid = r.json()["place"]["place_id"]
await client.post(f"/v1/place/{pid}/verify", headers=headers, json={"external_place_id": kakao})
return pid
async def _crawl(client, headers, pid, key, value, source=SourceType.CRAWL):
return (await client.post(f"/v1/place/{pid}/fact", headers=headers, json={
"key": key, "value": value, "source_type": source.value, "source_url": OTA,
})).json()
async def _own(client, headers, pid, key, value):
return (await client.post(f"/v1/place/{pid}/fact", headers=headers, json={
"key": key, "value": value, "source_type": SourceType.OWNER.value,
})).json()
async def _facts(client, headers, pid, **params):
return (await client.get(f"/v1/place/{pid}/fact/list", headers=headers, params=params)).json()
# ── 입력 검증 ────────────────────────────────────────────────────────────
async def test_schema_endpoint_returns_category_fields(auth_headers, client):
"""검증: 숙박 사업장의 fact 스키마 조회.
기대결과: 업종=lodging, 체크인시간이 critical=True 로 내려온다(관리 화면 폼 생성용)."""
h = await auth_headers("u1")
pid = await _verified_place(client, h)
body = (await client.get(f"/v1/place/{pid}/fact/schema", headers=h)).json()
assert body["category"] == "lodging"
check_in = next(f for f in body["fields"] if f["key"] == "check_in_time")
assert check_in["critical"] is True
assert check_in["allow_llm"] is False
async def test_key_outside_category_schema_is_rejected(auth_headers, client):
"""검증: 카페 사업장에 숙박 전용 key(check_in_time)를 기록한다.
기대결과: FACT_INVALID_KEY — 업종에 없는 필드는 저장되지 않는다."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, PlaceCategory.CAFE, kakao="c1")
body = await _own(client, h, pid, "check_in_time", "15:00")
assert body["result"]["code"] == ErrorType.FACT_INVALID_KEY.value
async def test_non_owner_source_requires_source_url(auth_headers, client):
"""검증: crawl 출처인데 source_url 없이 기록한다.
기대결과: FACT_SOURCE_REQUIRED — 출처 없는 사실은 받지 않는다."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="s1")
r = await client.post(f"/v1/place/{pid}/fact", headers=h, json={
"key": "check_in_time", "value": "15:00", "source_type": SourceType.CRAWL.value})
assert r.json()["result"]["code"] == ErrorType.FACT_SOURCE_REQUIRED.value
async def test_llm_cannot_write_non_sentence_fields(auth_headers, client):
"""검증: LLM 출처로 체크인시간(allow_llm=False)을 기록한다.
기대결과: 거부 — ★ LLM 은 사실을 만들지 않는다. 소개문(intro)에는 쓸 수 있다."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="l1")
body = await _crawl(client, h, pid, "check_in_time", "15:00", SourceType.LLM)
assert body["result"]["code"] == ErrorType.FACT_INVALID_KEY.value
body = await _crawl(client, h, pid, "intro", "조용한 숙소입니다.", SourceType.LLM)
assert body["result"]["success"] is True
# ── 생성 프로세스 ─────────────────────────────────────────────────────────
async def test_crawled_fact_starts_as_candidate_and_is_not_published(auth_headers, client):
"""검증: 크롤링으로 처음 들어온 값.
기대결과: UNVERIFIED 후보로 남고 사이트에 안 나간다 — ★ 절대규칙 1."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="p1")
body = await _crawl(client, h, pid, "check_in_time", "15:00")
assert body["outcome"] == FactWriteOutcome.CANDIDATE_CREATED.value
assert body["fact"]["status"] == FactStatus.UNVERIFIED.value
listed = await _facts(client, h, pid)
assert listed["publishable"] == 0
assert (await _facts(client, h, pid, publishable_only=True)).get("facts", []) == []
async def test_approving_candidate_publishes_it(auth_headers, client):
"""검증: 후보를 VERIFIED 로 승인한다.
기대결과: verified_at 이 찍히고 사이트에 나갈 수 있게 된다."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="p2")
fid = (await _crawl(client, h, pid, "check_in_time", "15:00"))["fact"]["fact_id"]
body = (await client.post(f"/v1/place/{pid}/fact/{fid}/transition", headers=h,
json={"status": FactStatus.VERIFIED.value})).json()
assert body["result"]["success"] is True
assert body["fact"]["status"] == FactStatus.VERIFIED.value
assert body["fact"]["verified_at"] is not None
assert (await _facts(client, h, pid))["publishable"] == 1
async def test_illegal_transition_is_rejected(auth_headers, client):
"""검증: UNVERIFIED → CORRECTED 처럼 전이표에 없는 이동.
기대결과: FACT_INVALID_TRANSITION — 확인을 건너뛴 '정정본'은 만들 수 없다."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="p3")
fid = (await _crawl(client, h, pid, "check_in_time", "15:00"))["fact"]["fact_id"]
r = await client.post(f"/v1/place/{pid}/fact/{fid}/transition", headers=h,
json={"status": FactStatus.CORRECTED.value, "value": "16:00"})
assert r.json()["result"]["code"] == ErrorType.FACT_INVALID_TRANSITION.value
# ── 업데이트(재수집) 프로세스 ★ ───────────────────────────────────────────
async def test_recrawl_same_value_keeps_verification(auth_headers, client):
"""검증: 확인된 값과 똑같은 값을 재수집한다.
기대결과: REFRESHED — 검증이 유지되고 사이트에서 사실이 사라지지 않는다."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="r1")
fid = (await _crawl(client, h, pid, "check_in_time", "15:00"))["fact"]["fact_id"]
await client.post(f"/v1/place/{pid}/fact/{fid}/transition", headers=h, json={"status": FactStatus.VERIFIED.value})
assert (await _facts(client, h, pid))["publishable"] == 1
body = await _crawl(client, h, pid, "check_in_time", "15:00")
assert body["outcome"] == FactWriteOutcome.REFRESHED.value
assert body["fact"]["status"] == FactStatus.VERIFIED.value
assert (await _facts(client, h, pid))["publishable"] == 1, "★ 재수집이 검증을 초기화하면 안 된다"
async def test_recrawl_changed_value_keeps_site_and_queues_candidate(auth_headers, client):
"""검증: 확인된 값과 다른 값을 재수집한다(OTA 가 바뀐 경우).
기대결과: ★ 노출값은 그대로 살아 있고, 새 값은 PENDING_OWNER 후보로만 쌓인다."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="r2")
fid = (await _crawl(client, h, pid, "check_in_time", "15:00"))["fact"]["fact_id"]
await client.post(f"/v1/place/{pid}/fact/{fid}/transition", headers=h, json={"status": FactStatus.VERIFIED.value})
body = await _crawl(client, h, pid, "check_in_time", "16:00")
assert body["outcome"] == FactWriteOutcome.CANDIDATE_CREATED.value
assert body["fact"]["status"] == FactStatus.PENDING_OWNER.value
listed = await _facts(client, h, pid)
assert listed["publishable"] == 1, "★ 재수집 중에도 사이트에는 확인된 값이 계속 나가야 한다"
assert listed["pending_review"] == 1
published = (await _facts(client, h, pid, publishable_only=True))["facts"]
assert published[0]["value"] == "15:00"
async def test_approving_candidate_replaces_published_value(auth_headers, client):
"""검증: 쌓인 후보를 사람이 승인한다(업데이트의 마지막 단계).
기대결과: 후보가 노출값이 되고 옛 값은 EXPIRED 이력으로 내려간다. 노출값은 여전히 1건."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="r3")
old = (await _crawl(client, h, pid, "check_in_time", "15:00"))["fact"]["fact_id"]
await client.post(f"/v1/place/{pid}/fact/{old}/transition", headers=h, json={"status": FactStatus.VERIFIED.value})
new = (await _crawl(client, h, pid, "check_in_time", "16:00"))["fact"]["fact_id"]
body = (await client.post(f"/v1/place/{pid}/fact/{new}/transition", headers=h,
json={"status": FactStatus.VERIFIED.value})).json()
assert body["result"]["success"] is True
assert body["outcome"] == FactWriteOutcome.PUBLISHED_REPLACED.value
published = (await _facts(client, h, pid, publishable_only=True))["facts"]
assert len(published) == 1 and published[0]["value"] == "16:00"
assert (await _facts(client, h, pid))["pending_review"] == 0
async def test_repeated_recrawl_does_not_pile_up_candidates(auth_headers, client):
"""검증: 같은 출처로 재수집을 세 번 반복한다.
기대결과: 후보가 쌓이지 않고 하나가 갱신된다(사람 확인 큐가 중복으로 넘치지 않게)."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="r4")
fid = (await _crawl(client, h, pid, "check_in_time", "15:00"))["fact"]["fact_id"]
await client.post(f"/v1/place/{pid}/fact/{fid}/transition", headers=h, json={"status": FactStatus.VERIFIED.value})
await _crawl(client, h, pid, "check_in_time", "16:00")
body = await _crawl(client, h, pid, "check_in_time", "17:00")
assert body["outcome"] == FactWriteOutcome.CANDIDATE_UPDATED.value
listed = await _facts(client, h, pid)
assert listed["pending_review"] == 1
candidate = next(f for f in listed["facts"] if f["status"] == FactStatus.PENDING_OWNER.value)
assert candidate["value"] == "17:00"
async def test_recrawl_cannot_overwrite_corrected_value(auth_headers, client):
"""검증: 사람이 정정한 값(CORRECTED)에 다른 값이 재수집된다.
기대결과: ★ 노출값은 정정본 그대로. 크롤링 값은 버려지지 않고 후보로 남아 불일치가 보인다."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="r5")
fid = (await _crawl(client, h, pid, "check_in_time", "15:00"))["fact"]["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:00"})
body = await _crawl(client, h, pid, "check_in_time", "15:00")
assert body["result"]["success"] is True
assert body["fact"]["status"] == FactStatus.PENDING_OWNER.value
published = (await _facts(client, h, pid, publishable_only=True))["facts"]
assert len(published) == 1
assert published[0]["value"] == "16:00", "★ 자동 수집이 사람 정정본을 덮어쓰면 안 된다"
assert published[0]["status"] == FactStatus.CORRECTED.value
# OTA 가 아직 15:00 이라는 사실은 후보로 남아 있어야 한다(신호를 버리지 않는다).
assert (await _facts(client, h, pid))["pending_review"] == 1
# ── 수정 프로세스 ────────────────────────────────────────────────────────
async def test_owner_input_publishes_immediately(auth_headers, client):
"""검증: 사람이 직접 값을 넣는다.
기대결과: 즉시 노출값이 된다 — 넣은 사람이 곧 출처이자 책임 주체다."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="o1")
body = await _own(client, h, pid, "check_in_time", "15:00")
assert body["outcome"] == FactWriteOutcome.PUBLISHED_CREATED.value
assert body["fact"]["status"] == FactStatus.VERIFIED.value
assert (await _facts(client, h, pid))["publishable"] == 1
async def test_owner_can_overwrite_own_value(auth_headers, client):
"""검증: 사람이 자기가 넣은 값을 다시 고친다.
기대결과: 노출값이 교체되고(PUBLISHED_REPLACED) 옛 값은 이력으로 내려간다."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="o2")
await _own(client, h, pid, "check_in_time", "15:00")
body = await _own(client, h, pid, "check_in_time", "17:00")
assert body["outcome"] == FactWriteOutcome.PUBLISHED_REPLACED.value
published = (await _facts(client, h, pid, publishable_only=True))["facts"]
assert len(published) == 1 and published[0]["value"] == "17:00"
async def test_publishing_marks_place_for_rebuild(auth_headers, client):
"""검증: 노출값이 바뀐 뒤 사업장의 content_updated_at.
기대결과: 값이 찍힌다 — ★ 이 사업장만 재빌드하면 된다는 표시(전체 재빌드 금지)."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="o3")
assert (await client.get(f"/v1/place/{pid}", headers=h)).json()["place"].get("content_updated_at") is None
await _own(client, h, pid, "check_in_time", "15:00")
place = (await client.get(f"/v1/place/{pid}", headers=h)).json()["place"]
assert place["content_updated_at"] is not None
async def test_crawl_only_does_not_mark_rebuild(auth_headers, client):
"""검증: 크롤링이 후보만 쌓았을 때 재빌드 표시.
기대결과: 안 찍힌다 — 사이트에 나가는 내용이 안 바뀌었으니 재빌드가 필요 없다."""
h = await auth_headers("u1")
pid = await _verified_place(client, h, kakao="o4")
await _crawl(client, h, pid, "check_in_time", "15:00")
place = (await client.get(f"/v1/place/{pid}", headers=h)).json()["place"]
assert place.get("content_updated_at") is None
async def test_facts_are_scoped_to_company(auth_headers, client, other_company_id):
"""검증: 다른 회사 계정으로 남의 사업장 fact 를 조회한다.
기대결과: PLACE_NOT_FOUND — 사업장이 안 보이니 fact 도 안 보인다."""
h1 = await auth_headers("o1")
pid = await _verified_place(client, h1, kakao="p6")
h2 = await auth_headers("o2", other_company_id)
r = await client.get(f"/v1/place/{pid}/fact/list", headers=h2)
assert r.json()["result"]["code"] == ErrorType.PLACE_NOT_FOUND.value
Reference in New Issue View Git Blame Copy Permalink