1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
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'))
|
