# -*- coding: utf-8 -*- # # test/test_runner.py # Part of ‘python-daemon’, an implementation of PEP 3143. # # Copyright © 2009–2015 Ben Finney # # 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 :