Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-05-08 01:16:31 +03:00
Code Issues Packages Projects Releases Wiki Activity

refactor bot-setup and testing into a helper function

Browse Source
This commit is contained in:
holger krekel
2020-03-07 09:45:39 +01:00
parent 6c3a8448cf
commit d4ba09c753
2 changed files with 43 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
python/examples/test_examples.py
View File

@@ -1,5 +1,4 @@
import threading
import pytest import pytest
import py import py
import echo_and_quit import echo_and_quit
@@ -18,56 +17,32 @@ def datadir():
def test_echo_quit_plugin(acfactory): def test_echo_quit_plugin(acfactory):
bot_ac, bot_cfg = acfactory.get_online_config() botproc = acfactory.run_bot_process(echo_and_quit)
def run_bot():
print("*"*20 + " starting bot")
echo_and_quit.main([
"echo",
"--show-ffi",
"--db", bot_ac.db_path,
"--email", bot_cfg["addr"],
"--password", bot_cfg["mail_pw"],
])
t = threading.Thread(target=run_bot)
t.start()
ac1 = acfactory.get_one_online_account() ac1 = acfactory.get_one_online_account()
bot_contact = ac1.create_contact(bot_cfg["addr"]) bot_contact = ac1.create_contact(botproc.addr)
ch1 = ac1.create_chat_by_contact(bot_contact) ch1 = ac1.create_chat_by_contact(bot_contact)
ch1.send_text("hello") ch1.send_text("hello")
reply = ac1._evtracker.wait_next_incoming_message() reply = ac1._evtracker.wait_next_incoming_message()
assert "hello" in reply.text assert "hello" in reply.text
assert reply.chat == ch1 assert reply.chat == ch1
ch1.send_text("/quit") ch1.send_text("/quit")
t.join() botproc.wait()
@pytest.mark.skip(reason="not implemented") @pytest.mark.skip(reason="botproc-matching not implementing")
def test_group_tracking_plugin(acfactory): def test_group_tracking_plugin(acfactory):
bot_ac, bot_cfg = acfactory.get_online_config() botproc = acfactory.run_bot_process(group_tracking)
def run_bot():
print("*"*20 + " starting bot")
print("*"*20 + " bot_ac.dbpath", bot_ac.db_path)
group_tracking.main([
"group-tracking",
"--show-ffi", bot_ac.db_path,
"--db", bot_ac.db_path,
"--email", bot_cfg["addr"],
"--password", bot_cfg["mail_pw"],
])
t = threading.Thread(target=run_bot)
t.setDaemon(1)
t.start()
ac1 = acfactory.get_one_online_account() ac1 = acfactory.get_one_online_account()
bot_contact = ac1.create_contact(bot_cfg["addr"]) bot_contact = ac1.create_contact(botproc.addr)
ch1 = ac1.create_chat_by_contact(bot_contact) ch1 = ac1.create_group_chat("bot test group")
ch1.add_contact(bot_contact)
ch1.send_text("hello") ch1.send_text("hello")
ch1.add_contact(ac1.create_contact("x@example.org")) ch1.add_contact(ac1.create_contact("x@example.org"))
# XXX wait for bot to receive things botproc.fnmatch_lines("""
t.join() *member_added x@example.org*
""")
botproc.kill()

30
python/src/deltachat/testplugin.py
View File

@@ -1,6 +1,7 @@
from __future__ import print_function from __future__ import print_function
import os import os
import sys import sys
import subprocess
import pytest import pytest
import requests import requests
import time import time
@@ -241,11 +242,40 @@ def acfactory(pytestconfig, tmpdir, request, session_liveconfig, datadir):
ac.start() ac.start()
return ac return ac
def run_bot_process(self, module):
fn = module.__file__
bot_ac, bot_cfg = self.get_online_config()
popen = subprocess.Popen([
sys.executable,
fn,
"--show-ffi",
"--db", bot_ac.db_path,
"--email", bot_cfg["addr"],
"--password", bot_cfg["mail_pw"],
])
bot = BotProcess(popen, bot_cfg)
self._finalizers.append(bot.kill)
return bot
am = AccountMaker() am = AccountMaker()
request.addfinalizer(am.finalize) request.addfinalizer(am.finalize)
return am return am
class BotProcess:
def __init__(self, popen, bot_cfg):
self.popen = popen
self.addr = bot_cfg["addr"]
def kill(self):
self.popen.kill()
def wait(self, timeout=30):
self.popen.wait(timeout=timeout)
@pytest.fixture @pytest.fixture
def tmp_db_path(tmpdir): def tmp_db_path(tmpdir):
return tmpdir.join("test.db").strpath return tmpdir.join("test.db").strpath