Senad Uka
e855650f09
- FastAPI skeleton, SQLAlchemy models (§13), Alembic initial migration - SchedulingProvider interface with google_calendar (free/busy read-only), partner_api (Appendix B client) and mock implementations - Proposal engine: create → provider-routed delivery → owner actions (resolve/confirm+SMS/reject) → expiry + reminders (§9) - Signed single-use action links, .ics METHOD:REQUEST attachment - Partner outcome webhook with HMAC verification + polling fallback - SmsProvider (console) with Bosnian templates (§5.5), EmailProvider (console/SMTP) - Fake partner API server in tests/ — Appendix B reference implementation - 43 tests: slot math, proposal lifecycle, action links, partner contract Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
134 lines
4.4 KiB
Python
134 lines
4.4 KiB
Python
"""Owner email action links over HTTP (§9.1-9.2, idempotency per §15)."""
|
|
|
|
import re
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from gogo.domain import BookingStatus
|
|
from tests.test_proposal_lifecycle import create
|
|
|
|
|
|
def extract_links(text: str) -> dict[str, str]:
|
|
"""Pull action URLs out of the plain-text proposal email."""
|
|
links = {}
|
|
resolve = re.search(r"RIJEŠENO[^\n]*\n\s+(http\S+)", text)
|
|
confirm = re.search(r"POTVRDI[^\n]*\n\s+(http\S+)", text)
|
|
reject = re.search(r"ODBIJ[^\n]*\n\s+(http\S+)", text)
|
|
if resolve:
|
|
links["resolve"] = resolve.group(1)
|
|
if confirm:
|
|
links["confirm"] = confirm.group(1)
|
|
if reject:
|
|
links["reject"] = reject.group(1)
|
|
return links
|
|
|
|
|
|
async def test_resolve_link_closes_request_no_sms(session, tenant, emails, sms, gogo_client):
|
|
req, _ = await create(session, tenant)
|
|
await session.commit()
|
|
links = extract_links(emails[0].text)
|
|
|
|
resp = await gogo_client.get(links["resolve"])
|
|
assert resp.status_code == 200
|
|
assert "Označeno kao riješeno" in resp.text
|
|
|
|
await session.refresh(req)
|
|
assert req.status == BookingStatus.resolved_by_owner.value
|
|
assert sms == []
|
|
|
|
# idempotent: second click shows friendly "already handled" page
|
|
resp2 = await gogo_client.get(links["resolve"])
|
|
assert resp2.status_code == 200
|
|
assert "već obrađen" in resp2.text
|
|
|
|
|
|
async def test_confirm_link_sends_sms(session, tenant, emails, sms, gogo_client):
|
|
req, _ = await create(session, tenant)
|
|
await session.commit()
|
|
links = extract_links(emails[0].text)
|
|
|
|
resp = await gogo_client.get(links["confirm"])
|
|
assert resp.status_code == 200
|
|
assert "Termin potvrđen" in resp.text
|
|
|
|
await session.refresh(req)
|
|
assert req.status == BookingStatus.confirmed.value
|
|
assert len(sms) == 1
|
|
assert "Potvrđen termin" in sms[0][1]
|
|
|
|
|
|
async def test_reject_link_shows_editable_form_then_sends(
|
|
session, tenant, emails, sms, gogo_client
|
|
):
|
|
req, _ = await create(session, tenant)
|
|
await session.commit()
|
|
links = extract_links(emails[0].text)
|
|
|
|
# GET shows the form with the default template prefilled — nothing sent yet
|
|
resp = await gogo_client.get(links["reject"])
|
|
assert resp.status_code == 200
|
|
assert "<textarea" in resp.text
|
|
assert sms == []
|
|
|
|
resp = await gogo_client.post(
|
|
links["reject"] + "/reject", data={"sms_body": "Nažalost, popunjeni smo. Salon Merima"}
|
|
)
|
|
assert resp.status_code == 200
|
|
await session.refresh(req)
|
|
assert req.status == BookingStatus.rejected.value
|
|
assert sms == [("+38765123456", "Nažalost, popunjeni smo. Salon Merima")]
|
|
|
|
|
|
async def test_forged_token_rejected(gogo_client, session, tenant):
|
|
resp = await gogo_client.get("/a/forged-token-value")
|
|
assert resp.status_code == 400
|
|
assert "Nevažeći link" in resp.text
|
|
|
|
|
|
async def test_action_after_expiry_shows_status(session, tenant, emails, sms, gogo_client):
|
|
from gogo.proposals import engine as pe
|
|
|
|
req, _ = await create(session, tenant)
|
|
req.expires_at = datetime.now(UTC) - timedelta(minutes=1)
|
|
await pe.process_expirations(session)
|
|
await session.commit()
|
|
sms.clear()
|
|
|
|
links = extract_links(emails[0].text)
|
|
resp = await gogo_client.get(links["confirm"])
|
|
assert resp.status_code == 200
|
|
assert "već obrađen" in resp.text
|
|
assert "istekao" in resp.text
|
|
assert sms == []
|
|
|
|
|
|
async def test_confirm_link_warns_when_slot_taken(session, tenant, emails, sms, gogo_client):
|
|
"""Recheck finds slot busy → warning page with force-confirm, no SMS yet."""
|
|
from gogo.scheduling.mock import MockProvider
|
|
|
|
req, _ = await create(session, tenant)
|
|
await session.commit()
|
|
links = extract_links(emails[0].text)
|
|
|
|
async def is_slot_free(self, service_id, slot_):
|
|
return False
|
|
|
|
MockProvider.is_slot_free = is_slot_free
|
|
try:
|
|
resp = await gogo_client.get(links["confirm"])
|
|
assert resp.status_code == 200
|
|
assert "u međuvremenu zauzet" in resp.text
|
|
await session.refresh(req)
|
|
assert req.status == BookingStatus.pending.value
|
|
assert sms == []
|
|
|
|
# owner forces the confirmation anyway
|
|
resp = await gogo_client.post(links["confirm"] + "/force-confirm")
|
|
assert resp.status_code == 200
|
|
assert "Termin potvrđen" in resp.text
|
|
finally:
|
|
del MockProvider.is_slot_free
|
|
|
|
await session.refresh(req)
|
|
assert req.status == BookingStatus.confirmed.value
|
|
assert len(sms) == 1
|