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

1import logging 

2from typing import TYPE_CHECKING, Any, Optional 

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( 

37 self, jid: OptJid, node: Optional[str], ifrom: OptJid, data: Any 

38 ): 

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

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

41 

42 if ifrom is None: 

43 raise XMPPError("subscription-required") 

44 

45 assert jid is not None 

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

47 

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

49 

50 entity = await session.get_contact_or_group_or_participant(jid) 

51 

52 if entity is None: 

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

54 

55 return await entity.get_disco_info(jid, node) 

56 

57 async def get_items( 

58 self, jid: OptJid, node: Optional[str], ifrom: OptJid, data: Any 

59 ): 

60 if ifrom is None: 

61 raise XMPPError("bad-request") 

62 

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

64 return DiscoItems() 

65 

66 assert ifrom is not None 

67 session = await self._get_session_from_jid(ifrom) 

68 

69 d = DiscoItems() 

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

71 for room in orm.execute( 

72 sa.select(Room) 

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

74 .filter_by(user=session.user) 

75 .order_by(Room.name) 

76 ).scalars(): 

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

78 

79 return d 

80 

81 

82log = logging.getLogger(__name__)