hbyang
e35ce311d3
주소가 500자였던 것은 곁가지고, 진짜 문제는 그 auto= 가 빌더 액세스 토큰 통짜(sub 에 UserInfo 전체 — role 포함)였다는 것이다. 메일 전달 한 번이 그날 자정까지의 권한 양도였고, 브라우저 히스토리·프록시 로그·Referer 에도 남았다. SNS 승인 흐름에서는 같은 이유로 "기존 액세스 토큰을 승인 링크에 얹지 않는다" 를 원칙으로 박아 뒀는데 이 경로에만 남아 있었다. - 0023: place_posts.edit_token_hash. 승인 토큰과 같은 규약(평문은 메일에만) - GET /v1/site/post/edit?t=<코드> → 검증 후 day-pass 를 그 자리에서 만들어 /blog?...#auto=<JWT> 로 303. ★ 프래그먼트는 서버 로그·Referer 에 안 남는다 - 프론트는 hash 에서 읽고 **주소창에서 지운다**. 쿼리도 계속 받는다 — 이미 나간 메일이 자정까지 살아 있고 그걸 깨면 그 링크들이 통째로 죽는다 - 두 코드는 서로 다른 칸에 산다. 하나로 둘 다 되면 일회성이 무의미해진다 - blog_service.app_origin() 으로 오리진 계산을 옮겼다 — 라우터도 같은 값을 쓴다 링크 길이 약 500자 → 약 75자. test_blog_owner 45 passed (신규 6건: 길이·프래그먼트·소유자·만료·없는 코드· 승인 코드 교차 사용). test_upcoming_only_returns_next_week_in_date_order 1건은 기준(stash)에서도 동일하게 실패하는 기존 건. npm run lint 통과 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1042 lines
48 KiB
Python
1042 lines
48 KiB
Python
"""미니 블로그 — 빌더 앱 로그인 화면(이번 달 생성된 글). 기획: docs/MINI_BLOG.md
|
|
|
|
★ 이 파일이 지키는 것:
|
|
- 로그인한 사장님은 자기 사업장의 글만 본다(남의 가게 글이 섞이면 안 된다)
|
|
- 아직 메일이 안 나간 REVIEWED 글도 로그인 화면에서 바로 고칠 수 있다 — 단, 저장만
|
|
한다. 승인은 이메일 승인 링크 또는 로그인 "바로 발행" 버튼, 두 경로 중 하나를
|
|
명시적으로 눌러야 한다(2026-09-21, 사장님 지시: "이메일 승인으로도 발행 가능하고
|
|
바로발행버튼으로도 발행 가능하도록")
|
|
- 여기서도 금칙 게이트는 그대로 탄다 — 로그인했다고 우회되지 않는다
|
|
"""
|
|
import uuid
|
|
from datetime import date, datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import text
|
|
|
|
from common.enums import JobStatus, JobType, PostStatus, PostTopicKind, SiteStatus
|
|
|
|
BODY = (
|
|
"비가 한 차례 지나간 뒤 마당 돌이 검게 젖었습니다. 이런 날에는 대청마루에 앉아 빗소리만 들어도 "
|
|
"하루가 지나갑니다. 우산은 현관에 넉넉히 두었으니 편하게 다녀오세요. 젖은 길은 미끄러우니 "
|
|
"편한 신발을 권합니다. 마당 평상은 비가 그치면 금세 마릅니다."
|
|
)
|
|
|
|
|
|
async def _place(client, headers, name="블로그펜션") -> str:
|
|
r = await client.post("/v1/place", headers=headers, json={"name": name, "category": 1})
|
|
return r.json()["place"]["place_id"]
|
|
|
|
|
|
async def _seed_post(db_engine, place_id, *, status=PostStatus.REVIEWED, scheduled=None) -> str:
|
|
post_id = uuid.uuid4()
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO place_posts (post_id, place_id, body, topic_kind, topic_key, status, scheduled_date) "
|
|
"VALUES (:id, :pid, :body, :kind, :key, :st, :sched)"),
|
|
{"id": post_id, "pid": place_id, "body": BODY, "kind": PostTopicKind.WEATHER.value,
|
|
"key": f"weather:{post_id.hex[:6]}", "st": status.value, "sched": scheduled or date.today()},
|
|
)
|
|
return str(post_id)
|
|
|
|
|
|
async def _status(db_engine, post_id) -> int:
|
|
async with db_engine.begin() as conn:
|
|
row = await conn.execute(text("SELECT status FROM place_posts WHERE post_id = :id"), {"id": post_id})
|
|
return row.scalar()
|
|
|
|
|
|
async def test_owner_sees_this_months_posts(client, db_engine, auth_headers):
|
|
h = await auth_headers("blogowner1")
|
|
place_id = await _place(client, h)
|
|
await _seed_post(db_engine, place_id)
|
|
|
|
res = await client.get(f"/v1/place/{place_id}/post", headers=h)
|
|
|
|
body = res.json()
|
|
assert body["result"]["code"] == 0
|
|
assert len(body["posts"]) == 1
|
|
assert body["posts"][0]["body"] == BODY
|
|
|
|
|
|
async def test_owner_cannot_see_someone_elses_posts(client, db_engine, auth_headers):
|
|
owner = await auth_headers("blogowner2")
|
|
other = await auth_headers("blogowner3")
|
|
place_id = await _place(client, owner)
|
|
await _seed_post(db_engine, place_id)
|
|
|
|
res = await client.get(f"/v1/place/{place_id}/post", headers=other)
|
|
|
|
body = res.json()
|
|
assert body["result"]["success"] is False
|
|
assert body["posts"] == []
|
|
|
|
|
|
async def test_owner_edit_saves_body_without_approving(client, db_engine, auth_headers):
|
|
"""로그인 화면에서 바로 고칠 수는 있지만, 저장만 한다 — 승인은 이메일 링크로만 일어난다
|
|
(2026-09-21, 사장님 지시: "승인되야 올라가도록 해야 한다")."""
|
|
h = await auth_headers("blogowner4")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
|
|
new_body = BODY.replace("빗소리", "새소리")
|
|
|
|
res = await client.put(f"/v1/place/{place_id}/post/{post_id}", headers=h, json={"body": new_body})
|
|
|
|
assert res.json()["result"]["success"] is True
|
|
assert await _status(db_engine, post_id) == PostStatus.REVIEWED.value
|
|
async with db_engine.begin() as conn:
|
|
saved_body = (await conn.execute(
|
|
text("SELECT body FROM place_posts WHERE post_id = :id"), {"id": post_id},
|
|
)).scalar()
|
|
assert saved_body == new_body
|
|
|
|
|
|
async def test_owner_edit_rejects_unverifiable_claims(client, db_engine, auth_headers):
|
|
h = await auth_headers("blogowner5")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id)
|
|
bad_body = BODY[:120] + " 주중 198,000원입니다."
|
|
|
|
res = await client.put(f"/v1/place/{place_id}/post/{post_id}", headers=h, json={"body": bad_body})
|
|
|
|
body = res.json()
|
|
assert body["result"]["success"] is False
|
|
assert "198,000원" in body["msg"]
|
|
assert await _status(db_engine, post_id) == PostStatus.REVIEWED.value
|
|
|
|
|
|
async def test_owner_can_delete_a_post(client, db_engine, auth_headers):
|
|
h = await auth_headers("blogdel1")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
|
|
|
|
res = await client.delete(f"/v1/place/{place_id}/post/{post_id}", headers=h)
|
|
|
|
assert res.json()["result"]["success"] is True
|
|
async with db_engine.begin() as conn:
|
|
deleted = (await conn.execute(
|
|
text("SELECT deleted FROM place_posts WHERE post_id = :id"), {"id": post_id},
|
|
)).scalar()
|
|
assert deleted is True
|
|
|
|
|
|
async def test_owner_cannot_delete_someone_elses_post(client, db_engine, auth_headers):
|
|
owner = await auth_headers("blogdel2")
|
|
other = await auth_headers("blogdel3")
|
|
place_id = await _place(client, owner)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
|
|
|
|
res = await client.delete(f"/v1/place/{place_id}/post/{post_id}", headers=other)
|
|
|
|
assert res.json()["result"]["success"] is False
|
|
async with db_engine.begin() as conn:
|
|
deleted = (await conn.execute(
|
|
text("SELECT deleted FROM place_posts WHERE post_id = :id"), {"id": post_id},
|
|
)).scalar()
|
|
assert deleted is False
|
|
|
|
|
|
async def test_deleting_a_post_frees_its_date_for_regeneration(client, db_engine, auth_headers, monkeypatch):
|
|
"""삭제는 소프트 삭제라 (place_id, scheduled_date) 유니크가 풀린다 — 지운 날짜에
|
|
바로 다시 생성할 수 있어야 한다."""
|
|
from services import blog_service
|
|
|
|
async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
|
|
return ("새로 만든 문구입니다. " + BODY, "gemini-test-model")
|
|
|
|
monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
|
|
|
|
h = await auth_headers("blogdel4")
|
|
place_id = await _place(client, h, name="재생성펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
today = date.today()
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED, scheduled=today)
|
|
|
|
del_res = await client.delete(f"/v1/place/{place_id}/post/{post_id}", headers=h)
|
|
assert del_res.json()["result"]["success"] is True
|
|
|
|
gen_res = await client.post(
|
|
f"/v1/place/{place_id}/post/generate-one", headers=h, params={"date": today.isoformat()},
|
|
)
|
|
|
|
assert gen_res.json()["result"]["success"] is True
|
|
|
|
|
|
async def test_deleting_a_published_post_enqueues_rebuild(client, db_engine, auth_headers):
|
|
h = await auth_headers("blogdel5")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.PUBLISHED)
|
|
|
|
res = await client.delete(f"/v1/place/{place_id}/post/{post_id}", headers=h)
|
|
|
|
assert res.json()["result"]["success"] is True
|
|
async with db_engine.begin() as conn:
|
|
payload = (await conn.execute(
|
|
text("SELECT payload FROM jobs WHERE job_type = :jt ORDER BY created_at DESC LIMIT 1"),
|
|
{"jt": JobType.BUILD.value},
|
|
)).scalar()
|
|
assert payload["place_id"] == place_id
|
|
|
|
|
|
async def test_deleting_a_draft_post_does_not_enqueue_rebuild(client, db_engine, auth_headers):
|
|
h = await auth_headers("blogdel6")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
|
|
|
|
res = await client.delete(f"/v1/place/{place_id}/post/{post_id}", headers=h)
|
|
|
|
assert res.json()["result"]["success"] is True
|
|
async with db_engine.begin() as conn:
|
|
count = (await conn.execute(
|
|
text("SELECT count(*) FROM jobs WHERE job_type = :jt"), {"jt": JobType.BUILD.value},
|
|
)).scalar()
|
|
assert count == 0
|
|
|
|
|
|
async def test_generate_now_creates_posts_for_published_site(client, db_engine, auth_headers, monkeypatch):
|
|
"""새벽 크론(04:10)을 기다리지 않고, 사장님이 고른 구간을 그 자리에서 채운다(2026-09-17,
|
|
사장님 지시: "지금 생성하기에서 시작이랑 끝 날짜를 정해야하지 않을까") — 발행된 사이트일 때만."""
|
|
from services import blog_service
|
|
|
|
async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
|
|
return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
|
|
|
|
monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
|
|
|
|
h = await auth_headers("bloggen1")
|
|
place_id = await _place(client, h, name="즉시생성펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
start = date.today()
|
|
end = start + timedelta(days=29)
|
|
|
|
res = await client.post(
|
|
f"/v1/place/{place_id}/post/generate", headers=h,
|
|
params={"start": start.isoformat(), "end": end.isoformat()},
|
|
)
|
|
|
|
body = res.json()
|
|
assert body["result"]["success"] is True
|
|
assert body["requested"] == 30
|
|
assert body["created"] > 0
|
|
|
|
async with db_engine.begin() as conn:
|
|
count = (await conn.execute(
|
|
text("SELECT count(*) FROM place_posts WHERE place_id = :pid"), {"pid": place_id},
|
|
)).scalar()
|
|
dates = [row[0] for row in (await conn.execute(
|
|
text("SELECT scheduled_date FROM place_posts WHERE place_id = :pid ORDER BY scheduled_date"),
|
|
{"pid": place_id},
|
|
)).all()]
|
|
assert count == body["created"]
|
|
# 구간 안, 오늘부터 순서대로, 겹치는 날짜 없이.
|
|
assert dates[0] == start
|
|
assert dates == sorted(set(dates))
|
|
assert all(start <= d <= end for d in dates)
|
|
|
|
|
|
async def test_generate_now_writes_each_post_for_its_own_date(client, db_engine, auth_headers, monkeypatch):
|
|
"""글마다 배정된 날짜를 게시일로 받아 쓰고, 그 날짜의 절기 소재가 붙는다(2026-09-23,
|
|
사장님 지시: "날짜에 맞는 글이 생성 되도록")."""
|
|
from services import blog_service
|
|
|
|
calls = []
|
|
|
|
async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
|
|
calls.append((post_date, topic_kind, material))
|
|
return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
|
|
|
|
monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
|
|
|
|
h = await auth_headers("bloggen18")
|
|
place_id = await _place(client, h, name="날짜맞춤펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
winter = date(date.today().year + 1, 1, 10)
|
|
|
|
res = await client.post(
|
|
f"/v1/place/{place_id}/post/generate", headers=h,
|
|
params={"start": winter.isoformat(), "end": winter.isoformat()},
|
|
)
|
|
|
|
assert res.json()["created"] == 1
|
|
assert calls[0][0] == winter
|
|
# 지역 소재가 없는 업장이라 첫 후보는 그 날짜의 절기다.
|
|
assert calls[0][1] == PostTopicKind.SEASON.value
|
|
assert calls[0][2] == "한겨울"
|
|
async with db_engine.begin() as conn:
|
|
row = (await conn.execute(
|
|
text("SELECT scheduled_date, topic_key FROM place_posts WHERE place_id = :pid"), {"pid": place_id},
|
|
)).one()
|
|
assert row[0] == winter
|
|
assert row[1] == f"season:{winter.year}:한겨울"
|
|
|
|
|
|
async def test_generate_now_does_not_append_a_publish_link(client, db_engine, auth_headers, monkeypatch):
|
|
"""미니 블로그에 올라가는 글에는 링크를 붙이지 않는다(2026-09-21, 사장님 지시).
|
|
site_payload.publish_url() 자체는 남겨둔다 — 나중에 쓰레드 연동에서 따로 쓸 수 있게."""
|
|
from services import blog_service
|
|
|
|
async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
|
|
return (f"테스트로 만든 문구입니다. {BODY}", "gemini-test-model")
|
|
|
|
monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
|
|
|
|
h = await auth_headers("bloglink1")
|
|
place_id = await _place(client, h, name="링크확인펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
today = date.today()
|
|
|
|
res = await client.post(
|
|
f"/v1/place/{place_id}/post/generate", headers=h,
|
|
params={"start": today.isoformat(), "end": today.isoformat()},
|
|
)
|
|
assert res.json()["created"] == 1
|
|
|
|
async with db_engine.begin() as conn:
|
|
body = (await conn.execute(
|
|
text("SELECT body FROM place_posts WHERE place_id = :pid"), {"pid": place_id},
|
|
)).scalar()
|
|
assert body == f"테스트로 만든 문구입니다. {BODY}"
|
|
|
|
|
|
async def test_send_reviewed_only_mails_posts_due_today(client, db_engine, auth_headers, monkeypatch):
|
|
"""미래 날짜로 배정된 글은 그날이 오기 전엔 메일이 안 나간다."""
|
|
from services import blog_jobs, mail_service
|
|
|
|
monkeypatch.setattr(mail_service, "is_configured", lambda: True)
|
|
sent_calls = []
|
|
monkeypatch.setattr(mail_service, "send", lambda **kwargs: sent_calls.append(kwargs) or True)
|
|
|
|
h = await auth_headers("bloggen3")
|
|
place_id = await _place(client, h, name="예약펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
await conn.execute(text("UPDATE users SET email = :e WHERE id = :id"), {"e": "owner@example.com", "id": "bloggen3"})
|
|
|
|
today_post = await _seed_post(db_engine, place_id, scheduled=date.today())
|
|
future_post = await _seed_post(db_engine, place_id, scheduled=date.today() + timedelta(days=5))
|
|
|
|
sent = await blog_jobs.send_reviewed()
|
|
|
|
assert sent == 1
|
|
assert len(sent_calls) == 1
|
|
assert await _status(db_engine, today_post) == PostStatus.SENT.value
|
|
assert await _status(db_engine, future_post) == PostStatus.REVIEWED.value
|
|
|
|
|
|
async def test_send_reviewed_prefers_place_notify_email_over_account_email(client, db_engine, auth_headers, monkeypatch):
|
|
"""places.notify_email 이 있으면 계정 로그인 이메일(users.email) 대신 그 주소로 보낸다
|
|
(2026-09-21, 사장님 요청 — 사장님 한 명이 사이트를 여러 개 가질 수 있어 업장별로
|
|
다른 담당자에게 보낼 수 있어야 한다)."""
|
|
from services import blog_jobs, mail_service
|
|
|
|
monkeypatch.setattr(mail_service, "is_configured", lambda: True)
|
|
sent_calls = []
|
|
monkeypatch.setattr(mail_service, "send", lambda **kwargs: sent_calls.append(kwargs) or True)
|
|
|
|
h = await auth_headers("bloggen4")
|
|
place_id = await _place(client, h, name="알림이메일펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
await conn.execute(text("UPDATE users SET email = :e WHERE id = :id"), {"e": "owner@example.com", "id": "bloggen4"})
|
|
|
|
patch_res = await client.patch(f"/v1/place/{place_id}", headers=h, json={"notify_email": "manager@example.com"})
|
|
assert patch_res.json()["result"]["success"] is True
|
|
|
|
await _seed_post(db_engine, place_id, scheduled=date.today())
|
|
sent = await blog_jobs.send_reviewed()
|
|
|
|
assert sent == 1
|
|
assert sent_calls[0]["to"] == "manager@example.com"
|
|
|
|
|
|
async def test_update_place_rejects_malformed_notify_email(client, db_engine, auth_headers):
|
|
h = await auth_headers("bloggen5")
|
|
place_id = await _place(client, h, name="잘못된이메일펜션")
|
|
|
|
res = await client.patch(f"/v1/place/{place_id}", headers=h, json={"notify_email": "not-an-email"})
|
|
|
|
assert res.json()["result"]["success"] is False
|
|
|
|
|
|
async def test_send_now_mails_todays_due_post_immediately(client, db_engine, auth_headers, monkeypatch):
|
|
"""빌더 화면의 '승인 알림보내기' — 아침 9시 스윕을 기다리지 않고 이 업장의 오늘 몫을
|
|
바로 보낸다(2026-09-21, 사장님 요청)."""
|
|
from services import blog_jobs, mail_service
|
|
|
|
monkeypatch.setattr(mail_service, "is_configured", lambda: True)
|
|
sent_calls = []
|
|
monkeypatch.setattr(mail_service, "send", lambda **kwargs: sent_calls.append(kwargs) or True)
|
|
|
|
h = await auth_headers("bloggen6")
|
|
place_id = await _place(client, h, name="즉시발송펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
await conn.execute(text("UPDATE users SET email = :e WHERE id = :id"), {"e": "owner6@example.com", "id": "bloggen6"})
|
|
post_id = await _seed_post(db_engine, place_id, scheduled=date.today())
|
|
|
|
res = await client.post(f"/v1/place/{place_id}/post/send-now", headers=h)
|
|
|
|
assert res.json()["result"]["success"] is True
|
|
assert len(sent_calls) == 1
|
|
assert sent_calls[0]["to"] == "owner6@example.com"
|
|
assert await _status(db_engine, post_id) == PostStatus.SENT.value
|
|
|
|
|
|
async def test_send_now_is_a_noop_when_nothing_is_due(client, db_engine, auth_headers, monkeypatch):
|
|
from services import mail_service
|
|
|
|
monkeypatch.setattr(mail_service, "is_configured", lambda: True)
|
|
sent_calls = []
|
|
monkeypatch.setattr(mail_service, "send", lambda **kwargs: sent_calls.append(kwargs) or True)
|
|
|
|
h = await auth_headers("bloggen7")
|
|
place_id = await _place(client, h, name="빈발송펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
await _seed_post(db_engine, place_id, scheduled=date.today() + timedelta(days=3))
|
|
|
|
res = await client.post(f"/v1/place/{place_id}/post/send-now", headers=h)
|
|
|
|
body = res.json()
|
|
assert body["result"]["success"] is True
|
|
assert "없습니다" in body["msg"]
|
|
assert sent_calls == []
|
|
|
|
|
|
async def test_send_now_fails_clearly_when_mail_is_not_configured(client, db_engine, auth_headers, monkeypatch):
|
|
from services import mail_service
|
|
|
|
monkeypatch.setattr(mail_service, "is_configured", lambda: False)
|
|
|
|
h = await auth_headers("bloggen8")
|
|
place_id = await _place(client, h, name="메일미설정펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
await _seed_post(db_engine, place_id, scheduled=date.today())
|
|
|
|
res = await client.post(f"/v1/place/{place_id}/post/send-now", headers=h)
|
|
|
|
assert res.json()["result"]["success"] is False
|
|
|
|
|
|
async def test_send_now_cannot_be_triggered_for_someone_elses_place(client, db_engine, auth_headers):
|
|
owner = await auth_headers("bloggen9")
|
|
other = await auth_headers("bloggen10")
|
|
place_id = await _place(client, owner, name="남의발송펜션")
|
|
|
|
res = await client.post(f"/v1/place/{place_id}/post/send-now", headers=other)
|
|
|
|
assert res.json()["result"]["success"] is False
|
|
|
|
|
|
async def _day_pass_headers(db_engine, login_id: str) -> dict:
|
|
"""메일의 '수정하려면' 링크가 주는 것과 같은 종류의 day-pass 토큰(2026-09-21,
|
|
사장님 지시: "메일 링크에서 들어와 수정한 후에는 승인할 수 있어야 한다")."""
|
|
from common.models.gmodel import UserInfo
|
|
from router.v1.validator.dependencies import CreateDayPassToken
|
|
|
|
async with db_engine.begin() as conn:
|
|
row = (await conn.execute(
|
|
text("SELECT user_id, role, token_version FROM users WHERE id = :id"), {"id": login_id},
|
|
)).first()
|
|
user_info = UserInfo(user_id=str(row[0]), id=login_id, role=row[1], token_version=row[2])
|
|
token = CreateDayPassToken(user_info)
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
async def test_approve_by_owner_shares_to_threads_when_linked(client, db_engine, auth_headers, monkeypatch):
|
|
"""'바로 발행' 버튼으로 승인하면, 쓰레드가 연동돼 있을 때 같은 문구가 쓰레드로도 나간다
|
|
(2026-09-21, 사장님 지시: "쓰레드에 연동되어 있으면 같이 업로드 되는 기능")."""
|
|
from services import post_service
|
|
|
|
calls = []
|
|
|
|
async def fake_publish_reused_text(user_id, place_id, body):
|
|
calls.append((user_id, place_id, body))
|
|
return uuid.uuid4()
|
|
|
|
monkeypatch.setattr(post_service.social_service, "publish_reused_text", fake_publish_reused_text)
|
|
|
|
h = await auth_headers("blogowner7")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
|
|
|
|
res = await client.post(f"/v1/place/{place_id}/post/{post_id}/approve", headers=h)
|
|
|
|
assert res.json()["result"]["success"] is True
|
|
assert len(calls) == 1
|
|
assert calls[0][2] == BODY
|
|
assert str(calls[0][1]) == place_id
|
|
|
|
|
|
async def test_email_token_approve_also_shares_to_threads(client, db_engine, auth_headers, monkeypatch):
|
|
"""이메일 GET 승인 경로도 '바로 발행' 버튼과 동일하게 쓰레드 연동을 태운다."""
|
|
from services import blog_service, post_service
|
|
|
|
calls = []
|
|
|
|
async def fake_publish_reused_text(user_id, place_id, body):
|
|
calls.append((user_id, place_id, body))
|
|
return uuid.uuid4()
|
|
|
|
monkeypatch.setattr(post_service.social_service, "publish_reused_text", fake_publish_reused_text)
|
|
|
|
h = await auth_headers("blogowner8")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.SENT)
|
|
token, token_hash, expires = blog_service.issue_token()
|
|
aware_expires = expires.replace(tzinfo=timezone.utc) if expires.tzinfo is None else expires
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("UPDATE place_posts SET approve_token_hash=:h, token_expires_at=:e WHERE post_id=:id"),
|
|
{"h": token_hash, "e": aware_expires, "id": post_id},
|
|
)
|
|
|
|
res = await client.get(f"/v1/site/post/approve?t={token}")
|
|
|
|
assert res.status_code == 200
|
|
assert await _status(db_engine, post_id) == PostStatus.APPROVED.value
|
|
assert len(calls) == 1
|
|
assert calls[0][2] == BODY
|
|
|
|
|
|
async def test_threads_share_failure_does_not_block_approval(client, db_engine, auth_headers, monkeypatch):
|
|
"""쓰레드 연동이 예외를 던져도 미니블로그 승인 자체는 그대로 끝나야 한다."""
|
|
from services import post_service
|
|
|
|
async def failing_publish_reused_text(user_id, place_id, body):
|
|
raise RuntimeError("threads down")
|
|
|
|
monkeypatch.setattr(post_service.social_service, "publish_reused_text", failing_publish_reused_text)
|
|
|
|
h = await auth_headers("blogowner9")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
|
|
|
|
res = await client.post(f"/v1/place/{place_id}/post/{post_id}/approve", headers=h)
|
|
|
|
assert res.json()["result"]["success"] is True
|
|
assert await _status(db_engine, post_id) == PostStatus.APPROVED.value
|
|
|
|
|
|
async def test_owner_can_publish_as_is_without_editing(client, db_engine, auth_headers):
|
|
"""'바로 발행' — 로그인 세션만으로, 본문을 안 고쳐도 승인되고 재발행 잡이 걸린다
|
|
(2026-09-21, 사장님 지시: "이메일 승인으로도 발행 가능하고 바로발행버튼으로도
|
|
발행 가능하도록") — 이메일 승인 링크와 별개의 두 번째 경로다."""
|
|
h = await auth_headers("blogowner6")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
|
|
|
|
res = await client.post(f"/v1/place/{place_id}/post/{post_id}/approve", headers=h)
|
|
|
|
assert res.json()["result"]["success"] is True
|
|
assert await _status(db_engine, post_id) == PostStatus.APPROVED.value
|
|
async with db_engine.begin() as conn:
|
|
body_row = (await conn.execute(
|
|
text("SELECT body FROM place_posts WHERE post_id = :id"), {"id": post_id},
|
|
)).scalar()
|
|
assert body_row == BODY
|
|
|
|
|
|
async def test_owner_cannot_publish_someone_elses_post(client, db_engine, auth_headers):
|
|
owner = await auth_headers("blogowner6e")
|
|
other = await auth_headers("blogowner6f")
|
|
place_id = await _place(client, owner)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
|
|
|
|
res = await client.post(f"/v1/place/{place_id}/post/{post_id}/approve", headers=other)
|
|
|
|
assert res.json()["result"]["success"] is False
|
|
assert await _status(db_engine, post_id) == PostStatus.REVIEWED.value
|
|
|
|
|
|
async def test_daypass_session_can_approve_after_editing(client, db_engine, auth_headers):
|
|
"""메일 '수정하려면' 링크(day-pass)로 들어와 고친 뒤에는, 다시 메일을 뒤지지 않고
|
|
그 자리에서 승인할 수 있어야 한다(2026-09-21, 사장님 지시)."""
|
|
h = await auth_headers("blogowner6b")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
|
|
day_pass = await _day_pass_headers(db_engine, "blogowner6b")
|
|
new_body = BODY.replace("빗소리", "새소리")
|
|
|
|
edit_res = await client.put(f"/v1/place/{place_id}/post/{post_id}", headers=day_pass, json={"body": new_body})
|
|
assert edit_res.json()["result"]["success"] is True
|
|
assert await _status(db_engine, post_id) == PostStatus.REVIEWED.value
|
|
|
|
approve_res = await client.post(f"/v1/place/{place_id}/post/{post_id}/approve", headers=day_pass)
|
|
|
|
assert approve_res.json()["result"]["success"] is True
|
|
assert await _status(db_engine, post_id) == PostStatus.APPROVED.value
|
|
async with db_engine.begin() as conn:
|
|
saved_body = (await conn.execute(
|
|
text("SELECT body FROM place_posts WHERE post_id = :id"), {"id": post_id},
|
|
)).scalar()
|
|
assert saved_body == new_body
|
|
|
|
|
|
async def test_daypass_approve_cannot_be_used_for_someone_elses_place(client, db_engine, auth_headers):
|
|
owner = await auth_headers("blogowner6c")
|
|
place_id = await _place(client, owner)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
|
|
await auth_headers("blogowner6d")
|
|
other_day_pass = await _day_pass_headers(db_engine, "blogowner6d")
|
|
|
|
res = await client.post(f"/v1/place/{place_id}/post/{post_id}/approve", headers=other_day_pass)
|
|
|
|
assert res.json()["result"]["success"] is False
|
|
assert await _status(db_engine, post_id) == PostStatus.REVIEWED.value
|
|
|
|
|
|
async def test_generate_now_is_noop_for_unpublished_site(client, db_engine, auth_headers):
|
|
"""발행 전 사업장은 생성 스윕 대상이 아니다(blog_jobs._published_places) — 0건이어야 한다."""
|
|
h = await auth_headers("bloggen2")
|
|
place_id = await _place(client, h, name="발행전펜션")
|
|
start = date.today()
|
|
|
|
res = await client.post(
|
|
f"/v1/place/{place_id}/post/generate", headers=h,
|
|
params={"start": start.isoformat(), "end": start.isoformat()},
|
|
)
|
|
|
|
body = res.json()
|
|
assert body["result"]["success"] is True
|
|
assert body["created"] == 0
|
|
|
|
|
|
async def test_generate_now_is_noop_for_site_without_domain(client, db_engine, auth_headers, monkeypatch):
|
|
"""domain 미확정(임시 주소) 사이트도 생성 스윕 대상이 아니다 — 쓰레드 연동 요구사항과
|
|
맞춘다(2026-09-21). PUBLISHED 여도 domain 이 없으면 0건이어야 한다. generate_one 을
|
|
실제로 성공하도록 목킹해 둬야 "그냥 LLM 이 설정 안 돼서 0건"과 구분된다."""
|
|
from services import blog_service
|
|
|
|
async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
|
|
return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
|
|
|
|
monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
|
|
|
|
h = await auth_headers("bloggen17")
|
|
place_id = await _place(client, h, name="도메인미정펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, status) VALUES (:sid, :pid, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
start = date.today()
|
|
|
|
res = await client.post(
|
|
f"/v1/place/{place_id}/post/generate", headers=h,
|
|
params={"start": start.isoformat(), "end": start.isoformat()},
|
|
)
|
|
|
|
body = res.json()
|
|
assert body["result"]["success"] is True
|
|
assert body["created"] == 0
|
|
|
|
|
|
async def test_generate_now_rejects_end_before_start(client, db_engine, auth_headers):
|
|
h = await auth_headers("bloggen16")
|
|
place_id = await _place(client, h, name="구간역순펜션")
|
|
start = date.today()
|
|
|
|
res = await client.post(
|
|
f"/v1/place/{place_id}/post/generate", headers=h,
|
|
params={"start": start.isoformat(), "end": (start - timedelta(days=1)).isoformat()},
|
|
)
|
|
|
|
body = res.json()
|
|
assert body["result"]["success"] is False
|
|
|
|
|
|
async def test_approved_post_flags_build_failed_when_job_dead(client, db_engine, auth_headers):
|
|
"""화면은 발행완료/발행실패만 본다(사장님 지시) — 승인됐는데 BUILD 잡이 dead-letter 면
|
|
build_failed=true, 그 외(대기 중인 잡·아직 승인 전)에는 false 로 남는다."""
|
|
h = await auth_headers("bloggen4")
|
|
place_id = await _place(client, h, name="실패펜션")
|
|
failed_post = await _seed_post(db_engine, place_id, status=PostStatus.APPROVED)
|
|
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO jobs (job_type, status, payload) VALUES (:jt, :st, :pl)"),
|
|
{"jt": JobType.BUILD.value, "st": JobStatus.DEAD.value, "pl": f'{{"place_id": "{place_id}"}}'},
|
|
)
|
|
|
|
res = await client.get(f"/v1/place/{place_id}/post", headers=h)
|
|
|
|
posts = {p["post_id"]: p for p in res.json()["posts"]}
|
|
assert posts[failed_post]["build_failed"] is True
|
|
|
|
|
|
async def test_pending_build_job_does_not_flag_failure(client, db_engine, auth_headers):
|
|
h = await auth_headers("bloggen5")
|
|
place_id = await _place(client, h, name="대기펜션")
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.APPROVED)
|
|
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO jobs (job_type, status, payload) VALUES (:jt, :st, :pl)"),
|
|
{"jt": JobType.BUILD.value, "st": JobStatus.PENDING.value, "pl": f'{{"place_id": "{place_id}"}}'},
|
|
)
|
|
|
|
res = await client.get(f"/v1/place/{place_id}/post", headers=h)
|
|
|
|
posts = {p["post_id"]: p for p in res.json()["posts"]}
|
|
assert posts[post_id].get("build_failed", False) is False
|
|
|
|
|
|
async def test_upcoming_only_returns_next_week_in_date_order(client, db_engine, auth_headers):
|
|
"""상단 카로셀 — 오늘부터 N일치만, 날짜 오름차순. 그 뒤 배정분은 안 보인다."""
|
|
h = await auth_headers("bloggen6")
|
|
place_id = await _place(client, h, name="주간펜션")
|
|
far = await _seed_post(db_engine, place_id, scheduled=date.today() + timedelta(days=20))
|
|
tomorrow = await _seed_post(db_engine, place_id, scheduled=date.today() + timedelta(days=1))
|
|
today = await _seed_post(db_engine, place_id, scheduled=date.today())
|
|
|
|
res = await client.get(f"/v1/place/{place_id}/post/upcoming", headers=h, params={"days": 7})
|
|
|
|
ids = [p["post_id"] for p in res.json()["posts"]]
|
|
assert ids == [today, tomorrow]
|
|
assert far not in ids
|
|
|
|
|
|
async def test_get_post_by_id_for_mail_edit_link(client, db_engine, auth_headers):
|
|
"""메일 '수정하기' 링크(자동 로그인)가 postId 하나로 그 글을 바로 찾는 경로."""
|
|
h = await auth_headers("bloggen7")
|
|
place_id = await _place(client, h, name="단건조회펜션")
|
|
post_id = await _seed_post(db_engine, place_id)
|
|
|
|
res = await client.get(f"/v1/place/{place_id}/post/{post_id}", headers=h)
|
|
|
|
body = res.json()
|
|
assert body["result"]["success"] is True
|
|
assert len(body["posts"]) == 1
|
|
assert body["posts"][0]["post_id"] == post_id
|
|
|
|
|
|
async def test_get_post_by_id_scoped_to_owner(client, db_engine, auth_headers):
|
|
owner = await auth_headers("bloggen8")
|
|
other = await auth_headers("bloggen9")
|
|
place_id = await _place(client, owner, name="타인조회펜션")
|
|
post_id = await _seed_post(db_engine, place_id)
|
|
|
|
res = await client.get(f"/v1/place/{place_id}/post/{post_id}", headers=other)
|
|
|
|
body = res.json()
|
|
assert body["result"]["success"] is False
|
|
assert body["posts"] == []
|
|
|
|
|
|
async def test_generation_history_counts_by_batch(client, db_engine, auth_headers, monkeypatch):
|
|
"""생성 이력 — 한 번에 몇 건 · 어느 모델(사장님 지시: "생성이력도 있어야해 몇개
|
|
생성했는지" / "어느 모델썼는지 등등" → JSONB 한 칸(generation_meta)에 담는다)."""
|
|
from services import blog_service
|
|
|
|
async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
|
|
return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
|
|
|
|
monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
|
|
|
|
h = await auth_headers("bloggen10")
|
|
place_id = await _place(client, h, name="이력펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
start = date.today()
|
|
generate_res = await client.post(
|
|
f"/v1/place/{place_id}/post/generate", headers=h,
|
|
params={"start": start.isoformat(), "end": (start + timedelta(days=29)).isoformat()},
|
|
)
|
|
created = generate_res.json()["created"]
|
|
assert created > 0
|
|
|
|
res = await client.get(f"/v1/place/{place_id}/post/history", headers=h)
|
|
|
|
batches = res.json()["batches"]
|
|
assert len(batches) == 1
|
|
assert batches[0]["count"] == created
|
|
assert batches[0]["model"] == "gemini-test-model"
|
|
|
|
|
|
async def test_mail_has_one_click_approve_and_autologin_edit_links(client, db_engine, auth_headers, monkeypatch):
|
|
"""사장님 지시: "승인이랑 수정하기 있어야해" — 승인은 토큰 링크 하나, 수정은
|
|
그날짜리 자동 로그인 토큰을 실은 빌더 앱 링크."""
|
|
from services import blog_jobs, mail_service
|
|
|
|
monkeypatch.setattr(mail_service, "is_configured", lambda: True)
|
|
sent_calls = []
|
|
monkeypatch.setattr(mail_service, "send", lambda **kwargs: sent_calls.append(kwargs) or True)
|
|
|
|
h = await auth_headers("bloggen11")
|
|
place_id = await _place(client, h, name="메일링크펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
await conn.execute(
|
|
text("UPDATE users SET email = :e WHERE id = :id"), {"e": "owner@example.com", "id": "bloggen11"},
|
|
)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED, scheduled=date.today())
|
|
|
|
sent = await blog_jobs.send_reviewed()
|
|
|
|
assert sent == 1
|
|
mail_text = sent_calls[0]["text"]
|
|
assert "/v1/site/post/approve?t=" in mail_text
|
|
# ★ 수정 링크도 일회용 코드다(2026-09-28). 예전에는 여기 빌더 액세스 토큰을 통짜로 실어
|
|
# `/blog?placeId=..&postId=..&auto=<JWT>` 를 보냈다 — 주소가 500자였던 것은 곁가지고,
|
|
# 진짜 문제는 **메일 전달 한 번이 그날 자정까지의 권한 양도**였다는 것이다.
|
|
assert "/v1/site/post/edit?t=" in mail_text
|
|
assert "auto=" not in mail_text
|
|
assert "eyJ" not in mail_text
|
|
|
|
|
|
async def test_mail_links_use_the_builder_app_origin_not_the_published_site_origin(
|
|
client, db_engine, auth_headers, monkeypatch,
|
|
):
|
|
"""수정·승인 링크는 빌더 앱(SOCIAL_APP_ORIGIN)으로 가야 한다 — 발행된 사이트 오리진
|
|
(site_payload.publish_origin, 로컬에선 solution-site 정적 서버 포트 80)으로 가면
|
|
404가 난다(2026-09-21 실측: 메일의 '수정하려면' 링크가 거기로 가서 404)."""
|
|
from services import blog_jobs, blog_service, mail_service, site_payload
|
|
|
|
monkeypatch.setattr(mail_service, "is_configured", lambda: True)
|
|
sent_calls = []
|
|
monkeypatch.setattr(mail_service, "send", lambda **kwargs: sent_calls.append(kwargs) or True)
|
|
monkeypatch.setattr(site_payload, "publish_origin", lambda: "http://published-site-origin")
|
|
# ★ 오리진 계산은 blog_service.app_origin() 으로 옮겼다 — 라우터(수정 링크 리다이렉트)도
|
|
# 같은 값을 써야 해서다. 그래서 패치 대상도 거기다.
|
|
monkeypatch.setattr(
|
|
blog_service.social_config, "get",
|
|
lambda name, default="": "http://builder-app-origin" if name == "SOCIAL_APP_ORIGIN" else default,
|
|
)
|
|
|
|
h = await auth_headers("bloggen12")
|
|
place_id = await _place(client, h, name="오리진확인펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
await conn.execute(text("UPDATE users SET email = :e WHERE id = :id"), {"e": "owner12@example.com", "id": "bloggen12"})
|
|
await _seed_post(db_engine, place_id, scheduled=date.today())
|
|
|
|
sent = await blog_jobs.send_reviewed()
|
|
|
|
assert sent == 1
|
|
mail_text = sent_calls[0]["text"]
|
|
# 두 링크 다 빌더 앱 오리진이어야 한다. 수정 링크는 일회용 코드를 거쳐 /blog 로
|
|
# 리다이렉트되므로(router/v1/site/post.py edit_redirect) 메일에는 /v1/site/post/edit 이 실린다.
|
|
assert "http://builder-app-origin/v1/site/post/edit?t=" in mail_text
|
|
assert "http://builder-app-origin/v1/site/post/approve?t=" in mail_text
|
|
assert "published-site-origin" not in mail_text
|
|
|
|
|
|
async def test_generate_one_fills_a_specific_empty_date(client, db_engine, auth_headers, monkeypatch):
|
|
"""사장님 지시: "개별적으로 새로 만들수있게 해줘" — 달력에서 빈 날짜 하나만 콕 집어 채운다."""
|
|
from services import blog_service
|
|
|
|
async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
|
|
return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
|
|
|
|
monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
|
|
|
|
h = await auth_headers("bloggen12")
|
|
place_id = await _place(client, h, name="개별생성펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
target = date.today() + timedelta(days=3)
|
|
|
|
res = await client.post(
|
|
f"/v1/place/{place_id}/post/generate-one", headers=h, params={"date": target.isoformat()},
|
|
)
|
|
|
|
body = res.json()
|
|
assert body["result"]["success"] is True
|
|
assert body["post"]["scheduled_date"] == target.isoformat()
|
|
assert await _status(db_engine, body["post"]["post_id"]) == PostStatus.REVIEWED.value
|
|
|
|
|
|
async def test_generate_one_fails_when_date_already_taken(client, db_engine, auth_headers, monkeypatch):
|
|
"""이미 그 날짜에 글이 있으면(유니크 충돌) 조용히 덮지 않고 실패로 답한다."""
|
|
from services import blog_service
|
|
|
|
async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
|
|
return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
|
|
|
|
monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
|
|
|
|
h = await auth_headers("bloggen13")
|
|
place_id = await _place(client, h, name="중복날짜펜션")
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
|
|
{"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
|
|
)
|
|
target = date.today() + timedelta(days=3)
|
|
await _seed_post(db_engine, place_id, scheduled=target)
|
|
|
|
res = await client.post(
|
|
f"/v1/place/{place_id}/post/generate-one", headers=h, params={"date": target.isoformat()},
|
|
)
|
|
|
|
body = res.json()
|
|
assert body["result"]["success"] is False
|
|
assert body.get("post") is None
|
|
|
|
|
|
async def test_generate_one_is_scoped_to_owner(client, db_engine, auth_headers):
|
|
owner = await auth_headers("bloggen14")
|
|
other = await auth_headers("bloggen15")
|
|
place_id = await _place(client, owner, name="타인개별생성펜션")
|
|
|
|
res = await client.post(
|
|
f"/v1/place/{place_id}/post/generate-one", headers=other,
|
|
params={"date": date.today().isoformat()},
|
|
)
|
|
|
|
assert res.json()["result"]["success"] is False
|
|
|
|
|
|
# ── 메일 '고쳐서 올리려면' — 일회용 코드 ────────────────────────────────
|
|
|
|
async def _seed_edit_token(db_engine, post_id, *, expired=False):
|
|
from services import blog_service
|
|
|
|
token, token_hash, expires = blog_service.issue_token()
|
|
if expired:
|
|
expires = datetime.now(timezone.utc) - timedelta(minutes=1)
|
|
aware = expires.replace(tzinfo=timezone.utc) if expires.tzinfo is None else expires
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("UPDATE place_posts SET edit_token_hash=:h, token_expires_at=:e WHERE post_id=:id"),
|
|
{"h": token_hash, "e": aware, "id": post_id},
|
|
)
|
|
return token
|
|
|
|
|
|
async def test_수정_링크는_짧고_액세스_토큰을_싣지_않는다(client, db_engine, auth_headers):
|
|
"""★ 예전에는 이 링크에 빌더 액세스 토큰을 통짜로 실었다 — 메일 전달 한 번이
|
|
그날 자정까지의 권한 양도였고, 주소는 500자였다."""
|
|
h = await auth_headers("blogedit1")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.SENT)
|
|
token = await _seed_edit_token(db_engine, post_id)
|
|
|
|
res = await client.get(f"/v1/site/post/edit?t={token}", follow_redirects=False)
|
|
assert res.status_code == 303
|
|
location = res.headers["location"]
|
|
|
|
# 액세스 토큰은 **프래그먼트**로만 간다 — 서버 로그·Referer 에 남지 않는다.
|
|
path_and_query = location.split("#", 1)[0]
|
|
assert "auto=" not in path_and_query
|
|
assert "#auto=" in location
|
|
assert str(post_id) in path_and_query and str(place_id) in path_and_query
|
|
assert res.headers["Referrer-Policy"] == "no-referrer"
|
|
|
|
|
|
async def test_수정_링크의_토큰은_그_사장님_것이다(client, db_engine, auth_headers):
|
|
"""URL 에 누구인지 싣지 않는다 — 토큰이 가리키는 글에서 사람을 끌어낸다."""
|
|
from router.v1.validator.dependencies import DecodeAccessToken
|
|
|
|
h = await auth_headers("blogedit2")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.SENT)
|
|
token = await _seed_edit_token(db_engine, post_id)
|
|
|
|
res = await client.get(f"/v1/site/post/edit?t={token}", follow_redirects=False)
|
|
auto = res.headers["location"].split("#auto=", 1)[1]
|
|
async with db_engine.begin() as conn:
|
|
owner = (
|
|
await conn.execute(text("SELECT owner_user_id FROM places WHERE place_id=:p"), {"p": place_id})
|
|
).scalar_one()
|
|
assert DecodeAccessToken(auto).user_id == str(owner)
|
|
|
|
|
|
async def test_만료된_수정_링크는_안_먹는다(client, db_engine, auth_headers):
|
|
h = await auth_headers("blogedit3")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.SENT)
|
|
token = await _seed_edit_token(db_engine, post_id, expired=True)
|
|
|
|
res = await client.get(f"/v1/site/post/edit?t={token}", follow_redirects=False)
|
|
assert res.status_code == 200 # 만료 안내 화면
|
|
assert "auto=" not in res.text
|
|
|
|
|
|
async def test_없는_수정_코드는_이유를_구분해_답하지_않는다(client, db_engine):
|
|
res = await client.get("/v1/site/post/edit?t=ZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ", follow_redirects=False)
|
|
assert res.status_code == 200
|
|
assert "auto=" not in res.text
|
|
|
|
|
|
async def test_승인_코드로는_수정_링크를_열_수_없다(client, db_engine, auth_headers):
|
|
"""★ 두 코드는 서로 다른 칸에 산다. 하나로 둘 다 되면 일회성이 무의미해진다."""
|
|
from services import blog_service
|
|
|
|
h = await auth_headers("blogedit4")
|
|
place_id = await _place(client, h)
|
|
post_id = await _seed_post(db_engine, place_id, status=PostStatus.SENT)
|
|
approve_token, approve_hash, expires = blog_service.issue_token()
|
|
aware = expires.replace(tzinfo=timezone.utc) if expires.tzinfo is None else expires
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("UPDATE place_posts SET approve_token_hash=:h, token_expires_at=:e WHERE post_id=:id"),
|
|
{"h": approve_hash, "e": aware, "id": post_id},
|
|
)
|
|
|
|
res = await client.get(f"/v1/site/post/edit?t={approve_token}", follow_redirects=False)
|
|
assert res.status_code == 200
|
|
assert "auto=" not in res.text
|
|
|
|
|
|
def test_메일_본문에_액세스_토큰이_없다():
|
|
"""★ 본문이 곧 유출 경로다 — 전달 한 번으로 권한이 넘어가면 안 된다."""
|
|
from types import SimpleNamespace
|
|
from services import blog_jobs
|
|
|
|
post = SimpleNamespace(post_id=uuid.uuid4(), place_id=uuid.uuid4(), body=BODY)
|
|
user = SimpleNamespace(user_id=uuid.uuid4(), id="owner", role=1, token_version=1, email="a@b.c")
|
|
body = blog_jobs._mail_body(
|
|
place_name="테스트", post=post, user=user, origin="https://example.com",
|
|
approve_token="APPROVE_CODE", edit_token="EDIT_CODE",
|
|
)
|
|
assert "APPROVE_CODE" in body and "EDIT_CODE" in body
|
|
assert "auto=" not in body # 액세스 토큰이 실리던 자리
|
|
assert "eyJ" not in body # JWT 의 머리글자
|
|
longest = max(len(word) for word in body.split())
|
|
assert longest < 120, f"링크가 아직 길다: {longest}자"
|