¡@
OpenStack Study: test_linux_daemon.py**** CubicPower OpenStack Study ****
# vim: tabstop=4 shiftwidth=4 softtabstop=4 # # Copyright 2012 New Dream Network, LLC (DreamHost) # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. # # @author: Mark McClain, DreamHost import os import sys import mock import testtools from neutron.agent.linux import daemon from neutron.tests import base FAKE_FD = 8
**** CubicPower OpenStack Study **** class TestPidfile(base.BaseTestCase):
**** CubicPower OpenStack Study **** def setUp(self): super(TestPidfile, self).setUp() self.os_p = mock.patch.object(daemon, 'os') self.os = self.os_p.start() self.addCleanup(self.os_p.stop) self.os.open.return_value = FAKE_FD self.fcntl_p = mock.patch.object(daemon, 'fcntl') self.fcntl = self.fcntl_p.start() self.addCleanup(self.fcntl_p.stop) self.fcntl.flock.return_value = 0
**** CubicPower OpenStack Study **** def test_init(self): self.os.O_CREAT = os.O_CREAT self.os.O_RDWR = os.O_RDWR daemon.Pidfile('thefile', 'python') self.os.open.assert_called_once_with('thefile', os.O_CREAT | os.O_RDWR) self.fcntl.flock.assert_called_once_with(FAKE_FD, self.fcntl.LOCK_EX)
**** CubicPower OpenStack Study **** def test_init_open_fail(self): self.os.open.side_effect = IOError with mock.patch.object(daemon.sys, 'stderr'): with testtools.ExpectedException(SystemExit): daemon.Pidfile('thefile', 'python') sys.assert_has_calls([ mock.call.stderr.write(mock.ANY), mock.call.exit(1)] )
**** CubicPower OpenStack Study **** def test_unlock(self): p = daemon.Pidfile('thefile', 'python') p.unlock() self.fcntl.flock.assert_has_calls([ mock.call(FAKE_FD, self.fcntl.LOCK_EX), mock.call(FAKE_FD, self.fcntl.LOCK_UN)] )
**** CubicPower OpenStack Study **** def test_write(self): p = daemon.Pidfile('thefile', 'python') p.write(34) self.os.assert_has_calls([ mock.call.ftruncate(FAKE_FD, 0), mock.call.write(FAKE_FD, '34'), mock.call.fsync(FAKE_FD)] )
**** CubicPower OpenStack Study **** def test_read(self): self.os.read.return_value = '34' p = daemon.Pidfile('thefile', 'python') self.assertEqual(34, p.read())
**** CubicPower OpenStack Study **** def test_is_running(self): with mock.patch('__builtin__.open') as mock_open: p = daemon.Pidfile('thefile', 'python') mock_open.return_value.__enter__ = lambda s: s mock_open.return_value.__exit__ = mock.Mock() mock_open.return_value.readline.return_value = 'python' with mock.patch.object(p, 'read') as read: read.return_value = 34 self.assertTrue(p.is_running()) mock_open.assert_called_once_with('/proc/34/cmdline', 'r')
**** CubicPower OpenStack Study **** def test_is_running_uuid_true(self): with mock.patch('__builtin__.open') as mock_open: p = daemon.Pidfile('thefile', 'python', uuid='1234') mock_open.return_value.__enter__ = lambda s: s mock_open.return_value.__exit__ = mock.Mock() mock_open.return_value.readline.return_value = 'python 1234' with mock.patch.object(p, 'read') as read: read.return_value = 34 self.assertTrue(p.is_running()) mock_open.assert_called_once_with('/proc/34/cmdline', 'r')
**** CubicPower OpenStack Study **** def test_is_running_uuid_false(self): with mock.patch('__builtin__.open') as mock_open: p = daemon.Pidfile('thefile', 'python', uuid='6789') mock_open.return_value.__enter__ = lambda s: s mock_open.return_value.__exit__ = mock.Mock() mock_open.return_value.readline.return_value = 'python 1234' with mock.patch.object(p, 'read') as read: read.return_value = 34 self.assertFalse(p.is_running()) mock_open.assert_called_once_with('/proc/34/cmdline', 'r')
**** CubicPower OpenStack Study **** class TestDaemon(base.BaseTestCase):
**** CubicPower OpenStack Study **** def setUp(self): super(TestDaemon, self).setUp() self.os_p = mock.patch.object(daemon, 'os') self.os = self.os_p.start() self.pidfile_p = mock.patch.object(daemon, 'Pidfile') self.pidfile = self.pidfile_p.start()
**** CubicPower OpenStack Study **** def tearDown(self): self.pidfile_p.stop() self.os_p.stop() super(TestDaemon, self).tearDown()
**** CubicPower OpenStack Study **** def test_init(self): d = daemon.Daemon('pidfile') self.assertEqual(d.procname, 'python')
**** CubicPower OpenStack Study **** def test_fork_parent(self): self.os.fork.return_value = 1 with testtools.ExpectedException(SystemExit): d = daemon.Daemon('pidfile') d._fork()
**** CubicPower OpenStack Study **** def test_fork_child(self): self.os.fork.return_value = 0 d = daemon.Daemon('pidfile') self.assertIsNone(d._fork())
**** CubicPower OpenStack Study **** def test_fork_error(self): self.os.fork.side_effect = lambda: OSError(1) with mock.patch.object(daemon.sys, 'stderr'): with testtools.ExpectedException(SystemExit): d = daemon.Daemon('pidfile', 'stdin') d._fork()
**** CubicPower OpenStack Study **** def test_daemonize(self): d = daemon.Daemon('pidfile') with mock.patch.object(d, '_fork') as fork: with mock.patch.object(daemon, 'atexit') as atexit: with mock.patch.object(daemon, 'signal') as signal: signal.SIGTERM = 15 with mock.patch.object(daemon, 'sys') as sys: sys.stdin.fileno.return_value = 0 sys.stdout.fileno.return_value = 1 sys.stderr.fileno.return_value = 2 d.daemonize() signal.signal.assert_called_once_with(15, d.handle_sigterm) atexit.register.assert_called_once_with(d.delete_pid) fork.assert_has_calls([mock.call(), mock.call()]) self.os.assert_has_calls([ mock.call.chdir('/'), mock.call.setsid(), mock.call.umask(0), mock.call.dup2(mock.ANY, 0), mock.call.dup2(mock.ANY, 1), mock.call.dup2(mock.ANY, 2), mock.call.getpid()] )
**** CubicPower OpenStack Study **** def test_delete_pid(self): self.pidfile.return_value.__str__.return_value = 'pidfile' d = daemon.Daemon('pidfile') d.delete_pid() self.os.remove.assert_called_once_with('pidfile')
**** CubicPower OpenStack Study **** def test_handle_sigterm(self): d = daemon.Daemon('pidfile') with mock.patch.object(daemon, 'sys') as sys: d.handle_sigterm(15, 1234) sys.exit.assert_called_once_with(0)
**** CubicPower OpenStack Study **** def test_start(self): self.pidfile.return_value.is_running.return_value = False d = daemon.Daemon('pidfile') with mock.patch.object(d, 'daemonize') as daemonize: with mock.patch.object(d, 'run') as run: d.start() run.assert_called_once_with() daemonize.assert_called_once_with()
**** CubicPower OpenStack Study **** def test_start_running(self): self.pidfile.return_value.is_running.return_value = True d = daemon.Daemon('pidfile') with mock.patch.object(daemon.sys, 'stderr'): with mock.patch.object(d, 'daemonize') as daemonize: with testtools.ExpectedException(SystemExit): d.start() self.assertFalse(daemonize.called)
|