Add tests that all events have all the info we need (#2211)

* Add tests that all events have all the info we need

* Reorder

* Fix import
This commit is contained in:
Adrià Casajús 2024-09-10 11:05:24 +02:00 committed by GitHub
parent 7856706da1
commit b61a171de3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 155 additions and 30 deletions

View File

@ -30,14 +30,30 @@ class PostgresDispatcher(Dispatcher):
return PostgresDispatcher() return PostgresDispatcher()
class GlobalDispatcher:
__dispatcher: Optional[Dispatcher] = None
@staticmethod
def get_dispatcher() -> Dispatcher:
if not GlobalDispatcher.__dispatcher:
GlobalDispatcher.__dispatcher = PostgresDispatcher.get()
return GlobalDispatcher.__dispatcher
@staticmethod
def set_dispatcher(dispatcher: Optional[Dispatcher]):
GlobalDispatcher.__dispatcher = dispatcher
class EventDispatcher: class EventDispatcher:
@staticmethod @staticmethod
def send_event( def send_event(
user: User, user: User,
content: event_pb2.EventContent, content: event_pb2.EventContent,
dispatcher: Dispatcher = PostgresDispatcher.get(), dispatcher: Optional[Dispatcher] = None,
skip_if_webhook_missing: bool = True, skip_if_webhook_missing: bool = True,
): ):
if dispatcher is None:
dispatcher = GlobalDispatcher.get_dispatcher()
if config.EVENT_WEBHOOK_DISABLE: if config.EVENT_WEBHOOK_DISABLE:
LOG.i("Not sending events because webhook is disabled") LOG.i("Not sending events because webhook is disabled")
return return

View File

@ -0,0 +1,32 @@
from app.events.event_dispatcher import Dispatcher
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 clear(self):
self.memory = []
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

View File

@ -1,33 +1,10 @@
from app.events.event_dispatcher import EventDispatcher, Dispatcher from app.events.event_dispatcher import EventDispatcher
from app.events.generated.event_pb2 import EventContent, UserDeleted from app.events.generated.event_pb2 import EventContent, UserDeleted
from app.models import PartnerUser, User from .event_test_utils import (
from app.proton.utils import get_proton_partner _create_unlinked_user,
from tests.utils import create_new_user, random_token OnMemoryDispatcher,
from typing import Tuple _create_linked_user,
)
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(): def test_event_dispatcher_stores_events():

View File

@ -0,0 +1,100 @@
from app import config, alias_utils
from app.db import Session
from app.events.event_dispatcher import GlobalDispatcher
from app.events.generated import event_pb2
from app.models import Alias, User, PartnerUser
from tests.utils import random_token
from .event_test_utils import (
OnMemoryDispatcher,
_create_linked_user,
)
on_memory_dispatcher = OnMemoryDispatcher()
def setup_module():
GlobalDispatcher.set_dispatcher(on_memory_dispatcher)
config.EVENT_WEBHOOK = "http://test"
def teardown_module():
GlobalDispatcher.set_dispatcher(None)
config.EVENT_WEBHOOK = None
def setup_function(func):
on_memory_dispatcher.clear()
def _get_event_from_string(
data: str, user: User, pu: PartnerUser
) -> event_pb2.EventContent:
event = event_pb2.Event()
event.ParseFromString(data)
assert user.id == event.user_id
assert pu.external_user_id == event.external_user_id
assert pu.partner_id == event.partner_id
return event.content
def test_fire_event_on_alias_creation():
(user, pu) = _create_linked_user()
alias = Alias.create_new_random(user)
Session.flush()
assert len(on_memory_dispatcher.memory) == 1
event_data = on_memory_dispatcher.memory[0]
event_content = _get_event_from_string(event_data, user, pu)
assert event_content.alias_created is not None
alias_created = event_content.alias_created
assert alias.id == alias_created.alias_id
assert alias.email == alias_created.alias_email
assert "" == alias_created.alias_note
assert alias.enabled == alias_created.enabled
def test_fire_event_on_alias_creation_with_note():
(user, pu) = _create_linked_user()
note = random_token(10)
alias = Alias.create_new_random(user, note=note)
Session.flush()
assert len(on_memory_dispatcher.memory) == 1
event_data = on_memory_dispatcher.memory[0]
event_content = _get_event_from_string(event_data, user, pu)
assert event_content.alias_created is not None
alias_created = event_content.alias_created
assert alias.id == alias_created.alias_id
assert alias.email == alias_created.alias_email
assert note == alias_created.alias_note
assert alias.enabled == alias_created.enabled
def test_fire_event_on_alias_deletion():
(user, pu) = _create_linked_user()
alias = Alias.create_new_random(user)
alias_id = alias.id
Session.flush()
on_memory_dispatcher.clear()
alias_utils.delete_alias(alias, user)
assert len(on_memory_dispatcher.memory) == 1
event_data = on_memory_dispatcher.memory[0]
event_content = _get_event_from_string(event_data, user, pu)
assert event_content.alias_deleted is not None
alias_deleted = event_content.alias_deleted
assert alias_id == alias_deleted.alias_id
assert alias.email == alias_deleted.alias_email
def test_fire_event_on_alias_status_change():
(user, pu) = _create_linked_user()
alias = Alias.create_new_random(user)
Session.flush()
on_memory_dispatcher.clear()
alias_utils.change_alias_status(alias, True)
assert len(on_memory_dispatcher.memory) == 1
event_data = on_memory_dispatcher.memory[0]
event_content = _get_event_from_string(event_data, user, pu)
assert event_content.alias_status_change is not None
event = event_content.alias_status_change
assert alias.id == event.alias_id
assert alias.email == event.alias_email
assert event.enabled