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

1from collections.abc import Mapping 

2from typing import TYPE_CHECKING, Any, ClassVar 

3 

4from slixmpp.plugins.xep_0004.stanza.form import Form 

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

6from slixmpp.plugins.xep_0030.stanza.items import DiscoItems 

7from slixmpp.types import OptJid 

8 

9from .base import Base 

10 

11if TYPE_CHECKING: 

12 from slidge.command.base import ContactCommand, MUCCommand 

13 

14 

15class BaseDiscoMixin(Base): 

16 DISCO_TYPE: str = NotImplemented 

17 DISCO_CATEGORY: str = NotImplemented 

18 DISCO_NAME: str = NotImplemented 

19 DISCO_LANG = None 

20 

21 commands: ClassVar[ 

22 Mapping[str, "type[ContactCommand[Any]] | type[MUCCommand[Any]]"] 

23 ] 

24 

25 def _get_disco_name(self) -> str | None: 

26 if self.DISCO_NAME is NotImplemented: 

27 return self.xmpp.COMPONENT_NAME 

28 return self.DISCO_NAME or self.xmpp.COMPONENT_NAME 

29 

30 def features(self) -> list[str]: 

31 return [] 

32 

33 async def extended_features(self) -> list[Form] | None: 

34 return None 

35 

36 async def get_disco_info( 

37 self, jid: OptJid = None, node: str | None = None 

38 ) -> DiscoInfo: 

39 info = DiscoInfo() 

40 if node == "http://jabber.org/protocol/commands": 

41 info.add_identity(category="automation", itype="command-list") 

42 elif node and node in self.commands: 

43 info.add_identity( 

44 category="automation", 

45 itype="command-node", 

46 name=self.commands[node].NAME, 

47 ) 

48 info.add_feature("http://jabber.org/protocol/commands") 

49 info.add_feature("jabber:x:data") 

50 else: 

51 for feature in self.features(): 

52 info.add_feature(feature) 

53 info.add_identity( 

54 category=self.DISCO_CATEGORY, 

55 itype=self.DISCO_TYPE, 

56 name=self._get_disco_name(), 

57 lang=self.DISCO_LANG, 

58 ) 

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

60 for form in forms: 

61 info.append(form) 

62 return info 

63 

64 async def get_caps_ver(self, jid: OptJid = None, node: str | None = None) -> str: 

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

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

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

68 return ver # type:ignore[no-any-return] 

69 

70 async def get_disco_items(self, node: str | None) -> DiscoItems: 

71 items = DiscoItems() 

72 if node == "http://jabber.org/protocol/commands": 

73 for node, command in self.commands.items(): 

74 items.add_item(jid=self.jid, node=node, name=command.NAME) 

75 

76 return items 

77 

78 

79class ChatterDiscoMixin(BaseDiscoMixin): 

80 AVATAR = True 

81 RECEIPTS = True 

82 MARKS = True 

83 CHAT_STATES = True 

84 UPLOAD = True 

85 CORRECTION = True 

86 REACTION = True 

87 RETRACTION = True 

88 REPLIES = True 

89 INVITATION_RECIPIENT = False 

90 

91 DISCO_TYPE = "pc" 

92 DISCO_CATEGORY = "client" 

93 DISCO_NAME = "" 

94 

95 is_participant: bool 

96 

97 def features(self) -> list[str]: 

98 features = [] 

99 if self.CHAT_STATES: 

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

101 if self.RECEIPTS: 

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

103 if self.CORRECTION: 

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

105 if self.MARKS: 

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

107 if self.UPLOAD: 

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

109 if self.REACTION: 

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

111 if self.RETRACTION: 

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

113 if self.REPLIES: 

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

115 if self.INVITATION_RECIPIENT: 

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

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

118 if not self.is_participant: 

119 features.append("http://jabber.org/protocol/commands") 

120 return features 

121 

122 async def extended_features(self) -> list[Form] | None: 

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

124 if f is None: 

125 return None 

126 

127 e = await f() 

128 if not e: 

129 return None 

130 

131 return [e] 

132 

133 

134class ContactAccountDiscoMixin(BaseDiscoMixin): 

135 async def get_disco_info( 

136 self, jid: OptJid = None, node: str | None = None 

137 ) -> DiscoInfo: 

138 if jid and jid.resource: 

139 return await super().get_disco_info(jid, node) 

140 info = DiscoInfo() 

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

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

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

144 info.add_identity( 

145 category="account", 

146 itype="registered", 

147 name=self._get_disco_name(), 

148 lang=self.DISCO_LANG, 

149 ) 

150 info.add_identity( 

151 category="pubsub", 

152 itype="pep", 

153 name=self._get_disco_name(), 

154 lang=self.DISCO_LANG, 

155 ) 

156 return info