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

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 __slots__: list[str] = [] 

17 

18 def __init__(self, xmpp: "BaseGateway") -> None: 

19 super().__init__(xmpp) 

20 self.__mam_cleanup_task = xmpp.loop.create_task(self.__mam_cleanup()) # type:ignore[misc] 

21 xmpp.register_handler( 

22 CoroutineCallback( 

23 "MAM_query", 

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

25 self.__handle_mam, 

26 ) 

27 ) 

28 xmpp.register_handler( 

29 CoroutineCallback( 

30 "MAM_get_from", 

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

32 self.__handle_mam_get_form, 

33 ) 

34 ) 

35 xmpp.register_handler( 

36 CoroutineCallback( 

37 "MAM_get_meta", 

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

39 self.__handle_mam_metadata, 

40 ) 

41 ) 

42 

43 async def __mam_cleanup(self) -> None: 

44 if not config.MAM_MAX_DAYS: 

45 return 

46 while True: 

47 await asyncio.sleep(3600 * 6) 

48 with self.xmpp.store.session() as orm: 

49 self.xmpp.store.mam.nuke_older_than(orm, config.MAM_MAX_DAYS) 

50 orm.commit() 

51 

52 @exceptions_to_xmpp_errors 

53 async def __handle_mam(self, iq: Iq) -> None: 

54 muc = await self.get_muc_from_stanza(iq) 

55 await muc.send_mam(iq) 

56 

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

58 assert isinstance(iq, Iq) 

59 ito = iq.get_to() 

60 

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

62 raise XMPPError( 

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

64 ) 

65 

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

67 await session.bookmarks.by_jid(ito) 

68 

69 reply = iq.reply() 

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

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

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

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

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

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

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

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

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

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

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

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

82 reply.send() 

83 

84 @exceptions_to_xmpp_errors 

85 async def __handle_mam_metadata(self, iq: Iq) -> None: 

86 muc = await self.get_muc_from_stanza(iq) 

87 await muc.send_mam_metadata(iq)