diff options
author | 2015-08-24 18:44:56 +0300 | |
---|---|---|
committer | 2015-08-24 18:44:56 +0300 | |
commit | 98bc71b1c21452c3d868c4a4d418a8aeaa5a43af (patch) | |
tree | 2e66b9a775f0a5c545caa76f62890f5a5546e231 /scripts/automation/trex_control_plane/python_lib/python-daemon-2.0.5/test/test_daemon.py | |
parent | 20eb7d362f9bce1951bd61ad3f78cf8f4267d1d5 (diff) | |
parent | 36c6c87fe4380b214f8ff8a45dc0213fa109821c (diff) |
Merge branch 'master' into dan_stateless
Diffstat (limited to 'scripts/automation/trex_control_plane/python_lib/python-daemon-2.0.5/test/test_daemon.py')
-rwxr-xr-x | scripts/automation/trex_control_plane/python_lib/python-daemon-2.0.5/test/test_daemon.py | 1744 |
1 files changed, 0 insertions, 1744 deletions
diff --git a/scripts/automation/trex_control_plane/python_lib/python-daemon-2.0.5/test/test_daemon.py b/scripts/automation/trex_control_plane/python_lib/python-daemon-2.0.5/test/test_daemon.py deleted file mode 100755 index a911858a..00000000 --- a/scripts/automation/trex_control_plane/python_lib/python-daemon-2.0.5/test/test_daemon.py +++ /dev/null @@ -1,1744 +0,0 @@ -# -*- coding: utf-8 -*- -# -# test/test_daemon.py -# Part of ‘python-daemon’, an implementation of PEP 3143. -# -# Copyright © 2008–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 ‘daemon’ module. - """ - -from __future__ import (absolute_import, unicode_literals) - -import os -import sys -import tempfile -import resource -import errno -import signal -import socket -from types import ModuleType -import collections -import functools -try: - # Standard library of Python 2.7 and later. - from io import StringIO -except ImportError: - # Standard library of Python 2.6 and earlier. - from StringIO import StringIO - -import mock - -from . import scaffold -from .scaffold import (basestring, unicode) -from .test_pidfile import ( - FakeFileDescriptorStringIO, - setup_pidfile_fixtures, - ) - -import daemon - - -class ModuleExceptions_TestCase(scaffold.Exception_TestCase): - """ Test cases for module exception classes. """ - - scenarios = scaffold.make_exception_scenarios([ - ('daemon.daemon.DaemonError', dict( - exc_type = daemon.daemon.DaemonError, - min_args = 1, - types = [Exception], - )), - ('daemon.daemon.DaemonOSEnvironmentError', dict( - exc_type = daemon.daemon.DaemonOSEnvironmentError, - min_args = 1, - types = [daemon.daemon.DaemonError, OSError], - )), - ('daemon.daemon.DaemonProcessDetachError', dict( - exc_type = daemon.daemon.DaemonProcessDetachError, - min_args = 1, - types = [daemon.daemon.DaemonError, OSError], - )), - ]) - - -def setup_daemon_context_fixtures(testcase): - """ Set up common test fixtures for DaemonContext test case. - - :param testcase: A ``TestCase`` instance to decorate. - :return: ``None``. - - Decorate the `testcase` with fixtures for tests involving - `DaemonContext`. - - """ - setup_streams_fixtures(testcase) - - setup_pidfile_fixtures(testcase) - - testcase.fake_pidfile_path = tempfile.mktemp() - testcase.mock_pidlockfile = mock.MagicMock() - testcase.mock_pidlockfile.path = testcase.fake_pidfile_path - - testcase.daemon_context_args = dict( - stdin=testcase.stream_files_by_name['stdin'], - stdout=testcase.stream_files_by_name['stdout'], - stderr=testcase.stream_files_by_name['stderr'], - ) - testcase.test_instance = daemon.DaemonContext( - **testcase.daemon_context_args) - -fake_default_signal_map = object() - -@mock.patch.object( - daemon.daemon, "is_detach_process_context_required", - new=(lambda: True)) -@mock.patch.object( - daemon.daemon, "make_default_signal_map", - new=(lambda: fake_default_signal_map)) -@mock.patch.object(os, "setgid", new=(lambda x: object())) -@mock.patch.object(os, "setuid", new=(lambda x: object())) -class DaemonContext_BaseTestCase(scaffold.TestCase): - """ Base class for DaemonContext test case classes. """ - - def setUp(self): - """ Set up test fixtures. """ - super(DaemonContext_BaseTestCase, self).setUp() - - setup_daemon_context_fixtures(self) - - -class DaemonContext_TestCase(DaemonContext_BaseTestCase): - """ Test cases for DaemonContext class. """ - - def test_instantiate(self): - """ New instance of DaemonContext should be created. """ - self.assertIsInstance( - self.test_instance, daemon.daemon.DaemonContext) - - def test_minimum_zero_arguments(self): - """ Initialiser should not require any arguments. """ - instance = daemon.daemon.DaemonContext() - self.assertIsNot(instance, None) - - def test_has_specified_chroot_directory(self): - """ Should have specified chroot_directory option. """ - args = dict( - chroot_directory=object(), - ) - expected_directory = args['chroot_directory'] - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_directory, instance.chroot_directory) - - def test_has_specified_working_directory(self): - """ Should have specified working_directory option. """ - args = dict( - working_directory=object(), - ) - expected_directory = args['working_directory'] - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_directory, instance.working_directory) - - def test_has_default_working_directory(self): - """ Should have default working_directory option. """ - args = dict() - expected_directory = "/" - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_directory, instance.working_directory) - - def test_has_specified_creation_mask(self): - """ Should have specified umask option. """ - args = dict( - umask=object(), - ) - expected_mask = args['umask'] - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_mask, instance.umask) - - def test_has_default_creation_mask(self): - """ Should have default umask option. """ - args = dict() - expected_mask = 0 - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_mask, instance.umask) - - def test_has_specified_uid(self): - """ Should have specified uid option. """ - args = dict( - uid=object(), - ) - expected_id = args['uid'] - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_id, instance.uid) - - def test_has_derived_uid(self): - """ Should have uid option derived from process. """ - args = dict() - expected_id = os.getuid() - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_id, instance.uid) - - def test_has_specified_gid(self): - """ Should have specified gid option. """ - args = dict( - gid=object(), - ) - expected_id = args['gid'] - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_id, instance.gid) - - def test_has_derived_gid(self): - """ Should have gid option derived from process. """ - args = dict() - expected_id = os.getgid() - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_id, instance.gid) - - def test_has_specified_detach_process(self): - """ Should have specified detach_process option. """ - args = dict( - detach_process=object(), - ) - expected_value = args['detach_process'] - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_value, instance.detach_process) - - def test_has_derived_detach_process(self): - """ Should have detach_process option derived from environment. """ - args = dict() - func = daemon.daemon.is_detach_process_context_required - expected_value = func() - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_value, instance.detach_process) - - def test_has_specified_files_preserve(self): - """ Should have specified files_preserve option. """ - args = dict( - files_preserve=object(), - ) - expected_files_preserve = args['files_preserve'] - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_files_preserve, instance.files_preserve) - - def test_has_specified_pidfile(self): - """ Should have the specified pidfile. """ - args = dict( - pidfile=object(), - ) - expected_pidfile = args['pidfile'] - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_pidfile, instance.pidfile) - - def test_has_specified_stdin(self): - """ Should have specified stdin option. """ - args = dict( - stdin=object(), - ) - expected_file = args['stdin'] - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_file, instance.stdin) - - def test_has_specified_stdout(self): - """ Should have specified stdout option. """ - args = dict( - stdout=object(), - ) - expected_file = args['stdout'] - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_file, instance.stdout) - - def test_has_specified_stderr(self): - """ Should have specified stderr option. """ - args = dict( - stderr=object(), - ) - expected_file = args['stderr'] - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_file, instance.stderr) - - def test_has_specified_signal_map(self): - """ Should have specified signal_map option. """ - args = dict( - signal_map=object(), - ) - expected_signal_map = args['signal_map'] - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_signal_map, instance.signal_map) - - def test_has_derived_signal_map(self): - """ Should have signal_map option derived from system. """ - args = dict() - expected_signal_map = daemon.daemon.make_default_signal_map() - instance = daemon.daemon.DaemonContext(**args) - self.assertEqual(expected_signal_map, instance.signal_map) - - -class DaemonContext_is_open_TestCase(DaemonContext_BaseTestCase): - """ Test cases for DaemonContext.is_open property. """ - - def test_begin_false(self): - """ Initial value of is_open should be False. """ - instance = self.test_instance - self.assertEqual(False, instance.is_open) - - def test_write_fails(self): - """ Writing to is_open should fail. """ - instance = self.test_instance - self.assertRaises( - AttributeError, - setattr, instance, 'is_open', object()) - - -class DaemonContext_open_TestCase(DaemonContext_BaseTestCase): - """ Test cases for DaemonContext.open method. """ - - def setUp(self): - """ Set up test fixtures. """ - super(DaemonContext_open_TestCase, self).setUp() - - self.test_instance._is_open = False - - self.mock_module_daemon = mock.MagicMock() - daemon_func_patchers = dict( - (func_name, mock.patch.object( - daemon.daemon, func_name)) - for func_name in [ - "detach_process_context", - "change_working_directory", - "change_root_directory", - "change_file_creation_mask", - "change_process_owner", - "prevent_core_dump", - "close_all_open_files", - "redirect_stream", - "set_signal_handlers", - "register_atexit_function", - ]) - for (func_name, patcher) in daemon_func_patchers.items(): - mock_func = patcher.start() - self.addCleanup(patcher.stop) - self.mock_module_daemon.attach_mock(mock_func, func_name) - - self.mock_module_daemon.attach_mock(mock.Mock(), 'DaemonContext') - - self.test_files_preserve_fds = object() - self.test_signal_handler_map = object() - daemoncontext_method_return_values = { - '_get_exclude_file_descriptors': - self.test_files_preserve_fds, - '_make_signal_handler_map': - self.test_signal_handler_map, - } - daemoncontext_func_patchers = dict( - (func_name, mock.patch.object( - daemon.daemon.DaemonContext, - func_name, - return_value=return_value)) - for (func_name, return_value) in - daemoncontext_method_return_values.items()) - for (func_name, patcher) in daemoncontext_func_patchers.items(): - mock_func = patcher.start() - self.addCleanup(patcher.stop) - self.mock_module_daemon.DaemonContext.attach_mock( - mock_func, func_name) - - def test_performs_steps_in_expected_sequence(self): - """ Should perform daemonisation steps in expected sequence. """ - instance = self.test_instance - instance.chroot_directory = object() - instance.detach_process = True - instance.pidfile = self.mock_pidlockfile - self.mock_module_daemon.attach_mock( - self.mock_pidlockfile, 'pidlockfile') - expected_calls = [ - mock.call.change_root_directory(mock.ANY), - mock.call.prevent_core_dump(), - mock.call.change_file_creation_mask(mock.ANY), - mock.call.change_working_directory(mock.ANY), - mock.call.change_process_owner(mock.ANY, mock.ANY), - mock.call.detach_process_context(), - mock.call.DaemonContext._make_signal_handler_map(), - mock.call.set_signal_handlers(mock.ANY), - mock.call.DaemonContext._get_exclude_file_descriptors(), - mock.call.close_all_open_files(exclude=mock.ANY), - mock.call.redirect_stream(mock.ANY, mock.ANY), - mock.call.redirect_stream(mock.ANY, mock.ANY), - mock.call.redirect_stream(mock.ANY, mock.ANY), - mock.call.pidlockfile.__enter__(), - mock.call.register_atexit_function(mock.ANY), - ] - instance.open() - self.mock_module_daemon.assert_has_calls(expected_calls) - - def test_returns_immediately_if_is_open(self): - """ Should return immediately if is_open property is true. """ - instance = self.test_instance - instance._is_open = True - instance.open() - self.assertEqual(0, len(self.mock_module_daemon.mock_calls)) - - def test_changes_root_directory_to_chroot_directory(self): - """ Should change root directory to `chroot_directory` option. """ - instance = self.test_instance - chroot_directory = object() - instance.chroot_directory = chroot_directory - instance.open() - self.mock_module_daemon.change_root_directory.assert_called_with( - chroot_directory) - - def test_omits_chroot_if_no_chroot_directory(self): - """ Should omit changing root directory if no `chroot_directory`. """ - instance = self.test_instance - instance.chroot_directory = None - instance.open() - self.assertFalse(self.mock_module_daemon.change_root_directory.called) - - def test_prevents_core_dump(self): - """ Should request prevention of core dumps. """ - instance = self.test_instance - instance.open() - self.mock_module_daemon.prevent_core_dump.assert_called_with() - - def test_omits_prevent_core_dump_if_prevent_core_false(self): - """ Should omit preventing core dumps if `prevent_core` is false. """ - instance = self.test_instance - instance.prevent_core = False - instance.open() - self.assertFalse(self.mock_module_daemon.prevent_core_dump.called) - - def test_closes_open_files(self): - """ Should close all open files, excluding `files_preserve`. """ - instance = self.test_instance - expected_exclude = self.test_files_preserve_fds - instance.open() - self.mock_module_daemon.close_all_open_files.assert_called_with( - exclude=expected_exclude) - - def test_changes_directory_to_working_directory(self): - """ Should change current directory to `working_directory` option. """ - instance = self.test_instance - working_directory = object() - instance.working_directory = working_directory - instance.open() - self.mock_module_daemon.change_working_directory.assert_called_with( - working_directory) - - def test_changes_creation_mask_to_umask(self): - """ Should change file creation mask to `umask` option. """ - instance = self.test_instance - umask = object() - instance.umask = umask - instance.open() - self.mock_module_daemon.change_file_creation_mask.assert_called_with( - umask) - - def test_changes_owner_to_specified_uid_and_gid(self): - """ Should change process UID and GID to `uid` and `gid` options. """ - instance = self.test_instance - uid = object() - gid = object() - instance.uid = uid - instance.gid = gid - instance.open() - self.mock_module_daemon.change_process_owner.assert_called_with( - uid, gid) - - def test_detaches_process_context(self): - """ Should request detach of process context. """ - instance = self.test_instance - instance.open() - self.mock_module_daemon.detach_process_context.assert_called_with() - - def test_omits_process_detach_if_not_required(self): - """ Should omit detach of process context if not required. """ - instance = self.test_instance - instance.detach_process = False - instance.open() - self.assertFalse(self.mock_module_daemon.detach_process_context.called) - - def test_sets_signal_handlers_from_signal_map(self): - """ Should set signal handlers according to `signal_map`. """ - instance = self.test_instance - instance.signal_map = object() - expected_signal_handler_map = self.test_signal_handler_map - instance.open() - self.mock_module_daemon.set_signal_handlers.assert_called_with( - expected_signal_handler_map) - - def test_redirects_standard_streams(self): - """ Should request redirection of standard stream files. """ - instance = self.test_instance - (system_stdin, system_stdout, system_stderr) = ( - sys.stdin, sys.stdout, sys.stderr) - (target_stdin, target_stdout, target_stderr) = ( - self.stream_files_by_name[name] - for name in ['stdin', 'stdout', 'stderr']) - expected_calls = [ - mock.call(system_stdin, target_stdin), - mock.call(system_stdout, target_stdout), - mock.call(system_stderr, target_stderr), - ] - instance.open() - self.mock_module_daemon.redirect_stream.assert_has_calls( - expected_calls, any_order=True) - - def test_enters_pidfile_context(self): - """ Should enter the PID file context manager. """ - instance = self.test_instance - instance.pidfile = self.mock_pidlockfile - instance.open() - self.mock_pidlockfile.__enter__.assert_called_with() - - def test_sets_is_open_true(self): - """ Should set the `is_open` property to True. """ - instance = self.test_instance - instance.open() - self.assertEqual(True, instance.is_open) - - def test_registers_close_method_for_atexit(self): - """ Should register the `close` method for atexit processing. """ - instance = self.test_instance - close_method = instance.close - instance.open() - self.mock_module_daemon.register_atexit_function.assert_called_with( - close_method) - - -class DaemonContext_close_TestCase(DaemonContext_BaseTestCase): - """ Test cases for DaemonContext.close method. """ - - def setUp(self): - """ Set up test fixtures. """ - super(DaemonContext_close_TestCase, self).setUp() - - self.test_instance._is_open = True - - def test_returns_immediately_if_not_is_open(self): - """ Should return immediately if is_open property is false. """ - instance = self.test_instance - instance._is_open = False - instance.pidfile = object() - instance.close() - self.assertFalse(self.mock_pidlockfile.__exit__.called) - - def test_exits_pidfile_context(self): - """ Should exit the PID file context manager. """ - instance = self.test_instance - instance.pidfile = self.mock_pidlockfile - instance.close() - self.mock_pidlockfile.__exit__.assert_called_with(None, None, None) - - def test_returns_none(self): - """ Should return None. """ - instance = self.test_instance - expected_result = None - result = instance.close() - self.assertIs(result, expected_result) - - def test_sets_is_open_false(self): - """ Should set the `is_open` property to False. """ - instance = self.test_instance - instance.close() - self.assertEqual(False, instance.is_open) - - -@mock.patch.object(daemon.daemon.DaemonContext, "open") -class DaemonContext_context_manager_enter_TestCase(DaemonContext_BaseTestCase): - """ Test cases for DaemonContext.__enter__ method. """ - - def test_opens_daemon_context(self, mock_func_daemoncontext_open): - """ Should open the DaemonContext. """ - instance = self.test_instance - instance.__enter__() - mock_func_daemoncontext_open.assert_called_with() - - def test_returns_self_instance(self, mock_func_daemoncontext_open): - """ Should return DaemonContext instance. """ - instance = self.test_instance - expected_result = instance - result = instance.__enter__() - self.assertIs(result, expected_result) - - -@mock.patch.object(daemon.daemon.DaemonContext, "close") -class DaemonContext_context_manager_exit_TestCase(DaemonContext_BaseTestCase): - """ Test cases for DaemonContext.__exit__ method. """ - - def setUp(self): - """ Set up test fixtures. """ - super(DaemonContext_context_manager_exit_TestCase, self).setUp() - - self.test_args = dict( - exc_type=object(), - exc_value=object(), - traceback=object(), - ) - - def test_closes_daemon_context(self, mock_func_daemoncontext_close): - """ Should close the DaemonContext. """ - instance = self.test_instance - args = self.test_args - instance.__exit__(**args) - mock_func_daemoncontext_close.assert_called_with() - - def test_returns_none(self, mock_func_daemoncontext_close): - """ Should return None, indicating exception was not handled. """ - instance = self.test_instance - args = self.test_args - expected_result = None - result = instance.__exit__(**args) - self.assertIs(result, expected_result) - - -class DaemonContext_terminate_TestCase(DaemonContext_BaseTestCase): - """ Test cases for DaemonContext.terminate method. """ - - def setUp(self): - """ Set up test fixtures. """ - super(DaemonContext_terminate_TestCase, self).setUp() - - self.test_signal = signal.SIGTERM - self.test_frame = None - self.test_args = (self.test_signal, self.test_frame) - - def test_raises_system_exit(self): - """ Should raise SystemExit. """ - instance = self.test_instance - args = self.test_args - expected_exception = SystemExit - self.assertRaises( - expected_exception, - instance.terminate, *args) - - def test_exception_message_contains_signal_number(self): - """ Should raise exception with a message containing signal number. """ - instance = self.test_instance - args = self.test_args - signal_number = self.test_signal - expected_exception = SystemExit - exc = self.assertRaises( - expected_exception, - instance.terminate, *args) - self.assertIn(unicode(signal_number), unicode(exc)) - - -class DaemonContext_get_exclude_file_descriptors_TestCase( - DaemonContext_BaseTestCase): - """ Test cases for DaemonContext._get_exclude_file_descriptors function. """ - - def setUp(self): - """ Set up test fixtures. """ - super( - DaemonContext_get_exclude_file_descriptors_TestCase, - self).setUp() - - self.test_files = { - 2: FakeFileDescriptorStringIO(), - 5: 5, - 11: FakeFileDescriptorStringIO(), - 17: None, - 23: FakeFileDescriptorStringIO(), - 37: 37, - 42: FakeFileDescriptorStringIO(), - } - for (fileno, item) in self.test_files.items(): - if hasattr(item, '_fileno'): - item._fileno = fileno - self.test_file_descriptors = set( - fd for (fd, item) in self.test_files.items() - if item is not None) - self.test_file_descriptors.update( - self.stream_files_by_name[name].fileno() - for name in ['stdin', 'stdout', 'stderr'] - ) - - def test_returns_expected_file_descriptors(self): - """ Should return expected set of file descriptors. """ - instance = self.test_instance - instance.files_preserve = list(self.test_files.values()) - expected_result = self.test_file_descriptors - result = instance._get_exclude_file_descriptors() - self.assertEqual(expected_result, result) - - def test_returns_stream_redirects_if_no_files_preserve(self): - """ Should return only stream redirects if no files_preserve. """ - instance = self.test_instance - instance.files_preserve = None - expected_result = set( - stream.fileno() - for stream in self.stream_files_by_name.values()) - result = instance._get_exclude_file_descriptors() - self.assertEqual(expected_result, result) - - def test_returns_empty_set_if_no_files(self): - """ Should return empty set if no file options. """ - instance = self.test_instance - for name in ['files_preserve', 'stdin', 'stdout', 'stderr']: - setattr(instance, name, None) - expected_result = set() - result = instance._get_exclude_file_descriptors() - self.assertEqual(expected_result, result) - - def test_omits_non_file_streams(self): - """ Should omit non-file stream attributes. """ - instance = self.test_instance - instance.files_preserve = list(self.test_files.values()) - stream_files = self.stream_files_by_name - expected_result = self.test_file_descriptors.copy() - for (pseudo_stream_name, pseudo_stream) in stream_files.items(): - test_non_file_object = object() - setattr(instance, pseudo_stream_name, test_non_file_object) - stream_fd = pseudo_stream.fileno() - expected_result.discard(stream_fd) - result = instance._get_exclude_file_descriptors() - self.assertEqual(expected_result, result) - - def test_includes_verbatim_streams_without_file_descriptor(self): - """ Should include verbatim any stream without a file descriptor. """ - instance = self.test_instance - instance.files_preserve = list(self.test_files.values()) - stream_files = self.stream_files_by_name - mock_fileno_method = mock.MagicMock( - spec=sys.__stdin__.fileno, - side_effect=ValueError) - expected_result = self.test_file_descriptors.copy() - for (pseudo_stream_name, pseudo_stream) in stream_files.items(): - test_non_fd_stream = StringIO() - if not hasattr(test_non_fd_stream, 'fileno'): - # Python < 3 StringIO doesn't have ‘fileno’ at all. - # Add a method which raises an exception. - test_non_fd_stream.fileno = mock_fileno_method - setattr(instance, pseudo_stream_name, test_non_fd_stream) - stream_fd = pseudo_stream.fileno() - expected_result.discard(stream_fd) - expected_result.add(test_non_fd_stream) - result = instance._get_exclude_file_descriptors() - self.assertEqual(expected_result, result) - - def test_omits_none_streams(self): - """ Should omit any stream attribute which is None. """ - instance = self.test_instance - instance.files_preserve = list(self.test_files.values()) - stream_files = self.stream_files_by_name - expected_result = self.test_file_descriptors.copy() - for (pseudo_stream_name, pseudo_stream) in stream_files.items(): - setattr(instance, pseudo_stream_name, None) - stream_fd = pseudo_stream.fileno() - expected_result.discard(stream_fd) - result = instance._get_exclude_file_descriptors() - self.assertEqual(expected_result, result) - - -class DaemonContext_make_signal_handler_TestCase(DaemonContext_BaseTestCase): - """ Test cases for DaemonContext._make_signal_handler function. """ - - def test_returns_ignore_for_none(self): - """ Should return SIG_IGN when None handler specified. """ - instance = self.test_instance - target = None - expected_result = signal.SIG_IGN - result = instance._make_signal_handler(target) - self.assertEqual(expected_result, result) - - def test_returns_method_for_name(self): - """ Should return method of DaemonContext when name specified. """ - instance = self.test_instance - target = 'terminate' - expected_result = instance.terminate - result = instance._make_signal_handler(target) - self.assertEqual(expected_result, result) - - def test_raises_error_for_unknown_name(self): - """ Should raise AttributeError for unknown method name. """ - instance = self.test_instance - target = 'b0gUs' - expected_error = AttributeError - self.assertRaises( - expected_error, - instance._make_signal_handler, target) - - def test_returns_object_for_object(self): - """ Should return same object for any other object. """ - instance = self.test_instance - target = object() - expected_result = target - result = instance._make_signal_handler(target) - self.assertEqual(expected_result, result) - - -class DaemonContext_make_signal_handler_map_TestCase( - DaemonContext_BaseTestCase): - """ Test cases for DaemonContext._make_signal_handler_map function. """ - - def setUp(self): - """ Set up test fixtures. """ - super(DaemonContext_make_signal_handler_map_TestCase, self).setUp() - - self.test_instance.signal_map = { - object(): object(), - object(): object(), - object(): object(), - } - - self.test_signal_handlers = dict( - (key, object()) - for key in self.test_instance.signal_map.values()) - self.test_signal_handler_map = dict( - (key, self.test_signal_handlers[target]) - for (key, target) in self.test_instance.signal_map.items()) - - def fake_make_signal_handler(target): - return self.test_signal_handlers[target] - - func_patcher_make_signal_handler = mock.patch.object( - daemon.daemon.DaemonContext, "_make_signal_handler", - side_effect=fake_make_signal_handler) - self.mock_func_make_signal_handler = ( - func_patcher_make_signal_handler.start()) - self.addCleanup(func_patcher_make_signal_handler.stop) - - def test_returns_constructed_signal_handler_items(self): - """ Should return items as constructed via make_signal_handler. """ - instance = self.test_instance - expected_result = self.test_signal_handler_map - result = instance._make_signal_handler_map() - self.assertEqual(expected_result, result) - - -try: - FileNotFoundError -except NameError: - # Python 2 uses IOError. - FileNotFoundError = functools.partial(IOError, errno.ENOENT) - - -@mock.patch.object(os, "chdir") -class change_working_directory_TestCase(scaffold.TestCase): - """ Test cases for change_working_directory function. """ - - def setUp(self): - """ Set up test fixtures. """ - super(change_working_directory_TestCase, self).setUp() - - self.test_directory = object() - self.test_args = dict( - directory=self.test_directory, - ) - - def test_changes_working_directory_to_specified_directory( - self, - mock_func_os_chdir): - """ Should change working directory to specified directory. """ - args = self.test_args - directory = self.test_directory - daemon.daemon.change_working_directory(**args) - mock_func_os_chdir.assert_called_with(directory) - - def test_raises_daemon_error_on_os_error( - self, - mock_func_os_chdir): - """ Should raise a DaemonError on receiving an IOError. """ - args = self.test_args - test_error = FileNotFoundError("No such directory") - mock_func_os_chdir.side_effect = test_error - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.change_working_directory, **args) - self.assertEqual(test_error, exc.__cause__) - - def test_error_message_contains_original_error_message( - self, - mock_func_os_chdir): - """ Should raise a DaemonError with original message. """ - args = self.test_args - test_error = FileNotFoundError("No such directory") - mock_func_os_chdir.side_effect = test_error - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.change_working_directory, **args) - self.assertIn(unicode(test_error), unicode(exc)) - - -@mock.patch.object(os, "chroot") -@mock.patch.object(os, "chdir") -class change_root_directory_TestCase(scaffold.TestCase): - """ Test cases for change_root_directory function. """ - - def setUp(self): - """ Set up test fixtures. """ - super(change_root_directory_TestCase, self).setUp() - - self.test_directory = object() - self.test_args = dict( - directory=self.test_directory, - ) - - def test_changes_working_directory_to_specified_directory( - self, - mock_func_os_chdir, mock_func_os_chroot): - """ Should change working directory to specified directory. """ - args = self.test_args - directory = self.test_directory - daemon.daemon.change_root_directory(**args) - mock_func_os_chdir.assert_called_with(directory) - - def test_changes_root_directory_to_specified_directory( - self, - mock_func_os_chdir, mock_func_os_chroot): - """ Should change root directory to specified directory. """ - args = self.test_args - directory = self.test_directory - daemon.daemon.change_root_directory(**args) - mock_func_os_chroot.assert_called_with(directory) - - def test_raises_daemon_error_on_os_error_from_chdir( - self, - mock_func_os_chdir, mock_func_os_chroot): - """ Should raise a DaemonError on receiving an IOError from chdir. """ - args = self.test_args - test_error = FileNotFoundError("No such directory") - mock_func_os_chdir.side_effect = test_error - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.change_root_directory, **args) - self.assertEqual(test_error, exc.__cause__) - - def test_raises_daemon_error_on_os_error_from_chroot( - self, - mock_func_os_chdir, mock_func_os_chroot): - """ Should raise a DaemonError on receiving an OSError from chroot. """ - args = self.test_args - test_error = OSError(errno.EPERM, "No chroot for you!") - mock_func_os_chroot.side_effect = test_error - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.change_root_directory, **args) - self.assertEqual(test_error, exc.__cause__) - - def test_error_message_contains_original_error_message( - self, - mock_func_os_chdir, mock_func_os_chroot): - """ Should raise a DaemonError with original message. """ - args = self.test_args - test_error = FileNotFoundError("No such directory") - mock_func_os_chdir.side_effect = test_error - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.change_root_directory, **args) - self.assertIn(unicode(test_error), unicode(exc)) - - -@mock.patch.object(os, "umask") -class change_file_creation_mask_TestCase(scaffold.TestCase): - """ Test cases for change_file_creation_mask function. """ - - def setUp(self): - """ Set up test fixtures. """ - super(change_file_creation_mask_TestCase, self).setUp() - - self.test_mask = object() - self.test_args = dict( - mask=self.test_mask, - ) - - def test_changes_umask_to_specified_mask(self, mock_func_os_umask): - """ Should change working directory to specified directory. """ - args = self.test_args - mask = self.test_mask - daemon.daemon.change_file_creation_mask(**args) - mock_func_os_umask.assert_called_with(mask) - - def test_raises_daemon_error_on_os_error_from_chdir( - self, - mock_func_os_umask): - """ Should raise a DaemonError on receiving an OSError from umask. """ - args = self.test_args - test_error = OSError(errno.EINVAL, "Whatchoo talkin' 'bout?") - mock_func_os_umask.side_effect = test_error - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.change_file_creation_mask, **args) - self.assertEqual(test_error, exc.__cause__) - - def test_error_message_contains_original_error_message( - self, - mock_func_os_umask): - """ Should raise a DaemonError with original message. """ - args = self.test_args - test_error = FileNotFoundError("No such directory") - mock_func_os_umask.side_effect = test_error - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.change_file_creation_mask, **args) - self.assertIn(unicode(test_error), unicode(exc)) - - -@mock.patch.object(os, "setgid") -@mock.patch.object(os, "setuid") -class change_process_owner_TestCase(scaffold.TestCase): - """ Test cases for change_process_owner function. """ - - def setUp(self): - """ Set up test fixtures. """ - super(change_process_owner_TestCase, self).setUp() - - self.test_uid = object() - self.test_gid = object() - self.test_args = dict( - uid=self.test_uid, - gid=self.test_gid, - ) - - def test_changes_gid_and_uid_in_order( - self, - mock_func_os_setuid, mock_func_os_setgid): - """ Should change process GID and UID in correct order. - - Since the process requires appropriate privilege to use - either of `setuid` or `setgid`, changing the UID must be - done last. - - """ - args = self.test_args - daemon.daemon.change_process_owner(**args) - mock_func_os_setuid.assert_called() - mock_func_os_setgid.assert_called() - - def test_changes_group_id_to_gid( - self, - mock_func_os_setuid, mock_func_os_setgid): - """ Should change process GID to specified value. """ - args = self.test_args - gid = self.test_gid - daemon.daemon.change_process_owner(**args) - mock_func_os_setgid.assert_called(gid) - - def test_changes_user_id_to_uid( - self, - mock_func_os_setuid, mock_func_os_setgid): - """ Should change process UID to specified value. """ - args = self.test_args - uid = self.test_uid - daemon.daemon.change_process_owner(**args) - mock_func_os_setuid.assert_called(uid) - - def test_raises_daemon_error_on_os_error_from_setgid( - self, - mock_func_os_setuid, mock_func_os_setgid): - """ Should raise a DaemonError on receiving an OSError from setgid. """ - args = self.test_args - test_error = OSError(errno.EPERM, "No switching for you!") - mock_func_os_setgid.side_effect = test_error - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.change_process_owner, **args) - self.assertEqual(test_error, exc.__cause__) - - def test_raises_daemon_error_on_os_error_from_setuid( - self, - mock_func_os_setuid, mock_func_os_setgid): - """ Should raise a DaemonError on receiving an OSError from setuid. """ - args = self.test_args - test_error = OSError(errno.EPERM, "No switching for you!") - mock_func_os_setuid.side_effect = test_error - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.change_process_owner, **args) - self.assertEqual(test_error, exc.__cause__) - - def test_error_message_contains_original_error_message( - self, - mock_func_os_setuid, mock_func_os_setgid): - """ Should raise a DaemonError with original message. """ - args = self.test_args - test_error = OSError(errno.EINVAL, "Whatchoo talkin' 'bout?") - mock_func_os_setuid.side_effect = test_error - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.change_process_owner, **args) - self.assertIn(unicode(test_error), unicode(exc)) - - -RLimitResult = collections.namedtuple('RLimitResult', ['soft', 'hard']) - -fake_RLIMIT_CORE = object() - -@mock.patch.object(resource, "RLIMIT_CORE", new=fake_RLIMIT_CORE) -@mock.patch.object(resource, "setrlimit", side_effect=(lambda x, y: None)) -@mock.patch.object(resource, "getrlimit", side_effect=(lambda x: None)) -class prevent_core_dump_TestCase(scaffold.TestCase): - """ Test cases for prevent_core_dump function. """ - - def setUp(self): - """ Set up test fixtures. """ - super(prevent_core_dump_TestCase, self).setUp() - - def test_sets_core_limit_to_zero( - self, - mock_func_resource_getrlimit, mock_func_resource_setrlimit): - """ Should set the RLIMIT_CORE resource to zero. """ - expected_resource = fake_RLIMIT_CORE - expected_limit = tuple(RLimitResult(soft=0, hard=0)) - daemon.daemon.prevent_core_dump() - mock_func_resource_getrlimit.assert_called_with(expected_resource) - mock_func_resource_setrlimit.assert_called_with( - expected_resource, expected_limit) - - def test_raises_error_when_no_core_resource( - self, - mock_func_resource_getrlimit, mock_func_resource_setrlimit): - """ Should raise DaemonError if no RLIMIT_CORE resource. """ - test_error = ValueError("Bogus platform doesn't have RLIMIT_CORE") - def fake_getrlimit(res): - if res == resource.RLIMIT_CORE: - raise test_error - else: - return None - mock_func_resource_getrlimit.side_effect = fake_getrlimit - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.prevent_core_dump) - self.assertEqual(test_error, exc.__cause__) - - -@mock.patch.object(os, "close") -class close_file_descriptor_if_open_TestCase(scaffold.TestCase): - """ Test cases for close_file_descriptor_if_open function. """ - - def setUp(self): - """ Set up test fixtures. """ - super(close_file_descriptor_if_open_TestCase, self).setUp() - - self.fake_fd = 274 - - def test_requests_file_descriptor_close(self, mock_func_os_close): - """ Should request close of file descriptor. """ - fd = self.fake_fd - daemon.daemon.close_file_descriptor_if_open(fd) - mock_func_os_close.assert_called_with(fd) - - def test_ignores_badfd_error_on_close(self, mock_func_os_close): - """ Should ignore OSError EBADF when closing. """ - fd = self.fake_fd - test_error = OSError(errno.EBADF, "Bad file descriptor") - def fake_os_close(fd): - raise test_error - mock_func_os_close.side_effect = fake_os_close - daemon.daemon.close_file_descriptor_if_open(fd) - mock_func_os_close.assert_called_with(fd) - - def test_raises_error_if_oserror_on_close(self, mock_func_os_close): - """ Should raise DaemonError if an OSError occurs when closing. """ - fd = self.fake_fd - test_error = OSError(object(), "Unexpected error") - def fake_os_close(fd): - raise test_error - mock_func_os_close.side_effect = fake_os_close - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.close_file_descriptor_if_open, fd) - self.assertEqual(test_error, exc.__cause__) - - def test_raises_error_if_ioerror_on_close(self, mock_func_os_close): - """ Should raise DaemonError if an IOError occurs when closing. """ - fd = self.fake_fd - test_error = IOError(object(), "Unexpected error") - def fake_os_close(fd): - raise test_error - mock_func_os_close.side_effect = fake_os_close - expected_error = daemon.daemon.DaemonOSEnvironmentError - exc = self.assertRaises( - expected_error, - daemon.daemon.close_file_descriptor_if_open, fd) - self.assertEqual(test_error, exc.__cause__) - - -class maxfd_TestCase(scaffold.TestCase): - """ Test cases for module MAXFD constant. """ - - def test_positive(self): - """ Should be a positive number. """ - maxfd = daemon.daemon.MAXFD - self.assertTrue(maxfd > 0) - - def test_integer(self): - """ Should be an integer. """ - maxfd = daemon.daemon.MAXFD - self.assertEqual(int(maxfd), maxfd) - - def test_reasonably_high(self): - """ Should be reasonably high for default open files limit. - - If the system reports a limit of “infinity” on maximum - file descriptors, we still need a finite number in order - to close “all” of them. Ensure this is reasonably high - to catch most use cases. - - """ - expected_minimum = 2048 - maxfd = daemon.daemon.MAXFD - self.assertTrue( - expected_minimum <= maxfd, - msg=( - "MAXFD should be at least {minimum!r}" - " (got {maxfd!r})".format( - minimum=expected_minimum, maxfd=maxfd))) - - -fake_default_maxfd = 8 -fake_RLIMIT_NOFILE = object() -fake_RLIM_INFINITY = object() -fake_rlimit_nofile_large = 2468 - -def fake_getrlimit_nofile_soft_infinity(resource): - result = RLimitResult(soft=fake_RLIM_INFINITY, hard=object()) - if resource != fake_RLIMIT_NOFILE: - result = NotImplemented - return result - -def fake_getrlimit_nofile_hard_infinity(resource): - result = RLimitResult(soft=object(), hard=fake_RLIM_INFINITY) - if resource != fake_RLIMIT_NOFILE: - result = NotImplemented - return result - -def fake_getrlimit_nofile_hard_large(resource): - result = RLimitResult(soft=object(), hard=fake_rlimit_nofile_large) - if resource != fake_RLIMIT_NOFILE: - result = NotImplemented - return result - -@mock.patch.object(daemon.daemon, "MAXFD", new=fake_default_maxfd) -@mock.patch.object(resource, "RLIMIT_NOFILE", new=fake_RLIMIT_NOFILE) -@mock.patch.object(resource, "RLIM_INFINITY", new=fake_RLIM_INFINITY) -@mock.patch.object( - resource, "getrlimit", - side_effect=fake_getrlimit_nofile_hard_large) -class get_maximum_file_descriptors_TestCase(scaffold.TestCase): - """ Test cases for get_maximum_file_descriptors function. """ - - def test_returns_system_hard_limit(self, mock_func_resource_getrlimit): - """ Should return process hard limit on number of files. """ - expected_result = fake_rlimit_nofile_large - result = daemon.daemon.get_maximum_file_descriptors() - self.assertEqual(expected_result, result) - - def test_returns_module_default_if_hard_limit_infinity( - self, mock_func_resource_getrlimit): - """ Should return module MAXFD if hard limit is infinity. """ - mock_func_resource_getrlimit.side_effect = ( - fake_getrlimit_nofile_hard_infinity) - expected_result = fake_default_maxfd - result = daemon.daemon.get_maximum_file_descriptors() - self.assertEqual(expected_result, result) - - -def fake_get_maximum_file_descriptors(): - return fake_default_maxfd - -@mock.patch.object(resource, "RLIMIT_NOFILE", new=fake_RLIMIT_NOFILE) -@mock.patch.object(resource, "RLIM_INFINITY", new=fake_RLIM_INFINITY) -@mock.patch.object( - resource, "getrlimit", - new=fake_getrlimit_nofile_soft_infinity) -@mock.patch.object( - daemon.daemon, "get_maximum_file_descriptors", - new=fake_get_maximum_file_descriptors) -@mock.patch.object(daemon.daemon, "close_file_descriptor_if_open") -class close_all_open_files_TestCase(scaffold.TestCase): - """ Test cases for close_all_open_files function. """ - - def test_requests_all_open_files_to_close( - self, mock_func_close_file_descriptor_if_open): - """ Should request close of all open files. """ - expected_file_descriptors = range(fake_default_maxfd) - expected_calls = [ - mock.call(fd) for fd in expected_file_descriptors] - daemon.daemon.close_all_open_files() - mock_func_close_file_descriptor_if_open.assert_has_calls( - expected_calls, any_order=True) - - def test_requests_all_but_excluded_files_to_close( - self, mock_func_close_file_descriptor_if_open): - """ Should request close of all open files but those excluded. """ - test_exclude = set([3, 7]) - args = dict( - exclude=test_exclude, - ) - expected_file_descriptors = set( - fd for fd in range(fake_default_maxfd) - if fd not in test_exclude) - expected_calls = [ - mock.call(fd) for fd in expected_file_descriptors] - daemon.daemon.close_all_open_files(**args) - mock_func_close_file_descriptor_if_open.assert_has_calls( - expected_calls, any_order=True) - - -class detach_process_context_TestCase(scaffold.TestCase): - """ Test cases for detach_process_context function. """ - - class FakeOSExit(SystemExit): - """ Fake exception raised for os._exit(). """ - - def setUp(self): - """ Set up test fixtures. """ - super(detach_process_context_TestCase, self).setUp() - - self.mock_module_os = mock.MagicMock(wraps=os) - - fake_pids = [0, 0] - func_patcher_os_fork = mock.patch.object( - os, "fork", - side_effect=iter(fake_pids)) - self.mock_func_os_fork = func_patcher_os_fork.start() - self.addCleanup(func_patcher_os_fork.stop) - self.mock_module_os.attach_mock(self.mock_func_os_fork, "fork") - - func_patcher_os_setsid = mock.patch.object(os, "setsid") - self.mock_func_os_setsid = func_patcher_os_setsid.start() - self.addCleanup(func_patcher_os_setsid.stop) - self.mock_module_os.attach_mock(self.mock_func_os_setsid, "setsid") - - def raise_os_exit(status=None): - raise self.FakeOSExit(status) - - func_patcher_os_force_exit = mock.patch.object( - os, "_exit", - side_effect=raise_os_exit) - self.mock_func_os_force_exit = func_patcher_os_force_exit.start() - self.addCleanup(func_patcher_os_force_exit.stop) - self.mock_module_os.attach_mock(self.mock_func_os_force_exit, "_exit") - - def test_parent_exits(self): - """ Parent process should exit. """ - parent_pid = 23 - self.mock_func_os_fork.side_effect = iter([parent_pid]) - self.assertRaises( - self.FakeOSExit, - daemon.daemon.detach_process_context) - self.mock_module_os.assert_has_calls([ - mock.call.fork(), - mock.call._exit(0), - ]) - - def test_first_fork_error_raises_error(self): - """ Error on first fork should raise DaemonProcessDetachError. """ - fork_errno = 13 - fork_strerror = "Bad stuff happened" - test_error = OSError(fork_errno, fork_strerror) - test_pids_iter = iter([test_error]) - - def fake_fork(): - next_item = next(test_pids_iter) - if isinstance(next_item, Exception): - raise next_item - else: - return next_item - - self.mock_func_os_fork.side_effect = fake_fork - exc = self.assertRaises( - daemon.daemon.DaemonProcessDetachError, - daemon.daemon.detach_process_context) - self.assertEqual(test_error, exc.__cause__) - self.mock_module_os.assert_has_calls([ - mock.call.fork(), - ]) - - def test_child_starts_new_process_group(self): - """ Child should start new process group. """ - daemon.daemon.detach_process_context() - self.mock_module_os.assert_has_calls([ - mock.call.fork(), - mock.call.setsid(), - ]) - - def test_child_forks_next_parent_exits(self): - """ Child should fork, then exit if parent. """ - fake_pids = [0, 42] - self.mock_func_os_fork.side_effect = iter(fake_pids) - self.assertRaises( - self.FakeOSExit, - daemon.daemon.detach_process_context) - self.mock_module_os.assert_has_calls([ - mock.call.fork(), - mock.call.setsid(), - mock.call.fork(), - mock.call._exit(0), - ]) - - def test_second_fork_error_reports_to_stderr(self): - """ Error on second fork should cause report to stderr. """ - fork_errno = 17 - fork_strerror = "Nasty stuff happened" - test_error = OSError(fork_errno, fork_strerror) - test_pids_iter = iter([0, test_error]) - - def fake_fork(): - next_item = next(test_pids_iter) - if isinstance(next_item, Exception): - raise next_item - else: - return next_item - - self.mock_func_os_fork.side_effect = fake_fork - exc = self.assertRaises( - daemon.daemon.DaemonProcessDetachError, - daemon.daemon.detach_process_context) - self.assertEqual(test_error, exc.__cause__) - self.mock_module_os.assert_has_calls([ - mock.call.fork(), - mock.call.setsid(), - mock.call.fork(), - ]) - - def test_child_forks_next_child_continues(self): - """ Child should fork, then continue if child. """ - daemon.daemon.detach_process_context() - self.mock_module_os.assert_has_calls([ - mock.call.fork(), - mock.call.setsid(), - mock.call.fork(), - ]) - - -@mock.patch("os.getppid", return_value=765) -class is_process_started_by_init_TestCase(scaffold.TestCase): - """ Test cases for is_process_started_by_init function. """ - - def test_returns_false_by_default(self, mock_func_os_getppid): - """ Should return False under normal circumstances. """ - expected_result = False - result = daemon.daemon.is_process_started_by_init() - self.assertIs(result, expected_result) - - def test_returns_true_if_parent_process_is_init( - self, mock_func_os_getppid): - """ Should return True if parent process is `init`. """ - init_pid = 1 - mock_func_os_getppid.return_value = init_pid - expected_result = True - result = daemon.daemon.is_process_started_by_init() - self.assertIs(result, expected_result) - - -class is_socket_TestCase(scaffold.TestCase): - """ Test cases for is_socket function. """ - - def setUp(self): - """ Set up test fixtures. """ - super(is_socket_TestCase, self).setUp() - - def fake_getsockopt(level, optname, buflen=None): - result = object() - if optname is socket.SO_TYPE: - result = socket.SOCK_RAW - return result - - self.fake_socket_getsockopt_func = fake_getsockopt - - self.fake_socket_error = socket.error( - errno.ENOTSOCK, - "Socket operation on non-socket") - - self.mock_socket = mock.MagicMock(spec=socket.socket) - self.mock_socket.getsockopt.side_effect = self.fake_socket_error - - def fake_socket_fromfd(fd, family, type, proto=None): - return self.mock_socket - - func_patcher_socket_fromfd = mock.patch.object( - socket, "fromfd", - side_effect=fake_socket_fromfd) - func_patcher_socket_fromfd.start() - self.addCleanup(func_patcher_socket_fromfd.stop) - - def test_returns_false_by_default(self): - """ Should return False under normal circumstances. """ - test_fd = 23 - expected_result = False - result = daemon.daemon.is_socket(test_fd) - self.assertIs(result, expected_result) - - def test_returns_true_if_stdin_is_socket(self): - """ Should return True if `stdin` is a socket. """ - test_fd = 23 - getsockopt = self.mock_socket.getsockopt - getsockopt.side_effect = self.fake_socket_getsockopt_func - expected_result = True - result = daemon.daemon.is_socket(test_fd) - self.assertIs(result, expected_result) - - def test_returns_false_if_stdin_socket_raises_error(self): - """ Should return True if `stdin` is a socket and raises error. """ - test_fd = 23 - getsockopt = self.mock_socket.getsockopt - getsockopt.side_effect = socket.error( - object(), "Weird socket stuff") - expected_result = True - result = daemon.daemon.is_socket(test_fd) - self.assertIs(result, expected_result) - - -class is_process_started_by_superserver_TestCase(scaffold.TestCase): - """ Test cases for is_process_started_by_superserver function. """ - - def setUp(self): - """ Set up test fixtures. """ - super(is_process_started_by_superserver_TestCase, self).setUp() - - def fake_is_socket(fd): - if sys.__stdin__.fileno() == fd: - result = self.fake_stdin_is_socket_func() - else: - result = False - return result - - self.fake_stdin_is_socket_func = (lambda: False) - - func_patcher_is_socket = mock.patch.object( - daemon.daemon, "is_socket", - side_effect=fake_is_socket) - func_patcher_is_socket.start() - self.addCleanup(func_patcher_is_socket.stop) - - def test_returns_false_by_default(self): - """ Should return False under normal circumstances. """ - expected_result = False - result = daemon.daemon.is_process_started_by_superserver() - self.assertIs(result, expected_result) - - def test_returns_true_if_stdin_is_socket(self): - """ Should return True if `stdin` is a socket. """ - self.fake_stdin_is_socket_func = (lambda: True) - expected_result = True - result = daemon.daemon.is_process_started_by_superserver() - self.assertIs(result, expected_result) - - -@mock.patch.object( - daemon.daemon, "is_process_started_by_superserver", - return_value=False) -@mock.patch.object( - daemon.daemon, "is_process_started_by_init", - return_value=False) -class is_detach_process_context_required_TestCase(scaffold.TestCase): - """ Test cases for is_detach_process_context_required function. """ - - def test_returns_true_by_default( - self, - mock_func_is_process_started_by_init, - mock_func_is_process_started_by_superserver): - """ Should return True under normal circumstances. """ - expected_result = True - result = daemon.daemon.is_detach_process_context_required() - self.assertIs(result, expected_result) - - def test_returns_false_if_started_by_init( - self, - mock_func_is_process_started_by_init, - mock_func_is_process_started_by_superserver): - """ Should return False if current process started by init. """ - mock_func_is_process_started_by_init.return_value = True - expected_result = False - result = daemon.daemon.is_detach_process_context_required() - self.assertIs(result, expected_result) - - def test_returns_true_if_started_by_superserver( - self, - mock_func_is_process_started_by_init, - mock_func_is_process_started_by_superserver): - """ Should return False if current process started by superserver. """ - mock_func_is_process_started_by_superserver.return_value = True - expected_result = False - result = daemon.daemon.is_detach_process_context_required() - self.assertIs(result, expected_result) - - -def setup_streams_fixtures(testcase): - """ Set up common test fixtures for standard streams. """ - testcase.stream_file_paths = dict( - stdin=tempfile.mktemp(), - stdout=tempfile.mktemp(), - stderr=tempfile.mktemp(), - ) - - testcase.stream_files_by_name = dict( - (name, FakeFileDescriptorStringIO()) - for name in ['stdin', 'stdout', 'stderr'] - ) - - testcase.stream_files_by_path = dict( - (testcase.stream_file_paths[name], - testcase.stream_files_by_name[name]) - for name in ['stdin', 'stdout', 'stderr'] - ) - - -@mock.patch.object(os, "dup2") -class redirect_stream_TestCase(scaffold.TestCase): - """ Test cases for redirect_stream function. """ - - def setUp(self): - """ Set up test fixtures. """ - super(redirect_stream_TestCase, self).setUp() - - self.test_system_stream = FakeFileDescriptorStringIO() - self.test_target_stream = FakeFileDescriptorStringIO() - self.test_null_file = FakeFileDescriptorStringIO() - - def fake_os_open(path, flag, mode=None): - if path == os.devnull: - result = self.test_null_file.fileno() - else: - raise FileNotFoundError("No such file", path) - return result - - func_patcher_os_open = mock.patch.object( - os, "open", - side_effect=fake_os_open) - self.mock_func_os_open = func_patcher_os_open.start() - self.addCleanup(func_patcher_os_open.stop) - - def test_duplicates_target_file_descriptor( - self, mock_func_os_dup2): - """ Should duplicate file descriptor from target to system stream. """ - system_stream = self.test_system_stream - system_fileno = system_stream.fileno() - target_stream = self.test_target_stream - target_fileno = target_stream.fileno() - daemon.daemon.redirect_stream(system_stream, target_stream) - mock_func_os_dup2.assert_called_with(target_fileno, system_fileno) - - def test_duplicates_null_file_descriptor_by_default( - self, mock_func_os_dup2): - """ Should by default duplicate the null file to the system stream. """ - system_stream = self.test_system_stream - system_fileno = system_stream.fileno() - target_stream = None - null_path = os.devnull - null_flag = os.O_RDWR - null_file = self.test_null_file - null_fileno = null_file.fileno() - daemon.daemon.redirect_stream(system_stream, target_stream) - self.mock_func_os_open.assert_called_with(null_path, null_flag) - mock_func_os_dup2.assert_called_with(null_fileno, system_fileno) - - -class make_default_signal_map_TestCase(scaffold.TestCase): - """ Test cases for make_default_signal_map function. """ - - def setUp(self): - """ Set up test fixtures. """ - super(make_default_signal_map_TestCase, self).setUp() - - # Use whatever default string type this Python version needs. - signal_module_name = str('signal') - self.fake_signal_module = ModuleType(signal_module_name) - - fake_signal_names = [ - 'SIGHUP', - 'SIGCLD', - 'SIGSEGV', - 'SIGTSTP', - 'SIGTTIN', - 'SIGTTOU', - 'SIGTERM', - ] - for name in fake_signal_names: - setattr(self.fake_signal_module, name, object()) - - module_patcher_signal = mock.patch.object( - daemon.daemon, "signal", new=self.fake_signal_module) - module_patcher_signal.start() - self.addCleanup(module_patcher_signal.stop) - - default_signal_map_by_name = { - 'SIGTSTP': None, - 'SIGTTIN': None, - 'SIGTTOU': None, - 'SIGTERM': 'terminate', - } - self.default_signal_map = dict( - (getattr(self.fake_signal_module, name), target) - for (name, target) in default_signal_map_by_name.items()) - - def test_returns_constructed_signal_map(self): - """ Should return map per default. """ - expected_result = self.default_signal_map - result = daemon.daemon.make_default_signal_map() - self.assertEqual(expected_result, result) - - def test_returns_signal_map_with_only_ids_in_signal_module(self): - """ Should return map with only signals in the `signal` module. - - The `signal` module is documented to only define those - signals which exist on the running system. Therefore the - default map should not contain any signals which are not - defined in the `signal` module. - - """ - del(self.default_signal_map[self.fake_signal_module.SIGTTOU]) - del(self.fake_signal_module.SIGTTOU) - expected_result = self.default_signal_map - result = daemon.daemon.make_default_signal_map() - self.assertEqual(expected_result, result) - - -@mock.patch.object(daemon.daemon.signal, "signal") -class set_signal_handlers_TestCase(scaffold.TestCase): - """ Test cases for set_signal_handlers function. """ - - def setUp(self): - """ Set up test fixtures. """ - super(set_signal_handlers_TestCase, self).setUp() - - self.signal_handler_map = { - signal.SIGQUIT: object(), - signal.SIGSEGV: object(), - signal.SIGINT: object(), - } - - def test_sets_signal_handler_for_each_item(self, mock_func_signal_signal): - """ Should set signal handler for each item in map. """ - signal_handler_map = self.signal_handler_map - expected_calls = [ - mock.call(signal_number, handler) - for (signal_number, handler) in signal_handler_map.items()] - daemon.daemon.set_signal_handlers(signal_handler_map) - self.assertEquals(expected_calls, mock_func_signal_signal.mock_calls) - - -@mock.patch.object(daemon.daemon.atexit, "register") -class register_atexit_function_TestCase(scaffold.TestCase): - """ Test cases for register_atexit_function function. """ - - def test_registers_function_for_atexit_processing( - self, mock_func_atexit_register): - """ Should register specified function for atexit processing. """ - func = object() - daemon.daemon.register_atexit_function(func) - mock_func_atexit_register.assert_called_with(func) - - -# Local variables: -# coding: utf-8 -# mode: python -# End: -# vim: fileencoding=utf-8 filetype=python : |