diff options
Diffstat (limited to 'external_libs/python/python-daemon-2.0.5/test/test_runner.py')
| -rw-r--r-- | external_libs/python/python-daemon-2.0.5/test/test_runner.py | 675 |
1 files changed, 0 insertions, 675 deletions
diff --git a/external_libs/python/python-daemon-2.0.5/test/test_runner.py b/external_libs/python/python-daemon-2.0.5/test/test_runner.py deleted file mode 100644 index 4c0c714a..00000000 --- a/external_libs/python/python-daemon-2.0.5/test/test_runner.py +++ /dev/null @@ -1,675 +0,0 @@ -# -*- 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 : |