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 ....group import LegacyMUC 

7from ..util import DispatcherMixin, exceptions_to_xmpp_errors 

8 

9if TYPE_CHECKING: 

10 from slidge.core.gateway import BaseGateway 

11 

12 

13class PingMixin(DispatcherMixin): 

14 __slots__: list[str] = [] 

15 

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

17 super().__init__(xmpp) 

18 

19 xmpp.remove_handler("Ping") 

20 xmpp.register_handler( 

21 CoroutineCallback( 

22 "Ping", 

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

24 self.__handle_ping, 

25 ) 

26 ) 

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

28 

29 @exceptions_to_xmpp_errors 

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

31 ito = iq.get_to() 

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

33 iq.reply().send() 

34 

35 session = await self._get_session(iq) 

36 

37 try: 

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

39 except XMPPError: 

40 pass 

41 else: 

42 self.__handle_muc_ping(muc, iq) 

43 return 

44 

45 try: 

46 await session.contacts.by_jid(ito) 

47 except XMPPError: 

48 pass 

49 else: 

50 iq.reply().send() 

51 return 

52 

53 raise XMPPError( 

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

55 ) 

56 

57 @staticmethod 

58 def __handle_muc_ping(muc: LegacyMUC, iq: Iq) -> None: 

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

60 iq.reply().send() 

61 else: 

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