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[Mapping[str, "type[ContactCommand[Any] | MUCCommand[Any]]"]] 

22 

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

24 if self.DISCO_NAME is NotImplemented: 

25 return self.xmpp.COMPONENT_NAME 

26 return self.DISCO_NAME or self.xmpp.COMPONENT_NAME 

27 

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

29 return [] 

30 

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

32 return None 

33 

34 async def get_disco_info( 

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

36 ) -> DiscoInfo: 

37 info = DiscoInfo() 

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

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

40 elif node and node in self.commands: 

41 info.add_identity( 

42 category="automation", 

43 itype="command-node", 

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

45 ) 

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

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

48 else: 

49 for feature in self.features(): 

50 info.add_feature(feature) 

51 info.add_identity( 

52 category=self.DISCO_CATEGORY, 

53 itype=self.DISCO_TYPE, 

54 name=self._get_disco_name(), 

55 lang=self.DISCO_LANG, 

56 ) 

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

58 for form in forms: 

59 info.append(form) 

60 return info 

61 

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

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

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

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

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

67 

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

69 items = DiscoItems() 

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

71 for node_, command in self.commands.items(): 

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

73 

74 return items 

75 

76 

77class ChatterDiscoMixin(BaseDiscoMixin): 

78 AVATAR = True 

79 RECEIPTS = True 

80 MARKS = True 

81 CHAT_STATES = True 

82 UPLOAD = True 

83 CORRECTION = True 

84 REACTION = True 

85 RETRACTION = True 

86 REPLIES = True 

87 INVITATION_RECIPIENT = False 

88 

89 DISCO_TYPE = "pc" 

90 DISCO_CATEGORY = "client" 

91 DISCO_NAME = "" 

92 

93 is_participant: bool 

94 

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

96 features = [] 

97 if self.CHAT_STATES: 

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

99 if self.RECEIPTS: 

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

101 if self.CORRECTION: 

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

103 if self.MARKS: 

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

105 if self.UPLOAD: 

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

107 if self.REACTION: 

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

109 if self.RETRACTION: 

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

111 if self.REPLIES: 

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

113 if self.INVITATION_RECIPIENT: 

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

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

116 if not self.is_participant: 

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

118 return features 

119 

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

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

122 if f is None: 

123 return None 

124 

125 e = await f() 

126 if not e: 

127 return None 

128 

129 return [e] 

130 

131 

132class ContactAccountDiscoMixin(BaseDiscoMixin): 

133 async def get_disco_info( 

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

135 ) -> DiscoInfo: 

136 if jid and jid.resource: 

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

138 info = DiscoInfo() 

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

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

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

142 info.add_identity( 

143 category="account", 

144 itype="registered", 

145 name=self._get_disco_name(), 

146 lang=self.DISCO_LANG, 

147 ) 

148 info.add_identity( 

149 category="pubsub", 

150 itype="pep", 

151 name=self._get_disco_name(), 

152 lang=self.DISCO_LANG, 

153 ) 

154 return info