Coverage for slidge/core/mixins/disco.py: 97%

1from typing import Optional 

2 

3from slixmpp.plugins.xep_0004 import Form 

4from slixmpp.plugins.xep_0030.stanza.info import DiscoInfo 

5from slixmpp.types import OptJid 

6 

7from .base import Base 

8 

9 

10class BaseDiscoMixin(Base): 

11 DISCO_TYPE: str = NotImplemented 

12 DISCO_CATEGORY: str = NotImplemented 

13 DISCO_NAME: str = NotImplemented 

14 DISCO_LANG = None 

15 

16 def _get_disco_name(self): 

17 if self.DISCO_NAME is NotImplemented: 

18 return self.xmpp.COMPONENT_NAME 

19 return self.DISCO_NAME or self.xmpp.COMPONENT_NAME 

20 

21 def features(self): 

22 return [] 

23 

24 async def extended_features(self) -> Optional[list[Form]]: 

25 return None 

26 

27 async def get_disco_info(self, jid: OptJid = None, node: Optional[str] = None): 

28 info = DiscoInfo() 

29 for feature in self.features(): 

30 info.add_feature(feature) 

31 info.add_identity( 

32 category=self.DISCO_CATEGORY, 

33 itype=self.DISCO_TYPE, 

34 name=self._get_disco_name(), 

35 lang=self.DISCO_LANG, 

36 ) 

37 if forms := await self.extended_features(): 

38 for form in forms: 

39 info.append(form) 

40 return info 

41 

42 async def get_caps_ver(self, jid: OptJid = None, node: Optional[str] = None): 

43 info = await self.get_disco_info(jid, node) 

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

45 ver = caps.generate_verstring(info, caps.hash) 

46 return ver 

47 

48 

49class ChatterDiscoMixin(BaseDiscoMixin): 

50 AVATAR = True 

51 RECEIPTS = True 

52 MARKS = True 

53 CHAT_STATES = True 

54 UPLOAD = True 

55 CORRECTION = True 

56 REACTION = True 

57 RETRACTION = True 

58 REPLIES = True 

59 INVITATION_RECIPIENT = False 

60 

61 DISCO_TYPE = "pc" 

62 DISCO_CATEGORY = "client" 

63 DISCO_NAME = "" 

64 

65 def features(self): 

66 features = [] 

67 if self.CHAT_STATES: 

68 features.append("http://jabber.org/protocol/chatstates") 

69 if self.RECEIPTS: 

70 features.append("urn:xmpp:receipts") 

71 if self.CORRECTION: 

72 features.append("urn:xmpp:message-correct:0") 

73 if self.MARKS: 

74 features.append("urn:xmpp:chat-markers:0") 

75 if self.UPLOAD: 

76 features.append("jabber:x:oob") 

77 if self.REACTION: 

78 features.append("urn:xmpp:reactions:0") 

79 if self.RETRACTION: 

80 features.append("urn:xmpp:message-retract:0") 

81 if self.REPLIES: 

82 features.append("urn:xmpp:reply:0") 

83 if self.INVITATION_RECIPIENT: 

84 features.append("jabber:x:conference") 

85 features.append("urn:ietf:params:xml:ns:vcard-4.0") 

86 return features 

87 

88 async def extended_features(self): 

89 f = getattr(self, "restricted_emoji_extended_feature", None) 

90 if f is None: 

91 return 

92 

93 e = await f() 

94 if not e: 

95 return 

96 

97 return [e] 

98 

99 

100class ContactAccountDiscoMixin(BaseDiscoMixin): 

101 async def get_disco_info(self, jid: OptJid = None, node: Optional[str] = None): 

102 if jid and jid.resource: 

103 return await super().get_disco_info() 

104 info = DiscoInfo() 

105 info.add_feature("http://jabber.org/protocol/pubsub") 

106 info.add_feature("http://jabber.org/protocol/pubsub#retrieve-items") 

107 info.add_feature("http://jabber.org/protocol/pubsub#subscribe") 

108 info.add_identity( 

109 category="account", 

110 itype="registered", 

111 name=self._get_disco_name(), 

112 lang=self.DISCO_LANG, 

113 ) 

114 info.add_identity( 

115 category="pubsub", 

116 itype="pep", 

117 name=self._get_disco_name(), 

118 lang=self.DISCO_LANG, 

119 ) 

120 return info