Coverage for slidge / slixfix / roster.py: 88%

1import logging 

2from collections.abc import Iterable 

3from typing import Any, Generic, TypeVar 

4 

5from slixmpp import JID 

6from slixmpp.types import JidStr, OptJidStr, RosterState 

7 

8GatewayType = TypeVar("GatewayType") 

9 

10 

11class YesSet(set): 

12 """ 

13 A pseudo-set which always test True for membership 

14 """ 

15 

16 def __contains__(self, item: object) -> bool: 

17 log.debug("Test in") 

18 return True 

19 

20 

21class RosterBackend(Generic[GatewayType]): 

22 """ 

23 A pseudo-roster for the gateway component. 

24 

25 If a user is in the user store, this will behave as if the user is part of the 

26 roster with subscription "both", and "none" otherwise. 

27 

28 This is rudimentary but the only sane way I could come up with so far. 

29 """ 

30 

31 xmpp: GatewayType 

32 

33 def __init__(self, xmpp: GatewayType) -> None: 

34 self.xmpp = xmpp 

35 

36 def entries( 

37 self, owner: OptJidStr, db_state: dict[str, Any] | None = None 

38 ) -> Iterable[str]: 

39 return YesSet() 

40 

41 def save( 

42 self, owner: JidStr, jid: JidStr, state: RosterState, db_state: dict[str, Any] 

43 ) -> None: 

44 pass 

45 

46 def load( 

47 self, owner: JidStr, jid: JidStr, db_state: dict[str, Any] 

48 ) -> RosterState | None: 

49 session = self.xmpp.get_session_from_jid(JID(jid)) 

50 if session is None: 

51 return { 

52 "name": "", 

53 "groups": [], 

54 "from": False, 

55 "to": False, 

56 "pending_in": False, 

57 "pending_out": False, 

58 "whitelisted": False, 

59 "subscription": "both", 

60 } 

61 else: 

62 return { 

63 "name": "", 

64 "groups": [], 

65 "from": True, 

66 "to": True, 

67 "pending_in": False, 

68 "pending_out": False, 

69 "whitelisted": False, 

70 "subscription": "none", 

71 } 

72 

73 

74log = logging.getLogger(__name__)