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

1import typing 

2 

3from slixmpp import CoroutineCallback, Iq, StanzaPath 

4from slixmpp.exceptions import XMPPError 

5from slixmpp.plugins.xep_0004.stanza.form import Form 

6from slixmpp.xmlstream import StanzaBase 

7 

8from ..util import DispatcherMixin, exceptions_to_xmpp_errors 

9 

10if typing.TYPE_CHECKING: 

11 from slidge.core.gateway import BaseGateway 

12 

13 

14class MucOwnerMixin(DispatcherMixin): 

15 __slots__: list[str] = [] 

16 

17 def __init__(self, xmpp: "BaseGateway") -> None: 

18 super().__init__(xmpp) 

19 xmpp.register_handler( 

20 CoroutineCallback( 

21 "MUCOwnerGet", 

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

23 self.on_muc_owner_query, 

24 ) 

25 ) 

26 xmpp.register_handler( 

27 CoroutineCallback( 

28 "MUCOwnerSet", 

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

30 self.on_muc_owner_set, 

31 ) 

32 ) 

33 

34 @exceptions_to_xmpp_errors 

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

36 assert isinstance(iq, Iq) 

37 muc = await self.get_muc_from_stanza(iq) 

38 

39 reply = iq.reply() 

40 

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

42 form["instructions"] = ( 

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

44 ) 

45 form.add_field( 

46 var="FORM_TYPE", 

47 type="hidden", 

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

49 ) 

50 form.add_field( 

51 var="muc#roomconfig_roomname", 

52 label="Natural-Language Room Name", 

53 type="text-single", 

54 value=muc.name, 

55 ) 

56 if muc.HAS_DESCRIPTION: 

57 form.add_field( 

58 var="muc#roomconfig_roomdesc", 

59 label="Short Description of Room", 

60 type="text-single", 

61 value=muc.description, 

62 ) 

63 

64 muc_owner = iq["mucowner_query"] 

65 muc_owner.append(form) 

66 reply.append(muc_owner) 

67 reply.send() 

68 

69 @exceptions_to_xmpp_errors 

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

71 assert isinstance(iq, Iq) 

72 muc = await self.get_muc_from_stanza(iq) 

73 query = iq["mucowner_query"] 

74 

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

76 values = form.get_values() 

77 await muc.on_set_config( 

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

79 description=( 

80 values.get("muc#roomconfig_roomdesc") 

81 if muc.HAS_DESCRIPTION 

82 else None 

83 ), 

84 ) 

85 form["type"] = "result" 

86 clear = False 

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

88 reason = destroy["reason"] or None 

89 await muc.on_destroy_request(reason) 

90 user_participant = await muc.get_user_participant() 

91 user_participant.stored.affiliation = "none" 

92 user_participant.stored.role = "none" 

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

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

95 if reason is not None: 

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

97 user_participant._send(presence) 

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

99 clear = True 

100 else: 

101 raise XMPPError("bad-request") 

102 

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