Coverage for slidge/slixfix/__init__.py: 80%

1# This module contains patches for slixmpp; some have pending requests upstream 

2# and should be removed on the next slixmpp release. 

3 

4# ruff: noqa: F401 

5 

6import slixmpp.plugins 

7from slixmpp import Iq, Message 

8from slixmpp.exceptions import XMPPError 

9from slixmpp.plugins.xep_0050 import XEP_0050, Command 

10from slixmpp.plugins.xep_0231 import XEP_0231 

11from slixmpp.plugins.xep_0425.stanza import Moderate 

12from slixmpp.xmlstream import StanzaBase 

13 

14from . import link_preview, xep_0077, xep_0100, xep_0153, xep_0292, xep_0356_old 

15 

16# TODO: remove this when the fix is included in slixmpp 

17Moderate.interfaces.add("id") 

18 

19 

20async def _handle_bob_iq(self, iq: Iq): 

21 cid = iq["bob"]["cid"] 

22 

23 if iq["type"] == "result": 

24 await self.api["set_bob"](iq["from"], None, iq["to"], args=iq["bob"]) 

25 self.xmpp.event("bob", iq) 

26 elif iq["type"] == "get": 

27 data = await self.api["get_bob"](iq["to"], None, iq["from"], args=cid) 

28 

29 if data is None: 

30 raise XMPPError( 

31 "item-not-found", 

32 f"Bits of binary '{cid}' is not available.", 

33 ) 

34 

35 if isinstance(data, Iq): 

36 data["id"] = iq["id"] 

37 data.send() 

38 return 

39 

40 iq = iq.reply() 

41 iq.append(data) 

42 iq.send() 

43 

44 

45XEP_0231._handle_bob_iq = _handle_bob_iq 

46 

47 

48def session_bind(self, jid): 

49 self.xmpp["xep_0030"].add_feature(Command.namespace) 

50 # awful hack to for the disco items: we need to comment this line 

51 # related issue: https://todo.sr.ht/~nicoco/slidge/131 

52 # self.xmpp['xep_0030'].set_items(node=Command.namespace, items=tuple()) 

53 

54 

55XEP_0050.session_bind = session_bind # type:ignore 

56 

57 

58def reply(self, body=None, clear=True): 

59 """ 

60 Overrides slixmpp's Message.reply(), since it strips to sender's resource 

61 for mtype=groupchat, and we do not want that, because when we raise an XMPPError, 

62 we actually want to preserve the resource. 

63 (this is called in RootStanza.exception() to handle XMPPErrors) 

64 """ 

65 new_message = StanzaBase.reply(self, clear) 

66 new_message["thread"] = self["thread"] 

67 new_message["parent_thread"] = self["parent_thread"] 

68 

69 del new_message["id"] 

70 if self.stream is not None and self.stream.use_message_ids: 

71 new_message["id"] = self.stream.new_id() 

72 

73 if body is not None: 

74 new_message["body"] = body 

75 return new_message 

76 

77 

78slixmpp.plugins.PLUGINS.extend( 

79 [ 

80 "link_preview", 

81 "xep_0292_provider", 

82 "xep_0356_old", 

83 ] 

84) 

85 

86 

87Message.reply = reply # type: ignore