workos-python/tests/test_events.py at dc2e04a8604f34f08aecc3baf3cf200e2ade997b · workos/workos-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
from typing import Union
import pytest

from tests.utils.fixtures.mock_event import MockEvent
from tests.utils.syncify import syncify
from workos.events import AsyncEvents, Events, EventsListResource
from workos.types.events import OrganizationMembershipCreatedEvent


@pytest.mark.sync_and_async(Events, AsyncEvents)
class TestEvents(object):
    @pytest.fixture
    def mock_events(self):
        events = [MockEvent(id=str(i)).dict() for i in range(10)]

        return {
            "object": "list",
            "data": events,
            "list_metadata": {
                "after": None,
            },
        }

    def test_list_events(
        self,
        module_instance: Union[Events, AsyncEvents],
        mock_events: EventsListResource,
        capture_and_mock_http_client_request,
    ):
        request_kwargs = capture_and_mock_http_client_request(
            http_client=module_instance._http_client,
            status_code=200,
            response_dict=mock_events,
        )

        events: EventsListResource = syncify(
            module_instance.list_events(events=["dsync.activated"])
        )

        assert request_kwargs["url"].endswith("/events")
        assert request_kwargs["method"] == "get"
        assert request_kwargs["params"] == {"events": ["dsync.activated"], "limit": 10}
        assert events.dict() == mock_events

    def test_list_events_organization_membership_missing_custom_attributes(
        self,
        module_instance: Union[Events, AsyncEvents],
        capture_and_mock_http_client_request,
    ):
        mock_response = {
            "object": "list",
            "data": [
                {
                    "object": "event",
                    "id": "event_01234",
                    "event": "organization_membership.created",
                    "data": {
                        "object": "organization_membership",
                        "id": "om_01234",
                        "user_id": "user_01234",
                        "organization_id": "org_01234",
                        "role": {"slug": "member"},
                        "status": "active",
                        "created_at": "2024-01-01T00:00:00.000Z",
                        "updated_at": "2024-01-01T00:00:00.000Z",
                    },
                    "created_at": "2024-01-01T00:00:00.000Z",
                }
            ],
            "list_metadata": {
                "after": None,
            },
        }

        capture_and_mock_http_client_request(
            http_client=module_instance._http_client,
            status_code=200,
            response_dict=mock_response,
        )

        events: EventsListResource = syncify(
            module_instance.list_events(
                events=["organization_membership.created"]
            )
        )

        event = events.data[0]
        assert isinstance(event, OrganizationMembershipCreatedEvent)
        assert event.data.custom_attributes == {}