summaryrefslogtreecommitdiffstats
path: root/scripts/automation/trex_control_plane/python_lib/python-daemon-2.0.5/test/test_runner.py
diff options
context:
space:
mode:
Diffstat (limited to 'scripts/automation/trex_control_plane/python_lib/python-daemon-2.0.5/test/test_runner.py')
-rwxr-xr-xscripts/automation/trex_control_plane/python_lib/python-daemon-2.0.5/test/test_runner.py675
1 files changed, 675 insertions, 0 deletions
diff --git a/scripts/automation/trex_control_plane/python_lib/python-daemon-2.0.5/test/test_runner.py b/scripts/automation/trex_control_plane/python_lib/python-daemon-2.0.5/test/test_runner.py
new file mode 100755
index 00000000..4c0c714a
--- /dev/null
+++ b/scripts/automation/trex_control_plane/python_lib/python-daemon-2.0.5/test/test_runner.py
@@ -0,0 +1,675 @@
+# -*- coding: utf-8 -*-
+#
+# test/test_runner.py
+# Part of ‘python-daemon’, an implementation of PEP 3143.
+#
+# Copyright © 2009–2015 Ben Finney <ben+python@benfinney.id.au>
+#
+# This is free software: you may copy, modify, and/or distribute this work
+# under the terms of the Apache License, version 2.0 as published by the
+# Apache Software Foundation.
+# No warranty expressed or implied. See the file ‘LICENSE.ASF-2’ for details.
+
+""" Unit test for ‘runner’ module.
+ """
+
+from __future__ import (absolute_import, unicode_literals)
+
+try:
+ # Python 3 standard library.
+ import builtins
+except ImportError:
+ # Python 2 standard library.
+ import __builtin__ as builtins
+import os
+import os.path
+import sys
+import tempfile
+import errno
+import signal
+import functools
+
+import lockfile
+import mock
+import testtools
+
+from . import scaffold
+from .scaffold import (basestring, unicode)
+from .test_pidfile import (
+ FakeFileDescriptorStringIO,
+ setup_pidfile_fixtures,
+ make_pidlockfile_scenarios,
+ apply_lockfile_method_mocks,
+ )
+from .test_daemon import (
+ setup_streams_fixtures,
+ )
+
+import daemon.daemon
+import daemon.runner
+import daemon.pidfile
+
+
+class ModuleExceptions_TestCase(scaffold.Exception_TestCase):
+ """ Test cases for module exception classes. """
+
+ scenarios = scaffold.make_exception_scenarios([
+ ('daemon.runner.DaemonRunnerError', dict(
+ exc_type = daemon.runner.DaemonRunnerError,
+ min_args = 1,
+ types = [Exception],
+ )),
+ ('daemon.runner.DaemonRunnerInvalidActionError', dict(
+ exc_type = daemon.runner.DaemonRunnerInvalidActionError,
+ min_args = 1,
+ types = [daemon.runner.DaemonRunnerError, ValueError],
+ )),
+ ('daemon.runner.DaemonRunnerStartFailureError', dict(
+ exc_type = daemon.runner.DaemonRunnerStartFailureError,
+ min_args = 1,
+ types = [daemon.runner.DaemonRunnerError, RuntimeError],
+ )),
+ ('daemon.runner.DaemonRunnerStopFailureError', dict(
+ exc_type = daemon.runner.DaemonRunnerStopFailureError,
+ min_args = 1,
+ types = [daemon.runner.DaemonRunnerError, RuntimeError],
+ )),
+ ])
+
+
+def make_runner_scenarios():
+ """ Make a collection of scenarios for testing `DaemonRunner` instances.
+
+ :return: A collection of scenarios for tests involving
+ `DaemonRunner` instances.
+
+ The collection is a mapping from scenario name to a dictionary of
+ scenario attributes.
+
+ """
+
+ pidlockfile_scenarios = make_pidlockfile_scenarios()
+
+ scenarios = {
+ 'simple': {
+ 'pidlockfile_scenario_name': 'simple',
+ },
+ 'pidfile-locked': {
+ 'pidlockfile_scenario_name': 'exist-other-pid-locked',
+ },
+ }
+
+ for scenario in scenarios.values():
+ if 'pidlockfile_scenario_name' in scenario:
+ pidlockfile_scenario = pidlockfile_scenarios.pop(
+ scenario['pidlockfile_scenario_name'])
+ scenario['pid'] = pidlockfile_scenario['pid']
+ scenario['pidfile_path'] = pidlockfile_scenario['pidfile_path']
+ scenario['pidfile_timeout'] = 23
+ scenario['pidlockfile_scenario'] = pidlockfile_scenario
+
+ return scenarios
+
+
+def set_runner_scenario(testcase, scenario_name):
+ """ Set the DaemonRunner test scenario for the test case.
+
+ :param testcase: The `TestCase` instance to decorate.
+ :param scenario_name: The name of the scenario to use.
+
+ Set the `DaemonRunner` test scenario name and decorate the
+ `testcase` with the corresponding scenario fixtures.
+
+ """
+ scenarios = testcase.runner_scenarios
+ testcase.scenario = scenarios[scenario_name]
+ apply_lockfile_method_mocks(
+ testcase.mock_runner_lockfile,
+ testcase,
+ testcase.scenario['pidlockfile_scenario'])
+
+
+def setup_runner_fixtures(testcase):
+ """ Set up common fixtures for `DaemonRunner` test cases.
+
+ :param testcase: A `TestCase` instance to decorate.
+
+ Decorate the `testcase` with attributes to be fixtures for tests
+ involving `DaemonRunner` instances.
+
+ """
+ setup_pidfile_fixtures(testcase)
+ setup_streams_fixtures(testcase)
+
+ testcase.runner_scenarios = make_runner_scenarios()
+
+ patcher_stderr = mock.patch.object(
+ sys, "stderr",
+ new=FakeFileDescriptorStringIO())
+ testcase.fake_stderr = patcher_stderr.start()
+ testcase.addCleanup(patcher_stderr.stop)
+
+ simple_scenario = testcase.runner_scenarios['simple']
+
+ testcase.mock_runner_lockfile = mock.MagicMock(
+ spec=daemon.pidfile.TimeoutPIDLockFile)
+ apply_lockfile_method_mocks(
+ testcase.mock_runner_lockfile,
+ testcase,
+ simple_scenario['pidlockfile_scenario'])
+ testcase.mock_runner_lockfile.path = simple_scenario['pidfile_path']
+
+ patcher_lockfile_class = mock.patch.object(
+ daemon.pidfile, "TimeoutPIDLockFile",
+ return_value=testcase.mock_runner_lockfile)
+ patcher_lockfile_class.start()
+ testcase.addCleanup(patcher_lockfile_class.stop)
+
+ class TestApp(object):
+
+ def __init__(self):
+ self.stdin_path = testcase.stream_file_paths['stdin']
+ self.stdout_path = testcase.stream_file_paths['stdout']
+ self.stderr_path = testcase.stream_file_paths['stderr']
+ self.pidfile_path = simple_scenario['pidfile_path']
+ self.pidfile_timeout = simple_scenario['pidfile_timeout']
+
+ run = mock.MagicMock(name="TestApp.run")
+
+ testcase.TestApp = TestApp
+
+ patcher_runner_daemoncontext = mock.patch.object(
+ daemon.runner, "DaemonContext", autospec=True)
+ patcher_runner_daemoncontext.start()
+ testcase.addCleanup(patcher_runner_daemoncontext.stop)
+
+ testcase.test_app = testcase.TestApp()
+
+ testcase.test_program_name = "bazprog"
+ testcase.test_program_path = os.path.join(
+ "/foo/bar", testcase.test_program_name)
+ testcase.valid_argv_params = {
+ 'start': [testcase.test_program_path, 'start'],
+ 'stop': [testcase.test_program_path, 'stop'],
+ 'restart': [testcase.test_program_path, 'restart'],
+ }
+
+ def fake_open(filename, mode=None, buffering=None):
+ if filename in testcase.stream_files_by_path:
+ result = testcase.stream_files_by_path[filename]
+ else:
+ result = FakeFileDescriptorStringIO()
+ result.mode = mode
+ result.buffering = buffering
+ return result
+
+ mock_open = mock.mock_open()
+ mock_open.side_effect = fake_open
+
+ func_patcher_builtin_open = mock.patch.object(
+ builtins, "open",
+ new=mock_open)
+ func_patcher_builtin_open.start()
+ testcase.addCleanup(func_patcher_builtin_open.stop)
+
+ func_patcher_os_kill = mock.patch.object(os, "kill")
+ func_patcher_os_kill.start()
+ testcase.addCleanup(func_patcher_os_kill.stop)
+
+ patcher_sys_argv = mock.patch.object(
+ sys, "argv",
+ new=testcase.valid_argv_params['start'])
+ patcher_sys_argv.start()
+ testcase.addCleanup(patcher_sys_argv.stop)
+
+ testcase.test_instance = daemon.runner.DaemonRunner(testcase.test_app)
+
+ testcase.scenario = NotImplemented
+
+
+class DaemonRunner_BaseTestCase(scaffold.TestCase):
+ """ Base class for DaemonRunner test case classes. """
+
+ def setUp(self):
+ """ Set up test fixtures. """
+ super(DaemonRunner_BaseTestCase, self).setUp()
+
+ setup_runner_fixtures(self)
+ set_runner_scenario(self, 'simple')
+
+
+class DaemonRunner_TestCase(DaemonRunner_BaseTestCase):
+ """ Test cases for DaemonRunner class. """
+
+ def setUp(self):
+ """ Set up test fixtures. """
+ super(DaemonRunner_TestCase, self).setUp()
+
+ func_patcher_parse_args = mock.patch.object(
+ daemon.runner.DaemonRunner, "parse_args")
+ func_patcher_parse_args.start()
+ self.addCleanup(func_patcher_parse_args.stop)
+
+ # Create a new instance now with our custom patches.
+ self.test_instance = daemon.runner.DaemonRunner(self.test_app)
+
+ def test_instantiate(self):
+ """ New instance of DaemonRunner should be created. """
+ self.assertIsInstance(self.test_instance, daemon.runner.DaemonRunner)
+
+ def test_parses_commandline_args(self):
+ """ Should parse commandline arguments. """
+ self.test_instance.parse_args.assert_called_with()
+
+ def test_has_specified_app(self):
+ """ Should have specified application object. """
+ self.assertIs(self.test_app, self.test_instance.app)
+
+ def test_sets_pidfile_none_when_pidfile_path_is_none(self):
+ """ Should set ‘pidfile’ to ‘None’ when ‘pidfile_path’ is ‘None’. """
+ pidfile_path = None
+ self.test_app.pidfile_path = pidfile_path
+ expected_pidfile = None
+ instance = daemon.runner.DaemonRunner(self.test_app)
+ self.assertIs(expected_pidfile, instance.pidfile)
+
+ def test_error_when_pidfile_path_not_string(self):
+ """ Should raise ValueError when PID file path not a string. """
+ pidfile_path = object()
+ self.test_app.pidfile_path = pidfile_path
+ expected_error = ValueError
+ self.assertRaises(
+ expected_error,
+ daemon.runner.DaemonRunner, self.test_app)
+
+ def test_error_when_pidfile_path_not_absolute(self):
+ """ Should raise ValueError when PID file path not absolute. """
+ pidfile_path = "foo/bar.pid"
+ self.test_app.pidfile_path = pidfile_path
+ expected_error = ValueError
+ self.assertRaises(
+ expected_error,
+ daemon.runner.DaemonRunner, self.test_app)
+
+ def test_creates_lock_with_specified_parameters(self):
+ """ Should create a TimeoutPIDLockFile with specified params. """
+ pidfile_path = self.scenario['pidfile_path']
+ pidfile_timeout = self.scenario['pidfile_timeout']
+ daemon.pidfile.TimeoutPIDLockFile.assert_called_with(
+ pidfile_path, pidfile_timeout)
+
+ def test_has_created_pidfile(self):
+ """ Should have new PID lock file as `pidfile` attribute. """
+ expected_pidfile = self.mock_runner_lockfile
+ instance = self.test_instance
+ self.assertIs(
+ expected_pidfile, instance.pidfile)
+
+ def test_daemon_context_has_created_pidfile(self):
+ """ DaemonContext component should have new PID lock file. """
+ expected_pidfile = self.mock_runner_lockfile
+ daemon_context = self.test_instance.daemon_context
+ self.assertIs(
+ expected_pidfile, daemon_context.pidfile)
+
+ def test_daemon_context_has_specified_stdin_stream(self):
+ """ DaemonContext component should have specified stdin file. """
+ test_app = self.test_app
+ expected_file = self.stream_files_by_name['stdin']
+ daemon_context = self.test_instance.daemon_context
+ self.assertEqual(expected_file, daemon_context.stdin)
+
+ def test_daemon_context_has_stdin_in_read_mode(self):
+ """ DaemonContext component should open stdin file for read. """
+ expected_mode = 'rt'
+ daemon_context = self.test_instance.daemon_context
+ self.assertIn(expected_mode, daemon_context.stdin.mode)
+
+ def test_daemon_context_has_specified_stdout_stream(self):
+ """ DaemonContext component should have specified stdout file. """
+ test_app = self.test_app
+ expected_file = self.stream_files_by_name['stdout']
+ daemon_context = self.test_instance.daemon_context
+ self.assertEqual(expected_file, daemon_context.stdout)
+
+ def test_daemon_context_has_stdout_in_append_mode(self):
+ """ DaemonContext component should open stdout file for append. """
+ expected_mode = 'w+t'
+ daemon_context = self.test_instance.daemon_context
+ self.assertIn(expected_mode, daemon_context.stdout.mode)
+
+ def test_daemon_context_has_specified_stderr_stream(self):
+ """ DaemonContext component should have specified stderr file. """
+ test_app = self.test_app
+ expected_file = self.stream_files_by_name['stderr']
+ daemon_context = self.test_instance.daemon_context
+ self.assertEqual(expected_file, daemon_context.stderr)
+
+ def test_daemon_context_has_stderr_in_append_mode(self):
+ """ DaemonContext component should open stderr file for append. """
+ expected_mode = 'w+t'
+ daemon_context = self.test_instance.daemon_context
+ self.assertIn(expected_mode, daemon_context.stderr.mode)
+
+ def test_daemon_context_has_stderr_with_no_buffering(self):
+ """ DaemonContext component should open stderr file unbuffered. """
+ expected_buffering = 0
+ daemon_context = self.test_instance.daemon_context
+ self.assertEqual(
+ expected_buffering, daemon_context.stderr.buffering)
+
+
+class DaemonRunner_usage_exit_TestCase(DaemonRunner_BaseTestCase):
+ """ Test cases for DaemonRunner.usage_exit method. """
+
+ def test_raises_system_exit(self):
+ """ Should raise SystemExit exception. """
+ instance = self.test_instance
+ argv = [self.test_program_path]
+ self.assertRaises(
+ SystemExit,
+ instance._usage_exit, argv)
+
+ def test_message_follows_conventional_format(self):
+ """ Should emit a conventional usage message. """
+ instance = self.test_instance
+ argv = [self.test_program_path]
+ expected_stderr_output = """\
+ usage: {progname} ...
+ """.format(
+ progname=self.test_program_name)
+ self.assertRaises(
+ SystemExit,
+ instance._usage_exit, argv)
+ self.assertOutputCheckerMatch(
+ expected_stderr_output, self.fake_stderr.getvalue())
+
+
+class DaemonRunner_parse_args_TestCase(DaemonRunner_BaseTestCase):
+ """ Test cases for DaemonRunner.parse_args method. """
+
+ def setUp(self):
+ """ Set up test fixtures. """
+ super(DaemonRunner_parse_args_TestCase, self).setUp()
+
+ func_patcher_usage_exit = mock.patch.object(
+ daemon.runner.DaemonRunner, "_usage_exit",
+ side_effect=NotImplementedError)
+ func_patcher_usage_exit.start()
+ self.addCleanup(func_patcher_usage_exit.stop)
+
+ def test_emits_usage_message_if_insufficient_args(self):
+ """ Should emit a usage message and exit if too few arguments. """
+ instance = self.test_instance
+ argv = [self.test_program_path]
+ exc = self.assertRaises(
+ NotImplementedError,
+ instance.parse_args, argv)
+ daemon.runner.DaemonRunner._usage_exit.assert_called_with(argv)
+
+ def test_emits_usage_message_if_unknown_action_arg(self):
+ """ Should emit a usage message and exit if unknown action. """
+ instance = self.test_instance
+ progname = self.test_program_name
+ argv = [self.test_program_path, 'bogus']
+ exc = self.assertRaises(
+ NotImplementedError,
+ instance.parse_args, argv)
+ daemon.runner.DaemonRunner._usage_exit.assert_called_with(argv)
+
+ def test_should_parse_system_argv_by_default(self):
+ """ Should parse sys.argv by default. """
+ instance = self.test_instance
+ expected_action = 'start'
+ argv = self.valid_argv_params['start']
+ with mock.patch.object(sys, "argv", new=argv):
+ instance.parse_args()
+ self.assertEqual(expected_action, instance.action)
+
+ def test_sets_action_from_first_argument(self):
+ """ Should set action from first commandline argument. """
+ instance = self.test_instance
+ for name, argv in self.valid_argv_params.items():
+ expected_action = name
+ instance.parse_args(argv)
+ self.assertEqual(expected_action, instance.action)
+
+
+try:
+ ProcessLookupError
+except NameError:
+ # Python 2 uses OSError.
+ ProcessLookupError = functools.partial(OSError, errno.ESRCH)
+
+class DaemonRunner_do_action_TestCase(DaemonRunner_BaseTestCase):
+ """ Test cases for DaemonRunner.do_action method. """
+
+ def test_raises_error_if_unknown_action(self):
+ """ Should emit a usage message and exit if action is unknown. """
+ instance = self.test_instance
+ instance.action = 'bogus'
+ expected_error = daemon.runner.DaemonRunnerInvalidActionError
+ self.assertRaises(
+ expected_error,
+ instance.do_action)
+
+
+class DaemonRunner_do_action_start_TestCase(DaemonRunner_BaseTestCase):
+ """ Test cases for DaemonRunner.do_action method, action 'start'. """
+
+ def setUp(self):
+ """ Set up test fixtures. """
+ super(DaemonRunner_do_action_start_TestCase, self).setUp()
+
+ self.test_instance.action = 'start'
+
+ def test_raises_error_if_pidfile_locked(self):
+ """ Should raise error if PID file is locked. """
+
+ instance = self.test_instance
+ instance.daemon_context.open.side_effect = lockfile.AlreadyLocked
+ pidfile_path = self.scenario['pidfile_path']
+ expected_error = daemon.runner.DaemonRunnerStartFailureError
+ expected_message_content = pidfile_path
+ exc = self.assertRaises(
+ expected_error,
+ instance.do_action)
+ self.assertIn(expected_message_content, unicode(exc))
+
+ def test_breaks_lock_if_no_such_process(self):
+ """ Should request breaking lock if PID file process is not running. """
+ set_runner_scenario(self, 'pidfile-locked')
+ instance = self.test_instance
+ self.mock_runner_lockfile.read_pid.return_value = (
+ self.scenario['pidlockfile_scenario']['pidfile_pid'])
+ pidfile_path = self.scenario['pidfile_path']
+ test_pid = self.scenario['pidlockfile_scenario']['pidfile_pid']
+ expected_signal = signal.SIG_DFL
+ test_error = ProcessLookupError("Not running")
+ os.kill.side_effect = test_error
+ instance.do_action()
+ os.kill.assert_called_with(test_pid, expected_signal)
+ self.mock_runner_lockfile.break_lock.assert_called_with()
+
+ def test_requests_daemon_context_open(self):
+ """ Should request the daemon context to open. """
+ instance = self.test_instance
+ instance.do_action()
+ instance.daemon_context.open.assert_called_with()
+
+ def test_emits_start_message_to_stderr(self):
+ """ Should emit start message to stderr. """
+ instance = self.test_instance
+ expected_stderr = """\
+ started with pid {pid:d}
+ """.format(
+ pid=self.scenario['pid'])
+ instance.do_action()
+ self.assertOutputCheckerMatch(
+ expected_stderr, self.fake_stderr.getvalue())
+
+ def test_requests_app_run(self):
+ """ Should request the application to run. """
+ instance = self.test_instance
+ instance.do_action()
+ self.test_app.run.assert_called_with()
+
+
+class DaemonRunner_do_action_stop_TestCase(DaemonRunner_BaseTestCase):
+ """ Test cases for DaemonRunner.do_action method, action 'stop'. """
+
+ def setUp(self):
+ """ Set up test fixtures. """
+ super(DaemonRunner_do_action_stop_TestCase, self).setUp()
+
+ set_runner_scenario(self, 'pidfile-locked')
+
+ self.test_instance.action = 'stop'
+
+ self.mock_runner_lockfile.is_locked.return_value = True
+ self.mock_runner_lockfile.i_am_locking.return_value = False
+ self.mock_runner_lockfile.read_pid.return_value = (
+ self.scenario['pidlockfile_scenario']['pidfile_pid'])
+
+ def test_raises_error_if_pidfile_not_locked(self):
+ """ Should raise error if PID file is not locked. """
+ set_runner_scenario(self, 'simple')
+ instance = self.test_instance
+ self.mock_runner_lockfile.is_locked.return_value = False
+ self.mock_runner_lockfile.i_am_locking.return_value = False
+ self.mock_runner_lockfile.read_pid.return_value = (
+ self.scenario['pidlockfile_scenario']['pidfile_pid'])
+ pidfile_path = self.scenario['pidfile_path']
+ expected_error = daemon.runner.DaemonRunnerStopFailureError
+ expected_message_content = pidfile_path
+ exc = self.assertRaises(
+ expected_error,
+ instance.do_action)
+ self.assertIn(expected_message_content, unicode(exc))
+
+ def test_breaks_lock_if_pidfile_stale(self):
+ """ Should break lock if PID file is stale. """
+ instance = self.test_instance
+ pidfile_path = self.scenario['pidfile_path']
+ test_pid = self.scenario['pidlockfile_scenario']['pidfile_pid']
+ expected_signal = signal.SIG_DFL
+ test_error = OSError(errno.ESRCH, "Not running")
+ os.kill.side_effect = test_error
+ instance.do_action()
+ self.mock_runner_lockfile.break_lock.assert_called_with()
+
+ def test_sends_terminate_signal_to_process_from_pidfile(self):
+ """ Should send SIGTERM to the daemon process. """
+ instance = self.test_instance
+ test_pid = self.scenario['pidlockfile_scenario']['pidfile_pid']
+ expected_signal = signal.SIGTERM
+ instance.do_action()
+ os.kill.assert_called_with(test_pid, expected_signal)
+
+ def test_raises_error_if_cannot_send_signal_to_process(self):
+ """ Should raise error if cannot send signal to daemon process. """
+ instance = self.test_instance
+ test_pid = self.scenario['pidlockfile_scenario']['pidfile_pid']
+ pidfile_path = self.scenario['pidfile_path']
+ test_error = OSError(errno.EPERM, "Nice try")
+ os.kill.side_effect = test_error
+ expected_error = daemon.runner.DaemonRunnerStopFailureError
+ expected_message_content = unicode(test_pid)
+ exc = self.assertRaises(
+ expected_error,
+ instance.do_action)
+ self.assertIn(expected_message_content, unicode(exc))
+
+
+@mock.patch.object(daemon.runner.DaemonRunner, "_start")
+@mock.patch.object(daemon.runner.DaemonRunner, "_stop")
+class DaemonRunner_do_action_restart_TestCase(DaemonRunner_BaseTestCase):
+ """ Test cases for DaemonRunner.do_action method, action 'restart'. """
+
+ def setUp(self):
+ """ Set up test fixtures. """
+ super(DaemonRunner_do_action_restart_TestCase, self).setUp()
+
+ set_runner_scenario(self, 'pidfile-locked')
+
+ self.test_instance.action = 'restart'
+
+ def test_requests_stop_then_start(
+ self,
+ mock_func_daemonrunner_start, mock_func_daemonrunner_stop):
+ """ Should request stop, then start. """
+ instance = self.test_instance
+ instance.do_action()
+ mock_func_daemonrunner_start.assert_called_with()
+ mock_func_daemonrunner_stop.assert_called_with()
+
+
+@mock.patch.object(sys, "stderr")
+class emit_message_TestCase(scaffold.TestCase):
+ """ Test cases for ‘emit_message’ function. """
+
+ def test_writes_specified_message_to_stream(self, mock_stderr):
+ """ Should write specified message to stream. """
+ test_message = self.getUniqueString()
+ expected_content = "{message}\n".format(message=test_message)
+ daemon.runner.emit_message(test_message, stream=mock_stderr)
+ mock_stderr.write.assert_called_with(expected_content)
+
+ def test_writes_to_specified_stream(self, mock_stderr):
+ """ Should write message to specified stream. """
+ test_message = self.getUniqueString()
+ mock_stream = mock.MagicMock()
+ daemon.runner.emit_message(test_message, stream=mock_stream)
+ mock_stream.write.assert_called_with(mock.ANY)
+
+ def test_writes_to_stderr_by_default(self, mock_stderr):
+ """ Should write message to ‘sys.stderr’ by default. """
+ test_message = self.getUniqueString()
+ daemon.runner.emit_message(test_message)
+ mock_stderr.write.assert_called_with(mock.ANY)
+
+
+class is_pidfile_stale_TestCase(scaffold.TestCase):
+ """ Test cases for ‘is_pidfile_stale’ function. """
+
+ def setUp(self):
+ """ Set up test fixtures. """
+ super(is_pidfile_stale_TestCase, self).setUp()
+
+ func_patcher_os_kill = mock.patch.object(os, "kill")
+ func_patcher_os_kill.start()
+ self.addCleanup(func_patcher_os_kill.stop)
+ os.kill.return_value = None
+
+ self.test_pid = self.getUniqueInteger()
+ self.test_pidfile = mock.MagicMock(daemon.pidfile.TimeoutPIDLockFile)
+ self.test_pidfile.read_pid.return_value = self.test_pid
+
+ def test_returns_false_if_no_pid_in_file(self):
+ """ Should return False if the pidfile contains no PID. """
+ self.test_pidfile.read_pid.return_value = None
+ expected_result = False
+ result = daemon.runner.is_pidfile_stale(self.test_pidfile)
+ self.assertEqual(expected_result, result)
+
+ def test_returns_false_if_process_exists(self):
+ """ Should return False if the process with its PID exists. """
+ expected_result = False
+ result = daemon.runner.is_pidfile_stale(self.test_pidfile)
+ self.assertEqual(expected_result, result)
+
+ def test_returns_true_if_process_does_not_exist(self):
+ """ Should return True if the process does not exist. """
+ test_error = ProcessLookupError("No such process")
+ del os.kill.return_value
+ os.kill.side_effect = test_error
+ expected_result = True
+ result = daemon.runner.is_pidfile_stale(self.test_pidfile)
+ self.assertEqual(expected_result, result)
+
+
+# Local variables:
+# coding: utf-8
+# mode: python
+# End:
+# vim: fileencoding=utf-8 filetype=python :