summary refs log tree commit diff
path: root/tests/server_notices/test_resource_limits_server_notices.py
blob: b1551df7ca01fd63783c8ca675dd4e3f5e449504 (plain) (blame)
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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
from mock import Mock

from twisted.internet import defer

from synapse.api.constants import EventTypes, ServerNoticeMsgType
from synapse.api.errors import ResourceLimitError
from synapse.server_notices.resource_limits_server_notices import (
    ResourceLimitsServerNotices,
)

from tests import unittest
from tests.utils import setup_test_homeserver


class TestResourceLimitsServerNotices(unittest.TestCase):
    @defer.inlineCallbacks
    def setUp(self):
        self.hs = yield setup_test_homeserver(self.addCleanup)
        self.server_notices_sender = self.hs.get_server_notices_sender()

        # relying on [1] is far from ideal, but the only case where
        # ResourceLimitsServerNotices class needs to be isolated is this test,
        # general code should never have a reason to do so ...
        self._rlsn = self.server_notices_sender._server_notices[1]
        if not isinstance(self._rlsn, ResourceLimitsServerNotices):
            raise Exception("Failed to find reference to ResourceLimitsServerNotices")

        self._rlsn._store.user_last_seen_monthly_active = Mock(
            return_value=defer.succeed(1000)
        )
        self._send_notice = self._rlsn._server_notices_manager.send_notice
        self._rlsn._server_notices_manager.send_notice = Mock()
        self._rlsn._state.get_current_state = Mock(return_value=defer.succeed(None))
        self._rlsn._store.get_events = Mock(return_value=defer.succeed({}))

        self._send_notice = self._rlsn._server_notices_manager.send_notice

        self.hs.config.limit_usage_by_mau = True
        self.user_id = "@user_id:test"

        # self.server_notices_mxid = "@server:test"
        # self.server_notices_mxid_display_name = None
        # self.server_notices_mxid_avatar_url = None
        # self.server_notices_room_name = "Server Notices"

        self._rlsn._server_notices_manager.get_notice_room_for_user = Mock(
            returnValue=""
        )
        self._rlsn._store.add_tag_to_room = Mock()
        self._rlsn._store.get_tags_for_room = Mock(return_value={})
        self.hs.config.admin_contact = "mailto:user@test.com"

    @defer.inlineCallbacks
    def test_maybe_send_server_notice_to_user_flag_off(self):
        """Tests cases where the flags indicate nothing to do"""
        # test hs disabled case
        self.hs.config.hs_disabled = True

        yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)

        self._send_notice.assert_not_called()
        # Test when mau limiting disabled
        self.hs.config.hs_disabled = False
        self.hs.limit_usage_by_mau = False
        yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)

        self._send_notice.assert_not_called()

    @defer.inlineCallbacks
    def test_maybe_send_server_notice_to_user_remove_blocked_notice(self):
        """Test when user has blocked notice, but should have it removed"""

        self._rlsn._auth.check_auth_blocking = Mock()
        mock_event = Mock(
            type=EventTypes.Message, content={"msgtype": ServerNoticeMsgType}
        )
        self._rlsn._store.get_events = Mock(
            return_value=defer.succeed({"123": mock_event})
        )

        yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
        # Would be better to check the content, but once == remove blocking event
        self._send_notice.assert_called_once()

    @defer.inlineCallbacks
    def test_maybe_send_server_notice_to_user_remove_blocked_notice_noop(self):
        """Test when user has blocked notice, but notice ought to be there (NOOP)"""
        self._rlsn._auth.check_auth_blocking = Mock(
            side_effect=ResourceLimitError(403, 'foo')
        )

        mock_event = Mock(
            type=EventTypes.Message, content={"msgtype": ServerNoticeMsgType}
        )
        self._rlsn._store.get_events = Mock(
            return_value=defer.succeed({"123": mock_event})
        )
        yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)

        self._send_notice.assert_not_called()

    @defer.inlineCallbacks
    def test_maybe_send_server_notice_to_user_add_blocked_notice(self):
        """Test when user does not have blocked notice, but should have one"""

        self._rlsn._auth.check_auth_blocking = Mock(
            side_effect=ResourceLimitError(403, 'foo')
        )
        yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)

        # Would be better to check contents, but 2 calls == set blocking event
        self.assertTrue(self._send_notice.call_count == 2)

    @defer.inlineCallbacks
    def test_maybe_send_server_notice_to_user_add_blocked_notice_noop(self):
        """Test when user does not have blocked notice, nor should they (NOOP)"""

        self._rlsn._auth.check_auth_blocking = Mock()

        yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)

        self._send_notice.assert_not_called()

    @defer.inlineCallbacks
    def test_maybe_send_server_notice_to_user_not_in_mau_cohort(self):

        """Test when user is not part of the MAU cohort - this should not ever
        happen - but ...
        """

        self._rlsn._auth.check_auth_blocking = Mock()
        self._rlsn._store.user_last_seen_monthly_active = Mock(
            return_value=defer.succeed(None)
        )
        yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)

        self._send_notice.assert_not_called()


class TestResourceLimitsServerNoticesWithRealRooms(unittest.TestCase):
    @defer.inlineCallbacks
    def setUp(self):
        self.hs = yield setup_test_homeserver(self.addCleanup)
        self.store = self.hs.get_datastore()
        self.server_notices_sender = self.hs.get_server_notices_sender()
        self.server_notices_manager = self.hs.get_server_notices_manager()
        self.event_source = self.hs.get_event_sources()

        # relying on [1] is far from ideal, but the only case where
        # ResourceLimitsServerNotices class needs to be isolated is this test,
        # general code should never have a reason to do so ...
        self._rlsn = self.server_notices_sender._server_notices[1]
        if not isinstance(self._rlsn, ResourceLimitsServerNotices):
            raise Exception("Failed to find reference to ResourceLimitsServerNotices")

        self.hs.config.limit_usage_by_mau = True
        self.hs.config.hs_disabled = False
        self.hs.config.max_mau_value = 5
        self.hs.config.server_notices_mxid = "@server:test"
        self.hs.config.server_notices_mxid_display_name = None
        self.hs.config.server_notices_mxid_avatar_url = None
        self.hs.config.server_notices_room_name = "Test Server Notice Room"

        self.user_id = "@user_id:test"

        self.hs.config.admin_contact = "mailto:user@test.com"

    @defer.inlineCallbacks
    def test_server_notice_only_sent_once(self):
        self.store.get_monthly_active_count = Mock(return_value=1000)

        self.store.user_last_seen_monthly_active = Mock(return_value=1000)

        # Call the function multiple times to ensure we only send the notice once
        yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
        yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
        yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)

        # Now lets get the last load of messages in the service notice room and
        # check that there is only one server notice
        room_id = yield self.server_notices_manager.get_notice_room_for_user(
            self.user_id
        )

        token = yield self.event_source.get_current_token()
        events, _ = yield self.store.get_recent_events_for_room(
            room_id, limit=100, end_token=token.room_key
        )

        count = 0
        for event in events:
            if event.type != EventTypes.Message:
                continue
            if event.content.get("msgtype") != ServerNoticeMsgType:
                continue

            count += 1

        self.assertEqual(count, 1)