Coverage for slidge/core/dispatcher/muc/admin.py: 94%

1from slixmpp import JID, CoroutineCallback, Iq, StanzaPath 

2from slixmpp.exceptions import XMPPError 

3from slixmpp.xmlstream import StanzaBase 

4 

5from ..util import DispatcherMixin, exceptions_to_xmpp_errors 

6 

7 

8class MucAdminMixin(DispatcherMixin): 

9 __slots__: list[str] = [] 

10 

11 def __init__(self, xmpp) -> None: 

12 super().__init__(xmpp) 

13 self.xmpp.register_handler( 

14 CoroutineCallback( 

15 "MUCModerate", 

16 StanzaPath("iq/moderate"), 

17 self.on_user_moderation, 

18 ) 

19 ) 

20 self.xmpp.register_handler( 

21 CoroutineCallback( 

22 "MUCSetAffiliation", 

23 StanzaPath("iq@type=set/mucadmin_query"), 

24 self.on_user_set_affiliation, 

25 ) 

26 ) 

27 self.xmpp.register_handler( 

28 CoroutineCallback( 

29 "MUCGetAffiliation", 

30 StanzaPath("iq@type=get/mucadmin_query"), 

31 self.on_muc_admin_query_get, 

32 ) 

33 ) 

34 

35 @exceptions_to_xmpp_errors 

36 async def on_user_moderation(self, iq: StanzaBase) -> None: 

37 assert isinstance(iq, Iq) 

38 muc = await self.get_muc_from_stanza(iq) 

39 

40 moderate = iq["moderate"] 

41 xmpp_id = iq["moderate"]["id"] 

42 if not xmpp_id: 

43 raise XMPPError("bad-request", "Missing moderated message ID") 

44 

45 if not moderate["retract"]: 

46 raise XMPPError( 

47 "feature-not-implemented", 

48 "Slidge only implements moderation/retraction", 

49 ) 

50 

51 legacy_id = self._xmpp_msg_id_to_legacy(muc.session, xmpp_id, muc) 

52 await muc.session.on_moderate(muc, legacy_id, moderate["reason"] or None) 

53 iq.reply(clear=True).send() 

54 

55 @exceptions_to_xmpp_errors 

56 async def on_user_set_affiliation(self, iq: StanzaBase) -> None: 

57 assert isinstance(iq, Iq) 

58 muc = await self.get_muc_from_stanza(iq) 

59 

60 item = iq["mucadmin_query"]["item"] 

61 if item["jid"]: 

62 contact = await muc.session.contacts.by_jid(JID(item["jid"])) 

63 else: 

64 part = await muc.get_participant( 

65 item["nick"], fill_first=True, raise_if_not_found=True 

66 ) 

67 assert part.contact is not None 

68 contact = part.contact 

69 

70 if item["affiliation"]: 

71 await muc.on_set_affiliation( 

72 contact, 

73 item["affiliation"], 

74 item["reason"] or None, 

75 item["nick"] or None, 

76 ) 

77 elif item["role"] == "none": 

78 await muc.on_kick(contact, item["reason"] or None) 

79 

80 iq.reply(clear=True).send() 

81 

82 @exceptions_to_xmpp_errors 

83 async def on_muc_admin_query_get(self, iq: StanzaBase) -> None: 

84 assert isinstance(iq, Iq) 

85 affiliation = iq["mucadmin_query"]["item"]["affiliation"] 

86 

87 if not affiliation: 

88 raise XMPPError("bad-request") 

89 

90 session = await self._get_session(iq, 1, logged=True) 

91 muc = await session.bookmarks.by_jid(iq.get_to()) 

92 

93 reply = iq.reply() 

94 reply.enable("mucadmin_query") 

95 async for participant in muc.get_participants(affiliation): 

96 reply["mucadmin_query"].append(participant.mucadmin_item()) 

97 reply.send()