Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-04-19 06:26:30 +03:00
Code Issues Packages Projects Releases Wiki Activity

add Client.run_until()

Browse Source
This commit is contained in:
adbenitez
2022-12-11 03:31:29 -05:00
parent be63e18ebf
commit 2ebd3f54e6
3 changed files with 111 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
deltachat-rpc-client/tests/test_something.py
View File

@@ -1,6 +1,8 @@
from unittest.mock import MagicMock
import pytest
from deltachat_rpc_client import AttrDict, EventType, events
from deltachat_rpc_client import EventType, events
from deltachat_rpc_client.rpc import JsonRpcError
@@ -216,40 +218,46 @@ async def test_message(acfactory) -> None:
@pytest.mark.asyncio
async def test_bot(acfactory) -> None:
async def callback(e):
res.append(e)
def track(key):
async def wrapper(e):
mock.hook(e[key])
res = []
return wrapper
mock = MagicMock()
user = (await acfactory.get_online_accounts(1))[0]
bot = await acfactory.new_configured_bot()
assert await bot.is_configured()
assert await bot.account.get_config("bot") == "1"
bot.add_hook(callback, events.RawEvent(EventType.INFO))
info_event = AttrDict(account=bot.account, type=EventType.INFO, msg="info")
warn_event = AttrDict(account=bot.account, type=EventType.WARNING, msg="warning")
await bot._on_event(info_event)
await bot._on_event(warn_event)
assert info_event in res
assert warn_event not in res
assert len(res) == 1
hook = track("msg_id"), events.RawEvent(EventType.INCOMING_MSG)
bot.add_hook(*hook)
event = await acfactory.process_message(
from_account=user, to_client=bot, text="Hello!"
)
mock.hook.assert_called_once_with(event.id)
bot.remove_hook(*hook)
res = []
bot.add_hook(callback, events.NewMessage(r"hello"))
bot.add_hook(callback, events.NewMessage(command="/help"))
snapshot1 = AttrDict(text="hello", command=None)
snapshot2 = AttrDict(text="hello, world", command=None)
snapshot3 = AttrDict(text="hey!", command=None)
for snapshot in [snapshot1, snapshot2, snapshot3]:
await bot._on_event(snapshot, events.NewMessage)
assert len(res) == 2
assert snapshot1 in res
assert snapshot2 in res
assert snapshot3 not in res
mock.hook.reset_mock()
hook = track("id"), events.NewMessage(r"hello")
bot.add_hook(*hook)
bot.add_hook(track("id"), events.NewMessage(command="/help"))
event = await acfactory.process_message(
from_account=user, to_client=bot, text="hello"
)
mock.hook.assert_called_with(event.id)
event = await acfactory.process_message(
from_account=user, to_client=bot, text="hello!"
)
mock.hook.assert_called_with(event.id)
await acfactory.process_message(from_account=user, to_client=bot, text="hey!")
assert len(mock.hook.mock_calls) == 2
bot.remove_hook(*hook)
res = []
bot.remove_hook(callback, events.NewMessage(r"hello"))
snapshot4 = AttrDict(command="/help")
await bot._on_event(snapshot, events.NewMessage)
await bot._on_event(snapshot4, events.NewMessage)
assert len(res) == 1
assert snapshot4 in res
mock.hook.reset_mock()
await acfactory.process_message(from_account=user, to_client=bot, text="hello")
event = await acfactory.process_message(
from_account=user, to_client=bot, text="/help"
)
mock.hook.assert_called_once_with(event.id)