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 typing import Any 

13 

14 from slidge import BaseGateway 

15 

16 

17class MamMixin(DispatcherMixin): 

18 __slots__: list[str] = [] 

19 

20 def __init__(self, xmpp: "BaseGateway[Any]") -> None: 

21 super().__init__(xmpp) 

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

23 xmpp.register_handler( 

24 CoroutineCallback( 

25 "MAM_query", 

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

27 self.__handle_mam, 

28 ) 

29 ) 

30 xmpp.register_handler( 

31 CoroutineCallback( 

32 "MAM_get_from", 

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

34 self.__handle_mam_get_form, 

35 ) 

36 ) 

37 xmpp.register_handler( 

38 CoroutineCallback( 

39 "MAM_get_meta", 

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

41 self.__handle_mam_metadata, 

42 ) 

43 ) 

44 

45 async def __mam_cleanup(self) -> None: 

46 if not config.MAM_MAX_DAYS: 

47 return 

48 while True: 

49 await asyncio.sleep(3600 * 6) 

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

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

52 orm.commit() 

53 

54 @exceptions_to_xmpp_errors 

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

56 muc = await self.get_muc_from_stanza(iq) 

57 await muc.send_mam(iq) 

58 

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

60 assert isinstance(iq, Iq) 

61 ito = iq.get_to() 

62 

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

64 raise XMPPError( 

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

66 ) 

67 

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

69 await session.bookmarks.by_jid(ito) 

70 

71 reply = iq.reply() 

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

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

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

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

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

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

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

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

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

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

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

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

84 reply.send() 

85 

86 @exceptions_to_xmpp_errors 

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

88 muc = await self.get_muc_from_stanza(iq) 

89 await muc.send_mam_metadata(iq)