Coverage for slidge/core/dispatcher/caps.py: 91%

1import logging 

2from typing import TYPE_CHECKING 

3 

4from slixmpp import Presence 

5from slixmpp.exceptions import XMPPError 

6from slixmpp.xmlstream import StanzaBase 

7 

8from .util import DispatcherMixin 

9 

10if TYPE_CHECKING: 

11 from slidge.core.gateway import BaseGateway 

12 

13 

14class CapsMixin(DispatcherMixin): 

15 __slots__: list[str] = [] 

16 

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

18 super().__init__(xmpp) 

19 xmpp.del_filter("out", xmpp.plugin["xep_0115"]._filter_add_caps) 

20 xmpp.add_filter("out", self._filter_add_caps) # type:ignore 

21 

22 async def _filter_add_caps(self, stanza: StanzaBase) -> StanzaBase: 

23 # we rolled our own "add caps on presences" filter because 

24 # there is too much magic happening in slixmpp 

25 # anyway, we probably want to roll our own "dynamic disco"/caps 

26 # module in the long run, so it's a step in this direction 

27 if not isinstance(stanza, Presence): 

28 return stanza 

29 

30 if stanza.get_plugin("caps", check=True): 

31 return stanza 

32 

33 if stanza["type"] not in ("available", "chat", "away", "dnd", "xa"): 

34 return stanza 

35 

36 pfrom = stanza.get_from() 

37 

38 caps = self.xmpp.plugin["xep_0115"] 

39 

40 if pfrom != self.xmpp.boundjid.bare: 

41 try: 

42 session = self.xmpp.get_session_from_jid(stanza.get_to()) 

43 except XMPPError: 

44 log.debug("not adding caps 1") 

45 return stanza 

46 

47 if session is None: 

48 return stanza 

49 

50 await session.ready 

51 

52 try: 

53 contact = await session.contacts.by_jid(pfrom) 

54 except XMPPError: 

55 return stanza 

56 if contact.stored.caps_ver: 

57 ver = contact.stored.caps_ver 

58 else: 

59 ver = await contact.get_caps_ver(pfrom) 

60 contact.stored.caps_ver = ver 

61 contact.commit() 

62 else: 

63 ver = await caps.get_verstring(pfrom) 

64 

65 log.debug("Ver: %s", ver) 

66 

67 if ver: 

68 stanza["caps"]["node"] = caps.caps_node 

69 stanza["caps"]["hash"] = caps.hash 

70 stanza["caps"]["ver"] = ver 

71 return stanza 

72 

73 

74log = logging.getLogger(__name__)