Coverage for slidge/slixfix/xep_0313/stanza.py: 82%

1# Slixmpp: The Slick XMPP Library 

2# Copyright (C) 2012 Nathanael C. Fritz, Lance J.T. Stout 

3# This file is part of Slixmpp. 

4# See the file LICENSE for copying permissio 

5from datetime import datetime 

6from typing import Any, Iterable, List, Optional, Set, Union 

7 

8from slixmpp.jid import JID 

9from slixmpp.plugins import xep_0082 

10from slixmpp.stanza import Message 

11from slixmpp.xmlstream import ET, ElementBase 

12 

13 

14class MAM(ElementBase): 

15 """A MAM Query element. 

16 

17 .. code-block:: xml 

18 

19 <iq type='set' id='juliet1'> 

20 <query xmlns='urn:xmpp:mam:2'> 

21 <x xmlns='jabber:x:data' type='submit'> 

22 <field var='FORM_TYPE' type='hidden'> 

23 <value>urn:xmpp:mam:2</value> 

24 </field> 

25 <field var='with'> 

26 <value>juliet@capulet.lit</value> 

27 </field> 

28 </x> 

29 </query> 

30 </iq> 

31 

32 """ 

33 

34 name = "query" 

35 namespace = "urn:xmpp:mam:2" 

36 plugin_attrib = "mam" 

37 #: Available interfaces: 

38 #: 

39 #: - ``queryid``: The MAM query id 

40 #: - ``start`` and ``end``: Temporal boundaries of the query 

41 #: - ``with``: JID of the other entity the conversation is with 

42 #: - ``after_id``: Fetch stanzas after this specific ID 

43 #: - ``before_id``: Fetch stanzas before this specific ID 

44 #: - ``ids``: Fetch the stanzas matching those IDs 

45 #: - ``results``: pseudo-interface used to accumulate MAM results during 

46 #: fetch, not relevant for the stanza itself. 

47 interfaces = { 

48 "queryid", 

49 "start", 

50 "end", 

51 "with", 

52 "results", 

53 "before_id", 

54 "after_id", 

55 "ids", 

56 "flip_page", 

57 } 

58 sub_interfaces = { 

59 "start", 

60 "end", 

61 "with", 

62 "before_id", 

63 "after_id", 

64 "ids", 

65 "flip_page", 

66 } 

67 

68 def setup(self, xml=None): 

69 ElementBase.setup(self, xml) 

70 self._results: List[Message] = [] 

71 

72 def _setup_form(self): 

73 found = self.xml.find( 

74 "{jabber:x:data}x/" 

75 '{jabber:x:data}field[@var="FORM_TYPE"]/' 

76 "{jabber:x:data}value[.='urn:xmpp:mam:2']" 

77 ) 

78 if found is None: 

79 self["form"]["type"] = "submit" 

80 self["form"].add_field( 

81 var="FORM_TYPE", ftype="hidden", value="urn:xmpp:mam:2" 

82 ) 

83 

84 def get_fields(self): 

85 form = self.get_plugin("form", check=True) 

86 if not form: 

87 return {} 

88 return form.get_fields() 

89 

90 def get_start(self) -> Optional[datetime]: 

91 fields = self.get_fields() 

92 field = fields.get("start") 

93 if field and field["value"]: 

94 return xep_0082.parse(field["value"]) 

95 return None 

96 

97 def set_start(self, value: Union[str, datetime]): 

98 self._setup_form() 

99 if isinstance(value, datetime): 

100 value = xep_0082.format_datetime(value) 

101 self.set_custom_field("start", value) 

102 

103 def get_end(self) -> Optional[datetime]: 

104 fields = self.get_fields() 

105 field = fields.get("end") 

106 if field and field["value"]: 

107 return xep_0082.parse(field["value"]) 

108 return None 

109 

110 def set_end(self, value: Union[str, datetime]): 

111 if isinstance(value, datetime): 

112 value = xep_0082.format_datetime(value) 

113 self.set_custom_field("end", value) 

114 

115 def get_with(self) -> Optional[JID]: 

116 fields = self.get_fields() 

117 field = fields.get("with") 

118 if field: 

119 return JID(field["value"]) 

120 return None 

121 

122 def set_with(self, value: JID): 

123 self.set_custom_field("with", value) 

124 

125 def set_custom_field(self, fieldname: str, value: Any): 

126 self._setup_form() 

127 fields = self.get_fields() 

128 field = fields.get(fieldname) 

129 if field: 

130 field["value"] = str(value) 

131 else: 

132 field = self["form"].add_field(var=fieldname) 

133 field["value"] = str(value) 

134 

135 def get_custom_field(self, fieldname: str) -> Optional[str]: 

136 fields = self.get_fields() 

137 field = fields.get(fieldname) 

138 if field: 

139 return field["value"] 

140 return None 

141 

142 def set_before_id(self, value: str): 

143 self.set_custom_field("before-id", value) 

144 

145 def get_before_id(self): 

146 self.get_custom_field("before-id") 

147 

148 def set_after_id(self, value: str): 

149 self.set_custom_field("after-id", value) 

150 

151 def get_after_id(self): 

152 self.get_custom_field("after-id") 

153 

154 def set_ids(self, value: List[str]): 

155 self._setup_form() 

156 fields = self.get_fields() 

157 field = fields.get("ids") 

158 if field: 

159 field["ids"] = value 

160 else: 

161 field = self["form"].add_field(var="ids") 

162 field["value"] = value 

163 

164 def get_ids(self): 

165 self.get_custom_field("id") 

166 

167 # The results interface is meant only as an easy 

168 # way to access the set of collected message responses 

169 # from the query. 

170 

171 def get_results(self) -> List[Message]: 

172 return self._results 

173 

174 def set_results(self, values: List[Message]): 

175 self._results = values 

176 

177 def del_results(self): 

178 self._results = [] 

179 

180 def get_flip_page(self): 

181 return self.xml.find(f"{{{self.namespace}}}flip-page") is not None 

182 

183 

184class Fin(ElementBase): 

185 """A MAM fin element (end of query). 

186 

187 .. code-block:: xml 

188 

189 <iq type='result' id='juliet1'> 

190 <fin xmlns='urn:xmpp:mam:2'> 

191 <set xmlns='http://jabber.org/protocol/rsm'> 

192 <first index='0'>28482-98726-73623</first> 

193 <last>09af3-cc343-b409f</last> 

194 </set> 

195 </fin> 

196 </iq> 

197 

198 """ 

199 

200 name = "fin" 

201 namespace = "urn:xmpp:mam:2" 

202 plugin_attrib = "mam_fin" 

203 interfaces = {"results", "stable", "complete"} 

204 

205 def setup(self, xml=None): 

206 ElementBase.setup(self, xml) 

207 self._results: List[Message] = [] 

208 

209 # The results interface is meant only as an easy 

210 # way to access the set of collected message responses 

211 # from the query. 

212 

213 def get_results(self) -> List[Message]: 

214 return self._results 

215 

216 def set_results(self, values: List[Message]): 

217 self._results = values 

218 

219 def del_results(self): 

220 self._results = [] 

221 

222 

223class Result(ElementBase): 

224 """A MAM result payload. 

225 

226 .. code-block:: xml 

227 

228 <message id='aeb213' to='juliet@capulet.lit/chamber'> 

229 <result xmlns='urn:xmpp:mam:2' queryid='f27' id='28482-98726-73623'> 

230 <forwarded xmlns='urn:xmpp:forward:0'> 

231 <delay xmlns='urn:xmpp:delay' stamp='2010-07-10T23:08:25Z'/> 

232 <message xmlns='jabber:client' from="witch@shakespeare.lit" 

233 to="macbeth@shakespeare.lit"> 

234 <body>Hail to thee</body> 

235 </message> 

236 </forwarded> 

237 </result> 

238 </message> 

239 """ 

240 

241 name = "result" 

242 namespace = "urn:xmpp:mam:2" 

243 plugin_attrib = "mam_result" 

244 #: Available interfaces: 

245 #: 

246 #: - ``queryid``: MAM queryid 

247 #: - ``id``: ID of the result 

248 interfaces = {"queryid", "id"} 

249 

250 

251class Metadata(ElementBase): 

252 """Element containing archive metadata 

253 

254 .. code-block:: xml 

255 

256 <iq type='result' id='jui8921rr9'> 

257 <metadata xmlns='urn:xmpp:mam:2'> 

258 <start id='YWxwaGEg' timestamp='2008-08-22T21:09:04Z' /> 

259 <end id='b21lZ2Eg' timestamp='2020-04-20T14:34:21Z' /> 

260 </metadata> 

261 </iq> 

262 

263 """ 

264 

265 name = "metadata" 

266 namespace = "urn:xmpp:mam:2" 

267 plugin_attrib = "mam_metadata" 

268 

269 

270class Start(ElementBase): 

271 """Metadata about the start of an archive. 

272 

273 .. code-block:: xml 

274 

275 <iq type='result' id='jui8921rr9'> 

276 <metadata xmlns='urn:xmpp:mam:2'> 

277 <start id='YWxwaGEg' timestamp='2008-08-22T21:09:04Z' /> 

278 <end id='b21lZ2Eg' timestamp='2020-04-20T14:34:21Z' /> 

279 </metadata> 

280 </iq> 

281 

282 """ 

283 

284 name = "start" 

285 namespace = "urn:xmpp:mam:2" 

286 plugin_attrib = name 

287 #: Available interfaces: 

288 #: 

289 #: - ``id``: ID of the first message of the archive 

290 #: - ``timestamp`` (``datetime``): timestamp of the first message of the 

291 #: archive 

292 interfaces = {"id", "timestamp"} 

293 

294 def get_timestamp(self) -> Optional[datetime]: 

295 """Get the timestamp. 

296 

297 :returns: The timestamp. 

298 """ 

299 stamp = self.xml.attrib.get("timestamp", None) 

300 if stamp is not None: 

301 return xep_0082.parse(stamp) 

302 return stamp 

303 

304 def set_timestamp(self, value: Union[datetime, str]): 

305 """Set the timestamp. 

306 

307 :param value: Value of the timestamp (either a datetime or a 

308 XEP-0082 timestamp string. 

309 """ 

310 if isinstance(value, str): 

311 value = xep_0082.parse(value) 

312 value = xep_0082.format_datetime(value) 

313 self.xml.attrib["timestamp"] = value 

314 

315 

316class End(ElementBase): 

317 """Metadata about the end of an archive. 

318 

319 .. code-block:: xml 

320 

321 <iq type='result' id='jui8921rr9'> 

322 <metadata xmlns='urn:xmpp:mam:2'> 

323 <start id='YWxwaGEg' timestamp='2008-08-22T21:09:04Z' /> 

324 <end id='b21lZ2Eg' timestamp='2020-04-20T14:34:21Z' /> 

325 </metadata> 

326 </iq> 

327 

328 """ 

329 

330 name = "end" 

331 namespace = "urn:xmpp:mam:2" 

332 plugin_attrib = name 

333 #: Available interfaces: 

334 #: 

335 #: - ``id``: ID of the first message of the archive 

336 #: - ``timestamp`` (``datetime``): timestamp of the first message of the 

337 #: archive 

338 interfaces = {"id", "timestamp"} 

339 

340 def get_timestamp(self) -> Optional[datetime]: 

341 """Get the timestamp. 

342 

343 :returns: The timestamp. 

344 """ 

345 stamp = self.xml.attrib.get("timestamp", None) 

346 if stamp is not None: 

347 return xep_0082.parse(stamp) 

348 return stamp 

349 

350 def set_timestamp(self, value: Union[datetime, str]): 

351 """Set the timestamp. 

352 

353 :param value: Value of the timestamp (either a datetime or a 

354 XEP-0082 timestamp string. 

355 """ 

356 if isinstance(value, str): 

357 value = xep_0082.parse(value) 

358 value = xep_0082.format_datetime(value) 

359 self.xml.attrib["timestamp"] = value