4b48be7e81
user_of 는 처음 들어온 커밋(b1a34ba)부터 한 번도 호출되지 않았고, import 하는
decode_access_token 은 어느 브랜치 기록에도 정의된 적이 없다 — 부르는 순간 ImportError 다.
같은 일은 DecodeAccessToken 을 쓰는 _sections 가 이미 한다.
- tests/test_agent_runtime.py: user_of 삭제(호출처 0 — 레포 전체·문자열 참조까지 확인)
pyflakes 경고 없음, test_agent_runtime 84 passed
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
882 lines
48 KiB
Python
882 lines
48 KiB
Python
"""사장님 에이전트 런타임.
|
|
|
|
여기서 지키는 것 셋 — 나머지 검사는 전부 이 셋을 지탱한다.
|
|
1. 도구는 서비스 계층을 통과한다(게이트가 살아 있다)
|
|
2. 등급은 레지스트리가 정한다 — 모델이 확인 절차를 건너뛸 수 없다
|
|
3. 모호하면 실행하지 않고 되묻는다
|
|
"""
|
|
|
|
import uuid
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
|
|
from common.enums import PlaceCategory
|
|
from services.agent import runtime, tools
|
|
from services.agent.tools import ToolGrade
|
|
from services.llm.errors import LlmError
|
|
|
|
|
|
@pytest.fixture
|
|
def choose(monkeypatch):
|
|
"""LLM 을 대신한다 — 테스트는 절대 실제 모델을 부르지 않는다."""
|
|
|
|
def _set(payload):
|
|
monkeypatch.setattr(runtime, "_choose", AsyncMock(return_value=payload))
|
|
|
|
monkeypatch.setenv("AGENT_CHAT_ENABLED", "1")
|
|
monkeypatch.setattr(runtime, "is_configured", lambda: True)
|
|
return _set
|
|
|
|
|
|
async def seed(client, auth_headers, name="대화숙소"):
|
|
h = await auth_headers(f"agent-{uuid.uuid4().hex[:8]}")
|
|
res = await client.post("/v1/place", headers=h, json={"name": name, "category": 1})
|
|
return h, res.json()["place"]["place_id"]
|
|
|
|
|
|
# ── 1. 게이트가 살아 있다 ────────────────────────────────────────────────
|
|
|
|
def test_모든_도구는_서비스_계층을_통과한다():
|
|
"""★ 도구가 crud 를 직접 부르면 스키마 검증·출처·정정본 보호가 조용히 사라진다.
|
|
|
|
소스에 `_crud.` 직접 호출이 없는지 본다 — 주석이 아니라 코드로 못 박는 자리다."""
|
|
import inspect
|
|
|
|
source = inspect.getsource(tools)
|
|
body = source[source.index("# ── 읽기"):source.index("class ToolRejected")]
|
|
assert "fact_crud." not in body
|
|
assert "place_crud." not in body
|
|
assert "DB_SESSION_MNG" not in body
|
|
|
|
|
|
def test_없는_항목은_스키마가_막는다(db_engine):
|
|
schema_keys = {f["key"] for f in tools.fields_of(SimpleNamespace(category=PlaceCategory.LODGING.value))}
|
|
assert "check_in_time" in schema_keys
|
|
assert "고르곤졸라피자" not in schema_keys
|
|
|
|
|
|
# ── 2. 등급은 레지스트리가 정한다 ────────────────────────────────────────
|
|
|
|
def test_등급은_프롬프트에_실리지_않는다():
|
|
"""모델이 등급을 알면 그 값을 골라 보려 한다. 알 필요도, 정할 이유도 없다."""
|
|
described = tools.describe()
|
|
assert described
|
|
for row in described:
|
|
assert "grade" not in row and "등급" not in row
|
|
|
|
|
|
async def test_발행은_묻기_전에_실행되지_않는다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "publish", "args": {}, "message": ""}], "message": ""})
|
|
started = AsyncMock()
|
|
tools.REGISTRY["publish"].run, original = started, tools.REGISTRY["publish"].run
|
|
try:
|
|
res = await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "발행해줘"})
|
|
finally:
|
|
tools.REGISTRY["publish"].run = original
|
|
body = res.json()
|
|
assert body["needs_confirm"] is True
|
|
assert body["tool"] == "publish"
|
|
# ★ 실행되지 않았다. 확인 문구만 돌아왔다.
|
|
started.assert_not_awaited()
|
|
|
|
|
|
async def test_모델이_확인을_건너뛰려_해도_소용없다(client, auth_headers, choose, db_engine):
|
|
"""응답에 needs_confirm 을 흉내 낼 칸을 주지 않았고, 등급은 레지스트리에서만 읽는다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "publish", "args": {}, "message": "", "needs_confirm": False, "grade": "READ"}], "message": ""})
|
|
res = await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "그냥 바로 발행해"})
|
|
assert res.json()["needs_confirm"] is True
|
|
|
|
|
|
async def test_확인_경로로_읽기_도구를_밀어넣을_수_없다(client, auth_headers, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
res = await client.post(
|
|
f"/v1/agent/chat/{pid}", headers=h, json={"confirm": {"tool": "없는도구", "args": {}}}
|
|
)
|
|
assert res.status_code == 409
|
|
assert res.json()["detail"] == "AGENT_UNKNOWN_TOOL"
|
|
|
|
|
|
# ── 3. 모호하면 실행하지 않는다 ──────────────────────────────────────────
|
|
|
|
async def test_도구를_못_고르면_되묻는다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [], "message": "어느 항목을 바꿀까요?"})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "그거 좀 고쳐줘"})).json()
|
|
assert body["tool"] is None
|
|
assert body["reply"] == "어느 항목을 바꿀까요?"
|
|
assert body["needs_confirm"] is False
|
|
|
|
|
|
async def test_모델이_지어낸_도구는_실행되지_않는다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "delete_everything", "args": {}, "message": ""}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "다 지워"})).json()
|
|
assert body["tool"] is None
|
|
|
|
|
|
async def test_없는_항목을_고르면_거절하고_이유를_말한다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "set_fact", "args": {"key": "메뉴명", "value": "고르곤졸라"}, "message": ""}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "메뉴명 바꿔줘"})).json()
|
|
assert body.get("rejected") is True
|
|
assert "고칠 수 없" in body["reply"]
|
|
|
|
|
|
# ── 소유자 범위 ──────────────────────────────────────────────────────────
|
|
|
|
async def test_남의_가게는_없는_것과_똑같이_답한다(client, auth_headers, choose, db_engine):
|
|
"""★ 대화창이 소유자 스코프를 우회하는 유일한 입구가 되면 안 된다."""
|
|
_mine, pid = await seed(client, auth_headers, "내가게")
|
|
other = await auth_headers("agent-outsider")
|
|
choose({"actions": [{"tool": "list_facts", "args": {}, "message": ""}], "message": ""})
|
|
res = await client.post(f"/v1/agent/chat/{pid}", headers=other, json={"message": "정보 보여줘"})
|
|
assert res.status_code == 404
|
|
assert res.json()["detail"] == "PLACE_NOT_FOUND"
|
|
|
|
|
|
async def test_로그인_없이는_열리지_않는다(client):
|
|
res = await client.post(f"/v1/agent/chat/{uuid.uuid4()}", json={"message": "안녕"})
|
|
assert res.status_code in (401, 403)
|
|
|
|
|
|
# ── 실행 결과 문구 ───────────────────────────────────────────────────────
|
|
|
|
async def test_값을_바꾸면_재발행이_필요하다고_말한다(client, auth_headers, choose, db_engine):
|
|
"""★ 이 한 줄이 빠지면 사장님은 반영된 줄 알고 확인하러 갔다가 옛 값을 본다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "set_fact", "args": {"key": "check_in_time", "value": "15:00"}, "message": ""}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "체크인 3시로"})).json()
|
|
assert body.get("rejected") is not True, body["reply"]
|
|
assert "체크인 시간" in body["reply"]
|
|
assert "발행" in body["reply"]
|
|
|
|
async with db_engine.begin() as c:
|
|
stored = (
|
|
await c.execute(
|
|
text("SELECT value FROM place_facts WHERE place_id=:p AND key='check_in_time' AND deleted=false"),
|
|
{"p": uuid.UUID(pid)},
|
|
)
|
|
).scalars().all()
|
|
assert "15:00" in stored
|
|
|
|
|
|
async def test_결과_문구는_모델이_쓰지_않는다(client, auth_headers, choose, db_engine):
|
|
"""모델이 결과를 쓰면 하지 않은 일을 했다고 말할 수 있다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({
|
|
"actions": [{"tool": "set_fact", "args": {"key": "check_in_time", "value": "15:00"}}],
|
|
"message": "사이트까지 전부 반영을 끝냈습니다!",
|
|
})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "체크인 3시로"})).json()
|
|
assert "전부 반영을 끝냈습니다" not in body["reply"]
|
|
|
|
|
|
# ── 실패 처리 ────────────────────────────────────────────────────────────
|
|
|
|
async def test_LLM_실패는_502_로_나가고_원문을_흘리지_않는다(client, auth_headers, monkeypatch, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
monkeypatch.setattr(runtime, "is_configured", lambda: True)
|
|
monkeypatch.setattr(runtime, "_choose", AsyncMock(side_effect=LlmError("키가 sk-1234 라서 실패")))
|
|
res = await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "안녕"})
|
|
assert res.status_code == 502
|
|
assert res.json()["detail"] == "AGENT_CALL_FAILED"
|
|
assert "sk-1234" not in res.text
|
|
|
|
|
|
async def test_키가_없으면_대화창을_열지_않는다(client, auth_headers, monkeypatch, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
monkeypatch.setattr(runtime, "is_configured", lambda: False)
|
|
res = await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "안녕"})
|
|
assert res.status_code == 409
|
|
assert res.json()["detail"] == "AGENT_NOT_CONFIGURED"
|
|
assert (await client.get("/v1/agent/status", headers=h)).json()["enabled"] is False
|
|
|
|
|
|
async def test_너무_긴_발화는_모델을_부르기_전에_끊는다(client, auth_headers, monkeypatch, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
called = AsyncMock()
|
|
monkeypatch.setattr(runtime, "_choose", called)
|
|
res = await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "가" * (runtime.MAX_MESSAGE + 1)})
|
|
assert res.status_code == 422 # pydantic 이 먼저 막는다
|
|
called.assert_not_awaited()
|
|
|
|
|
|
def test_읽기_도구는_확인을_요구하지_않는다():
|
|
for name in ("get_site_status", "list_facts"):
|
|
assert tools.REGISTRY[name].grade == ToolGrade.READ
|
|
assert tools.REGISTRY["set_fact"].grade == ToolGrade.REVERSIBLE
|
|
assert tools.REGISTRY["publish"].grade == ToolGrade.SEMI
|
|
assert tools.REGISTRY["publish"].confirm
|
|
|
|
|
|
# ── 보류 스위치 ─────────────────────────────────────────────────────────
|
|
|
|
def test_스위치와_키를_둘_다_본다(monkeypatch):
|
|
"""★ 키만 보면 '잠시 닫아 두기' 를 키를 지워서 해야 하고, 그러면 소개문·사진분류까지
|
|
같이 꺼진다. 스위치만 보면 키 없는 환경에서 **눌러도 안 되는 입구**가 생긴다."""
|
|
monkeypatch.setattr(runtime.provider, "active", lambda: SimpleNamespace(is_configured=lambda: True))
|
|
monkeypatch.delenv("AGENT_CHAT_ENABLED", raising=False)
|
|
assert runtime.is_configured() is True # 기본은 켜짐(2026-09-22)
|
|
monkeypatch.setenv("AGENT_CHAT_ENABLED", "0")
|
|
assert runtime.is_configured() is False # 스위치로 닫을 수 있다
|
|
|
|
monkeypatch.setenv("AGENT_CHAT_ENABLED", "1")
|
|
monkeypatch.setattr(runtime.provider, "active", lambda: SimpleNamespace(is_configured=lambda: False))
|
|
assert runtime.is_configured() is False # 켜 뒀어도 키가 없으면 안 열린다
|
|
|
|
|
|
async def test_닫아_두면_대화_요청은_거절된다(client, auth_headers, monkeypatch, db_engine):
|
|
monkeypatch.setenv("AGENT_CHAT_ENABLED", "0")
|
|
h, pid = await seed(client, auth_headers)
|
|
res = await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "안녕"})
|
|
assert res.status_code == 409
|
|
assert res.json()["detail"] == "AGENT_NOT_CONFIGURED"
|
|
assert (await client.get("/v1/agent/status", headers=h)).json()["enabled"] is False
|
|
|
|
|
|
# ── 페이지 구성 (섹션 on/off · 순서) ─────────────────────────────────────
|
|
|
|
async def _sections(client, headers, place_id):
|
|
"""대화가 보는 것과 같은 목록 — 도구가 쓰는 함수를 그대로 쓴다."""
|
|
from services.agent.tools import ToolContext, sections_of
|
|
from router.v1.validator.dependencies import DecodeAccessToken
|
|
from crud.place_crud import PlaceCRUD
|
|
from common.database.db_session_manager import DB_SESSION_MNG
|
|
from common.database.model.models import places
|
|
from common.enums import DBWRType
|
|
|
|
user = DecodeAccessToken(headers["Authorization"].split(" ", 1)[1])
|
|
_err, place = await DB_SESSION_MNG.execute_lambda(
|
|
places.DBType(), DBWRType.DB_READ.value,
|
|
lambda s: PlaceCRUD().get_place(s, uuid.UUID(user.user_id), uuid.UUID(place_id)),
|
|
)
|
|
rows, theme = await sections_of(ToolContext(user=user, place_id=place_id, place=place))
|
|
return rows, theme
|
|
|
|
|
|
async def test_저장값이_없어도_업종_기본_구성이_보인다(client, auth_headers, choose, db_engine):
|
|
"""★ 아직 디자인을 한 번도 안 만진 사업장에서도 대화가 바로 통해야 한다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
rows, _theme = await _sections(client, h, pid)
|
|
names = {r["name"] for r in rows}
|
|
assert "히어로" in names and "객실 안내" in names
|
|
|
|
|
|
async def test_섹션을_끄면_순서는_그대로고_그_칸만_꺼진다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
before, _t = await _sections(client, h, pid)
|
|
order_before = [r["id"] for r in before]
|
|
|
|
choose({"actions": [{"tool": "toggle_section", "args": {"name": "사진 갤러리", "enabled": "false"}, "message": ""}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "사진 갤러리 빼줘"})).json()
|
|
assert body.get("rejected") is not True, body["reply"]
|
|
assert "발행" in body["reply"] # 재발행 안내가 따라붙는다
|
|
|
|
after, _t = await _sections(client, h, pid)
|
|
assert [r["id"] for r in after] == order_before
|
|
assert next(r for r in after if r["name"] == "사진 갤러리")["enabled"] is False
|
|
|
|
|
|
async def test_꼭_있어야_하는_부분은_끌_수_없다(client, auth_headers, choose, db_engine):
|
|
"""★ 잠긴 섹션은 발행본이 어차피 켜서 내보낸다 — 끌 수 있게 두면 화면만 거짓말한다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "toggle_section", "args": {"name": "히어로", "enabled": "false"}, "message": ""}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "히어로 빼줘"})).json()
|
|
assert body.get("rejected") is True
|
|
assert "끌 수 없" in body["reply"]
|
|
|
|
after, _t = await _sections(client, h, pid)
|
|
assert next(r for r in after if r["name"] == "히어로")["enabled"] is True
|
|
|
|
|
|
async def test_순서를_옮기면_배열_순서가_바뀐다(client, auth_headers, choose, db_engine):
|
|
"""★ 배열 순서가 곧 발행본의 섹션 순서다(site_payload._sections)."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "move_section", "args": {"name": "사진 갤러리", "to": "맨 위"}, "message": ""}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "사진 갤러리 맨 위로"})).json()
|
|
assert body.get("rejected") is not True, body["reply"]
|
|
|
|
after, _t = await _sections(client, h, pid)
|
|
assert after[0]["name"] == "사진 갤러리"
|
|
|
|
|
|
async def test_어느_것인지_모호하면_고르지_않는다(client, auth_headers, choose, db_engine):
|
|
"""★ 추측으로 고르면 엉뚱한 부분을 끄고, 사장님은 발행하고 나서야 안다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "toggle_section", "args": {"name": "정보", "enabled": "false"}, "message": ""}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "정보 빼줘"})).json()
|
|
assert body.get("rejected") is True
|
|
assert "못 찾았" in body["reply"]
|
|
|
|
|
|
async def test_색과_서체는_손대지_않는다(client, auth_headers, choose, db_engine):
|
|
"""★ sections 만 갈아끼운다 — 통째로 새로 쓰면 사장님이 고른 색이 말없이 사라진다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
await client.post(f"/v1/place/{pid}/site/theme", headers=h, json={
|
|
"theme": {"colors": {"accent": "#123456"}, "fontStyle": "고딕"},
|
|
})
|
|
choose({"actions": [{"tool": "toggle_section", "args": {"name": "사진 갤러리", "enabled": "false"}, "message": ""}], "message": ""})
|
|
await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "사진 갤러리 빼줘"})
|
|
|
|
_rows, theme = await _sections(client, h, pid)
|
|
assert theme["colors"]["accent"] == "#123456"
|
|
assert theme["fontStyle"] == "고딕"
|
|
|
|
|
|
def test_구성_도구의_등급(client=None):
|
|
assert tools.REGISTRY["list_sections"].grade == ToolGrade.READ
|
|
assert tools.REGISTRY["toggle_section"].grade == ToolGrade.REVERSIBLE
|
|
assert tools.REGISTRY["move_section"].grade == ToolGrade.REVERSIBLE
|
|
|
|
|
|
# ── 사진 (내리기 · 대표 지정) ────────────────────────────────────────────
|
|
|
|
async def _seed_photos(db_engine, place_id, rows):
|
|
"""(label, unit_id, sort_order) 로 사진을 심는다. alt 는 채운다 — 빈 alt 는 발행본이 안 그린다."""
|
|
from common.enums import MediaStatus, SourceType
|
|
|
|
async with db_engine.begin() as c:
|
|
for label, unit_id, order in rows:
|
|
await c.execute(
|
|
text("""INSERT INTO place_photos
|
|
(media_id, place_id, unit_id, url, source_type, label, alt_text, status, sort_order)
|
|
VALUES (:m, :p, :u, :url, :st, :l, :alt, :status, :o)"""),
|
|
{"m": uuid.uuid4(), "p": uuid.UUID(place_id), "u": unit_id,
|
|
"url": f"https://example.com/{label}.jpg", "st": SourceType.CRAWL.value,
|
|
"l": label, "alt": f"{label} 사진", "status": MediaStatus.APPROVED.value, "o": order},
|
|
)
|
|
|
|
|
|
async def _photo_rows(db_engine, place_id):
|
|
async with db_engine.begin() as c:
|
|
result = await c.execute(
|
|
text("""SELECT label, status, sort_order FROM place_photos
|
|
WHERE place_id=:p AND deleted=false
|
|
ORDER BY sort_order ASC, created_at ASC"""),
|
|
{"p": uuid.UUID(place_id)},
|
|
)
|
|
return result.all()
|
|
|
|
|
|
async def test_사진을_내리면_지우지_않고_내려간다(client, auth_headers, choose, db_engine):
|
|
"""★ 지우면 origin_url·source_type 이 사라져 재게시 권리 결론이 났을 때 되짚을 수 없다."""
|
|
from common.enums import MediaStatus
|
|
|
|
h, pid = await seed(client, auth_headers)
|
|
await _seed_photos(db_engine, pid, [("외관", None, 0), ("객실", None, 1)])
|
|
|
|
choose({"actions": [{"tool": "hide_photo", "args": {"name": "객실"}, "message": ""}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "객실 사진 내려줘"})).json()
|
|
assert body.get("rejected") is not True, body["reply"]
|
|
assert "발행" in body["reply"]
|
|
|
|
rows = await _photo_rows(db_engine, pid)
|
|
hidden = next(r for r in rows if r.label == "객실")
|
|
assert hidden.status == MediaStatus.REJECTED.value
|
|
assert len(rows) == 2 # 행은 남는다
|
|
|
|
|
|
async def test_대표_사진은_목록의_첫_장이_된다(client, auth_headers, choose, db_engine):
|
|
"""★ 별도 칸을 두지 않는다 — site_payload.primary_media 가 '첫 장' 을 쓴다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
await _seed_photos(db_engine, pid, [("외관", None, 0), ("수영장", None, 1)])
|
|
|
|
choose({"actions": [{"tool": "set_primary_photo", "args": {"name": "수영장"}, "message": ""}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "대표 사진 수영장으로"})).json()
|
|
assert body.get("rejected") is not True, body["reply"]
|
|
|
|
rows = await _photo_rows(db_engine, pid)
|
|
assert rows[0].label == "수영장"
|
|
|
|
|
|
async def test_객실_전용_사진은_대표가_될_수_없다(client, auth_headers, choose, db_engine):
|
|
"""★ primary_media 가 unit_id 있는 사진을 건너뛴다 — 지정하게 두면 화면만 거짓말한다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
await _seed_photos(db_engine, pid, [("외관", None, 0), ("침대", uuid.uuid4(), 1)])
|
|
|
|
choose({"actions": [{"tool": "set_primary_photo", "args": {"name": "침대"}, "message": ""}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "대표 사진 침대로"})).json()
|
|
assert body.get("rejected") is True
|
|
assert "대표로 쓸 수 없" in body["reply"]
|
|
|
|
rows = await _photo_rows(db_engine, pid)
|
|
assert rows[0].label == "외관"
|
|
|
|
|
|
async def test_어느_사진인지_모호하면_고르지_않는다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
await _seed_photos(db_engine, pid, [("객실 A", None, 0), ("객실 B", None, 1)])
|
|
|
|
choose({"actions": [{"tool": "hide_photo", "args": {"name": "객실"}, "message": ""}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "객실 사진 내려줘"})).json()
|
|
assert body.get("rejected") is True
|
|
assert "못 찾았" in body["reply"]
|
|
|
|
rows = await _photo_rows(db_engine, pid)
|
|
assert all(r.status != 3 for r in rows) # 아무것도 안 내려갔다
|
|
|
|
|
|
async def test_남의_가게_사진에는_닿지_못한다(client, auth_headers, db_engine):
|
|
"""서비스가 사업장 스코프를 먼저 건다 — 없는 것과 남의 것을 똑같이 답한다."""
|
|
from crud.media_crud import MediaCRUD
|
|
from crud.place_crud import PlaceCRUD
|
|
from router.v1.validator.dependencies import DecodeAccessToken
|
|
from services.media_service import MediaService
|
|
|
|
h, pid = await seed(client, auth_headers, "내가게")
|
|
await _seed_photos(db_engine, pid, [("외관", None, 0)])
|
|
async with db_engine.begin() as c:
|
|
media_id = (
|
|
await c.execute(text("SELECT media_id FROM place_photos WHERE place_id=:p"), {"p": uuid.UUID(pid)})
|
|
).scalar_one()
|
|
|
|
other = await auth_headers("photo-outsider")
|
|
outsider = DecodeAccessToken(other["Authorization"].split(" ", 1)[1])
|
|
res = await MediaService(MediaCRUD(), PlaceCRUD()).hide_media(outsider, pid, str(media_id))
|
|
assert res.result.success is False
|
|
|
|
rows = await _photo_rows(db_engine, pid)
|
|
assert rows[0].status != 3
|
|
|
|
|
|
async def test_업로드_도구는_만들지_않았다():
|
|
"""★ 이미지 재게시 권리가 미결이라 저장 경로를 일부러 안 만들어 뒀다(DECISIONS 1-2·5-3).
|
|
도구가 생기면 그 결정을 코드가 먼저 풀어 버린다."""
|
|
names = set(tools.REGISTRY)
|
|
assert not {n for n in names if "upload" in n or "replace" in n}
|
|
assert tools.REGISTRY["list_photos"].grade == ToolGrade.READ
|
|
assert tools.REGISTRY["hide_photo"].grade == ToolGrade.REVERSIBLE
|
|
assert tools.REGISTRY["set_primary_photo"].grade == ToolGrade.REVERSIBLE
|
|
|
|
|
|
def test_도구가_선언한_인자는_응답_스키마에_있다():
|
|
"""★ 스키마에 없는 이름은 모델이 채울 자리가 없다 — 도구가 선언해도 빈손으로 온다.
|
|
|
|
실측(2026-09-28, 킹서버): move_section 이 {name,to} 를 받는데 스키마에 그 칸이 없어
|
|
{key,value} 로 왔고 도구는 "못 찾았어요" 로 끝났다. **도구 선택은 6/6 정확했는데도** 그랬다 —
|
|
다른 테스트는 `_choose` 를 monkeypatch 해서 이 층을 통째로 건너뛰므로 아무도 못 잡는다."""
|
|
from services.prompts import agent as prompt
|
|
|
|
action = prompt.RESPONSE_SCHEMA["properties"]["actions"]["items"]
|
|
allowed = set(action["properties"]["args"]["properties"])
|
|
for name, tool in tools.REGISTRY.items():
|
|
missing = set(tool.args) - allowed
|
|
assert not missing, f"{name} 이 선언한 인자가 응답 스키마에 없다: {missing}"
|
|
|
|
|
|
# ── 한 발화에 여러 가지 ──────────────────────────────────────────────────
|
|
|
|
async def test_한_번에_두_가지를_시키면_둘_다_한다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [
|
|
{"tool": "set_fact", "args": {"key": "check_in_time", "value": "15:00"}},
|
|
{"tool": "toggle_section", "args": {"name": "사진 갤러리", "enabled": "false"}},
|
|
], "message": ""})
|
|
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h,
|
|
json={"message": "체크인 3시로 바꾸고 사진 갤러리도 빼줘"})).json()
|
|
assert body.get("rejected") is not True, body["reply"]
|
|
assert "체크인 시간" in body["reply"] and "사진 갤러리" in body["reply"]
|
|
|
|
rows, _t = await _sections(client, h, pid)
|
|
assert next(r for r in rows if r["name"] == "사진 갤러리")["enabled"] is False
|
|
|
|
|
|
async def test_재발행_안내는_한_번만_붙는다(client, auth_headers, choose, db_engine):
|
|
"""★ 도구마다 문장에 박아 두면 셋을 고쳤을 때 같은 말이 세 번 나온다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [
|
|
{"tool": "set_fact", "args": {"key": "check_in_time", "value": "15:00"}},
|
|
{"tool": "toggle_section", "args": {"name": "사진 갤러리", "enabled": "false"}},
|
|
{"tool": "move_section", "args": {"name": "소개", "to": "맨 위"}},
|
|
], "message": ""})
|
|
|
|
reply = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "셋 다 해줘"})).json()["reply"]
|
|
assert reply.count(runtime.REPUBLISH_NOTICE) == 1
|
|
|
|
|
|
async def test_중간에_실패하면_앞의_것은_남기고_거기서_멈춘다(client, auth_headers, choose, db_engine):
|
|
"""★ 되돌리지 않는다(2026-09-28 결정). 대신 무엇이 됐고 무엇이 안 됐는지 그대로 말한다 —
|
|
부분 성공을 뭉뚱그리면 사장님은 전부 된 줄 안다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [
|
|
{"tool": "set_fact", "args": {"key": "check_in_time", "value": "15:00"}},
|
|
{"tool": "toggle_section", "args": {"name": "없는섹션", "enabled": "false"}},
|
|
{"tool": "move_section", "args": {"name": "소개", "to": "맨 위"}},
|
|
], "message": ""})
|
|
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "셋 다 해줘"})).json()
|
|
assert body["rejected"] is True
|
|
assert "체크인 시간" in body["reply"] # 앞의 것은 됐다고 말한다
|
|
assert "멈췄습니다" in body["reply"] # 어디서 멈췄는지도 말한다
|
|
|
|
async with db_engine.begin() as c:
|
|
stored = (await c.execute(
|
|
text("SELECT value FROM place_facts WHERE place_id=:p AND key='check_in_time' AND deleted=false"),
|
|
{"p": uuid.UUID(pid)},
|
|
)).scalars().all()
|
|
assert "15:00" in stored # ★ 되돌리지 않았다
|
|
|
|
rows, _t = await _sections(client, h, pid)
|
|
assert rows[0]["name"] != "소개" # 뒤의 것은 하지 않았다
|
|
|
|
|
|
async def test_발행이_섞이면_앞까지만_하고_확인을_받는다(client, auth_headers, choose, db_engine):
|
|
"""★ 확인이 필요한 행위를 다른 일에 묻어 실행하면 확인의 의미가 없다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
started = AsyncMock()
|
|
tools.REGISTRY["publish"].run, original = started, tools.REGISTRY["publish"].run
|
|
try:
|
|
choose({"actions": [
|
|
{"tool": "set_fact", "args": {"key": "check_in_time", "value": "15:00"}},
|
|
{"tool": "publish", "args": {}},
|
|
], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h,
|
|
json={"message": "체크인 3시로 바꾸고 발행해줘"})).json()
|
|
finally:
|
|
tools.REGISTRY["publish"].run = original
|
|
|
|
assert body["needs_confirm"] is True
|
|
assert body["tool"] == "publish"
|
|
assert "체크인 시간" in body["reply"] # 앞서 한 일을 함께 말한다
|
|
started.assert_not_awaited() # ★ 발행은 실행되지 않았다
|
|
|
|
|
|
async def test_한_번에_다섯_가지까지만_한다(client, auth_headers, choose, db_engine):
|
|
"""★ 무한정 허용하면 '다 지워줘' 한 마디에 연쇄로 실행된다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
names = ["사진 갤러리", "소개", "예약 안내", "자주 묻는 질문", "날씨", "지역 정보"]
|
|
choose({"actions": [
|
|
{"tool": "toggle_section", "args": {"name": n, "enabled": "false"}} for n in names
|
|
], "message": ""})
|
|
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "다 빼줘"})).json()
|
|
assert f"{runtime.MAX_ACTIONS}가지까지" in body["reply"]
|
|
|
|
rows, _t = await _sections(client, h, pid)
|
|
off = {r["name"] for r in rows if not r["enabled"]}
|
|
assert names[-1] not in off # 여섯 번째는 하지 않았다
|
|
|
|
|
|
async def test_모델이_지어낸_도구는_건너뛰고_나머지는_한다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [
|
|
{"tool": "delete_everything", "args": {}},
|
|
{"tool": "set_fact", "args": {"key": "check_in_time", "value": "15:00"}},
|
|
], "message": ""})
|
|
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "다 지우고 체크인 3시로"})).json()
|
|
assert "체크인 시간" in body["reply"]
|
|
assert body.get("rejected") is not True
|
|
|
|
|
|
# ── 값 형식 (bool · time · number) ──────────────────────────────────────
|
|
|
|
async def _fact_value(db_engine, place_id, key):
|
|
async with db_engine.begin() as c:
|
|
return (await c.execute(
|
|
text("SELECT value FROM place_facts WHERE place_id=:p AND key=:k AND deleted=false"),
|
|
{"p": uuid.UUID(place_id), "k": key},
|
|
)).scalars().all()
|
|
|
|
|
|
@pytest.mark.parametrize("said, stored", [
|
|
("가능", "true"), ("돼요", "true"), ("있음", "true"), ("True", "true"),
|
|
("불가", "false"), ("안 돼요", "false"), ("없음", "false"),
|
|
])
|
|
async def test_예_아니오_항목은_true_false_로_저장한다(said, stored, client, auth_headers, choose, db_engine):
|
|
"""★ 렌더러(shared/src/lib/facts.ts factBool)는 'true' 만 참으로 읽는다. "가능" 으로 저장하면
|
|
화면에는 "가능" 이 뜨는데 구조화 데이터는 거짓이 된다 — 조용히 틀리는 종류다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "set_fact", "args": {"key": "pet_allowed", "value": said}}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "반려동물 바꿔줘"})).json()
|
|
assert body.get("rejected") is not True, body["reply"]
|
|
assert await _fact_value(db_engine, pid, "pet_allowed") == [stored]
|
|
assert "true" not in body["reply"] and "false" not in body["reply"] # 사장님께는 가능·불가로 말한다
|
|
|
|
|
|
async def test_예_아니오를_알아볼_수_없으면_저장하지_않는다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "set_fact", "args": {"key": "pet_allowed", "value": "소형견만"}}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "소형견만 돼"})).json()
|
|
assert body["rejected"] is True
|
|
assert await _fact_value(db_engine, pid, "pet_allowed") == []
|
|
|
|
|
|
@pytest.mark.parametrize("said, stored", [
|
|
("15:00", "15:00"), ("9:30", "09:30"), ("오후 3시", "15:00"), ("15시 30분", "15:30"),
|
|
("오후 3시 반", "15:30"), ("오전 11시", "11:00"), ("12시", "12:00"),
|
|
])
|
|
async def test_시각은_HH_MM_으로_저장한다(said, stored, client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "set_fact", "args": {"key": "check_in_time", "value": said}}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "체크인 바꿔줘"})).json()
|
|
assert body.get("rejected") is not True, body["reply"]
|
|
assert await _fact_value(db_engine, pid, "check_in_time") == [stored]
|
|
|
|
|
|
@pytest.mark.parametrize("said", ["3시", "25:00", "곧"])
|
|
async def test_오전_오후가_모호하거나_시각이_아니면_되묻는다(said, client, auth_headers, choose, db_engine):
|
|
"""★ '3시' 를 03:00 으로 넣으면 손님이 새벽에 온다. 추측하지 않고 묻는다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "set_fact", "args": {"key": "check_in_time", "value": said}}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "체크인 바꿔줘"})).json()
|
|
assert body["rejected"] is True
|
|
assert await _fact_value(db_engine, pid, "check_in_time") == []
|
|
|
|
|
|
@pytest.mark.parametrize("said, stored", [
|
|
("20000", "20000"), ("2만원", "20000"), ("2만 5천원", "25000"), ("20,000원", "20000"), ("무료", "0"),
|
|
])
|
|
async def test_숫자_항목은_숫자만_저장한다(said, stored, client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "set_fact", "args": {"key": "extra_person_fee", "value": said}}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "추가 요금 바꿔줘"})).json()
|
|
assert body.get("rejected") is not True, body["reply"]
|
|
assert await _fact_value(db_engine, pid, "extra_person_fee") == [stored]
|
|
|
|
|
|
async def test_숫자가_아니면_저장하지_않는다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "set_fact", "args": {"key": "extra_person_fee", "value": "문의"}}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "추가 요금은 문의"})).json()
|
|
assert body["rejected"] is True
|
|
assert await _fact_value(db_engine, pid, "extra_person_fee") == []
|
|
|
|
|
|
def test_프롬프트에_항목_형식이_실린다():
|
|
"""★ 형식을 안 알려 주면 모델은 '주차 가능' 에 "가능" 을 쓴다 — 코드가 고쳐 주기 전에 맞게 오게 한다."""
|
|
from services.prompts import agent as prompt
|
|
|
|
fields = tools.fields_of(SimpleNamespace(category=PlaceCategory.LODGING.value))
|
|
text_ = prompt.build_prompt(place_name="가게", tools=tools.describe(), fields=fields, facts=[],
|
|
sections=[], photos=[], message="안녕")
|
|
assert "pet_allowed: 반려동물 동반 (true/false)" in text_
|
|
assert "check_in_time: 체크인 시간 (HH:MM" in text_
|
|
assert "extra_person_fee: 인원 추가 요금 (숫자만" in text_
|
|
|
|
|
|
async def test_모델이_문자열이_아닌_값을_줘도_터지지_않는다(client, auth_headers, choose, db_engine):
|
|
"""모델은 스키마를 어길 수 있다 — true 를 불리언으로, 인자를 배열로 보낸다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [
|
|
{"tool": "set_fact", "args": {"key": "pet_allowed", "value": True}},
|
|
{"tool": "toggle_section", "args": ["사진 갤러리"]},
|
|
], "message": ""})
|
|
res = await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "둘 다"})
|
|
assert res.status_code == 200, res.text
|
|
assert await _fact_value(db_engine, pid, "pet_allowed") == ["true"]
|
|
|
|
|
|
# ── 켜기·끄기가 비었을 때 ────────────────────────────────────────────────
|
|
|
|
async def test_켤지_끌지_모르면_끄지_않고_되묻는다(client, auth_headers, choose, db_engine):
|
|
"""★ 스키마가 모든 인자를 필수로 받아서 모델이 enabled 를 "" 로 채울 수 있다.
|
|
그걸 '끄기' 로 읽으면 "후기 다시 보여줘" 가 후기를 끈다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "toggle_section", "args": {"name": "사진 갤러리", "enabled": ""}}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "사진 갤러리 다시"})).json()
|
|
assert body["rejected"] is True
|
|
assert "켤지 끌지" in body["reply"]
|
|
|
|
rows, _t = await _sections(client, h, pid)
|
|
assert next(r for r in rows if r["name"] == "사진 갤러리")["enabled"] is True
|
|
|
|
|
|
@pytest.mark.parametrize("said", ["false", "끄기", "off", "빼기"])
|
|
async def test_끄는_말은_끈다(said, client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "toggle_section", "args": {"name": "사진 갤러리", "enabled": said}}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "빼줘"})).json()
|
|
assert body.get("rejected") is not True, body["reply"]
|
|
rows, _t = await _sections(client, h, pid)
|
|
assert next(r for r in rows if r["name"] == "사진 갤러리")["enabled"] is False
|
|
|
|
|
|
# ── 사진 — 사이트에 안 나가는 사진 ───────────────────────────────────────
|
|
|
|
async def _hide(db_engine, place_id, label):
|
|
from common.enums import MediaStatus
|
|
|
|
async with db_engine.begin() as c:
|
|
await c.execute(
|
|
text("UPDATE place_photos SET status=:s WHERE place_id=:p AND label=:l"),
|
|
{"s": MediaStatus.REJECTED.value, "p": uuid.UUID(place_id), "l": label},
|
|
)
|
|
|
|
|
|
async def test_내린_사진은_대표가_될_수_없다(client, auth_headers, choose, db_engine):
|
|
"""★ 내린 사진의 순서만 당기면 "바꿨습니다" 라고 말하는데 발행본의 대표는 그대로다 —
|
|
화면만 거짓말한다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
await _seed_photos(db_engine, pid, [("외관", None, 0), ("수영장", None, 1)])
|
|
await _hide(db_engine, pid, "수영장")
|
|
|
|
choose({"actions": [{"tool": "set_primary_photo", "args": {"name": "수영장"}}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "대표 사진 수영장으로"})).json()
|
|
assert body["rejected"] is True
|
|
assert "안 나가는 사진" in body["reply"]
|
|
|
|
rows = await _photo_rows(db_engine, pid)
|
|
assert rows[0].label == "외관"
|
|
|
|
|
|
async def test_이미_내린_사진을_또_내리라면_그렇다고_말한다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
await _seed_photos(db_engine, pid, [("외관", None, 0), ("객실", None, 1)])
|
|
await _hide(db_engine, pid, "객실")
|
|
|
|
choose({"actions": [{"tool": "hide_photo", "args": {"name": "객실"}}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "객실 사진 내려줘"})).json()
|
|
assert "이미" in body["reply"]
|
|
|
|
|
|
async def test_사진_목록의_대표는_실제로_나가는_사진이다(client, auth_headers, choose, db_engine):
|
|
"""★ 맨 앞 사진이 내려가 있으면 그건 대표가 아니다 — primary_media 는 나가는 사진에서 고른다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
await _seed_photos(db_engine, pid, [("외관", None, 0), ("수영장", None, 1)])
|
|
await _hide(db_engine, pid, "외관")
|
|
|
|
choose({"actions": [{"tool": "list_photos", "args": {}}], "message": ""})
|
|
reply = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "사진 보여줘"})).json()["reply"]
|
|
assert "대표 수영장" in reply
|
|
assert "대표 외관" not in reply
|
|
|
|
|
|
async def test_프롬프트에는_나가는_사진만_대표를_맨_앞에_싣는다(client, auth_headers, db_engine):
|
|
"""모델은 '맨 앞이 대표' 로 읽는다. 객실 전용 사진이 앞에 있거나 내린 사진이 섞이면 틀리게 읽는다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
await _seed_photos(db_engine, pid, [("침대", uuid.uuid4(), 0), ("외관", None, 1), ("주차장", None, 2)])
|
|
await _hide(db_engine, pid, "주차장")
|
|
|
|
from router.v1.validator.dependencies import DecodeAccessToken
|
|
|
|
user = DecodeAccessToken(h["Authorization"].split(" ", 1)[1])
|
|
place = await runtime._load_place(user, pid)
|
|
names = await tools.photo_names(tools.ToolContext(user=user, place_id=pid, place=place))
|
|
assert names[0] == "외관"
|
|
assert "주차장" not in names
|
|
|
|
|
|
async def test_이름이_정확히_맞으면_그_사진을_고른다(client, auth_headers, choose, db_engine):
|
|
"""'객실' 과 '객실 욕실' 이 있을 때 '객실' 은 모호하지 않다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
await _seed_photos(db_engine, pid, [("객실", None, 0), ("객실 욕실", None, 1)])
|
|
|
|
choose({"actions": [{"tool": "hide_photo", "args": {"name": "객실"}}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "객실 사진 내려줘"})).json()
|
|
assert body.get("rejected") is not True, body["reply"]
|
|
from common.enums import MediaStatus
|
|
|
|
rows = {r.label: r.status for r in await _photo_rows(db_engine, pid)}
|
|
assert rows["객실"] == MediaStatus.REJECTED.value
|
|
assert rows["객실 욕실"] != MediaStatus.REJECTED.value
|
|
|
|
|
|
# ── 한 발화에 여러 가지 — 못 한 것·남은 것·겹친 것 ─────────────────────
|
|
|
|
async def test_할_수_없는_요청이_섞이면_그것도_말한다(client, auth_headers, choose, db_engine):
|
|
"""★ 되는 것만 하고 입을 다물면 사장님은 전부 된 줄 안다. 문장은 코드가 만든다 —
|
|
모델은 '무엇을 못 했는지' 이름만 준다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "set_fact", "args": {"key": "check_in_time", "value": "15:00"}}],
|
|
"skipped": ["전화번호 변경"], "message": ""})
|
|
reply = (await client.post(f"/v1/agent/chat/{pid}", headers=h,
|
|
json={"message": "체크인 3시로 바꾸고 전화번호도 바꿔줘"})).json()["reply"]
|
|
assert "체크인 시간" in reply
|
|
assert "'전화번호 변경' 은(는) 대화로는 아직 할 수 없어요." in reply
|
|
|
|
|
|
async def test_할_수_없는_요청뿐이면_그렇다고_말한다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [], "skipped": ["주소 변경"], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "주소 바꿔줘"})).json()
|
|
assert "'주소 변경' 은(는) 대화로는 아직 할 수 없어요." in body["reply"]
|
|
assert body["tool"] is None
|
|
|
|
|
|
async def test_지어낸_도구는_건너뛰었다고_말한다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [
|
|
{"tool": "delete_everything", "args": {}},
|
|
{"tool": "set_fact", "args": {"key": "check_in_time", "value": "15:00"}},
|
|
], "message": ""})
|
|
reply = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "다 지우고 체크인 3시로"})).json()["reply"]
|
|
assert "알아듣지 못한 요청 1가지" in reply
|
|
|
|
|
|
async def test_중간에_멈추면_남은_요청을_알려준다(client, auth_headers, choose, db_engine):
|
|
"""★ 멈춘 뒤의 요청을 말없이 버리면 사장님은 그것도 된 줄 안다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [
|
|
{"tool": "set_fact", "args": {"key": "check_in_time", "value": "15:00"}},
|
|
{"tool": "toggle_section", "args": {"name": "없는섹션", "enabled": "false"}},
|
|
{"tool": "move_section", "args": {"name": "소개", "to": "맨 위"}},
|
|
{"tool": "set_fact", "args": {"key": "check_out_time", "value": "11:00"}},
|
|
], "message": ""})
|
|
reply = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "넷 다 해줘"})).json()["reply"]
|
|
assert "소개 옮기기, 체크아웃 시간 변경 은(는) 아직 하지 않았어요." in reply
|
|
|
|
|
|
async def test_발행에서_멈추면_뒤의_요청도_알려준다(client, auth_headers, choose, db_engine):
|
|
"""★ 발행 뒤의 요청은 확인을 눌러도 실행되지 않는다 — 확인 전에 알려야 한다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [
|
|
{"tool": "toggle_section", "args": {"name": "사진 갤러리", "enabled": "false"}},
|
|
{"tool": "publish", "args": {}},
|
|
{"tool": "set_fact", "args": {"key": "check_in_time", "value": "15:00"}},
|
|
], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h,
|
|
json={"message": "갤러리 빼고 발행하고 체크인 3시로"})).json()
|
|
assert body["needs_confirm"] is True
|
|
assert "체크인 시간 변경 은(는) 아직 하지 않았어요." in body["reply"]
|
|
assert body["reply"].endswith(tools.REGISTRY["publish"].confirm) # 묻는 말은 맨 끝에 선다
|
|
assert await _fact_value(db_engine, pid, "check_in_time") == []
|
|
|
|
|
|
async def test_같은_항목을_고쳐_말하면_마지막_값만_남긴다(client, auth_headers, choose, db_engine):
|
|
""""체크인 3시… 아니 4시로" — 둘 다 하면 문구에 두 값이 함께 서서 어느 쪽이 남았는지 모른다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [
|
|
{"tool": "set_fact", "args": {"key": "check_in_time", "value": "15:00"}},
|
|
{"tool": "toggle_section", "args": {"name": "사진 갤러리", "enabled": "false"}},
|
|
{"tool": "set_fact", "args": {"key": "check_in_time", "value": "16:00"}},
|
|
], "message": ""})
|
|
reply = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "3시 아니 4시"})).json()["reply"]
|
|
assert reply.count("체크인 시간") == 1
|
|
assert "16:00" in reply and "15:00" not in reply
|
|
assert await _fact_value(db_engine, pid, "check_in_time") == ["16:00"]
|
|
|
|
|
|
async def test_겹친_요청은_상한을_세기_전에_합친다(client, auth_headers, choose, db_engine):
|
|
h, pid = await seed(client, auth_headers)
|
|
names = ["사진 갤러리", "소개", "예약 안내", "자주 묻는 질문", "날씨"]
|
|
choose({"actions": [
|
|
{"tool": "toggle_section", "args": {"name": n, "enabled": "false"}} for n in names
|
|
] + [{"tool": "toggle_section", "args": {"name": "날씨", "enabled": "false"}}], "message": ""})
|
|
reply = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "다 빼줘"})).json()["reply"]
|
|
assert f"{runtime.MAX_ACTIONS}가지까지" not in reply
|
|
|
|
|
|
async def test_바뀐_것이_없으면_재발행을_권하지_않는다(client, auth_headers, choose, db_engine):
|
|
"""★ "이미 켜져 있어요" 에 "다시 발행해야 해요" 가 붙으면, 사장님은 무언가 바뀐 줄 안다."""
|
|
h, pid = await seed(client, auth_headers)
|
|
choose({"actions": [{"tool": "toggle_section", "args": {"name": "사진 갤러리", "enabled": "true"}}], "message": ""})
|
|
body = (await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "사진 갤러리 켜줘"})).json()
|
|
assert "이미" in body["reply"]
|
|
assert runtime.REPUBLISH_NOTICE not in body["reply"]
|
|
assert body["done"] is False
|
|
|
|
|
|
def test_응답_스키마에_못_한_요청_칸이_있다():
|
|
"""OpenAI strict 모드는 properties 를 전부 required 로 요구한다."""
|
|
from services.prompts import agent as prompt
|
|
|
|
assert "skipped" in prompt.RESPONSE_SCHEMA["properties"]
|
|
assert "skipped" in prompt.RESPONSE_SCHEMA["required"]
|