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
|
import datetime
import unittest.mock
import sqlalchemy as sa
from slixmpp import Message
from slidge import BaseGateway, BaseSession
from slidge.core.session import _sessions
from slidge.db.models import ArchivedMessage
from slidge.util.archive_msg import HistoryMessage
from slidge.util.test import SlidgeTest
class Gateway(BaseGateway):
COMPONENT_NAME = "A test"
class Session(BaseSession):
async def login(self):
return "YUP"
class TestBackfill(SlidgeTest):
plugin = globals()
xmpp: Gateway
def setUp(self):
super().setUp()
self.setup_logged_session()
self.xmpp.LEGACY_MSG_ID_TYPE = int
def tearDown(self):
with self.xmpp.store.session() as orm:
orm.execute(sa.delete(ArchivedMessage))
super().tearDown()
def test_empty_archive(self):
with unittest.mock.patch("slidge.group.LegacyMUC.backfill") as backfill:
self.run_coro(self.room._LegacyMUC__fill_history())
backfill.assert_awaited_with(None, None)
def test_live_no_id_before_backfill(self):
self.first_witch.send_text("BODY 1")
self.first_witch.send_text("BODY 2")
self.first_witch.send_text(
"BODY 3", when=datetime.datetime.now(tz=datetime.timezone.utc)
)
with unittest.mock.patch("slidge.group.LegacyMUC.backfill") as backfill:
self.run_coro(self.room._LegacyMUC__fill_history())
backfill.assert_awaited_with(None, None)
def test_live_with_id_before_backfill(self):
now = datetime.datetime.now(tz=datetime.timezone.utc)
self.first_witch.send_text("BODY 2", 222, when=now)
self.first_witch.send_text(
"BODY 1", 111, when=now - datetime.timedelta(hours=1)
)
self.first_witch.send_text(
"BODY 3", 333, when=now + datetime.timedelta(hours=1)
)
with unittest.mock.patch("slidge.group.LegacyMUC.backfill") as backfill:
self.run_coro(self.room._LegacyMUC__fill_history())
backfill.assert_awaited_once()
after, before = backfill.call_args[0]
assert before.id == 111
assert before.timestamp == now - datetime.timedelta(hours=1)
assert after is None
def _add_back_filled_msg(self, legacy_id=None, when=None):
with self.xmpp.store.session() as orm:
self.xmpp.store.mam.add_message(
orm,
self.room.stored.id,
HistoryMessage(Message(), when),
archive_only=True,
legacy_msg_id=legacy_id,
)
orm.commit()
def test_pre_backfilled_no_id(self):
self._add_back_filled_msg()
with unittest.mock.patch("slidge.group.LegacyMUC.backfill") as backfill:
self.run_coro(self.room._LegacyMUC__fill_history())
backfill.assert_awaited_with(None, None)
def test_pre_backfilled_with_id(self):
self._add_back_filled_msg(None)
self._add_back_filled_msg(111)
self._add_back_filled_msg(222)
self._add_back_filled_msg(None)
with unittest.mock.patch("slidge.group.LegacyMUC.backfill") as backfill:
self.run_coro(self.room._LegacyMUC__fill_history())
backfill.assert_awaited_once()
after, before = backfill.call_args[0]
assert before is None
assert after is not None
assert after.id == 222
def test_pre_backfilled_with_id_and_live(self):
now = datetime.datetime.now(tz=datetime.timezone.utc)
self._add_back_filled_msg(None, now - datetime.timedelta(days=5))
self._add_back_filled_msg(111, now - datetime.timedelta(days=4))
self._add_back_filled_msg(222, now - datetime.timedelta(days=3))
self._add_back_filled_msg(None, now - datetime.timedelta(days=2))
self.first_witch.send_text("BODY1", None)
self.first_witch.send_text("BODY2", 555)
self.first_witch.send_text("BODY3", None)
self.first_witch.send_text("BODY5", 666)
self.first_witch.send_text("BODY6", None)
self.first_witch.send_text("BODY7", None)
with unittest.mock.patch("slidge.group.LegacyMUC.backfill") as backfill:
self.run_coro(self.room._LegacyMUC__fill_history())
backfill.assert_awaited_once()
after, before = backfill.call_args[0]
assert before.id == 555
assert after.id == 222
|
