Coverage for slidge/core/mixins/recipient.py: 97%

1import abc 

2 

3from slixmpp import JID 

4from slixmpp.plugins.xep_0004.stanza.form import Form 

5 

6from ...util.types import ChatState 

7 

8 

9class ReactionMixin: 

10 REACTIONS_SINGLE_EMOJI = False 

11 

12 async def restricted_emoji_extended_feature(self) -> Form | None: 

13 available = await self.available_emojis() 

14 if not self.REACTIONS_SINGLE_EMOJI and available is None: 

15 return None 

16 

17 form = Form() 

18 form["type"] = "result" 

19 form.add_field("FORM_TYPE", "hidden", value="urn:xmpp:reactions:0:restrictions") 

20 if self.REACTIONS_SINGLE_EMOJI: 

21 form.add_field("max_reactions_per_user", value="1", type="text-single") 

22 if available: 

23 form.add_field("allowlist", value=list(available), type="text-multi") 

24 return form 

25 

26 @abc.abstractmethod 

27 async def available_emojis( 

28 self, legacy_msg_id: str | None = None 

29 ) -> set[str] | None: 

30 """ 

31 Override this to restrict the subset of reactions this recipient 

32 can handle. 

33 

34 :return: A set of emojis or None if any emoji is allowed 

35 """ 

36 return None 

37 

38 

39class ThreadMixin: 

40 @abc.abstractmethod 

41 async def create_thread(self, xmpp_id: str, root_legacy_id: str) -> str: 

42 """Create a legacy thread rooted at an existing message. 

43 

44 :param root_legacy_id: legacy_id of the message which shall be the root of the 

45 thread. 

46 """ 

47 return root_legacy_id 

48 

49 

50class MessageMixin: 

51 async def on_chat_state(self, chat_state: ChatState, thread: str | None) -> None: 

52 """ 

53 Triggered when the user sends a chat state (:xep:`0085`) to this 

54 :term:`Recipient`. 

55 

56 :param chat_state: The sticker sent by the user. 

57 """ 

58 raise NotImplementedError 

59 

60 async def on_displayed(self, legacy_msg_id: str, thread: str | None) -> None: 

61 """ 

62 Triggered when the user sends a read marker (:xep:`0333`) to this 

63 :term:`Recipient`. 

64 

65 This is only possible if a valid ``legacy_msg_id`` was passed when 

66 transmitting a message from a legacy chat to the user, eg in 

67 :meth:`slidge.contact.LegacyContact.send_text` 

68 or 

69 :meth:`slidge.group.LegacyParticipant.send_text`. 

70 

71 :param legacy_msg_id: Identifier of the message/ 

72 :param thread: 

73 """ 

74 raise NotImplementedError 

75 

76 async def on_react( 

77 self, legacy_msg_id: str, emojis: list[str], thread: str | None 

78 ) -> None: 

79 """ 

80 Triggered when the user sends an emoji reaction (:xep:`0444`) to 

81 this :term:`Recipient`. 

82 

83 :param legacy_msg_id: ID of the message the user reacts to 

84 :param emojis: Unicode characters representing reactions to the message ``legacy_msg_id``. 

85 An empty list means "no reaction", ie, remove all reactions if any were present before 

86 :param thread: 

87 """ 

88 raise NotImplementedError 

89 

90 async def on_retract(self, legacy_msg_id: str, thread: str | None) -> None: 

91 """ 

92 Triggered when the user retracts (:xep:`0424`) a message. 

93 

94 :param legacy_msg_id: Legacy ID of the retracted message 

95 :param thread: 

96 """ 

97 raise NotImplementedError 

98 

99 

100class RecipientMixin(ReactionMixin, ThreadMixin, MessageMixin): 

101 jid: JID 

102 is_group: bool 

103 

104 def __eq__(self, other: object) -> bool: 

105 return isinstance(other, self.__class__) and self.jid == other.jid