hbyang
1300652b09
실제로 붙여 보니 빠진 것이 드러났다 — 연결은 됐는데 **어느 홈페이지를 다루는 대화인지** 말해 주지 않았다. 가게가 하나면 말없이 자동 선택돼 더 모호했다. - 연결 직후 목록을 보여준다. 하나면 이름+발행 여부를, 여럿이면 바로가기 버튼으로 - 목록 줄에 발행 여부를 적는다 — 안 그러면 고친 것이 손님에게 보이는 줄 안다 - "목록"·"가게 바꿔줘" 로 언제든 돌아와 바꾼다. ★ 이 경로는 LLM 을 부르지 않는다: 대화가 막혔을 때 처음 찾는 길이라 늘 통해야 하고, 목록에 돈을 쓸 이유가 없다 - 사업장 목록이 아니라 list_my_sites 를 쓴다 — 사장님이 알아야 하는 건 "가게가 있다" 가 아니라 "발행돼 있나" 다(/sites 화면이 같은 이유로 그걸 쓴다) test_kakao_webhook.py 21 passed(목록·전환 4건 추가). 전체 845 passed / 53 failed — 53 은 이번 변경 전과 동일 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
321 lines
15 KiB
Python
321 lines
15 KiB
Python
"""카카오톡 채널 웹훅.
|
|
|
|
여기서 지키는 것 셋:
|
|
1. 시크릿 없는 요청은 아무것도 하지 못한다 — 오픈빌더가 서명을 주지 않으므로 이게 유일한 문이다
|
|
2. 연결되지 않은 발화자는 어떤 사장님도 되지 못한다
|
|
3. 확인(SEMI)은 **만료되면 안 먹는다** — 한참 뒤의 "네" 한 마디에 묵은 발행이 돌면 안 된다
|
|
"""
|
|
|
|
import uuid
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
|
|
from services import kakao_link_service as link_service
|
|
from services.agent import channel, runtime
|
|
|
|
SECRET = "test-webhook-secret-0123456789"
|
|
PATH = "/v1/agent/kakao/webhook"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def secret(monkeypatch):
|
|
monkeypatch.setenv("KAKAO_WEBHOOK_SECRET", SECRET)
|
|
monkeypatch.setenv("KAKAO_CHANNEL_PUBLIC_ID", "_testCh")
|
|
monkeypatch.setenv("AGENT_CHAT_ENABLED", "1")
|
|
monkeypatch.delenv("KAKAO_BOT_ID", raising=False)
|
|
|
|
|
|
def body(utterance, speaker="kakao-speaker-1", bot_id="bot-1"):
|
|
return {
|
|
"userRequest": {"utterance": utterance, "user": {"id": speaker, "type": "botUserKey"}},
|
|
"bot": {"id": bot_id, "name": "ADO2"},
|
|
}
|
|
|
|
|
|
def said(res) -> str:
|
|
return res.json()["template"]["outputs"][0]["simpleText"]["text"]
|
|
|
|
|
|
def quick(res) -> list:
|
|
return [q["label"] for q in res.json()["template"].get("quickReplies", [])]
|
|
|
|
|
|
async def owner_with_place(client, auth_headers, name="대화숙소"):
|
|
h = await auth_headers(f"kakao-{uuid.uuid4().hex[:8]}")
|
|
res = await client.post("/v1/place", headers=h, json={"name": name, "category": 1})
|
|
return h, res.json()["place"]["place_id"]
|
|
|
|
|
|
async def user_id_of(db_engine, place_id):
|
|
async with db_engine.begin() as c:
|
|
return (
|
|
await c.execute(text("SELECT owner_user_id FROM places WHERE place_id=:p"), {"p": uuid.UUID(place_id)})
|
|
).scalar_one()
|
|
|
|
|
|
async def link(db_engine, client, auth_headers, speaker, name="대화숙소"):
|
|
"""사장님 하나 + 가게 하나 + 그 사장님에 묶인 카톡 발화자."""
|
|
h, pid = await owner_with_place(client, auth_headers, name)
|
|
uid = await user_id_of(db_engine, pid)
|
|
await link_service.redeem((await link_service.issue_code(uid))["code"], speaker)
|
|
return h, pid, uid
|
|
|
|
|
|
# ── 1. 시크릿이 유일한 문이다 ────────────────────────────────────────────
|
|
|
|
async def test_시크릿이_없으면_존재를_알리지_않는다(client, monkeypatch):
|
|
"""★ 401 이 아니라 404 다. 401 은 '여기 뭔가 있다' 를 알려 준다."""
|
|
monkeypatch.setenv("KAKAO_WEBHOOK_SECRET", "")
|
|
assert (await client.post(PATH, json=body("안녕"))).status_code == 404
|
|
|
|
|
|
async def test_틀린_시크릿도_404(client):
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": "wrong-secret"}, json=body("안녕"))
|
|
assert res.status_code == 404
|
|
res = await client.post(PATH, json=body("안녕")) # 헤더 없음
|
|
assert res.status_code == 404
|
|
|
|
|
|
async def test_경로_시크릿도_받는다(client, db_engine):
|
|
res = await client.post(f"{PATH}/{SECRET}", json=body("안녕"))
|
|
assert res.status_code == 200
|
|
res = await client.post(f"{PATH}/wrong-secret", json=body("안녕"))
|
|
assert res.status_code == 404
|
|
|
|
|
|
async def test_다른_봇의_요청은_거절한다(client, monkeypatch, db_engine):
|
|
monkeypatch.setenv("KAKAO_BOT_ID", "bot-1")
|
|
ok = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("안녕", bot_id="bot-1"))
|
|
assert ok.status_code == 200
|
|
bad = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("안녕", bot_id="남의봇"))
|
|
assert bad.status_code == 404
|
|
|
|
|
|
# ── 2. 연결되지 않은 발화자 ──────────────────────────────────────────────
|
|
|
|
async def test_연결_전에는_어떤_도구도_돌지_않는다(client, monkeypatch, db_engine):
|
|
called = AsyncMock()
|
|
monkeypatch.setattr(runtime, "chat", called)
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("체크인 3시로 바꿔줘", "모르는-키"))
|
|
assert res.status_code == 200
|
|
assert "연결" in said(res)
|
|
called.assert_not_awaited()
|
|
|
|
|
|
async def test_발화자_id_를_위조해도_남의_가게에_닿지_않는다(client, auth_headers, db_engine, monkeypatch):
|
|
"""★ 신원 연결이 없으면 카톡 진입점만 소유자 범위 밖에 놓인다 — 그걸 막는 자리다."""
|
|
await link(db_engine, client, auth_headers, "진짜-사장님-키")
|
|
called = AsyncMock()
|
|
monkeypatch.setattr(runtime, "chat", called)
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("정보 보여줘", "위조한-키"))
|
|
assert "연결" in said(res)
|
|
called.assert_not_awaited()
|
|
|
|
|
|
async def test_코드를_보내면_연결된다(client, auth_headers, db_engine):
|
|
h, pid = await owner_with_place(client, auth_headers)
|
|
uid = await user_id_of(db_engine, pid)
|
|
code = (await link_service.issue_code(uid))["code"]
|
|
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body(code, "새-발화자"))
|
|
assert "연결됐습니다" in said(res)
|
|
assert await link_service.resolve("새-발화자") == uid
|
|
|
|
|
|
async def test_틀린_코드는_이유를_구분해_말하지_않는다(client, db_engine):
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("ZZZZZZ", "새-발화자2"))
|
|
assert "코드가 맞지 않거나" in said(res)
|
|
|
|
|
|
# ── 3. 확인은 만료되면 안 먹는다 ─────────────────────────────────────────
|
|
|
|
async def test_발행은_묻고_바로가기를_준다(client, auth_headers, db_engine, monkeypatch):
|
|
speaker = "확인-테스트-키"
|
|
await link(db_engine, client, auth_headers, speaker)
|
|
# 테스트는 실제 모델을 부르지 않는다 — 런타임만 열고 선택 결과를 대신 준다.
|
|
monkeypatch.setattr(runtime, "is_configured", lambda: True)
|
|
monkeypatch.setattr(runtime, "_choose", AsyncMock(return_value={"tool": "publish", "args": {}, "message": ""}))
|
|
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("발행해줘", speaker))
|
|
assert channel.CONFIRM_LABEL in quick(res)
|
|
async with db_engine.begin() as c:
|
|
pending = (
|
|
await c.execute(text("SELECT pending_tool FROM owner_kakao_links WHERE channel_user_key=:k"), {"k": speaker})
|
|
).scalar_one()
|
|
assert pending == "publish"
|
|
|
|
|
|
async def test_만료된_확인에_네_라고_해도_실행되지_않는다(client, auth_headers, db_engine, monkeypatch):
|
|
"""★ 이게 없으면 한참 뒤의 '네' 한 마디에 **묵은 발행**이 돈다."""
|
|
speaker = "만료-테스트-키"
|
|
await link(db_engine, client, auth_headers, speaker)
|
|
async with db_engine.begin() as c:
|
|
await c.execute(
|
|
text("""UPDATE owner_kakao_links
|
|
SET pending_tool='publish', pending_args='{}'::jsonb,
|
|
pending_expires_at = now() - interval '1 minute'
|
|
WHERE channel_user_key=:k"""),
|
|
{"k": speaker},
|
|
)
|
|
ran = AsyncMock()
|
|
monkeypatch.setattr(runtime, "chat", ran)
|
|
await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("네", speaker))
|
|
# 확인으로 실행된 적이 없다(다른 말로 취급돼 일반 경로로 갔을 수는 있다).
|
|
for call in ran.await_args_list:
|
|
assert call.kwargs.get("confirm") is None
|
|
|
|
|
|
async def test_바로가기_라벨은_예로_읽히는_말에_들어_있다():
|
|
"""★ 라벨과 _YES 가 어긋나면 **눌러도 안 먹는다** — 사장님은 버튼이 고장난 줄 안다."""
|
|
assert channel.CONFIRM_LABEL in channel._YES
|
|
assert channel.PUBLISH_LABEL in channel._YES
|
|
assert channel.DECLINE_LABEL in channel._NO
|
|
|
|
|
|
# ── 가게 고르기 ──────────────────────────────────────────────────────────
|
|
|
|
async def test_가게가_여럿이면_추측하지_않고_되묻는다(client, auth_headers, db_engine, monkeypatch):
|
|
"""★ 임의로 첫 가게를 고르면 사장님은 엉뚱한 가게를 고쳐 놓고도 모른다."""
|
|
speaker = "다가게-키"
|
|
h, _pid, _uid = await link(db_engine, client, auth_headers, speaker, "첫째가게")
|
|
await client.post("/v1/place", headers=h, json={"name": "둘째가게", "category": 1})
|
|
|
|
called = AsyncMock()
|
|
monkeypatch.setattr(runtime, "chat", called)
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("정보 보여줘", speaker))
|
|
assert "어느 가게" in said(res)
|
|
assert set(quick(res)) == {"첫째가게", "둘째가게"}
|
|
called.assert_not_awaited()
|
|
|
|
# 고르면 기억한다.
|
|
picked = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("둘째가게", speaker))
|
|
assert "둘째가게" in said(picked)
|
|
async with db_engine.begin() as c:
|
|
current = (
|
|
await c.execute(
|
|
text("SELECT current_place_id FROM owner_kakao_links WHERE channel_user_key=:k"), {"k": speaker}
|
|
)
|
|
).scalar_one()
|
|
assert current is not None
|
|
|
|
|
|
# ── 5초 벽 · 실패 ────────────────────────────────────────────────────────
|
|
|
|
async def test_느리면_침묵_대신_안내로_끊는다(client, monkeypatch, db_engine):
|
|
"""넘기면 카카오가 연결을 끊는다 — 사장님에게는 말없이 실패하는 봇이 된다."""
|
|
import router.v1.agent.kakao_bot as bot
|
|
|
|
async def slow(*_a, **_kw):
|
|
import asyncio
|
|
|
|
await asyncio.sleep(1)
|
|
|
|
monkeypatch.setattr(bot, "DEADLINE_SEC", 0.01)
|
|
monkeypatch.setattr(bot.channel, "handle", slow)
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("안녕"))
|
|
assert res.status_code == 200
|
|
assert "잠시 뒤" in said(res)
|
|
|
|
|
|
async def test_내부_오류도_200_으로_답한다(client, monkeypatch, db_engine):
|
|
"""메신저에서는 500 도 침묵으로 보인다 — 무슨 일이 있었는지 한 줄은 말해야 한다."""
|
|
import router.v1.agent.kakao_bot as bot
|
|
|
|
monkeypatch.setattr(bot.channel, "handle", AsyncMock(side_effect=RuntimeError("어딘가 터짐")))
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("안녕"))
|
|
assert res.status_code == 200
|
|
assert "어딘가 터짐" not in res.text
|
|
|
|
|
|
async def test_발화자가_없으면_거기서_끝낸다(client, db_engine):
|
|
res = await client.post(
|
|
PATH, headers={"X-Agent-Secret": SECRET}, json={"userRequest": {"utterance": "안녕", "user": {}}}
|
|
)
|
|
assert res.status_code == 200
|
|
assert "확인하지 못했" in said(res)
|
|
|
|
|
|
# ── 응답 형식 ────────────────────────────────────────────────────────────
|
|
|
|
def test_카카오_형식은_이_파일_밖으로_나가지_않는다():
|
|
"""★ 서비스 계층에 새면 다른 채널을 붙일 때 전부 걷어내야 한다."""
|
|
import ast
|
|
import inspect
|
|
|
|
# ★ 주석·docstring 에 이름이 나오는 것은 '샌' 것이 아니다 — 실제 코드만 본다.
|
|
tree = ast.parse(inspect.getsource(channel))
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str):
|
|
node.value.value = "" # docstring 비우기
|
|
dumped = ast.dump(tree)
|
|
for token in ("simpleText", "quickReplies", "userRequest", "2.0"):
|
|
assert token not in dumped, token
|
|
|
|
|
|
def test_바로가기는_열_개를_넘기지_않는다():
|
|
import router.v1.agent.kakao_bot as bot
|
|
|
|
out = bot._reply("안녕", [f"라벨{i}" for i in range(20)])
|
|
assert len(out["template"]["quickReplies"]) == 10
|
|
assert out["version"] == "2.0"
|
|
assert out["template"]["outputs"][0]["simpleText"]["text"] == "안녕"
|
|
|
|
|
|
# ── 사이트 목록 · 고르기 ─────────────────────────────────────────────────
|
|
|
|
async def test_연결되자마자_홈페이지_목록을_알려준다(client, auth_headers, db_engine):
|
|
"""★ 연결만 알리고 끝내면 사장님은 **어느 홈페이지를 다루는 대화인지** 모른 채 말을 건다."""
|
|
h, pid = await owner_with_place(client, auth_headers, "첫째가게")
|
|
await client.post("/v1/place", headers=h, json={"name": "둘째가게", "category": 1})
|
|
uid = await user_id_of(db_engine, pid)
|
|
code = (await link_service.issue_code(uid))["code"]
|
|
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body(code, "목록-발화자"))
|
|
text_out = said(res)
|
|
assert "연결됐습니다" in text_out
|
|
assert "첫째가게" in text_out and "둘째가게" in text_out
|
|
# 바로 고를 수 있어야 한다 — 이름을 외워 치게 하지 않는다.
|
|
assert set(quick(res)) == {"첫째가게", "둘째가게"}
|
|
|
|
|
|
async def test_가게가_하나면_그_이름을_말해_준다(client, auth_headers, db_engine):
|
|
h, pid = await owner_with_place(client, auth_headers, "혼자가게")
|
|
uid = await user_id_of(db_engine, pid)
|
|
code = (await link_service.issue_code(uid))["code"]
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body(code, "하나-발화자"))
|
|
assert "혼자가게" in said(res)
|
|
assert "발행" in said(res) # 발행 여부를 같이 말한다
|
|
|
|
|
|
async def test_목록이라고_하면_언제든_다시_보여주고_가게를_바꿀_수_있다(client, auth_headers, db_engine, monkeypatch):
|
|
"""★ 대화가 막혔을 때 사장님이 처음 찾는 길이다. LLM 을 부르지 않는다."""
|
|
speaker = "전환-발화자"
|
|
h, _pid, _uid = await link(db_engine, client, auth_headers, speaker, "첫째가게")
|
|
await client.post("/v1/place", headers=h, json={"name": "둘째가게", "category": 1})
|
|
|
|
called = AsyncMock()
|
|
monkeypatch.setattr(runtime, "chat", called)
|
|
|
|
# 먼저 한 곳을 고른다.
|
|
await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("첫째가게", speaker))
|
|
# 목록으로 돌아온다.
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("가게 바꿔줘", speaker))
|
|
assert "첫째가게" in said(res) and "둘째가게" in said(res)
|
|
assert set(quick(res)) == {"첫째가게", "둘째가게"}
|
|
called.assert_not_awaited() # 목록 보기에 모델을 부르지 않는다
|
|
|
|
# 다른 곳으로 바꾼다.
|
|
picked = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("둘째가게", speaker))
|
|
assert "둘째가게" in said(picked)
|
|
|
|
|
|
async def test_목록은_발행_여부를_같이_말한다(client, auth_headers, db_engine):
|
|
"""★ 안 그러면 사장님은 고친 것이 손님에게 보이는 줄 안다."""
|
|
speaker = "발행표시-발화자"
|
|
await link(db_engine, client, auth_headers, speaker, "미발행가게")
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("목록", speaker))
|
|
assert "아직 발행 전" in said(res)
|