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

1import abc 

2 

3from slixmpp import JID 

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

5 

6from ...util.types import ChatState, Sticker, XMPPMessage 

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) -> str: 

42 return xmpp_id 

43 

44 

45class MessageMixin: 

46 @abc.abstractmethod 

47 async def on_message(self, message: XMPPMessage) -> str | None: 

48 """ 

49 Triggered when the user sends a message to this :term:`Recipient`. 

50 """ 

51 raise NotImplementedError 

52 

53 @abc.abstractmethod 

54 async def on_sticker(self, sticker: Sticker) -> str | None: 

55 """ 

56 Triggered when the user sends a sticker to this :term:`Recipient`. 

57 

58 :param sticker: The sticker sent by the user. 

59 

60 :return: An ID of some sort that can be used later to ack and mark the 

61 message as read by the user 

62 """ 

63 raise NotImplementedError 

64 

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

66 """ 

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

68 :term:`Recipient`. 

69 

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

71 """ 

72 raise NotImplementedError 

73 

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

75 """ 

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

77 :term:`Recipient`. 

78 

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

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

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

82 or 

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

84 

85 :param legacy_msg_id: Identifier of the message/ 

86 :param thread: 

87 """ 

88 raise NotImplementedError 

89 

90 async def on_react( 

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

92 ) -> None: 

93 """ 

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

95 this :term:`Recipient`. 

96 

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

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

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

100 :param thread: 

101 """ 

102 raise NotImplementedError 

103 

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

105 """ 

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

107 

108 :param legacy_msg_id: Legacy ID of the retracted message 

109 :param thread: 

110 """ 

111 raise NotImplementedError 

112 

113 

114class RecipientMixin(ReactionMixin, ThreadMixin, MessageMixin): 

115 jid: JID 

116 

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

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