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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
# Copyright (C) 2010-2017 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 master watcher utilities."""
import os
import tempfile
import unittest
from click.testing import CliRunner
from contextlib import ExitStack, suppress
from datetime import timedelta
from flufl.lock import Lock, TimeOutError
from io import StringIO
from mailman.bin import master
from mailman.config import config
from mailman.testing.layers import ConfigLayer
from pkg_resources import resource_filename
from unittest.mock import patch
class FakeLock:
details = ('host.example.com', 9999, '/tmp/whatever')
def unlock(self):
pass
class TestMaster(unittest.TestCase):
layer = ConfigLayer
def setUp(self):
fd, self.lock_file = tempfile.mkstemp()
os.close(fd)
# The lock file should not exist before we try to acquire it.
os.remove(self.lock_file)
def tearDown(self):
# Unlocking removes the lock file, but just to be safe (i.e. in case
# of errors).
with suppress(FileNotFoundError):
os.remove(self.lock_file)
def test_acquire_lock_1(self):
lock = master.acquire_lock_1(False, self.lock_file)
is_locked = lock.is_locked
lock.unlock()
self.assertTrue(is_locked)
def test_master_state(self):
my_lock = Lock(self.lock_file)
# Mailman is not running.
state, lock = master.master_state(self.lock_file)
self.assertEqual(state, master.WatcherState.none)
# Acquire the lock as if another process had already started the
# master. Use a timeout to avoid this test deadlocking.
my_lock.lock(timedelta(seconds=60))
try:
state, lock = master.master_state(self.lock_file)
finally:
my_lock.unlock()
self.assertEqual(state, master.WatcherState.conflict)
def test_acquire_lock_timeout_reason_unknown(self):
stderr = StringIO()
with ExitStack() as resources:
resources.enter_context(patch(
'mailman.bin.master.acquire_lock_1',
side_effect=TimeOutError))
resources.enter_context(patch(
'mailman.bin.master.master_state',
return_value=(master.WatcherState.none, FakeLock())))
resources.enter_context(patch(
'mailman.bin.master.sys.stderr', stderr))
with self.assertRaises(SystemExit) as cm:
master.acquire_lock(False)
self.assertEqual(cm.exception.code, 1)
self.assertEqual(stderr.getvalue(), """\
For unknown reasons, the master lock could not be acquired.
Lock file: {}
Lock host: host.example.com
Exiting.
""".format(config.LOCK_FILE))
def test_main_cli(self):
command = CliRunner()
fake_lock = FakeLock()
with ExitStack() as resources:
config_file = resource_filename(
'mailman.testing', 'testing.cfg')
init_mock = resources.enter_context(patch(
'mailman.bin.master.initialize'))
lock_mock = resources.enter_context(patch(
'mailman.bin.master.acquire_lock',
return_value=fake_lock))
start_mock = resources.enter_context(patch.object(
master.Loop, 'start_runners'))
loop_mock = resources.enter_context(patch.object(
master.Loop, 'loop'))
command.invoke(
master.main,
('-C', config_file,
'--no-restart', '--force',
'-r', 'in:1:1', '--verbose'))
# We got initialized with the custom configuration file and the
# verbose flag.
init_mock.assert_called_once_with(config_file, True)
# We returned a lock that was force-acquired.
lock_mock.assert_called_once_with(True)
# We created a non-restartable loop.
start_mock.assert_called_once_with([('in', 1, 1)])
loop_mock.assert_called_once_with()
|
