13
13
# License for the specific language governing permissions and limitations
14
14
# under the License.
16
from oslo.config import cfg
16
from oslo_config import cfg
18
from neutron.agent.common import config
19
18
from neutron.agent.linux import external_process
20
19
from neutron.agent.linux import keepalived
21
from neutron.openstack.common import log as logging
22
from neutron.tests.functional import base as functional_base
20
from neutron.agent.linux import utils
21
from neutron.tests import base
23
22
from neutron.tests.unit.agent.linux import test_keepalived
25
LOG = logging.getLogger(__name__)
28
class KeepalivedManagerTestCase(functional_base.BaseSudoTestCase,
25
class KeepalivedManagerTestCase(base.BaseTestCase,
29
26
test_keepalived.KeepalivedConfBaseMixin):
31
29
super(KeepalivedManagerTestCase, self).setUp()
32
self.check_sudo_enabled()
30
cfg.CONF.set_override('check_child_processes_interval', 1, 'AGENT')
36
cfg.CONF.set_override('debug', False)
37
config.setup_logging()
32
self.expected_config = self._get_config()
33
self.process_monitor = external_process.ProcessMonitor(cfg.CONF,
35
self.manager = keepalived.KeepalivedManager(
36
'router1', self.expected_config, conf_path=cfg.CONF.state_path,
37
process_monitor=self.process_monitor)
38
self.addCleanup(self.manager.get_process().disable)
39
self.addCleanup(self.process_monitor.stop)
39
41
def test_keepalived_spawn(self):
40
expected_config = self._get_config()
41
manager = keepalived.KeepalivedManager('router1', expected_config,
42
conf_path=cfg.CONF.state_path,
43
root_helper=self.root_helper)
44
self.addCleanup(manager.disable)
47
43
process = external_process.ProcessManager(
52
47
pids_path=cfg.CONF.state_path)
53
48
self.assertTrue(process.active)
55
config_path = manager._get_full_config_file_path('keepalived.conf')
56
with open(config_path, 'r') as config_file:
57
config_contents = config_file.read()
58
self.assertEqual(expected_config.get_config_str(), config_contents)
50
self.assertEqual(self.expected_config.get_config_str(),
51
self.manager.get_conf_on_disk())
53
def test_keepalived_respawns(self):
55
process = self.manager.get_process()
56
self.assertTrue(process.active)
58
process.disable(sig='15')
60
utils.wait_until_true(
61
lambda: process.active,
64
exception=RuntimeError(_("Keepalived didn't respawn")))