firebase-functions-python/tests/test_firestore_fn.py at 689b564234fc4d082a890ec0bb7e576a96a2cd1a · firebase/firebase-functions-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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
"""
This module contains tests for the firestore_fn module.
"""

import json
from unittest import TestCase
from unittest.mock import MagicMock, Mock, patch

mocked_modules = {
    "google.cloud.firestore": MagicMock(),
    "google.cloud.firestore_v1": MagicMock(),
    "firebase_admin": MagicMock(),
}


class TestFirestore(TestCase):
    """
    firestore_fn tests.
    """

    def test_firestore_decorator_retry_option(self):
        with patch.dict("sys.modules", mocked_modules):
            from firebase_functions import firestore_fn

            func = Mock(__name__="example_func")
            decorated_func = firestore_fn.on_document_created(
                document="/foo/{bar}",
                retry=True,
            )(func)

            endpoint = decorated_func.__firebase_endpoint__

            self.assertIsNotNone(endpoint.eventTrigger)
            self.assertTrue(endpoint.eventTrigger["retry"])

    def test_firestore_decorator_retry_defaults_false(self):
        with patch.dict("sys.modules", mocked_modules):
            from firebase_functions import firestore_fn

            func = Mock(__name__="example_func")
            decorated_func = firestore_fn.on_document_created(document="/foo/{bar}")(func)

            endpoint = decorated_func.__firebase_endpoint__

            self.assertIsNotNone(endpoint.eventTrigger)
            self.assertFalse(endpoint.eventTrigger["retry"])

    def test_firestore_decorator_omits_empty_path_patterns(self):
        with patch.dict("sys.modules", mocked_modules):
            from firebase_functions import firestore_fn

            func = Mock(__name__="example_func")
            decorated_func = firestore_fn.on_document_created(document="/foo/bar")(func)

            endpoint = decorated_func.__firebase_endpoint__

            self.assertIsNotNone(endpoint.eventTrigger)
            self.assertNotIn("eventFilterPathPatterns", endpoint.eventTrigger)

    def test_firestore_endpoint_handler_calls_function_with_correct_args(self):
        with patch.dict("sys.modules", mocked_modules):
            from cloudevents.http import CloudEvent

            from firebase_functions.firestore_fn import (
                AuthEvent,
            )
            from firebase_functions.firestore_fn import (
                _event_type_created_with_auth_context as event_type,
            )
            from firebase_functions.firestore_fn import (
                _firestore_endpoint_handler as firestore_endpoint_handler,
            )
            from firebase_functions.private import path_pattern

            func = Mock(__name__="example_func")

            document_pattern = path_pattern.PathPattern("foo/{bar}")
            attributes = {
                "specversion": "1.0",
                "type": event_type,
                "source": "https://example.com/testevent",
                "time": "2023-03-11T13:25:37.403Z",
                "subject": "test_subject",
                "datacontenttype": "application/json",
                "location": "projects/project-id/databases/(default)/documents/foo/{bar}",
                "project": "project-id",
                "namespace": "(default)",
                "document": "foo/{bar}",
                "database": "projects/project-id/databases/(default)",
                "authtype": "unauthenticated",
                "authid": "foo",
            }
            raw_event = CloudEvent(attributes=attributes, data=json.dumps({}))

            firestore_endpoint_handler(
                func=func, event_type=event_type, document_pattern=document_pattern, raw=raw_event
            )

            func.assert_called_once()

            event = func.call_args.args[0]
            self.assertIsNotNone(event)
            self.assertIsInstance(event, AuthEvent)
            self.assertEqual(event.auth_type, "unauthenticated")
            self.assertEqual(event.auth_id, "foo")

    def test_calls_init_function(self):
        with patch.dict("sys.modules", mocked_modules):
            from cloudevents.http import CloudEvent

            from firebase_functions import core, firestore_fn

            func = Mock(__name__="example_func")

            hello = None

            @core.init
            def init():
                nonlocal hello
                hello = "world"

            attributes = {
                "specversion": "1.0",
                # pylint: disable=protected-access
                "type": firestore_fn._event_type_created,
                "source": "https://example.com/testevent",
                "time": "2023-03-11T13:25:37.403Z",
                "subject": "test_subject",
                "datacontenttype": "application/json",
                "location": "projects/project-id/databases/(default)/documents/foo/{bar}",
                "project": "project-id",
                "namespace": "(default)",
                "document": "foo/{bar}",
                "database": "projects/project-id/databases/(default)",
                "authtype": "unauthenticated",
                "authid": "foo",
            }
            raw_event = CloudEvent(attributes=attributes, data=json.dumps({}))
            decorated_func = firestore_fn.on_document_created(document="/foo/{bar}")(func)

            decorated_func(raw_event)

            self.assertEqual(hello, "world")