Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def base_message(chat, master_channel):
msg = EFBMsg()
msg.deliver_to = master_channel
msg.author = chat
msg.chat = chat
msg.text = "Message"
msg.uid = "0"
return msg
def test_pickle_message_removal(slave_channel, master_channel):
msg = EFBMsg()
msg.chat = slave_channel.alice
msg.uid = "uid"
msg_removal = EFBMessageRemoval(
source_channel=master_channel,
destination_channel=slave_channel,
message=msg
)
msg_removal_dup = pickle.loads(pickle.dumps(msg_removal))
# Assume EFBMsg is picklable
for i in ("source_channel", "destination_channel"):
assert getattr(msg_removal, i) == getattr(msg_removal_dup, i)
def test_verify_missing_chat(chat):
msg = EFBMsg()
msg.deliver_to = coordinator.master
msg.author = chat
msg.type = MsgType.Text
msg.text = "Message"
with pytest.raises(ValueError):
msg.verify()
def test_verify_missing_deliver_to(chat):
msg = EFBMsg()
msg.author = chat
msg.chat = chat
msg.type = MsgType.Text
msg.text = "Message"
with pytest.raises(ValueError):
msg.verify()
def test_verify_same_author_and_chat(patch_chat_0, master_channel):
msg = EFBMsg()
msg.deliver_to = master_channel
msg.author = patch_chat_0
msg.chat = patch_chat_0
msg.text = "Message"
msg.verify()
patch_chat_0.verify.assert_called_once()
def send_link_msg(self):
slave = next(iter(coordinator.slaves.values()))
alice = slave.get_chat('alice')
msg = EFBMsg()
msg.deliver_to = slave
msg.chat = alice
msg.author = EFBChat(self).self()
msg.type = MsgType.Link
msg.text = "Check it out."
msg.attributes = EFBMsgLinkAttribute(
title="Example",
url="https://example.com"
)
return coordinator.send_message(msg)
def send_text_msg(self):
slave = next(iter(coordinator.slaves.values()))
wonderland = slave.get_chat('wonderland001')
msg = EFBMsg()
msg.deliver_to = slave
msg.chat = wonderland
msg.author = EFBChat(self).self()
msg.type = MsgType.Text
msg.text = "Hello, world."
return coordinator.send_message(msg)
def send_location_msg(self):
slave = next(iter(coordinator.slaves.values()))
alice = slave.get_chat('alice')
msg = EFBMsg()
msg.deliver_to = slave
msg.chat = alice
msg.author = EFBChat(self).self()
msg.type = MsgType.Location
msg.text = "I'm not here."
msg.attributes = EFBMsgLocationAttribute(latitude=0.1, longitude=1.0)
return coordinator.send_message(msg)
def reply_message(self, message: EFBMsg, text: str):
reply = EFBMsg()
reply.text = text
reply.chat = coordinator.slaves[message.chat.channel_id].get_chat(message.chat.chat_uid)
reply.author = self.chat
reply.type = MsgType.Text
reply.deliver_to = coordinator.master
reply.target = message
reply.uid = str(uuid.uuid4())
coordinator.send_message(reply)
def verify(self):
if self.source_channel is None or not isinstance(self.source_channel, EFBChannel):
raise ValueError("Source channel is not valid.")
if self.destination_channel is None or not isinstance(self.destination_channel, EFBChannel):
raise ValueError("Destination channel is not valid.")
if self.message is None or not isinstance(self.message, EFBMsg):
raise ValueError("Message channel is not valid.")
if not self.message.chat.module_id or not self.message.chat.chat_uid or not self.message.uid:
raise ValueError("Message does not contain the minimum information required")