Skip to content
Permalink
Browse files

test: unit test (on just one method that I found I could write someth…

…ing)
  • Loading branch information
blueset committed Jan 1, 2020
1 parent 053fd49 commit adc2b660b8db735105bd7043dad1000db7502db8
@@ -321,7 +321,7 @@ def send_message(self, msg: EFBMsg) -> EFBMsg:
failed = 0
for i in msg_ids:
try:
ews_utils.message_to_dummy_message(i, self).recall()
ews_utils.message_id_to_dummy_message(i, self).recall()
except wxpy.ResponseError as e:
self.logger.error("[%s] Trying to recall message but failed: %s", msg.uid, e)
failed += 1
@@ -468,7 +468,7 @@ def send_status(self, status: EFBStatus):
)
for i in msg_ids:
try:
ews_utils.message_to_dummy_message(i, self).recall()
ews_utils.message_id_to_dummy_message(i, self).recall()
except wxpy.ResponseError:
failed += 1
if failed:
@@ -581,7 +581,7 @@ def force_log_out(self, _: str = "") -> str:
self.exit_callback()
return self._("Done.")

# Command functions
# region [Command functions]

def reauth(self, command=False):
msg = self._("Preparing to log in...")
@@ -592,6 +592,8 @@ def reauth(self, command=False):
threading.Thread(target=self.authenticate, args=(qr_reload,), name="EWS reauth thread").start()
return msg

# endregion [Command functions]

def authenticate(self, qr_reload, first_start=False):
qr_callback = getattr(self, qr_reload, self.master_qr_code)
if getattr(self, 'bot', None): # if a bot exists
@@ -1,3 +1,3 @@
# coding: utf-8

__version__ = "2.0.0a36.dev5"
__version__ = "2.0.0a36.dev6"
@@ -59,23 +59,6 @@ def __init__(self, channel: 'WeChatChannel'):
def bot(self):
return self.channel.bot

def get_chat_by_puid(self, puid: str) -> EFBChat:
if puid in wxpy.Chat.SYSTEM_ACCOUNTS:
efb_chat = self.wxpy_chat_to_efb_chat(wxpy.Chat(wxpy.utils.wrap_user_name(puid), self.bot))
assert efb_chat is not None
efb_chat.chat_type = ChatType.System
return efb_chat
try:
chat: wxpy.Chat = wxpy.ensure_one(self.bot.search(puid=puid))
return self.wxpy_chat_to_efb_chat(chat)
except ValueError:
try:
self.bot.chats(update=True)
chat = wxpy.ensure_one(self.bot.search(puid=puid))
return self.wxpy_chat_to_efb_chat(chat)
except ValueError:
return self.MISSING_CHAT

def get_wxpy_chat_by_uid(self, uid: str) -> wxpy.Chat:
if uid in wxpy.Chat.SYSTEM_ACCOUNTS:
return wxpy.Chat(wxpy.utils.wrap_user_name(uid), self.bot)
@@ -131,6 +114,7 @@ def wxpy_chat_to_efb_chat(self, chat: wxpy.Chat, recursive=True) -> EFBChat:
if isinstance(chat, wxpy.Member):
efb_chat.chat_type = ChatType.User
efb_chat.is_chat = False
efb_chat.has_self = False
if recursive:
efb_chat.group = self.wxpy_chat_to_efb_chat(chat.group, False)
elif isinstance(chat, wxpy.Group):
@@ -163,6 +147,7 @@ def get_chats(self) -> List[EFBChat]:
return l

def search_chat(self, uid: str, refresh: bool = False) -> EFBChat:
"""Search chat by temporary UserName."""
try:
if refresh:
self.bot.chats(True)
@@ -1,28 +1,28 @@
# coding: utf-8

import json
import logging
import re
import tempfile
import uuid
import threading
import re
import json
import uuid
from pathlib import Path
from typing import TYPE_CHECKING, Callable, Optional, Tuple, IO, Dict, List
from typing import TYPE_CHECKING, Callable, Optional, Tuple, IO, Dict
from xml.etree import ElementTree as ETree
from xml.etree.ElementTree import Element

import magic
import requests
from xml.etree import ElementTree as ETree
from PIL import Image

from ehforwarderbot import EFBMsg, MsgType, EFBChat, coordinator
from ehforwarderbot.status import EFBMessageRemoval
from ehforwarderbot.message import EFBMsgLocationAttribute, EFBMsgLinkAttribute, EFBMsgCommands, EFBMsgCommand, \
EFBMsgSubstitutions
from ehforwarderbot.status import EFBMessageRemoval
from ehforwarderbot.types import MessageID
from . import constants
from . import utils as ews_utils
from .vendor import wxpy, itchat
from .vendor import wxpy
from .vendor.wxpy.api import consts

if TYPE_CHECKING:
@@ -160,7 +160,7 @@ def generate_message_uid(messages: List[wxpy.SentMessage]) -> MessageID:
))


def message_to_dummy_message(message_uid: List[str], channel: 'WeChatChannel') -> wxpy.SentMessage:
def message_id_to_dummy_message(message_uid: List[str], channel: 'WeChatChannel') -> wxpy.SentMessage:
"""
Generate a wxpy.SentMessage object with minimum identifying information.
This is generally used to recall messages using WXPY's API without the message object
@@ -1,6 +1,5 @@
import logging
import queue
from queue import _T
from typing import Optional

from .templates import AttributeDict
@@ -173,7 +173,7 @@ def is_at(self):
"""
当消息来自群聊,且被 @ 时,为 True
"""
return self.raw.get('IsAt') or self.raw.get('isAt')
return self.raw.get('IsAt')

# misc

@@ -0,0 +1,20 @@
from xml.etree import ElementTree as ETree

from efb_wechat_slave.slave_message import SlaveMessageManager


def test_slave_message_get_node_text():
xml = """<?xml version="1.0" encoding="UTF-8"?>
<root>
<item>
<child>Text</child>
</item>
<emptyItem/>
</root>
"""

root = ETree.fromstring(xml)

assert SlaveMessageManager.get_node_text(root, "./item/child", "") == "Text"
assert SlaveMessageManager.get_node_text(root, "./item/non_existing", "fallback") == "fallback"
assert SlaveMessageManager.get_node_text(root, "./emptyItem", "fallback") == "fallback"

0 comments on commit adc2b66

Please sign in to comment.
You can’t perform that action at this time.