firebase-functions-python/tests/test_db.py at 9632e9cd2e92cce49e20f93856ea58d5f2662b2a · 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
"""
Tests for the db module.
"""

import datetime as dt
import unittest
from unittest import mock

from cloudevents.http import CloudEvent

from firebase_functions import core, db_fn


class TestDb(unittest.TestCase):
    """
    Tests for the db module.
    """

    def test_calls_init_function(self):
        hello = None

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

        func = mock.Mock(__name__="example_func")
        decorated_func = db_fn.on_value_created(reference="path")(func)

        event = CloudEvent(
            attributes={
                "specversion": "1.0",
                "id": "id",
                "source": "source",
                "subject": "subject",
                "type": "type",
                "time": "2024-04-10T12:00:00.000Z",
                "instance": "instance",
                "ref": "ref",
                "firebasedatabasehost": "firebasedatabasehost",
                "location": "location",
                "authtype": "app_user",
                "authid": "auth-id",
            },
            data={"delta": "delta"},
        )

        decorated_func(event)

        func.assert_called_once()
        event_arg = func.call_args.args[0]
        self.assertIsNotNone(event_arg)
        self.assertEqual(event_arg.auth_type, "app_user")
        self.assertEqual(event_arg.auth_id, "auth-id")

        self.assertEqual(hello, "world")

    def test_missing_auth_context(self):
        func = mock.Mock(__name__="example_func_no_auth")
        decorated_func = db_fn.on_value_created(reference="path")(func)

        event = CloudEvent(
            attributes={
                "specversion": "1.0",
                "id": "id",
                "source": "source",
                "subject": "subject",
                "type": "type",
                "time": "2024-04-10T12:00:00.000Z",
                "instance": "instance",
                "ref": "ref",
                "firebasedatabasehost": "firebasedatabasehost",
                "location": "location",
            },
            data={"delta": "delta"},
        )

        decorated_func(event)

        func.assert_called_once()
        event_arg = func.call_args.args[0]
        self.assertIsNotNone(event_arg)
        self.assertEqual(event_arg.auth_type, "unknown")
        self.assertIsNone(event_arg.auth_id)

    def test_written_event_parses_timestamp_without_microseconds(self):
        func = mock.Mock(__name__="example_func_no_microseconds")
        decorated_func = db_fn.on_value_written(reference="/items/{itemId}")(func)

        event = CloudEvent(
            attributes={
                "specversion": "1.0",
                "id": "issue-257-repro",
                "source": "//firebase.test/projects/demo-test/instances/my-instance/refs/items/123",
                "subject": "refs/items/123",
                "type": "google.firebase.database.ref.v1.written",
                "time": "2025-10-30T21:15:51Z",
                "instance": "my-instance",
                "ref": "/items/123",
                "firebasedatabasehost": "my-instance.firebaseio.com",
                "location": "location",
            },
            data={
                "data": {"existing": True},
                "delta": {"updated": True},
            },
        )

        decorated_func(event)

        func.assert_called_once()
        event_arg = func.call_args.args[0]
        self.assertEqual(
            event_arg.time,
            dt.datetime(2025, 10, 30, 21, 15, 51, tzinfo=dt.timezone.utc),
        )
        self.assertEqual(event_arg.data.after, {"existing": True, "updated": True})
        self.assertEqual(event_arg.params, {"itemId": "123"})