Commit 29f30422 authored by Barry Warsaw's avatar Barry Warsaw

* Remove an unused file.

* Port to Python 3.4
* Add `queues` resource.
* Clean up doctest.
* Move one corner case to unittest.
* Record HTTP traffic with vcr so that the test suite can be run off-line.
* Added `tox -e record` to run the vcr recorder.
parent d3f29968
......@@ -68,7 +68,7 @@ class _Connection:
self.basic_auth = None
else:
auth = '{0}:{1}'.format(name, password)
self.basic_auth = b64encode(auth)
self.basic_auth = b64encode(auth.encode('utf-8')).decode('utf-8')
def call(self, path, data=None, method=None):
"""Make a call to the Mailman REST API.
......@@ -144,6 +144,14 @@ class Client:
def preferences(self):
return _Preferences(self._connection, 'system/preferences')
@property
def queues(self):
response, content = self._connection.call('queues')
queues = {}
for entry in content['entries']:
queues[entry['name']] = _Queue(self._connection, entry)
return queues
@property
def lists(self):
response, content = self._connection.call('lists')
......@@ -975,3 +983,22 @@ class _Page:
self._page -= 1
self._create_page()
return self
class _Queue:
def __init__(self, connection, entry):
self._connection = connection
self.name = entry['name']
self.url = entry['self_link']
self.directory = entry['directory']
def __repr__(self):
return '<Queue: {}>'.format(self.name)
def inject(self, list_id, text):
self._connection.call(self.url, dict(list_id=list_id, text=text))
@property
def files(self):
response, content = self._connection.call(self.url)
return content['files']
......@@ -2,13 +2,6 @@
Mailman REST client
===================
>>> import os
>>> import time
>>> import smtplib
>>> import subprocess
>>> from mock import patch
This is the official Python bindings for the GNU Mailman REST API. In order
to talk to Mailman, the engine's REST server must be running. You begin by
instantiating a client object to access the root of the REST hierarchy,
......@@ -74,13 +67,6 @@ Additionally you can get an existing domain using its web host.
>>> print(example_dot_com.base_url)
http://example.com
But you cannot retrieve a non-existent domain.
>>> client.get_domain('example.org')
Traceback (most recent call last):
...
HTTPError: HTTP Error 404: 404 Not Found
After creating a few more domains, we can print the list of all domains.
>>> client.create_domain('example.net')
......@@ -309,6 +295,7 @@ If you use an address which is not a member of test_two `ValueError` is raised:
After a while, Anna decides to unsubscribe from the Test One mailing list,
though she keeps her Test Two membership active.
>>> import time
>>> time.sleep(2)
>>> test_one.unsubscribe('anna@example.com')
>>> for member in client.members:
......@@ -645,6 +632,9 @@ Moderation
Message Moderation
------------------
By injecting a message by a non-member into the incoming queue, we can
simulate a message being held for moderator approval.
>>> msg = """From: nomember@example.com
... To: test-one@example.com
... Subject: Something
......@@ -653,11 +643,27 @@ Message Moderation
... Some text.
...
... """
>>> inject_message('test-one@example.com', msg)
>>> inq = client.queues['in']
>>> inq.inject('test-one.example.com', msg)
Now wait until the message has been processed.
>>> while True:
... if len(inq.files) == 0:
... break
... time.sleep(0.1)
It might take a few moments for the message to show up in the moderation
queue.
>>> while True:
... held = test_one.held
... if len(held) > 0:
... break
... time.sleep(0.1)
Messages held for moderation can be listed on a per list basis.
>>> held = test_one.held
>>> print(held[0]['subject'])
Something
>>> print(held[0]['reason'])
......
This source diff could not be displayed because it is too large. You can view the blob instead.
# Copyright (C) 2015 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, version 3 of the License.
#
# 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 domain corner cases."""
from __future__ import absolute_import, print_function, unicode_literals
__metaclass__ = type
__all__ = [
'TestDomains',
]
import unittest
from mailmanclient import Client
from six.moves.urllib_error import HTTPError
class TestDomains(unittest.TestCase):
def setUp(self):
self._client = Client(
'http://localhost:9001/3.0', 'restadmin', 'restpass')
def test_no_domain(self):
# Trying to get a non-existent domain returns a 404.
#
# We can't use `with self.assertRaises()` until we drop Python 2.6
try:
self._client.get_domain('example.org')
except HTTPError as error:
self.assertEqual(error.code, 404)
else:
raise AssertionError('Expected HTTPError 404')
# Copyright (C) 2010-2015 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
import os
from urllib2 import HTTPError
from urlparse import urljoin
from zope.component import getUtility
from webtest import TestApp
from mailman.config import config
from mailman.core.chains import process
from mailman.interfaces.listmanager import IListManager
from mailman.interfaces.usermanager import IUserManager
from mailman.testing.helpers import specialized_message_from_string
from mailman.testing.layers import ConfigLayer
import mailmanclient
__metaclass__ = type
__all__ = [
"inject_message",
"FakeMailmanClient",
]
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, {})
#
# 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())
if self.basic_auth:
self.app.authorization = ('Basic', (self.name, self.password))
super(FakeConnection, self).__init__(baseurl, name, password)
def call(self, path, data=None, method=None):
self.called_paths.append(
{ "path": path, "data": data, "method": method })
if method is None:
if data is None:
method = 'GET'
else:
method = 'POST'
method_fn = getattr(self.app, method.lower())
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 mailmanclient.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
@classmethod
def setUp(self):
ConfigLayer.setUp()
@classmethod
def tearDown(self):
config.create_paths = False # or ConfigLayer.tearDown will create them
ConfigLayer.testTearDown()
ConfigLayer.tearDown()
reset_mailman_config()
def reset_mailman_config():
# This is necessary because ConfigLayer.setup/tearDown was designed to be
# run only once for the whole test suite, and thus does not reset
# everything afterwards
for prop in ("switchboards", "rules", "chains", "handlers",
"pipelines", "commands"):
getattr(config, prop).clear()
from mailman.model.user import uid_factory
uid_factory._lockobj = None
from mailman.model.member import uid_factory
uid_factory._lockobj = None
......@@ -9,3 +9,7 @@ deps =
mock
nose2
vcrpy==1.1.4
[testenv:record]
basepython = python3.4
commands = python -m nose2 -v -R
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