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

11 super().__init__(xmpp) 

12 xmpp.register_handler( 

13 CoroutineCallback( 

14 "MUCOwnerGet", 

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

16 self.on_muc_owner_query, 

17 ) 

18 ) 

19 xmpp.register_handler( 

20 CoroutineCallback( 

21 "MUCOwnerSet", 

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

23 self.on_muc_owner_set, 

24 ) 

25 ) 

26 

27 @exceptions_to_xmpp_errors 

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

29 assert isinstance(iq, Iq) 

30 muc = await self.get_muc_from_stanza(iq) 

31 

32 reply = iq.reply() 

33 

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

35 form["instructions"] = ( 

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

37 ) 

38 form.add_field( 

39 var="FORM_TYPE", 

40 type="hidden", 

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

42 ) 

43 form.add_field( 

44 var="muc#roomconfig_roomname", 

45 label="Natural-Language Room Name", 

46 type="text-single", 

47 value=muc.name, 

48 ) 

49 if muc.HAS_DESCRIPTION: 

50 form.add_field( 

51 var="muc#roomconfig_roomdesc", 

52 label="Short Description of Room", 

53 type="text-single", 

54 value=muc.description, 

55 ) 

56 

57 muc_owner = iq["mucowner_query"] 

58 muc_owner.append(form) 

59 reply.append(muc_owner) 

60 reply.send() 

61 

62 @exceptions_to_xmpp_errors 

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

64 assert isinstance(iq, Iq) 

65 muc = await self.get_muc_from_stanza(iq) 

66 query = iq["mucowner_query"] 

67 

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

69 values = form.get_values() 

70 await muc.on_set_config( 

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

72 description=( 

73 values.get("muc#roomconfig_roomdesc") 

74 if muc.HAS_DESCRIPTION 

75 else None 

76 ), 

77 ) 

78 form["type"] = "result" 

79 clear = False 

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

81 reason = destroy["reason"] or None 

82 await muc.on_destroy_request(reason) 

83 user_participant = await muc.get_user_participant() 

84 user_participant._affiliation = "none" 

85 user_participant._role = "none" 

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

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

88 if reason is not None: 

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

90 user_participant._send(presence) 

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

92 clear = True 

93 else: 

94 raise XMPPError("bad-request") 

95 

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