"""fact 테이블 제약 — '사이트에 나가는 값은 (사업장, 단위, key) 당 1건' 이 DB 레벨에서 지켜지는지. 유니크는 **노출 상태(VERIFIED·CORRECTED)에만** 걸린다. - 걸어야 하는 이유: 안 걸면 체크인 시간이 15시/16시 두 값으로 동시에 노출된다. - 활성 전체에 걸면 안 되는 이유: 재수집이 올 때마다 확인된 노출값을 밀어내야 하고, 그 순간 사이트에서 사실이 사라진다. 후보(UNVERIFIED·PENDING_OWNER)는 공존해야 한다. """ import uuid import pytest from sqlalchemy import text from sqlalchemy.exc import IntegrityError from common.enums import ( FACT_STATUS_TRANSITIONS, LOCKED_FACT_STATUSES, PUBLISHABLE_FACT_STATUSES, FactStatus, PlaceCategory, PlaceStatus, SourceType, ) async def _seed_place(db_engine, company_id) -> str: """검증까지 끝난 사업장 1개를 시드하고 place_id 를 돌려준다.""" pid = uuid.uuid4() async with db_engine.begin() as conn: await conn.execute( text( "INSERT INTO places (place_id, company_id, name, category, status, external_place_id, verified_at) " "VALUES (:pid, :cid, :name, :cat, :status, :kakao, now())" ), { "pid": pid, "cid": uuid.UUID(company_id), "name": "테스트펜션", "cat": PlaceCategory.LODGING.value, "status": PlaceStatus.DRAFT.value, "kakao": "12345678", }, ) return str(pid) async def _insert_fact(db_engine, place_id, key, value, status, unit_id=None): async with db_engine.begin() as conn: await conn.execute( text( "INSERT INTO facts (fact_id, place_id, unit_id, key, value, source_type, status, collected_at) " "VALUES (:fid, :pid, :uid, :key, :val, :src, :status, now())" ), { "fid": uuid.uuid4(), "pid": uuid.UUID(place_id), "uid": unit_id, "key": key, "val": value, "src": SourceType.CRAWL.value, "status": status.value, }, ) async def test_published_fact_is_unique_per_place_and_key(db_engine, company_id): """검증: 같은 사업장·같은 key 로 노출 상태 fact 를 두 번 넣는다. 기대결과: 두 번째 INSERT 가 유니크 인덱스에 막힌다(체크인 시간이 두 값으로 갈라지지 않는다).""" place_id = await _seed_place(db_engine, company_id) await _insert_fact(db_engine, place_id, "check_in_time", "15:00", FactStatus.VERIFIED) with pytest.raises(IntegrityError): await _insert_fact(db_engine, place_id, "check_in_time", "16:00", FactStatus.CORRECTED) async def test_candidates_coexist_with_published_value(db_engine, company_id): """검증: 노출값이 있는 상태에서 재수집 후보를 여러 건 넣는다. 기대결과: 전부 공존한다 — ★ 재수집이 노출 중인 사실을 밀어내지 않는다.""" place_id = await _seed_place(db_engine, company_id) await _insert_fact(db_engine, place_id, "check_in_time", "15:00", FactStatus.VERIFIED) await _insert_fact(db_engine, place_id, "check_in_time", "16:00", FactStatus.PENDING_OWNER) await _insert_fact(db_engine, place_id, "check_in_time", "14:00", FactStatus.UNVERIFIED) async with db_engine.begin() as conn: rows = (await conn.execute( text("SELECT status FROM facts WHERE place_id = :pid AND key = 'check_in_time'"), {"pid": uuid.UUID(place_id)}, )).all() assert len(rows) == 3, "노출값 1건 + 후보 2건이 공존해야 한다" async def test_rejected_fact_frees_the_key(db_engine, company_id): """검증: 기존 값을 REJECTED 로 내린 뒤 같은 key 를 새로 노출한다. 기대결과: 통과 — 틀린 값은 이력으로 남고, 새 값이 노출 자리를 차지한다.""" place_id = await _seed_place(db_engine, company_id) await _insert_fact(db_engine, place_id, "check_in_time", "15:00", FactStatus.REJECTED) await _insert_fact(db_engine, place_id, "check_in_time", "16:00", FactStatus.VERIFIED) async with db_engine.begin() as conn: rows = (await conn.execute( text("SELECT value, status FROM facts WHERE place_id = :pid ORDER BY status"), {"pid": uuid.UUID(place_id)}, )).all() assert len(rows) == 2, "REJECTED 이력과 새 값이 함께 남아야 한다" async def test_expired_fact_frees_the_key(db_engine, company_id): """검증: 유효기간이 지나 EXPIRED 로 내린 값과 새 수집값의 공존. 기대결과: 통과 — EXPIRED 도 유니크에서 빠진다.""" place_id = await _seed_place(db_engine, company_id) await _insert_fact(db_engine, place_id, "cancel_policy", "구 규정", FactStatus.EXPIRED) await _insert_fact(db_engine, place_id, "cancel_policy", "새 규정", FactStatus.VERIFIED) async def test_same_key_allowed_across_units(db_engine, company_id): """검증: 객실이 다르면 같은 key 를 각각 가질 수 있는지. 기대결과: 통과 — A동·B동이 각자의 기준 인원을 갖는다.""" place_id = await _seed_place(db_engine, company_id) unit_a, unit_b = uuid.uuid4(), uuid.uuid4() async with db_engine.begin() as conn: for uid, name in ((unit_a, "A동"), (unit_b, "B동")): await conn.execute( text("INSERT INTO units (unit_id, place_id, name) VALUES (:uid, :pid, :name)"), {"uid": uid, "pid": uuid.UUID(place_id), "name": name}, ) await _insert_fact(db_engine, place_id, "standard_capacity", "4", FactStatus.VERIFIED, unit_id=unit_a) await _insert_fact(db_engine, place_id, "standard_capacity", "2", FactStatus.VERIFIED, unit_id=unit_b) with pytest.raises(IntegrityError): # 같은 객실 안에서 노출값은 여전히 1건 await _insert_fact(db_engine, place_id, "standard_capacity", "6", FactStatus.CORRECTED, unit_id=unit_a) async def test_unit_fact_and_place_fact_are_separate(db_engine, company_id): """검증: 같은 key 를 사업장 단위와 객실 단위로 동시에 갖는 경우. 기대결과: 통과 — 부분 인덱스가 unit_id NULL 여부로 갈라져 있다.""" place_id = await _seed_place(db_engine, company_id) unit_id = uuid.uuid4() async with db_engine.begin() as conn: await conn.execute( text("INSERT INTO units (unit_id, place_id, name) VALUES (:uid, :pid, :name)"), {"uid": unit_id, "pid": uuid.UUID(place_id), "name": "A동"}, ) await _insert_fact(db_engine, place_id, "has_kitchen", "false", FactStatus.VERIFIED) await _insert_fact(db_engine, place_id, "has_kitchen", "true", FactStatus.VERIFIED, unit_id=unit_id) def test_only_verified_and_corrected_are_publishable(): """검증: 노출 가능 상태 집합(절대규칙 1). 기대결과: VERIFIED·CORRECTED 뿐. 미검증·반려·만료는 절대 사이트에 나가지 않는다.""" assert PUBLISHABLE_FACT_STATUSES == {FactStatus.VERIFIED, FactStatus.CORRECTED} for status in (FactStatus.UNVERIFIED, FactStatus.PENDING_OWNER, FactStatus.REJECTED, FactStatus.EXPIRED): assert status not in PUBLISHABLE_FACT_STATUSES def test_corrected_is_locked_against_auto_update(): """검증: 사람이 고친 값이 잠기는지(절대규칙 6). 기대결과: CORRECTED 는 잠금 상태이고, 전이표에서 자동 갱신 경로(UNVERIFIED 등)로 못 돌아간다.""" assert FactStatus.CORRECTED in LOCKED_FACT_STATUSES allowed = FACT_STATUS_TRANSITIONS[FactStatus.CORRECTED] assert FactStatus.UNVERIFIED not in allowed, "자동 수집이 사장님 수정본을 덮어쓸 수 있으면 안 된다" assert FactStatus.VERIFIED not in allowed def test_transition_table_covers_every_status(): """검증: 전이표가 모든 상태를 다루는지. 기대결과: 6개 상태 전부 키로 존재하고, 목적지도 전부 유효한 FactStatus.""" assert set(FACT_STATUS_TRANSITIONS) == set(FactStatus) for src, dests in FACT_STATUS_TRANSITIONS.items(): assert dests, f"{src.name}: 나갈 수 있는 상태가 없다" for dest in dests: assert isinstance(dest, FactStatus)