Coverage for slidge/core/dispatcher/muc/mam.py: 96%

1import asyncio 

2from typing import TYPE_CHECKING 

3 

4from slixmpp import CoroutineCallback, Iq, StanzaPath 

5from slixmpp.exceptions import XMPPError 

6from slixmpp.xmlstream import StanzaBase 

7 

8from ... import config 

9from ..util import DispatcherMixin, exceptions_to_xmpp_errors 

10 

11if TYPE_CHECKING: 

12 from slidge.core.gateway import BaseGateway 

13 

14 

15class MamMixin(DispatcherMixin): 

16 def __init__(self, xmpp: "BaseGateway"): 

17 super().__init__(xmpp) 

18 self.__mam_cleanup_task = xmpp.loop.create_task(self.__mam_cleanup()) 

19 xmpp.register_handler( 

20 CoroutineCallback( 

21 "MAM_query", 

22 StanzaPath("iq@type=set/mam"), 

23 self.__handle_mam, 

24 ) 

25 ) 

26 xmpp.register_handler( 

27 CoroutineCallback( 

28 "MAM_get_from", 

29 StanzaPath("iq@type=get/mam"), 

30 self.__handle_mam_get_form, 

31 ) 

32 ) 

33 xmpp.register_handler( 

34 CoroutineCallback( 

35 "MAM_get_meta", 

36 StanzaPath("iq@type=get/mam_metadata"), 

37 self.__handle_mam_metadata, 

38 ) 

39 ) 

40 

41 async def __mam_cleanup(self): 

42 if not config.MAM_MAX_DAYS: 

43 return 

44 while True: 

45 await asyncio.sleep(3600 * 6) 

46 self.xmpp.store.mam.nuke_older_than(config.MAM_MAX_DAYS) 

47 

48 @exceptions_to_xmpp_errors 

49 async def __handle_mam(self, iq: Iq): 

50 muc = await self.get_muc_from_stanza(iq) 

51 await muc.send_mam(iq) 

52 

53 async def __handle_mam_get_form(self, iq: StanzaBase): 

54 assert isinstance(iq, Iq) 

55 ito = iq.get_to() 

56 

57 if ito == self.xmpp.boundjid.bare: 

58 raise XMPPError( 

59 text="No MAM on the component itself, use a JID with a resource" 

60 ) 

61 

62 session = await self._get_session(iq, 0, logged=True) 

63 await session.bookmarks.by_jid(ito) 

64 

65 reply = iq.reply() 

66 form = self.xmpp.plugin["xep_0004"].make_form() 

67 form.add_field(ftype="hidden", var="FORM_TYPE", value="urn:xmpp:mam:2") 

68 form.add_field(ftype="jid-single", var="with") 

69 form.add_field(ftype="text-single", var="start") 

70 form.add_field(ftype="text-single", var="end") 

71 form.add_field(ftype="text-single", var="before-id") 

72 form.add_field(ftype="text-single", var="after-id") 

73 form.add_field(ftype="boolean", var="include-groupchat") 

74 field = form.add_field(ftype="list-multi", var="ids") 

75 field["validate"]["datatype"] = "xs:string" 

76 field["validate"]["open"] = True 

77 reply["mam"].append(form) 

78 reply.send() 

79 

80 @exceptions_to_xmpp_errors 

81 async def __handle_mam_metadata(self, iq: Iq): 

82 muc = await self.get_muc_from_stanza(iq) 

83 await muc.send_mam_metadata(iq)