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 def __init__(self, xmpp) -> None: 

10 super().__init__(xmpp) 

11 self.xmpp.register_handler( 

12 CoroutineCallback( 

13 "MUCModerate", 

14 StanzaPath("iq/apply_to/moderate"), 

15 self.on_user_moderation, 

16 ) 

17 ) 

18 self.xmpp.register_handler( 

19 CoroutineCallback( 

20 "MUCSetAffiliation", 

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

22 self.on_user_set_affiliation, 

23 ) 

24 ) 

25 self.xmpp.register_handler( 

26 CoroutineCallback( 

27 "MUCGetAffiliation", 

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

29 self.on_muc_admin_query_get, 

30 ) 

31 ) 

32 

33 @exceptions_to_xmpp_errors 

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

35 assert isinstance(iq, Iq) 

36 muc = await self.get_muc_from_stanza(iq) 

37 

38 apply_to = iq["apply_to"] 

39 xmpp_id = apply_to["id"] 

40 if not xmpp_id: 

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

42 

43 moderate = apply_to["moderate"] 

44 if not moderate["retract"]: 

45 raise XMPPError( 

46 "feature-not-implemented", 

47 "Slidge only implements moderation/retraction", 

48 ) 

49 

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

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

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

53 

54 @exceptions_to_xmpp_errors 

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

56 assert isinstance(iq, Iq) 

57 muc = await self.get_muc_from_stanza(iq) 

58 

59 item = iq["mucadmin_query"]["item"] 

60 if item["jid"]: 

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

62 else: 

63 part = await muc.get_participant( 

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

65 ) 

66 assert part.contact is not None 

67 contact = part.contact 

68 

69 if item["affiliation"]: 

70 await muc.on_set_affiliation( 

71 contact, 

72 item["affiliation"], 

73 item["reason"] or None, 

74 item["nick"] or None, 

75 ) 

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

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

78 

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

80 

81 @exceptions_to_xmpp_errors 

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

83 assert isinstance(iq, Iq) 

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

85 

86 if not affiliation: 

87 raise XMPPError("bad-request") 

88 

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

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

91 

92 reply = iq.reply() 

93 reply.enable("mucadmin_query") 

94 async for participant in muc.get_participants(): 

95 if not participant.affiliation == affiliation: 

96 continue 

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

98 reply.send()