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

1from __future__ import annotations 

2 

3import logging 

4from typing import TYPE_CHECKING 

5 

6from slixmpp import Presence 

7from slixmpp.exceptions import XMPPError 

8from slixmpp.xmlstream import StanzaBase 

9 

10from ...contact import LegacyContact 

11from ...util.types import ClientType 

12from .util import DispatcherMixin 

13 

14if TYPE_CHECKING: 

15 from slidge.util.types import AnyGateway 

16 

17 

18class CapsMixin(DispatcherMixin): 

19 __slots__: list[str] = [] 

20 __contact_vers: dict[ClientType, str] 

21 __gateway_ver: str | None 

22 

23 def __init__(self, xmpp: AnyGateway) -> None: 

24 super().__init__(xmpp) 

25 xmpp.del_filter("out", xmpp.plugin["xep_0115"]._filter_add_caps) # type:ignore[arg-type] 

26 xmpp.add_filter("out", self._filter_add_caps) 

27 self.__contact_vers: dict[ClientType, str] = {} # type:ignore[misc] 

28 self.__gateway_ver: str | None = None # type:ignore[misc] 

29 

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

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

32 # there is too much magic happening in slixmpp 

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

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

35 if not isinstance(stanza, Presence): 

36 return stanza 

37 

38 if "caps" in stanza: 

39 return stanza 

40 

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

42 return stanza 

43 

44 pfrom = stanza.get_from() 

45 

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

47 

48 if pfrom == self.xmpp.boundjid.bare: 

49 if self.__gateway_ver is None: 

50 ver = self.__gateway_ver = await caps.get_verstring(pfrom) # type:ignore[misc] 

51 else: 

52 ver = self.__gateway_ver 

53 else: 

54 try: 

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

56 except XMPPError: 

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

58 return stanza 

59 

60 if session is None: 

61 return stanza 

62 

63 await session.ready 

64 

65 try: 

66 contact = await session.contacts.get_or_create_by_jid(pfrom) 

67 except XMPPError: 

68 return stanza 

69 except Exception: 

70 log.exception("Could not add caps") 

71 return stanza 

72 

73 ver = self.__contact_vers.get(contact.client_type) 

74 

75 if ver is None: 

76 ver = self.__contact_vers[ 

77 contact.client_type 

78 ] = await contact.get_caps_ver(pfrom) 

79 

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

81 

82 if ver: 

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

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

85 stanza["caps"]["ver"] = ver 

86 return stanza 

87 

88 async def populate_cache(self, contact: LegacyContact) -> None: 

89 if contact.client_type not in self.__contact_vers: 

90 log.debug( 

91 "Populating contact caps cache for %r", contact._stored.client_type 

92 ) 

93 self.__contact_vers[contact.client_type] = await contact.get_caps_ver( 

94 contact.jid 

95 ) 

96 

97 def get_caps_ver(self, client_type: ClientType) -> str | None: 

98 return self.__contact_vers.get(client_type) 

99 

100 

101log = logging.getLogger(__name__)