Coverage for slidge / core / dispatcher / disco.py: 95%

1import logging 

2from typing import TYPE_CHECKING, Any 

3 

4import sqlalchemy as sa 

5from slixmpp.exceptions import XMPPError 

6from slixmpp.plugins.xep_0030.stanza.items import DiscoItems 

7from slixmpp.types import OptJid 

8 

9from ...db.models import Room 

10from .util import DispatcherMixin 

11 

12if TYPE_CHECKING: 

13 from slidge.core.gateway import BaseGateway 

14 

15 

16class DiscoMixin(DispatcherMixin): 

17 __slots__: list[str] = [] 

18 

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

20 super().__init__(xmpp) 

21 

22 xmpp.plugin["xep_0030"].set_node_handler( 

23 "get_info", 

24 jid=None, 

25 node=None, 

26 handler=self.get_info, 

27 ) 

28 

29 xmpp.plugin["xep_0030"].set_node_handler( 

30 "get_items", 

31 jid=None, 

32 node=None, 

33 handler=self.get_items, 

34 ) 

35 

36 async def get_info(self, jid: OptJid, node: str | None, ifrom: OptJid, data: Any): 

37 if ifrom == self.xmpp.boundjid.bare or jid in (self.xmpp.boundjid.bare, None): 

38 return self.xmpp.plugin["xep_0030"].static.get_info(jid, node, ifrom, data) 

39 

40 if ifrom is None: 

41 raise XMPPError("subscription-required") 

42 

43 assert jid is not None 

44 session = await self._get_session_from_jid(jid=ifrom) 

45 

46 log.debug("Looking for entity: %s", jid) 

47 

48 entity = await session.get_contact_or_group_or_participant(jid) 

49 

50 if entity is None: 

51 raise XMPPError("item-not-found") 

52 

53 return await entity.get_disco_info(jid, node) 

54 

55 async def get_items(self, jid: OptJid, node: str | None, ifrom: OptJid, data: Any): 

56 if ifrom is None: 

57 raise XMPPError("bad-request") 

58 

59 if jid != self.xmpp.boundjid.bare: 

60 return DiscoItems() 

61 

62 assert ifrom is not None 

63 session = await self._get_session_from_jid(ifrom) 

64 

65 d = DiscoItems() 

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

67 for room in orm.execute( 

68 sa.select(Room) 

69 .options(sa.orm.load_only(Room.jid, Room.name)) 

70 .filter_by(user=session.user) 

71 .order_by(Room.name) 

72 ).scalars(): 

73 d.add_item(room.jid, name=room.name) 

74 

75 return d 

76 

77 

78log = logging.getLogger(__name__)