Coverage for slidge / core / dispatcher / muc / ping.py: 72%

1from typing import TYPE_CHECKING 

2 

3from slixmpp import CoroutineCallback, Iq, StanzaPath 

4from slixmpp.exceptions import XMPPError 

5 

6from slidge.util.types import AnyMUC 

7 

8from ..util import DispatcherMixin, exceptions_to_xmpp_errors 

9 

10if TYPE_CHECKING: 

11 from slidge.core.gateway import BaseGateway 

12 

13 

14class PingMixin(DispatcherMixin): 

15 __slots__: list[str] = [] 

16 

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

18 super().__init__(xmpp) 

19 

20 xmpp.remove_handler("Ping") 

21 xmpp.register_handler( 

22 CoroutineCallback( 

23 "Ping", 

24 StanzaPath("iq@type=get/ping"), 

25 self.__handle_ping, 

26 ) 

27 ) 

28 xmpp.plugin["xep_0030"].add_feature("urn:xmpp:ping") 

29 

30 @exceptions_to_xmpp_errors 

31 async def __handle_ping(self, iq: Iq) -> None: 

32 ito = iq.get_to() 

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

34 iq.reply().send() 

35 

36 session = await self._get_session(iq) 

37 

38 try: 

39 muc = await session.bookmarks.by_jid(ito) 

40 except XMPPError: 

41 pass 

42 else: 

43 self.__handle_muc_ping(muc, iq) 

44 return 

45 

46 try: 

47 await session.contacts.by_jid(ito) 

48 except XMPPError: 

49 pass 

50 else: 

51 iq.reply().send() 

52 return 

53 

54 raise XMPPError( 

55 "item-not-found", f"This JID does not match anything slidge knows: {ito}" 

56 ) 

57 

58 @staticmethod 

59 def __handle_muc_ping(muc: AnyMUC, iq: Iq) -> None: 

60 if iq.get_from().resource in muc.get_user_resources(): 

61 iq.reply().send() 

62 else: 

63 raise XMPPError("not-acceptable", etype="cancel", by=muc.jid)