2 files changed, 396 insertions, 0 deletions

diff --git a/src/mailman/model/tests/test_cache.py b/src/mailman/model/tests/test_cache.py
new file mode 100644
index 000000000..2d910e202
--- /dev/null
+++ b/src/mailman/model/tests/test_cache.py
@@ -0,0 +1,110 @@
+# Copyright (C) 2016 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman 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 General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Test the cache."""
+
+import os
+import unittest
+
+from datetime import timedelta
+from mailman.config import config
+from mailman.interfaces.cache import ICacheManager
+from mailman.testing.helpers import configuration
+from mailman.testing.layers import ConfigLayer
+from mailman.utilities.datetime import factory
+from zope.component import getUtility
+
+
+class TestCache(unittest.TestCase):
+    layer = ConfigLayer
+
+    def setUp(self):
+        self._cachemgr = getUtility(ICacheManager)
+
+    def test_add_str_contents(self):
+        file_id = self._cachemgr.add('abc', 'xyz')
+        self.assertEqual(
+            file_id,
+            'ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad')
+        file_path = os.path.join(config.CACHE_DIR, 'ba', '78', file_id)
+        self.assertTrue(os.path.exists(file_path))
+        # The original content was a string.
+        with open(file_path, 'r', encoding='utf-8') as fp:
+            self.assertEqual(fp.read(), 'xyz')
+
+    def test_add_bytes_contents(self):
+        # No name is given so the file is cached by the hash of the contents.
+        file_id = self._cachemgr.add('abc', b'xyz')
+        self.assertEqual(
+            file_id,
+            'ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad')
+        file_path = os.path.join(config.CACHE_DIR, 'ba', '78', file_id)
+        self.assertTrue(os.path.exists(file_path))
+        # The original content was a string.
+        with open(file_path, 'br') as fp:
+            self.assertEqual(fp.read(), b'xyz')
+
+    def test_add_overwrite(self):
+        # If the file already exists and hasn't expired, a conflict exception
+        # is raised the second time we try to save it.
+        self._cachemgr.add('abc', 'xyz')
+        self.assertEqual(self._cachemgr.get('abc'), 'xyz')
+        self._cachemgr.add('abc', 'def')
+        self.assertEqual(self._cachemgr.get('abc'), 'def')
+
+    def test_get_str(self):
+        # Store a str, get a str.
+        self._cachemgr.add('abc', 'xyz')
+        contents = self._cachemgr.get('abc')
+        self.assertEqual(contents, 'xyz')
+
+    def test_get_bytes(self):
+        # Store a bytes, get a bytes.
+        self._cachemgr.add('abc', b'xyz')
+        contents = self._cachemgr.get('abc')
+        self.assertEqual(contents, b'xyz')
+
+    def test_get_str_expunge(self):
+        # When the entry is not expunged, it can be gotten multiple times.
+        # Once it's expunged, it's gone.
+        self._cachemgr.add('abc', 'xyz')
+        self.assertEqual(self._cachemgr.get('abc'), 'xyz')
+        self.assertEqual(self._cachemgr.get('abc', expunge=True), 'xyz')
+        self.assertIsNone(self._cachemgr.get('abc'))
+
+    @configuration('mailman', cache_life='1d')
+    def test_evict(self):
+        # Evicting all expired cache entries makes them inaccessible.
+        self._cachemgr.add('abc', 'xyz', lifetime=timedelta(hours=3))
+        self._cachemgr.add('def', 'uvw', lifetime=timedelta(days=3))
+        self.assertEqual(self._cachemgr.get('abc'), 'xyz')
+        self.assertEqual(self._cachemgr.get('def'), 'uvw')
+        factory.fast_forward(days=1)
+        self._cachemgr.evict()
+        self.assertIsNone(self._cachemgr.get('abc'))
+        self.assertEqual(self._cachemgr.get('def'), 'uvw')
+
+    def test_clear(self):
+        # Clearing the cache gets rid of all entries, regardless of lifetime.
+        self._cachemgr.add('abc', 'xyz', lifetime=timedelta(hours=3))
+        self._cachemgr.add('def', 'uvw')
+        self.assertEqual(self._cachemgr.get('abc'), 'xyz')
+        self.assertEqual(self._cachemgr.get('def'), 'uvw')
+        factory.fast_forward(days=1)
+        self._cachemgr.clear()
+        self.assertIsNone(self._cachemgr.get('abc'))
+        self.assertIsNone(self._cachemgr.get('xyz'))
diff --git a/src/mailman/model/tests/test_template.py b/src/mailman/model/tests/test_template.py
new file mode 100644
index 000000000..f16a4aa5c
--- /dev/null
+++ b/src/mailman/model/tests/test_template.py
@@ -0,0 +1,286 @@
+# Copyright (C) 2016 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman 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 General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Test the template manager."""
+
+import unittest
+import threading
+
+from contextlib import ExitStack
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from mailman.app.lifecycle import create_list
+from mailman.config import config
+from mailman.interfaces.domain import IDomainManager
+from mailman.interfaces.template import ITemplateLoader, ITemplateManager
+from mailman.testing.helpers import wait_for_webservice
+from mailman.testing.layers import ConfigLayer
+from mailman.utilities.i18n import find
+from requests import HTTPError
+from tempfile import TemporaryDirectory
+from urllib.error import URLError
+from zope.component import getUtility
+
+# New in Python 3.5.
+try:
+    from http import HTTPStatus
+except ImportError:
+    class HTTPStatus:
+        FORBIDDEN = 403
+        NOT_FOUND = 404
+        OK = 200
+
+
+# We need a web server to vend non-mailman: urls.
+class TestableHandler(BaseHTTPRequestHandler):
+    # Be quiet.
+    def log_request(*args, **kws):
+        pass
+
+    log_error = log_request
+
+    def do_GET(self):
+        if self.path == '/welcome_3.txt':
+            if self.headers['Authorization'] != 'Basic YW5uZTppcyBzcGVjaWFs':
+                self.send_error(HTTPStatus.FORBIDDEN)
+                return
+        response = TEXTS.get(self.path)
+        if response is None:
+            self.send_error(HTTPStatus.NOT_FOUND)
+            return
+        self.send_response(HTTPStatus.OK)
+        self.send_header('Content-Type', 'UTF-8')
+        self.end_headers()
+        self.wfile.write(response.encode('utf-8'))
+
+
+class HTTPLayer(ConfigLayer):
+    httpd = None
+
+    @classmethod
+    def setUp(cls):
+        assert cls.httpd is None, 'Layer already set up'
+        cls.httpd = HTTPServer(('localhost', 8180), TestableHandler)
+        cls._thread = threading.Thread(target=cls.httpd.serve_forever)
+        cls._thread.daemon = True
+        cls._thread.start()
+        wait_for_webservice('localhost', 8180)
+
+    @classmethod
+    def tearDown(cls):
+        assert cls.httpd is not None, 'Layer not set up'
+        cls.httpd.shutdown()
+        cls.httpd.server_close()
+        cls._thread.join()
+
+
+class TestTemplateCache(unittest.TestCase):
+    layer = HTTPLayer
+
+    def setUp(self):
+        self._templatemgr = getUtility(ITemplateManager)
+
+    def test_http_set_get(self):
+        self._templatemgr.set(
+            'list:user:notice:welcome', 'test.example.com',
+            'http://localhost:8180/welcome_1.txt')
+        contents = self._templatemgr.get(
+            'list:user:notice:welcome', 'test.example.com')
+        self.assertEqual(contents, WELCOME_1)
+
+    def test_http_set_override(self):
+        self._templatemgr.set(
+            'list:user:notice:welcome', 'test.example.com',
+            'http://localhost:8180/welcome_1.txt')
+        # Resetting the template with the same context and domain, but a
+        # different url overrides the previous value.
+        self._templatemgr.set(
+            'list:user:notice:welcome', 'test.example.com',
+            'http://localhost:8180/welcome_2.txt')
+        contents = self._templatemgr.get(
+            'list:user:notice:welcome', 'test.example.com')
+        self.assertEqual(contents, WELCOME_2)
+
+    def test_http_get_cached(self):
+        self._templatemgr.set(
+            'list:user:notice:welcome', 'test.example.com',
+            'http://localhost:8180/welcome_1.txt')
+        # The first one warms the cache.
+        self._templatemgr.get('list:user:notice:welcome', 'test.example.com')
+        # The second one hits the cache.
+        contents = self._templatemgr.get(
+            'list:user:notice:welcome', 'test.example.com')
+        self.assertEqual(contents, WELCOME_1)
+
+    def test_http_basic_auth(self):
+        # We get an HTTP error when we forget the username and password.
+        self._templatemgr.set(
+            'list:user:notice:welcome', 'test.example.com',
+            'http://localhost:8180/welcome_3.txt')
+        with self.assertRaises(HTTPError) as cm:
+            self._templatemgr.get(
+                'list:user:notice:welcome', 'test.example.com')
+        self.assertEqual(cm.exception.response.status_code, 403)
+        self.assertEqual(cm.exception.response.reason, 'Forbidden')
+        # But providing the basic auth information let's it work.
+        self._templatemgr.set(
+            'list:user:notice:welcome', 'test.example.com',
+            'http://localhost:8180/welcome_3.txt',
+            username='anne', password='is special')
+        contents = self._templatemgr.get(
+            'list:user:notice:welcome', 'test.example.com')
+        self.assertEqual(contents, WELCOME_3)
+
+    def test_delete(self):
+        self._templatemgr.set(
+            'list:user:notice:welcome', 'test.example.com',
+            'http://localhost:8180/welcome_1.txt')
+        self._templatemgr.get('list:user:notice:welcome', 'test.example.com')
+        self._templatemgr.delete(
+            'list:user:notice:welcome', 'test.example.com')
+        self.assertIsNone(
+            self._templatemgr.get(
+                'list:user:notice:welcome', 'test.example.com'))
+
+    def test_delete_missing(self):
+        self._templatemgr.delete(
+            'list:user:notice:welcome', 'test.example.com')
+        self.assertIsNone(
+            self._templatemgr.get(
+                'list:user:notice:welcome', 'test.example.com'))
+
+    def test_get_keywords(self):
+        self._templatemgr.set(
+            'list:user:notice:welcome', 'test.example.com',
+            'http://localhost:8180/${path}_${number}.txt')
+        contents = self._templatemgr.get(
+            'list:user:notice:welcome', 'test.example.com',
+            path='welcome', number='1')
+        self.assertEqual(contents, WELCOME_1)
+
+    def test_get_different_keywords(self):
+        self._templatemgr.set(
+            'list:user:notice:welcome', 'test.example.com',
+            'http://localhost:8180/${path}_${number}.txt')
+        contents = self._templatemgr.get(
+            'list:user:notice:welcome', 'test.example.com',
+            path='welcome', number='1')
+        self.assertEqual(contents, WELCOME_1)
+        contents = self._templatemgr.get(
+            'list:user:notice:welcome', 'test.example.com',
+            path='welcome', number='2')
+        self.assertEqual(contents, WELCOME_2)
+
+    def test_not_found(self):
+        # A 404 is treated specially, resulting in the empty string.
+        self._templatemgr.set(
+            'list:user:notice:welcome', 'test.example.com',
+            'http://localhost:8180/missing.txt')
+        contents = self._templatemgr.get(
+            'list:user:notice:welcome', 'test.example.com')
+        self.assertEqual(contents, '')
+
+
+class TestTemplateLoader(unittest.TestCase):
+    """Test the template downloader API."""
+
+    layer = HTTPLayer
+
+    def setUp(self):
+        resources = ExitStack()
+        self.addCleanup(resources.close)
+        var_dir = resources.enter_context(TemporaryDirectory())
+        config.push('template config', """\
+        [paths.testing]
+        var_dir: {}
+        """.format(var_dir))
+        resources.callback(config.pop, 'template config')
+        self._mlist = create_list('test@example.com')
+        self._loader = getUtility(ITemplateLoader)
+        self._manager = getUtility(ITemplateManager)
+
+    def test_domain_context(self):
+        self._manager.set(
+            'list:user:notice:welcome', 'example.com',
+            'http://localhost:8180/$domain_name/welcome_4.txt')
+        domain = getUtility(IDomainManager).get('example.com')
+        content = self._loader.get('list:user:notice:welcome', domain)
+        self.assertEqual(content, 'This is a domain welcome.\n')
+
+    def test_domain_content_fallback(self):
+        self._manager.set(
+            'list:user:notice:welcome', 'example.com',
+            'http://localhost:8180/$domain_name/welcome_4.txt')
+        content = self._loader.get('list:user:notice:welcome', self._mlist)
+        self.assertEqual(content, 'This is a domain welcome.\n')
+
+    def test_site_context(self):
+        self._manager.set(
+            'list:user:notice:welcome', None,
+            'http://localhost:8180/welcome_2.txt')
+        content = self._loader.get('list:user:notice:welcome')
+        self.assertEqual(content, "Sure, I guess you're welcome.\n")
+
+    def test_site_context_mailman(self):
+        self._manager.set(
+            'list:user:notice:welcome', None,
+            'mailman:///welcome.txt')
+        template_content = self._loader.get('list:user:notice:welcome')
+        path, fp = find('list:user:notice:welcome.txt')
+        try:
+            found_contents = fp.read()
+        finally:
+            fp.close()
+        self.assertEqual(template_content, found_contents)
+
+    def test_bad_context(self):
+        self.assertRaises(
+            ValueError, self._loader.get, 'list:user:notice:welcome', object())
+
+    def test_no_such_file(self):
+        self.assertRaises(URLError, self._loader.get, 'missing', self._mlist)
+
+    def test_403_forbidden(self):
+        # 404s are swallowed, but not 403s.
+        self._manager.set(
+            'forbidden', 'test.example.com',
+            'http://localhost:8180/welcome_3.txt')
+        self.assertRaises(URLError, self._loader.get, 'forbidden', self._mlist)
+
+
+# Response texts.
+WELCOME_1 = """\
+Welcome to the {fqdn_listname} mailing list!
+"""
+
+WELCOME_2 = """\
+Sure, I guess you're welcome.
+"""
+
+WELCOME_3 = """\
+Well?  Come.
+"""
+
+WELCOME_4 = """\
+This is a domain welcome.
+"""
+
+TEXTS = {
+    '/welcome_1.txt': WELCOME_1,
+    '/welcome_2.txt': WELCOME_2,
+    '/welcome_3.txt': WELCOME_3,
+    '/example.com/welcome_4.txt': WELCOME_4,
+    }