mirror of
https://github.com/chatmail/relay.git
synced 2026-05-18 07:48:58 +00:00
add support for using a second chatmail server
This commit is contained in:
@@ -1,6 +1,7 @@
|
|||||||
import os
|
import os
|
||||||
import io
|
import io
|
||||||
import random
|
import random
|
||||||
|
import contextlib
|
||||||
import subprocess
|
import subprocess
|
||||||
import imaplib
|
import imaplib
|
||||||
import smtplib
|
import smtplib
|
||||||
@@ -34,6 +35,19 @@ def sshdomain(maildomain):
|
|||||||
return os.environ.get("CHATMAIL_SSH", maildomain)
|
return os.environ.get("CHATMAIL_SSH", maildomain)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def maildomain2():
|
||||||
|
domain = os.environ.get("CHATMAIL_DOMAIN2")
|
||||||
|
if not domain:
|
||||||
|
pytest.skip("set CHATMAIL_DOMAIN2 to a ssh-reachable chatmail instance")
|
||||||
|
return domain
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def sshdomain2(maildomain2):
|
||||||
|
return os.environ.get("CHATMAIL_SSH2", maildomain2)
|
||||||
|
|
||||||
|
|
||||||
def pytest_report_header():
|
def pytest_report_header():
|
||||||
domain = os.environ.get("CHATMAIL_DOMAIN")
|
domain = os.environ.get("CHATMAIL_DOMAIN")
|
||||||
if domain:
|
if domain:
|
||||||
@@ -93,7 +107,9 @@ def gencreds(maildomain):
|
|||||||
count = itertools.count()
|
count = itertools.count()
|
||||||
next(count)
|
next(count)
|
||||||
|
|
||||||
def gen():
|
def gen(domain=None):
|
||||||
|
if domain is not None:
|
||||||
|
maildomain = domain
|
||||||
while 1:
|
while 1:
|
||||||
num = next(count)
|
num = next(count)
|
||||||
alphanumeric = "abcdefghijklmnopqrstuvwxyz1234567890"
|
alphanumeric = "abcdefghijklmnopqrstuvwxyz1234567890"
|
||||||
@@ -102,7 +118,7 @@ def gencreds(maildomain):
|
|||||||
password = "".join(random.choices(alphanumeric, k=10))
|
password = "".join(random.choices(alphanumeric, k=10))
|
||||||
yield f"{user}@{maildomain}", f"{password}"
|
yield f"{user}@{maildomain}", f"{password}"
|
||||||
|
|
||||||
return lambda: next(gen())
|
return lambda domain=None: next(gen(domain))
|
||||||
|
|
||||||
|
|
||||||
#
|
#
|
||||||
@@ -117,12 +133,13 @@ class ChatmailTestProcess:
|
|||||||
def __init__(self, pytestconfig, maildomain, gencreds):
|
def __init__(self, pytestconfig, maildomain, gencreds):
|
||||||
self.pytestconfig = pytestconfig
|
self.pytestconfig = pytestconfig
|
||||||
self.maildomain = maildomain
|
self.maildomain = maildomain
|
||||||
|
assert "." in self.maildomain, maildomain
|
||||||
self.gencreds = gencreds
|
self.gencreds = gencreds
|
||||||
self._addr2files = {}
|
self._addr2files = {}
|
||||||
|
|
||||||
def get_liveconfig_producer(self):
|
def get_liveconfig_producer(self):
|
||||||
while 1:
|
while 1:
|
||||||
user, password = self.gencreds()
|
user, password = self.gencreds(self.maildomain)
|
||||||
config = {
|
config = {
|
||||||
"addr": user,
|
"addr": user,
|
||||||
"mail_pw": password,
|
"mail_pw": password,
|
||||||
@@ -140,7 +157,25 @@ class ChatmailTestProcess:
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def cmfactory(request, maildomain, gencreds, tmpdir, data):
|
def switch_maildomain():
|
||||||
|
"""return a function that allows to switch an account factory temporarily
|
||||||
|
to another maildomain.
|
||||||
|
"""
|
||||||
|
|
||||||
|
# nb. a bit hacky
|
||||||
|
# would probably be better if deltachat's test machinery grows native support
|
||||||
|
@contextlib.contextmanager
|
||||||
|
def switch(acfactory, maildomain2):
|
||||||
|
old_domain = acfactory.testprocess.maildomain
|
||||||
|
acfactory.testprocess.maildomain = maildomain2
|
||||||
|
yield
|
||||||
|
acfactory.testprocess.maildomain = old_domain
|
||||||
|
|
||||||
|
return switch
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def cmfactory(request, gencreds, tmpdir, data, maildomain):
|
||||||
# cloned from deltachat.testplugin.amfactory
|
# cloned from deltachat.testplugin.amfactory
|
||||||
pytest.importorskip("deltachat")
|
pytest.importorskip("deltachat")
|
||||||
from deltachat.testplugin import ACFactory
|
from deltachat.testplugin import ACFactory
|
||||||
|
|||||||
@@ -2,3 +2,13 @@ def test_remotelog(remotelog, imap_or_smtp):
|
|||||||
lineproducer = remotelog.iter(imap_or_smtp.logunit)
|
lineproducer = remotelog.iter(imap_or_smtp.logunit)
|
||||||
imap_or_smtp.connect()
|
imap_or_smtp.connect()
|
||||||
assert imap_or_smtp.logunit in next(lineproducer)
|
assert imap_or_smtp.logunit in next(lineproducer)
|
||||||
|
|
||||||
|
|
||||||
|
def test_use_two_chatmailservers(cmfactory, maildomain2, switch_maildomain):
|
||||||
|
(ac1,) = cmfactory.get_online_accounts(1)
|
||||||
|
with switch_maildomain(cmfactory, maildomain2):
|
||||||
|
(ac2,) = cmfactory.get_online_accounts(1)
|
||||||
|
cmfactory.get_accepted_chat(ac1, ac2)
|
||||||
|
domain1 = ac1.get_config("addr").split("@")[1]
|
||||||
|
domain2 = ac2.get_config("addr").split("@")[1]
|
||||||
|
assert domain1 != domain2
|
||||||
|
|||||||
Reference in New Issue
Block a user