mirror of
https://github.com/simple-login/app.git
synced 2024-09-20 15:05:59 +08:00
3e0b7bb369
* 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
57 lines
1.8 KiB
Python
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
|