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, RoomCommand 

13 

14 

15class BaseDiscoMixin(Base): 

16 @property 

17 def DISCO_TYPE(self) -> str: 

18 raise NotImplementedError 

19 

20 DISCO_CATEGORY: str = NotImplemented 

21 DISCO_NAME: str = NotImplemented 

22 DISCO_LANG = None 

23 

24 _commands: ClassVar[Mapping[str, "type[ContactCommand[Any] | RoomCommand[Any]]"]] 

25 

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

27 if self.DISCO_NAME is NotImplemented: 

28 return self.xmpp.COMPONENT_NAME 

29 return self.DISCO_NAME or self.xmpp.COMPONENT_NAME 

30 

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

32 return [] 

33 

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

35 return None 

36 

37 async def get_disco_info( 

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

39 ) -> DiscoInfo: 

40 info = DiscoInfo() 

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

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

43 elif node and node in self._commands: 

44 info.add_identity( 

45 category="automation", 

46 itype="command-node", 

47 name=self._commands[node].NAME, 

48 ) 

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

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

51 else: 

52 for feature in self.features(): 

53 info.add_feature(feature) 

54 info.add_identity( 

55 category=self.DISCO_CATEGORY, 

56 itype=self.DISCO_TYPE, 

57 name=self._get_disco_name(), 

58 lang=self.DISCO_LANG, 

59 ) 

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

61 for form in forms: 

62 info.append(form) 

63 return info 

64 

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

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

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

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

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

70 

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

72 items = DiscoItems() 

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

74 for node_, command in self._commands.items(): 

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

76 

77 return items 

78 

79 

80class ChatterDiscoMixin(BaseDiscoMixin): 

81 AVATAR = True 

82 RECEIPTS = True 

83 MARKS = True 

84 CHAT_STATES = True 

85 UPLOAD = True 

86 CORRECTION = True 

87 REACTION = True 

88 RETRACTION = True 

89 REPLIES = True 

90 _INVITATION_RECIPIENT = False 

91 

92 @property 

93 def DISCO_TYPE(self) -> str: 

94 return "pc" 

95 

96 DISCO_CATEGORY = "client" 

97 DISCO_NAME = "" 

98 

99 is_participant: bool 

100 

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

102 features = [] 

103 if self.CHAT_STATES: 

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

105 if self.RECEIPTS: 

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

107 if self.CORRECTION: 

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

109 if self.MARKS: 

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

111 if self.UPLOAD: 

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

113 if self.REACTION: 

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

115 if self.RETRACTION: 

116 features.append("urn:xmpp:message-retract:1") 

117 if self.REPLIES: 

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

119 if self._INVITATION_RECIPIENT: 

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

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

122 if not self.is_participant: 

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

124 return features 

125 

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

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

128 if f is None: 

129 return None 

130 

131 e = await f() 

132 if not e: 

133 return None 

134 

135 return [e] 

136 

137 

138class ContactAccountDiscoMixin(BaseDiscoMixin): 

139 async def get_disco_info( 

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

141 ) -> DiscoInfo: 

142 if jid and jid.resource: 

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

144 info = DiscoInfo() 

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

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

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

148 info.add_identity( 

149 category="account", 

150 itype="registered", 

151 name=self._get_disco_name(), 

152 lang=self.DISCO_LANG, 

153 ) 

154 info.add_identity( 

155 category="pubsub", 

156 itype="pep", 

157 name=self._get_disco_name(), 

158 lang=self.DISCO_LANG, 

159 ) 

160 return info