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

42 return xmpp_id 

43 

44 

45class MessageMixin: 

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

47 """ 

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

49 :term:`Recipient`. 

50 

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

52 """ 

53 raise NotImplementedError 

54 

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

56 """ 

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

58 :term:`Recipient`. 

59 

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

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

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

63 or 

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

65 

66 :param legacy_msg_id: Identifier of the message/ 

67 :param thread: 

68 """ 

69 raise NotImplementedError 

70 

71 async def on_react( 

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

73 ) -> None: 

74 """ 

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

76 this :term:`Recipient`. 

77 

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

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

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

81 :param thread: 

82 """ 

83 raise NotImplementedError 

84 

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

86 """ 

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

88 

89 :param legacy_msg_id: Legacy ID of the retracted message 

90 :param thread: 

91 """ 

92 raise NotImplementedError 

93 

94 

95class RecipientMixin(ReactionMixin, ThreadMixin, MessageMixin): 

96 jid: JID 

97 is_group: bool 

98 

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

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