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/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 moderate = iq["moderate"] 

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

40 if not xmpp_id: 

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

42 

43 if not moderate["retract"]: 

44 raise XMPPError( 

45 "feature-not-implemented", 

46 "Slidge only implements moderation/retraction", 

47 ) 

48 

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

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

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

52 

53 @exceptions_to_xmpp_errors 

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

55 assert isinstance(iq, Iq) 

56 muc = await self.get_muc_from_stanza(iq) 

57 

58 item = iq["mucadmin_query"]["item"] 

59 if item["jid"]: 

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

61 else: 

62 part = await muc.get_participant( 

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

64 ) 

65 assert part.contact is not None 

66 contact = part.contact 

67 

68 if item["affiliation"]: 

69 await muc.on_set_affiliation( 

70 contact, 

71 item["affiliation"], 

72 item["reason"] or None, 

73 item["nick"] or None, 

74 ) 

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

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

77 

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

79 

80 @exceptions_to_xmpp_errors 

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

82 assert isinstance(iq, Iq) 

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

84 

85 if not affiliation: 

86 raise XMPPError("bad-request") 

87 

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

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

90 

91 reply = iq.reply() 

92 reply.enable("mucadmin_query") 

93 async for participant in muc.get_participants(): 

94 if not participant.affiliation == affiliation: 

95 continue 

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

97 reply.send()