Commit 843fd599 authored by Aurélien Bompard's avatar Aurélien Bompard

Add a FakeMailmanClient which connects to the in-memory REST server

parent 8599c9ae
......@@ -640,12 +640,7 @@ Message Moderation
... Some text.
...
... """
>>> server = smtplib.LMTP('localhost', 8024)
>>> server.sendmail('nomember@example.com', 'test-one@example.com', msg)
{}
>>> server.quit()
(221, 'Bye')
>>> time.sleep(2)
>>> inject_message('test-one@example.com', msg)
Messages held for moderation can be listed on a per list basis.
......
......@@ -17,7 +17,7 @@
"""Test harness for doctests."""
from __future__ import absolute_import, unicode_literals
from __future__ import absolute_import, unicode_literals, print_function
__metaclass__ = type
__all__ = [
......@@ -33,10 +33,21 @@ import doctest
import tempfile
import unittest
import subprocess
from textwrap import dedent
from mock import patch
# pylint: disable-msg=F0401
from pkg_resources import (
resource_filename, resource_exists, resource_listdir, cleanup_resources)
from zope.component import getUtility
from mailman.config import config
from mailman.testing.layers import ConfigLayer
from mailman.interfaces.listmanager import IListManager
from mailman.interfaces.usermanager import IUserManager
from mailman.core.chains import process
from mailman.testing.helpers import specialized_message_from_string
from mailmanclient.tests.utils import FakeMailmanClient
COMMASPACE = ', '
......@@ -50,17 +61,17 @@ DOCTEST_FLAGS = (
def dump(results):
if results is None:
print None
print(None)
return
for key in sorted(results):
if key == 'entries':
for i, entry in enumerate(results[key]):
# entry is a dictionary.
print 'entry %d:' % i
print('entry %d:' % i)
for entry_key in sorted(entry):
print ' {0}: {1}'.format(entry_key, entry[entry_key])
print(' {0}: {1}'.format(entry_key, entry[entry_key]))
else:
print '{0}: {1}'.format(key, results[key])
print('{0}: {1}'.format(key, results[key]))
......@@ -72,71 +83,47 @@ def stop():
pdb.set_trace()
def inject_message(fqdn_listname, msg):
mlist = getUtility(IListManager).get(fqdn_listname)
user_manager = getUtility(IUserManager)
msg = specialized_message_from_string(msg)
for sender in msg.senders:
if user_manager.get_address(sender) is None:
user_manager.create_address(sender)
process(mlist, msg, {})
def setup(testobj):
"""Test setup."""
# Create a unique database for the running version of Mailman, then start
# it up. It should not yet be running. This environment variable must be
# set to find the installation of Mailman we can run. Yes, this should be
# fixed.
testobj._bindir = os.environ.get('MAILMAN_TEST_BINDIR')
if testobj._bindir is None:
raise RuntimeError('Must set $MAILMAN_TEST_BINDIR to run tests')
vardir = testobj._vardir = tempfile.mkdtemp()
cfgfile = testobj._cfgfile = os.path.join(vardir, 'client_test.cfg')
with open(cfgfile, 'w') as fp:
print >> fp, """\
[mailman]
layout: tmpdir
[paths.tmpdir]
var_dir: {vardir}
log_dir: /tmp/mmclient/logs
[webservice]
port: 9001
[runner.archive]
start: no
[runner.bounces]
start: no
[runner.command]
start: yes
[runner.in]
start: yes
[runner.lmtp]
start: yes
[runner.news]
start: no
[runner.out]
start: yes
[runner.pipeline]
start: no
[runner.retry]
start: no
[runner.virgin]
start: yes
[runner.digest]
start: no
""".format(vardir=vardir)
mailman = os.path.join(testobj._bindir, 'mailman')
subprocess.call([mailman, '-C', cfgfile, 'start', '-q'])
time.sleep(3)
# Make sure future statements in our doctests match the Python code. When
# run with 2to3, the future import gets removed and these names are not
# defined.
try:
testobj.globs['absolute_import'] = absolute_import
testobj.globs['unicode_literals'] = unicode_literals
except NameError:
pass
testobj.globs['stop'] = stop
testobj.globs['dump'] = dump
testobj.globs['inject_message'] = inject_message
ConfigLayer.setUp()
# In unit tests, passwords aren't encrypted. Don't show this in the doctests
passlib_cfg = os.path.join(config.VAR_DIR, 'passlib.cfg')
with open(passlib_cfg, 'w') as fp:
print(dedent("""
[passlib]
schemes = sha512_crypt
"""), file=fp)
conf_hash_pw = dedent("""
[passwords]
configuration: {}
""".format(passlib_cfg))
config.push('conf_hash_pw', conf_hash_pw)
# Use the FakeMailmanClient
testobj.patcher = patch("mailmanclient.Client", FakeMailmanClient)
fmc = testobj.patcher.start()
def teardown(testobj):
"""Test teardown."""
mailman = os.path.join(testobj._bindir, 'mailman')
subprocess.call([mailman, '-C', testobj._cfgfile, 'stop', '-q'])
shutil.rmtree(testobj._vardir)
time.sleep(3)
testobj.patcher.stop()
config.pop('conf_hash_pw')
config.create_paths = False # or ConfigLayer.tearDown will create them
ConfigLayer.tearDown()
......
# Copyright (C) 2010-2014 by The Free Software Foundation, Inc.
#
# This file is part of mailman.client.
#
# mailman.client is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# mailman.client is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
# for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with mailman.client. If not, see <http://www.gnu.org/licenses/>.
"""Test tools."""
from __future__ import absolute_import, unicode_literals, print_function
from urllib2 import HTTPError
from urlparse import urljoin
from webtest import TestApp
import mailmanclient
__metaclass__ = type
__all__ = [
"FakeMailmanClient",
]
#
# Mocking mailman client
#
class FakeConnection(mailmanclient._client._Connection):
"""
Looks for information inside a dict instead of making HTTP requests.
Also, logs the called URLs as called_paths.
Very incomplete at the moment.
"""
def __init__(self, baseurl, name=None, password=None):
super(FakeConnection, self).__init__(baseurl, name, password)
self.called_paths = []
from mailman.rest.wsgiapp import make_application
self.app = TestApp(make_application())
def call(self, path, data=None, method=None):
self.called_paths.append(path)
if method is None:
if data is None:
method = 'GET'
else:
method = 'POST'
method_fn = getattr(self.app, method.lower())
if self.basic_auth:
self.app.authorization = ('Basic', (self.name, self.password))
url = urljoin(self.baseurl, path)
try:
kw = {"expect_errors": True}
if data:
kw["params"] = data
response = method_fn(url, **kw)
headers = response.headers
headers["status"] = response.status_int
content = unicode(response.body)
# If we did not get a 2xx status code, make this look like a
# urllib2 exception, for backward compatibility.
if response.status_int // 100 != 2:
raise HTTPError(url, response.status_int, content, headers, None)
if len(content) == 0:
return headers, None
# XXX Work around for http://bugs.python.org/issue10038
return headers, response.json
except HTTPError:
raise
except IOError:
raise MailmanConnectionError('Could not connect to Mailman API')
class FakeMailmanClient(mailmanclient.Client):
"""
Subclass of mailmanclient.Client to instantiate a FakeConnection object
instead of the real connection.
"""
def __init__(self, baseurl, name=None, password=None):
self._connection = FakeConnection(baseurl, name, password)
@property
def called_paths(self):
return self._connection.called_paths
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment