23 lines
1.7 KiB
Python
23 lines
1.7 KiB
Python
import io, json
|
|
from email_connect.transactional import SQLiteDeliveryStore, TransactionalApplication
|
|
|
|
class Provider:
|
|
def __init__(self): self.calls=[]
|
|
def send(self,*args): self.calls.append(args); return "message:1"
|
|
|
|
def event(): return {"id":"evt-1","type":"family_member.invited","source":"user-engine","data":{"primary_email":"person@example.test","invitation_id":"inv-1"}}
|
|
def invoke(app,payload,token="opaque",key="evt-1"):
|
|
raw=json.dumps(payload).encode(); result={}; body=b"".join(app({"PATH_INFO":"/v1/send","REQUEST_METHOD":"POST","CONTENT_LENGTH":str(len(raw)),"wsgi.input":io.BytesIO(raw),"HTTP_AUTHORIZATION":f"Bearer {token}","HTTP_IDEMPOTENCY_KEY":key},lambda s,h:result.update(status=s))); return result["status"],json.loads(body)
|
|
|
|
def test_sends_fixed_template_once(tmp_path):
|
|
provider=Provider(); app=TransactionalApplication(SQLiteDeliveryStore(str(tmp_path/"mail.db")),provider,"opaque","https://users.example")
|
|
assert invoke(app,event())[0].startswith("202")
|
|
assert invoke(app,event())[1]["status"] == "duplicate"
|
|
assert len(provider.calls)==1 and "/invitations/inv-1" in provider.calls[0][2]
|
|
|
|
def test_rejects_auth_recipient_template_and_key(tmp_path):
|
|
app=TransactionalApplication(SQLiteDeliveryStore(str(tmp_path/"mail.db")),Provider(),"opaque","https://users.example")
|
|
assert invoke(app,event(),token="bad")[0].startswith("401")
|
|
bad=event(); bad["data"]["primary_email"]="bad\n@example.test"; assert invoke(app,bad)[1]["error"]=="invalid_recipient"
|
|
bad=event(); bad["type"]="arbitrary.send"; assert invoke(app,bad)[1]["error"]=="template_not_allowed"
|
|
assert invoke(app,event(),key="other")[1]["error"]=="idempotency_key_mismatch"
|