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
|
from slidge import BaseGateway, BaseSession, LegacyContact, LegacyMUC, LegacyBookmarks, LegacyRoster, LegacyParticipant
from slidge.util.test import SlidgeTest
from slixmpp.exceptions import XMPPError
class Gateway(BaseGateway):
COMPONENT_NAME = "Megacorp Chat XMPP Gateway"
GROUPS = True
class Session(BaseSession):
async def login(self):
pass
class Contact(LegacyContact):
pass
class Roster(LegacyRoster):
async def legacy_id_to_jid_username(self, legacy_id: str) -> str:
if not legacy_id.startswith("room-"):
raise XMPPError
class Bookmarks(LegacyBookmarks):
async def fill(self):
pass
class TestSpaces(SlidgeTest):
plugin = globals()
def setUp(self):
super().setUp()
self._add_slidge_user()
def test_archive_correction_no_event_id(self) -> None:
muc = self.get_joined_muc("room-correction")
part: LegacyParticipant = self.run_coro(muc.get_participant("part"))
part.send_text("a body", "a-legacy-id")
self.next_sent() # part presence
msg = self.next_sent()
assert msg["body"] == "a body", msg
assert msg["stanza_id"]["id"] == "a-legacy-id", msg
part.correct("a-legacy-id", "a new body")
msg = self.next_sent()
assert msg["body"] == "a new body", msg
correction_stanza_id = msg["stanza_id"]["id"]
messages = list(muc.archive.get_all(before_id=correction_stanza_id))
assert len(messages) == 1
assert messages[0]
def test_archive_correction_specify_event_id(self) -> None:
muc = self.get_joined_muc("room-correction")
part: LegacyParticipant = self.run_coro(muc.get_participant("part"))
part.send_text("a body", "a-legacy-id")
self.next_sent() # part presence
msg = self.next_sent()
assert msg["body"] == "a body", msg
assert msg["stanza_id"]["id"] == "a-legacy-id", msg
part.correct("a-legacy-id", "a new body", correction_event_id="another-legacy-id")
msg = self.next_sent()
assert msg["body"] == "a new body", msg
correction_stanza_id = msg["stanza_id"]["id"]
assert correction_stanza_id == "another-legacy-id"
messages = list(muc.archive.get_all(before_id=correction_stanza_id))
assert len(messages) == 1
assert messages[0]
messages = list(muc.get_archived_messages(correction_stanza_id))
assert len(messages) == 1
def test_archive_correction_specify_existing_event_id(self) -> None:
muc = self.get_joined_muc("room-correction")
part: LegacyParticipant = self.run_coro(muc.get_participant("part"))
part.send_text("a body", "a-legacy-id")
self.next_sent() # part presence
msg = self.next_sent()
assert msg["body"] == "a body", msg
assert msg["stanza_id"]["id"] == "a-legacy-id", msg
part.correct("a-legacy-id", "a new body", correction_event_id="a-legacy-id")
msg = self.next_sent()
assert msg["body"] == "a new body", msg
correction_stanza_id = msg["stanza_id"]["id"]
assert correction_stanza_id == "a-legacy-id"
messages = list(muc.archive.get_all(before_id=correction_stanza_id))
assert not messages
|
