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 import BaseGateway 

13 

14 

15class MamMixin(DispatcherMixin): 

16 __slots__: list[str] = ["__mam_cleanup_task"] 

17 

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

19 super().__init__(xmpp) 

20 self.__mam_cleanup_task = xmpp.loop.create_task( 

21 self.__mam_cleanup(), 

22 name="MAM cleanup", 

23 ) 

24 xmpp.register_handler( 

25 CoroutineCallback( 

26 "MAM_query", 

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

28 self.__handle_mam, 

29 ) 

30 ) 

31 xmpp.register_handler( 

32 CoroutineCallback( 

33 "MAM_get_from", 

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

35 self.__handle_mam_get_form, 

36 ) 

37 ) 

38 xmpp.register_handler( 

39 CoroutineCallback( 

40 "MAM_get_meta", 

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

42 self.__handle_mam_metadata, 

43 ) 

44 ) 

45 

46 async def __mam_cleanup(self) -> None: 

47 if not config.MAM_MAX_DAYS: 

48 return 

49 while True: 

50 await asyncio.sleep(3600 * 6) 

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

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

53 orm.commit() 

54 

55 @exceptions_to_xmpp_errors 

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

57 muc = await self.get_muc_from_stanza(iq) 

58 await muc.send_mam(iq) 

59 

60 async def __handle_mam_get_form(self, iq: StanzaBase) -> None: 

61 assert isinstance(iq, Iq) 

62 ito = iq.get_to() 

63 

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

65 raise XMPPError( 

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

67 ) 

68 

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

70 await session.bookmarks.by_jid(ito) 

71 

72 reply = iq.reply() 

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

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

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

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

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

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

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

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

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

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

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

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

85 reply.send() 

86 

87 @exceptions_to_xmpp_errors 

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

89 muc = await self.get_muc_from_stanza(iq) 

90 await muc.send_mam_metadata(iq)