Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_verify_location_message(base_message):
msg = base_message
msg.type = MsgType.Location
msg.attributes = EFBMsgLocationAttribute(latitude=0.0, longitude=0.0)
msg.verify()
with pytest.raises(ValueError) as exec_info:
msg.attributes = EFBMsgLocationAttribute(latitude='0.0', longitude=1.0)
msg.verify()
assert 'Latitude' in exec_info.value.args[0]
with pytest.raises(ValueError) as exec_info:
msg.attributes = EFBMsgLocationAttribute(latitude=1.0, longitude=10)
msg.verify()
assert 'Longitude' in exec_info.value.args[0]
from typing import Set, Optional, List
from logging import getLogger
from ehforwarderbot import EFBChannel, EFBMsg, EFBStatus, ChannelType, MsgType, EFBChat, ChatType
from ehforwarderbot.exceptions import EFBChatNotFound
from ehforwarderbot.utils import extra
from ehforwarderbot.types import ModuleID, MessageID
class MockSlaveChannel(EFBChannel):
channel_name: str = "Mock Slave"
channel_emoji: str = "➖"
channel_id: ModuleID = ModuleID("tests.mocks.slave.MockSlaveChannel")
channel_type: ChannelType = ChannelType.Slave
supported_message_types: Set[MsgType] = {MsgType.Text, MsgType.Link}
__version__: str = '0.0.1'
logger = getLogger(channel_id)
polling = threading.Event()
__picture_dict = {
"alice": "A.png",
"bob": "B.png",
"carol": "C.png",
"dave": "D.png",
"wonderland001": "W.png"
}
def __init__(self, instance_id=None):
super().__init__(instance_id)
def compare_message(self, tg_msg: Message, efb_msg: EFBMessage) -> None:
assert efb_msg.type == MsgType.Sticker
assert efb_msg.file
assert efb_msg.file.seek(0, 2)
# Cannot compare file size as WebP pictures are converted to PNG here.
from . import __version__ as version
from .Clients.BaseClient import BaseClient
from .ClientMgr import ClientMgr
class QQMessengerChannel(EFBChannel):
channel_name: str = "QQ Slave"
channel_emoji: str = "🐧"
channel_id = "milkice.qq"
channel_type: ChannelType = ChannelType.Slave
__version__ = version.__version__
supported_message_types = {MsgType.Text, MsgType.Sticker, MsgType.Image,
MsgType.Link, MsgType.Voice, MsgType.Animation}
# todo supported_message can be dynamically defined by Client
config = dict()
QQClientMgr: ClientMgr
QQClient: BaseClient
logger: logging.Logger = logging.getLogger(__name__)
def __init__(self, instance_id: str = None):
super().__init__(instance_id)
"""
Load Config
"""
self.load_config()
def check_updates(self):
def qq_image_wrapper(self, data, chat: Chat = None):
efb_msg = Message()
if 'url' not in data:
efb_msg.type = MsgType.Text
efb_msg.text = self._('[Image Source missing]')
return [efb_msg]
efb_msg.file = cq_get_image(data['url'])
if efb_msg.file is None:
efb_msg.type = MsgType.Text
efb_msg.text = self._('[Download image failed, please check on your QQ client]')
return [efb_msg]
efb_msg.type = MsgType.Image
mime = magic.from_file(efb_msg.file.name, mime=True)
if isinstance(mime, bytes):
mime = mime.decode()
efb_msg.filename = data['file'] if 'file' in data else efb_msg.file.name
efb_msg.path = efb_msg.file.name
efb_msg.mime = mime
def wechat_shared_image_msg(self, msg: wxpy.Message, source: str, text: str = "", mode: str = "image") -> Message:
efb_msg = Message()
efb_msg.type = MsgType.Image
efb_msg.text = self._("Via {source}").format(source=source)
if text:
efb_msg.text = "%s\n%s" % (text, efb_msg.text)
efb_msg.path, efb_msg.mime, efb_msg.file = self.save_file(msg, app_message=mode)
return efb_msg
def build_etm_msg(self, chat_manager: ChatObjectCacheManager,
recur: bool = True) -> ETMMsg:
c_module, c_id, _ = chat_id_str_to_id(self.slave_origin_uid)
a_module, a_id, a_grp = chat_id_str_to_id(self.slave_member_uid)
chat: 'ETMChatType' = chat_manager.get_chat(c_module, c_id, build_dummy=True)
author: 'ETMChatMember' = chat_manager.get_chat_member(a_module, a_grp, a_id, build_dummy=True) # type: ignore
msg = ETMMsg(
uid=self.slave_message_id,
chat=chat,
author=author,
text=self.text,
type=MsgType(self.msg_type),
type_telegram=TGMsgType(self.media_type),
mime=self.mime or None,
file_id=self.file_id or None,
)
with suppress(NameError):
to_module = coordinator.get_module_by_id(self.sent_to)
if isinstance(to_module, Channel):
msg.deliver_to = to_module
# - ``target``: ``master_msg_id`` of the target message
# - ``is_system``
# - ``attributes``
# - ``commands``
# - ``substitutions``: ``Dict[Tuple[int, int], SlaveChatID]``
# - ``reactions``: ``Dict[str, Collection[SlaveChatID]]``
if self.pickle:
context['uid_prefix'] = 'group_request'
context['group_name'] = self._('[Request]') + self.get_group_info(context['group_id'])['group_name']
context['group_id_orig'] = context['group_id']
context['group_id'] = str(context['group_id']) + "_notification"
context['message_type'] = 'group'
context['event_description'] = '\u2139 New Group Join Request'
original_group = self.get_group_info(context['group_id'], False)
group_name = context['group_id']
if original_group is not None and 'group_name' in original_group:
group_name = original_group['group_name']
msg = Message()
msg.uid = 'group' + '_' + str(context['group_id'])
msg.author = self.chat_manager.build_efb_chat_as_system_user(context)
msg.chat = self.chat_manager.build_efb_chat_as_group(context)
msg.deliver_to = coordinator.master
msg.type = MsgType.Text
name = ""
if not self.get_friend_remark(context['user_id']):
name = "{}({})[{}] ".format(
self.get_stranger_info(context['user_id'])['nickname'], self.get_friend_remark(context['user_id']),
context['user_id'])
else:
name = "{}[{}] ".format(self.get_stranger_info(context['user_id'])['nickname'], context['user_id'])
msg.text = "{} wants to join the group {}({}). \nHere is the comment: {}".format(
name, group_name, context['group_id_orig'], context['comment']
)
msg.commands = MessageCommands([MessageCommand(
name=self._("Accept"),
callable_name="process_group_request",
kwargs={'result': 'accept',
'flag': context['flag'],
'sub_type': context['sub_type']}
def __init__(self, attributes: Optional[MessageAttribute] = None, author: ChatMember = None, chat: Chat = None,
commands: Optional[MessageCommands] = None, deliver_to: Channel = None, edit: bool = False,
edit_media: bool = False, file: Optional[BinaryIO] = None, filename: Optional[str] = None,
is_system: bool = False, mime: Optional[str] = None, path: Optional[Path] = None,
reactions: Reactions = None, substitutions: Optional[Substitutions] = None,
target: 'Optional[Message]' = None, text: str = "", type: MsgType = MsgType.Unsupported,
uid: Optional[MessageID] = None, vendor_specific: Dict[str, Any] = None,
type_telegram: TGMsgType = TGMsgType.System, file_id: Optional[str] = None):
super().__init__(attributes=attributes, chat=chat, author=author, commands=commands, deliver_to=deliver_to,
edit=edit, edit_media=edit_media, file=file, filename=filename, is_system=is_system, mime=mime,
path=path, reactions=reactions, substitutions=substitutions, target=target, text=text,
type=type, uid=uid, vendor_specific=vendor_specific)
self.__initialized = False
self.type_telegram = type_telegram
self.file_id = file_id