김성경
f1b473968d
결정(2026-09-29): 카톡과 메일 둘 다, 승인은 링크가 아니라 연결된 계정 신원으로. 이 커밋은 발송과 메시지 그리기까지고 [승인] 버튼은 다음 단계(2-2)다. - blog_jobs._send_one: 메일에 더해 Event API 로 보낸다. 하나라도 나가면 SENT, 아무 데도 안 나가면 SENT 로 표시하지 않아 다음 스윕이 다시 시도한다. 카톡은 채널 친구가 아니거나 차단했으면 실패하므로 메일을 빼지 않는다. send_now 도 같은 경로 - 카톡 params 로 post_id 와 수정용 일회용 코드(edit_token)를 넘긴다. 코드 평문은 발송 시점에만 알아서다. 로그에는 params 의 키만 남기고 값은 남기지 않는다 - channel.approval_notice: 발화자 키 → 사장님 → 그 글이 그 사장님 가게 것·미처리· 기한 전일 때만 본문과 [수정하기] 를 준다. 연결 안 됨·남의 글·처리됨·만료·이상한 ID 는 구분 없이 같은 안내(구분해 주면 글 ID 를 탐색할 수 있다) - kakao_bot: userRequest.params.post_id 가 있으면 승인 알림 요청으로 처리하고 링크 버튼은 본문과 따로 textCard 로 그린다(카드 설명 길이 제한을 피한다) - KAKAO_APPROVAL_PUSH_ENABLED(기본 0), KAKAO_APPROVAL_EVENT_NAME 추가 — 오픈빌더 이벤트 블록(스킬 연결)과 배포가 끝나기 전에는 켜지 않는다 test_blog_owner 6건·test_kakao_webhook 6건 추가, 카카오·미니블로그 스위트 119 passed, 인접 스위트 110 passed. 실제 카톡 수신은 콘솔 설정·운영 배포 뒤에 확인한다.
479 lines
23 KiB
Python
479 lines
23 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={"actions": [{"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)
|
|
|
|
|
|
# ── 콜백 (5초 벽 넘기) ───────────────────────────────────────────────────
|
|
|
|
async def test_콜백이_켜져_있으면_즉답하고_뒤에서_마저_만든다(client, auth_headers, db_engine, monkeypatch):
|
|
"""★ 오픈빌더 스킬 타임아웃은 5초다. 콜백을 쓰면 '잠시만 기다려 주세요' 를 먼저 주고
|
|
답이 완성되면 그 주소로 따로 보낸다 — 느린 답이 침묵이 되지 않는다."""
|
|
import router.v1.agent.kakao_bot as bot
|
|
|
|
sent = {}
|
|
|
|
async def fake_push(callback_url, utterance, speaker):
|
|
sent["url"] = callback_url
|
|
sent["payload"] = await bot._answer(utterance, speaker, 5.0)
|
|
|
|
monkeypatch.setattr(bot, "_push", fake_push)
|
|
|
|
payload = body("안녕", "콜백-발화자")
|
|
payload["userRequest"]["callbackUrl"] = "https://callback.example/one-shot"
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=payload)
|
|
|
|
# 즉답은 useCallback 이다 — template 을 싣지 않는다.
|
|
assert res.json()["useCallback"] is True
|
|
assert res.json()["data"]["text"]
|
|
assert "template" not in res.json()
|
|
|
|
# 실제 답은 콜백으로 간다.
|
|
assert sent["url"] == "https://callback.example/one-shot"
|
|
assert "연결" in sent["payload"]["template"]["outputs"][0]["simpleText"]["text"]
|
|
|
|
|
|
async def test_콜백이_없으면_예전처럼_동기로_답한다(client, db_engine):
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("안녕", "동기-발화자"))
|
|
assert "useCallback" not in res.json()
|
|
assert res.json()["template"]["outputs"]
|
|
|
|
|
|
async def test_콜백_전송이_실패해도_터지지_않는다(monkeypatch):
|
|
"""★ 주소는 1분 · 1회다. 재시도하지 않는다 — 두 번째 POST 는 어차피 거절된다."""
|
|
import router.v1.agent.kakao_bot as bot
|
|
|
|
monkeypatch.setattr(bot, "_answer", AsyncMock(return_value=bot._reply("답")))
|
|
|
|
class Boom:
|
|
async def __aenter__(self):
|
|
raise RuntimeError("네트워크 끊김")
|
|
|
|
async def __aexit__(self, *_):
|
|
return False
|
|
|
|
monkeypatch.setattr(bot.httpx, "AsyncClient", lambda **_kw: Boom())
|
|
await bot._push("https://callback.example/x", "안녕", "누구") # 예외가 새 나오지 않는다
|
|
|
|
|
|
# ── 승인 알림(Event API 로 시작된 스킬 요청) ─────────────────────────────
|
|
# ★ 이 자리가 지키는 것: 글 본문은 **연결된 본인의 글일 때만** 나간다. 이벤트 요청의
|
|
# params 는 카카오가 채우지만, 발화자 키로 사장님을 찾고 그 글이 그 사장님 가게 것인지
|
|
# 서버가 다시 본다 — 글 ID 만 알면 남의 글이 보이는 구멍이 되면 안 된다.
|
|
|
|
POST_BODY = "비가 그친 뒤 마당 돌이 촉촉합니다. 대청마루에 앉아 빗소리를 들어 보세요."
|
|
|
|
|
|
def event_body(post_id, edit_token="EDITCODE1234", speaker="kakao-speaker-1"):
|
|
payload = body("", speaker)
|
|
payload["userRequest"]["params"] = {"post_id": str(post_id)}
|
|
if edit_token:
|
|
payload["userRequest"]["params"]["edit_token"] = edit_token
|
|
return payload
|
|
|
|
|
|
async def seed_post(db_engine, place_id, status=3, expires=None):
|
|
"""status 3 = SENT. expires 는 tz 를 단 UTC 로 넣는다(raw text() 로 timestamptz 를 바인딩할 때의 함정)."""
|
|
post_id = uuid.uuid4()
|
|
async with db_engine.begin() as c:
|
|
await c.execute(
|
|
text("INSERT INTO place_posts (post_id, place_id, body, topic_kind, topic_key, status, token_expires_at) "
|
|
"VALUES (:id, :pid, :body, 1, :key, :st, :exp)"),
|
|
{"id": post_id, "pid": uuid.UUID(str(place_id)), "body": POST_BODY,
|
|
"key": f"weather:{post_id.hex[:6]}", "st": status, "exp": expires},
|
|
)
|
|
return post_id
|
|
|
|
|
|
def outputs(res):
|
|
return res.json()["template"]["outputs"]
|
|
|
|
|
|
async def test_승인_알림은_본인_글이면_본문과_수정_링크를_그린다(client, auth_headers, db_engine):
|
|
h, pid, uid = await link(db_engine, client, auth_headers, "승인-발화자", "알림숙소")
|
|
post_id = await seed_post(db_engine, pid)
|
|
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=event_body(post_id, speaker="승인-발화자"))
|
|
|
|
assert res.status_code == 200
|
|
first = outputs(res)[0]["simpleText"]["text"]
|
|
assert "알림숙소" in first and POST_BODY in first
|
|
button = outputs(res)[1]["textCard"]["buttons"][0]
|
|
assert button["label"] == "수정하기" and button["action"] == "webLink"
|
|
assert button["webLinkUrl"].endswith("/v1/site/post/edit?t=EDITCODE1234")
|
|
|
|
|
|
async def test_승인_알림은_수정_코드가_없으면_버튼_없이_본문만_준다(client, auth_headers, db_engine):
|
|
h, pid, uid = await link(db_engine, client, auth_headers, "코드없는-발화자")
|
|
post_id = await seed_post(db_engine, pid)
|
|
|
|
res = await client.post(
|
|
PATH, headers={"X-Agent-Secret": SECRET}, json=event_body(post_id, edit_token=None, speaker="코드없는-발화자")
|
|
)
|
|
|
|
assert POST_BODY in outputs(res)[0]["simpleText"]["text"]
|
|
assert len(outputs(res)) == 1
|
|
|
|
|
|
async def test_남의_글은_본문을_주지_않는다(client, auth_headers, db_engine):
|
|
"""★ 글 ID 를 알아도 그 글의 주인에게 연결된 발화자가 아니면 아무것도 안 나온다."""
|
|
_h1, pid1, _uid1 = await link(db_engine, client, auth_headers, "주인-발화자", "주인숙소")
|
|
await link(db_engine, client, auth_headers, "남-발화자", "남의숙소")
|
|
post_id = await seed_post(db_engine, pid1)
|
|
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=event_body(post_id, speaker="남-발화자"))
|
|
|
|
assert res.status_code == 200
|
|
assert POST_BODY not in res.text and "주인숙소" not in res.text
|
|
assert "EDITCODE1234" not in res.text
|
|
|
|
|
|
async def test_연결되지_않은_발화자에게는_본문을_주지_않는다(client, auth_headers, db_engine):
|
|
_h, pid, _uid = await link(db_engine, client, auth_headers, "연결된-발화자")
|
|
post_id = await seed_post(db_engine, pid)
|
|
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=event_body(post_id, speaker="모르는-발화자"))
|
|
|
|
assert res.status_code == 200
|
|
assert POST_BODY not in res.text and "EDITCODE1234" not in res.text
|
|
|
|
|
|
async def test_이미_처리했거나_기한이_지난_글은_안내로_끝난다(client, auth_headers, db_engine):
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
h, pid, uid = await link(db_engine, client, auth_headers, "지난-발화자")
|
|
approved = await seed_post(db_engine, pid, status=4)
|
|
expired = await seed_post(db_engine, pid, expires=datetime.now(timezone.utc) - timedelta(hours=1))
|
|
|
|
for post_id in (approved, expired):
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=event_body(post_id, speaker="지난-발화자"))
|
|
assert res.status_code == 200
|
|
assert POST_BODY not in res.text
|
|
assert "기한" in said(res) or "처리" in said(res)
|
|
|
|
|
|
async def test_글_id_가_이상해도_터지지_않는다(client, auth_headers, db_engine):
|
|
await link(db_engine, client, auth_headers, "이상한-발화자")
|
|
|
|
res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=event_body("not-a-uuid", speaker="이상한-발화자"))
|
|
|
|
assert res.status_code == 200
|
|
assert said(res)
|