Coverage for slidge / util / types.py: 95%

1""" 

2Typing stuff 

3""" 

4 

5from collections.abc import AsyncIterator, Hashable 

6from dataclasses import dataclass, fields 

7from datetime import datetime 

8from enum import IntEnum 

9from pathlib import Path 

10from typing import ( 

11 IO, 

12 TYPE_CHECKING, 

13 Any, 

14 Generic, 

15 Literal, 

16 NamedTuple, 

17 TypedDict, 

18 TypeVar, 

19 Union, 

20) 

21 

22from slixmpp import Message, Presence 

23from slixmpp.types import PresenceShows, PresenceTypes, ResourceDict # noqa: F401 

24 

25if TYPE_CHECKING: 

26 from ..contact import LegacyContact, LegacyRoster 

27 from ..core.session import BaseSession 

28 from ..group import LegacyBookmarks, LegacyMUC 

29 from ..group.participant import LegacyParticipant 

30 

31 AnySession = BaseSession[Any, Any] 

32 AnyContact = LegacyContact[Any] 

33 AnyMUC = LegacyMUC[Any, Any, Any, Any] 

34 AnyBookmarks = LegacyBookmarks[Any, Any] 

35 AnyRoster = LegacyRoster[Any, Any] 

36else: 

37 # Hack to work around circular import. 

38 AnySession = AnyContact = AnyMUC = AnyBookmarks = AnyRoster = int 

39 

40 

41LegacyGroupIdType = TypeVar("LegacyGroupIdType", bound=Hashable) 

42""" 

43Type of the unique identifier for groups, usually a str or an int, 

44but anything hashable should work. 

45""" 

46LegacyMessageType = TypeVar("LegacyMessageType", bound=Hashable) 

47LegacyThreadType = TypeVar("LegacyThreadType", bound=Hashable) 

48LegacyUserIdType = TypeVar("LegacyUserIdType", bound=Hashable) 

49 

50LegacyContactType = TypeVar("LegacyContactType", bound=AnyContact) 

51LegacyMUCType = TypeVar("LegacyMUCType", bound=AnyMUC) 

52LegacyParticipantType = TypeVar("LegacyParticipantType", bound="LegacyParticipant") 

53 

54SessionType = TypeVar("SessionType", bound=AnySession) 

55AnyRecipient = Union[AnyContact, AnyMUC] # noqa:UP007 because it messes up mypy to use | here 

56RecipientType = TypeVar("RecipientType", bound=AnyRecipient) 

57Sender = Union[AnyContact, "LegacyParticipant"] 

58LegacyFileIdType = int | str 

59 

60ChatState = Literal["active", "composing", "gone", "inactive", "paused"] 

61ProcessingHint = Literal["no-store", "markable", "store"] 

62Marker = Literal["acknowledged", "received", "displayed"] 

63FieldType = Literal[ 

64 "boolean", 

65 "fixed", 

66 "text-single", 

67 "text-multi", 

68 "jid-single", 

69 "jid-multi", 

70 "list-single", 

71 "list-multi", 

72 "text-private", 

73] 

74MucAffiliation = Literal["owner", "admin", "member", "outcast", "none"] 

75MucRole = Literal["visitor", "participant", "moderator", "none"] 

76# https://xmpp.org/registrar/disco-categories.html#client 

77ClientType = Literal[ 

78 "bot", "console", "game", "handheld", "pc", "phone", "sms", "tablet", "web" 

79] 

80AttachmentDisposition = Literal["attachment", "inline"] 

81 

82 

83@dataclass 

84class MessageReference(Generic[LegacyMessageType]): 

85 """ 

86 A "message reply", ie a "quoted message" (:xep:`0461`) 

87 

88 At the very minimum, the legacy message ID attribute must be set, but to 

89 ensure that the quote is displayed in all XMPP clients, the author must also 

90 be set (use the string "user" if the slidge user is the author of the referenced 

91 message). 

92 The body is used as a fallback for XMPP clients that do not support :xep:`0461` 

93 of that failed to find the referenced message. 

94 """ 

95 

96 legacy_id: LegacyMessageType 

97 author: Union[Literal["user"], "LegacyParticipant", AnyContact] | None = None 

98 body: str | None = None 

99 

100 

101AnyMessageReference = MessageReference[Any] 

102 

103 

104@dataclass 

105class LegacyAttachment: 

106 """ 

107 A file attachment to a message 

108 

109 At the minimum, one of the ``path``, ``steam``, ``data`` or ``url`` attribute 

110 has to be set 

111 

112 To be used with :meth:`.LegacyContact.send_files` or 

113 :meth:`.LegacyParticipant.send_files` 

114 """ 

115 

116 path: Path | str | None = None 

117 name: str | None = None 

118 stream: IO[bytes] | None = None 

119 aio_stream: AsyncIterator[bytes] | None = None 

120 data: bytes | None = None 

121 content_type: str | None = None 

122 legacy_file_id: str | int | None = None 

123 url: str | None = None 

124 caption: str | None = None 

125 disposition: AttachmentDisposition | None = None 

126 """ 

127 A caption for this specific image. For a global caption for a list of attachments, 

128 use the ``body`` parameter of :meth:`.AttachmentMixin.send_files` 

129 """ 

130 

131 def __post_init__(self) -> None: 

132 if all( 

133 x is None 

134 for x in (self.path, self.stream, self.data, self.url, self.aio_stream) 

135 ): 

136 raise TypeError("There is not data in this attachment", self) 

137 if isinstance(self.path, str): 

138 self.path = Path(self.path) 

139 

140 def format_for_user(self) -> str: 

141 if self.name: 

142 name = self.name 

143 elif self.path: 

144 name = self.path.name # type:ignore[union-attr] 

145 elif self.url: 

146 name = self.url 

147 else: 

148 name = "" 

149 

150 if self.caption: 

151 if name: 

152 name = f"{name}: {self.caption}" 

153 else: 

154 name = self.caption 

155 

156 return name 

157 

158 def __str__(self) -> str: 

159 attrs = ", ".join( 

160 f"{f.name}={getattr(self, f.name)!r}" 

161 for f in fields(self) 

162 if getattr(self, f.name) is not None and f.name != "data" 

163 ) 

164 if self.data is not None: 

165 data_str = f"data=<{len(self.data)} bytes>" 

166 to_join = (attrs, data_str) if attrs else (data_str,) 

167 attrs = ", ".join(to_join) 

168 return f"Attachment({attrs})" 

169 

170 

171class MucType(IntEnum): 

172 """ 

173 The type of group, private, public, anonymous or not. 

174 """ 

175 

176 GROUP = 0 

177 """ 

178 A private group, members-only and non-anonymous, eg a family group. 

179 """ 

180 CHANNEL = 1 

181 """ 

182 A public group, aka an anonymous channel. 

183 """ 

184 CHANNEL_NON_ANONYMOUS = 2 

185 """ 

186 A public group where participants' legacy IDs are visible to everybody. 

187 """ 

188 

189 

190PseudoPresenceShow = PresenceShows | Literal[""] 

191 

192 

193MessageOrPresenceTypeVar = TypeVar("MessageOrPresenceTypeVar", bound=Message | Presence) 

194 

195 

196class LinkPreview(NamedTuple): 

197 """ 

198 Embedded metadata from :xep:`0511`. 

199 

200 See <https://ogp.me/>_. 

201 """ 

202 

203 about: str 

204 """ 

205 URL of the link. 

206 """ 

207 title: str | None 

208 """ 

209 Title of the linked page. 

210 """ 

211 description: str | None 

212 """ 

213 A description of the page. 

214 """ 

215 url: str | None 

216 """ 

217 The canonical URL of the link. 

218 """ 

219 image: str | Path | bytes | None 

220 """ 

221 An image representing the link. If it is a string, it should represent a URL to an image. 

222 """ 

223 type: str | None 

224 """ 

225 Type of the link destination. 

226 """ 

227 site_name: str | None 

228 """ 

229 Name of the web site. 

230 """ 

231 

232 @property 

233 def is_empty(self) -> bool: 

234 return not any(x for x in self) 

235 

236 

237class Mention(NamedTuple): 

238 contact: "LegacyContact[Any]" 

239 start: int 

240 end: int 

241 

242 

243class Hat(NamedTuple): 

244 uri: str 

245 title: str 

246 hue: float | None = None 

247 

248 

249class UserPreferences(TypedDict): 

250 sync_avatar: bool 

251 sync_presence: bool 

252 

253 

254class MamMetadata(NamedTuple): 

255 id: str 

256 sent_on: datetime 

257 

258 

259class HoleBound(NamedTuple): 

260 id: int | str 

261 timestamp: datetime 

262 

263 

264class CachedPresence(NamedTuple): 

265 last_seen: datetime | None = None 

266 ptype: PresenceTypes | None = None 

267 pstatus: str | None = None 

268 pshow: PresenceShows | None = None 

269 

270 

271class Sticker(NamedTuple): 

272 path: Path 

273 content_type: str | None 

274 hashes: dict[str, str] 

275 

276 

277class Avatar(NamedTuple): 

278 path: Path | None = None 

279 unique_id: str | int | None = None 

280 url: str | None = None 

281 data: bytes | None = None