Coverage for slidge/core/dispatcher/muc/owner.py: 98%

1from slixmpp import CoroutineCallback, Iq, StanzaPath 

2from slixmpp.exceptions import XMPPError 

3from slixmpp.plugins.xep_0004 import Form 

4from slixmpp.xmlstream import StanzaBase 

5 

6from ..util import DispatcherMixin, exceptions_to_xmpp_errors 

7 

8 

9class MucOwnerMixin(DispatcherMixin): 

10 __slots__: list[str] = [] 

11 

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

13 super().__init__(xmpp) 

14 xmpp.register_handler( 

15 CoroutineCallback( 

16 "MUCOwnerGet", 

17 StanzaPath("iq@type=get/mucowner_query"), 

18 self.on_muc_owner_query, 

19 ) 

20 ) 

21 xmpp.register_handler( 

22 CoroutineCallback( 

23 "MUCOwnerSet", 

24 StanzaPath("iq@type=set/mucowner_query"), 

25 self.on_muc_owner_set, 

26 ) 

27 ) 

28 

29 @exceptions_to_xmpp_errors 

30 async def on_muc_owner_query(self, iq: StanzaBase) -> None: 

31 assert isinstance(iq, Iq) 

32 muc = await self.get_muc_from_stanza(iq) 

33 

34 reply = iq.reply() 

35 

36 form = Form(title="Slidge room configuration") 

37 form["instructions"] = ( 

38 "Complete this form to modify the configuration of your room." 

39 ) 

40 form.add_field( 

41 var="FORM_TYPE", 

42 type="hidden", 

43 value="http://jabber.org/protocol/muc#roomconfig", 

44 ) 

45 form.add_field( 

46 var="muc#roomconfig_roomname", 

47 label="Natural-Language Room Name", 

48 type="text-single", 

49 value=muc.name, 

50 ) 

51 if muc.HAS_DESCRIPTION: 

52 form.add_field( 

53 var="muc#roomconfig_roomdesc", 

54 label="Short Description of Room", 

55 type="text-single", 

56 value=muc.description, 

57 ) 

58 

59 muc_owner = iq["mucowner_query"] 

60 muc_owner.append(form) 

61 reply.append(muc_owner) 

62 reply.send() 

63 

64 @exceptions_to_xmpp_errors 

65 async def on_muc_owner_set(self, iq: StanzaBase) -> None: 

66 assert isinstance(iq, Iq) 

67 muc = await self.get_muc_from_stanza(iq) 

68 query = iq["mucowner_query"] 

69 

70 if form := query.get_plugin("form", check=True): 

71 values = form.get_values() 

72 await muc.on_set_config( 

73 name=values.get("muc#roomconfig_roomname"), 

74 description=( 

75 values.get("muc#roomconfig_roomdesc") 

76 if muc.HAS_DESCRIPTION 

77 else None 

78 ), 

79 ) 

80 form["type"] = "result" 

81 clear = False 

82 elif destroy := query.get_plugin("destroy", check=True): 

83 reason = destroy["reason"] or None 

84 await muc.on_destroy_request(reason) 

85 user_participant = await muc.get_user_participant() 

86 user_participant.stored.affiliation = "none" 

87 user_participant.stored.role = "none" 

88 presence = user_participant._make_presence(ptype="unavailable", force=True) 

89 presence["muc"].enable("destroy") 

90 if reason is not None: 

91 presence["muc"]["destroy"]["reason"] = reason 

92 user_participant._send(presence) 

93 await muc.session.bookmarks.remove(muc, kick=False) 

94 clear = True 

95 else: 

96 raise XMPPError("bad-request") 

97 

98 iq.reply(clear=clear).send()