app-MAIL-temp/tests/events/test_event_dispatcher.py
Carlos Quintana 3e0b7bb369
Add sync events (#2113)
* feat: add protocol buffers for events

* chore: add EventDispatcher

* chore: add WebhookEvent class

* chore: emit events

* feat: initial version of event listener

* chore: emit user plan change with new timestamp

* feat: emit metrics + add alias status to create event

* chore: add newrelic decorator to functions

* fix: event emitter fixes

* fix: take null end_time into account

* fix: avoid double-commits

* chore: move UserDeleted event to User.delete method

* db: add index to sync_event created_at and taken_time columns

* chore: add index to model
2024-05-23 10:27:08 +02:00

57 lines
1.8 KiB
Python

from app.events.event_dispatcher import EventDispatcher, Dispatcher
from app.events.generated.event_pb2 import EventContent, UserDeleted
from app.models import PartnerUser, User
from app.proton.utils import get_proton_partner
from tests.utils import create_new_user, random_token
from typing import Tuple
class OnMemoryDispatcher(Dispatcher):
def __init__(self):
self.memory = []
def send(self, event: bytes):
self.memory.append(event)
def _create_unlinked_user() -> User:
return create_new_user()
def _create_linked_user() -> Tuple[User, PartnerUser]:
user = _create_unlinked_user()
partner_user = PartnerUser.create(
partner_id=get_proton_partner().id,
user_id=user.id,
external_user_id=random_token(10),
flush=True,
)
return user, partner_user
def test_event_dispatcher_stores_events():
dispatcher = OnMemoryDispatcher()
(user, partner) = _create_linked_user()
content = EventContent(user_deleted=UserDeleted())
EventDispatcher.send_event(user, content, dispatcher, skip_if_webhook_missing=False)
assert len(dispatcher.memory) == 1
content = EventContent(user_deleted=UserDeleted())
EventDispatcher.send_event(user, content, dispatcher, skip_if_webhook_missing=False)
assert len(dispatcher.memory) == 2
def test_event_dispatcher_does_not_send_event_if_user_not_linked():
dispatcher = OnMemoryDispatcher()
user = _create_unlinked_user()
content = EventContent(user_deleted=UserDeleted())
EventDispatcher.send_event(user, content, dispatcher, skip_if_webhook_missing=False)
assert len(dispatcher.memory) == 0
content = EventContent(user_deleted=UserDeleted())
EventDispatcher.send_event(user, content, dispatcher, skip_if_webhook_missing=False)
assert len(dispatcher.memory) == 0