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

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.xmlstream import StanzaBase 

12 

13from . import ( # xep_0356, 

14 link_preview, 

15 xep_0077, 

16 xep_0100, 

17 xep_0153, 

18 xep_0264, 

19 xep_0292, 

20 xep_0313, 

21 xep_0317, 

22 xep_0356_old, 

23 xep_0424, 

24 xep_0490, 

25) 

26 

27 

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

29 cid = iq["bob"]["cid"] 

30 

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

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

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

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

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

36 

37 if data is None: 

38 raise XMPPError( 

39 "item-not-found", 

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

41 ) 

42 

43 if isinstance(data, Iq): 

44 data["id"] = iq["id"] 

45 data.send() 

46 return 

47 

48 iq = iq.reply() 

49 iq.append(data) 

50 iq.send() 

51 

52 

53XEP_0231._handle_bob_iq = _handle_bob_iq 

54 

55 

56def session_bind(self, jid): 

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

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

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

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

61 

62 

63XEP_0050.session_bind = session_bind # type:ignore 

64 

65 

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

67 """ 

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

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

70 we actually want to preserve the resource. 

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

72 """ 

73 new_message = StanzaBase.reply(self, clear) 

74 new_message["thread"] = self["thread"] 

75 new_message["parent_thread"] = self["parent_thread"] 

76 

77 del new_message["id"] 

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

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

80 

81 if body is not None: 

82 new_message["body"] = body 

83 return new_message 

84 

85 

86slixmpp.plugins.PLUGINS.extend( 

87 [ 

88 "link_preview", 

89 "xep_0264", 

90 "xep_0292_provider", 

91 "xep_0317", 

92 "xep_0356_old", 

93 "xep_0490", 

94 ] 

95) 

96 

97 

98Message.reply = reply # type: ignore