7d3578827d
Now that we no longer support py27, we can use the standard library unittest.mock module instead of the third party mock lib. Change-Id: Ic18f03681c14131217e3025e2cb0cca46ada84ee Signed-off-by: Sean McGinnis <sean.mcginnis@gmail.com>
178 lines
7.8 KiB
Python
178 lines
7.8 KiB
Python
# Copyright 2010 United States Government as represented by the
|
|
# Administrator of the National Aeronautics and Space Administration.
|
|
# All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
"""Unit Tests for eventlet backdoor."""
|
|
import errno
|
|
import os
|
|
import socket
|
|
from unittest import mock
|
|
|
|
import eventlet
|
|
|
|
from oslo_service import eventlet_backdoor
|
|
from oslo_service.tests import base
|
|
|
|
|
|
class BackdoorSocketPathTest(base.ServiceBaseTestCase):
|
|
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
@mock.patch.object(eventlet, 'listen')
|
|
def test_backdoor_path(self, listen_mock, spawn_mock):
|
|
self.config(backdoor_socket="/tmp/my_special_socket")
|
|
listen_mock.side_effect = mock.Mock()
|
|
path = eventlet_backdoor.initialize_if_enabled(self.conf)
|
|
self.assertEqual("/tmp/my_special_socket", path)
|
|
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
@mock.patch.object(eventlet, 'listen')
|
|
def test_backdoor_path_with_format_string(self, listen_mock, spawn_mock):
|
|
self.config(backdoor_socket="/tmp/my_special_socket-{pid}")
|
|
listen_mock.side_effect = mock.Mock()
|
|
path = eventlet_backdoor.initialize_if_enabled(self.conf)
|
|
expected_path = "/tmp/my_special_socket-{}".format(os.getpid())
|
|
self.assertEqual(expected_path, path)
|
|
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
@mock.patch.object(eventlet, 'listen')
|
|
def test_backdoor_path_with_broken_format_string(self, listen_mock,
|
|
spawn_mock):
|
|
broken_socket_paths = [
|
|
"/tmp/my_special_socket-{}",
|
|
"/tmp/my_special_socket-{broken",
|
|
"/tmp/my_special_socket-{broken}",
|
|
]
|
|
for socket_path in broken_socket_paths:
|
|
self.config(backdoor_socket=socket_path)
|
|
listen_mock.side_effect = mock.Mock()
|
|
path = eventlet_backdoor.initialize_if_enabled(self.conf)
|
|
self.assertEqual(socket_path, path)
|
|
|
|
@mock.patch.object(os, 'unlink')
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
@mock.patch.object(eventlet, 'listen')
|
|
def test_backdoor_path_already_exists(self, listen_mock,
|
|
spawn_mock, unlink_mock):
|
|
self.config(backdoor_socket="/tmp/my_special_socket")
|
|
sock = mock.Mock()
|
|
listen_mock.side_effect = [socket.error(errno.EADDRINUSE, ''), sock]
|
|
path = eventlet_backdoor.initialize_if_enabled(self.conf)
|
|
self.assertEqual("/tmp/my_special_socket", path)
|
|
unlink_mock.assert_called_with("/tmp/my_special_socket")
|
|
|
|
@mock.patch.object(os, 'unlink')
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
@mock.patch.object(eventlet, 'listen')
|
|
def test_backdoor_path_already_exists_and_gone(self, listen_mock,
|
|
spawn_mock, unlink_mock):
|
|
self.config(backdoor_socket="/tmp/my_special_socket")
|
|
sock = mock.Mock()
|
|
listen_mock.side_effect = [socket.error(errno.EADDRINUSE, ''), sock]
|
|
unlink_mock.side_effect = OSError(errno.ENOENT, '')
|
|
path = eventlet_backdoor.initialize_if_enabled(self.conf)
|
|
self.assertEqual("/tmp/my_special_socket", path)
|
|
unlink_mock.assert_called_with("/tmp/my_special_socket")
|
|
|
|
@mock.patch.object(os, 'unlink')
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
@mock.patch.object(eventlet, 'listen')
|
|
def test_backdoor_path_already_exists_and_not_gone(self, listen_mock,
|
|
spawn_mock,
|
|
unlink_mock):
|
|
self.config(backdoor_socket="/tmp/my_special_socket")
|
|
listen_mock.side_effect = socket.error(errno.EADDRINUSE, '')
|
|
unlink_mock.side_effect = OSError(errno.EPERM, '')
|
|
self.assertRaises(OSError, eventlet_backdoor.initialize_if_enabled,
|
|
self.conf)
|
|
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
@mock.patch.object(eventlet, 'listen')
|
|
def test_backdoor_path_no_perms(self, listen_mock, spawn_mock):
|
|
self.config(backdoor_socket="/tmp/my_special_socket")
|
|
listen_mock.side_effect = socket.error(errno.EPERM, '')
|
|
self.assertRaises(socket.error,
|
|
eventlet_backdoor.initialize_if_enabled,
|
|
self.conf)
|
|
|
|
|
|
class BackdoorPortTest(base.ServiceBaseTestCase):
|
|
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
@mock.patch.object(eventlet, 'listen')
|
|
def test_backdoor_port(self, listen_mock, spawn_mock):
|
|
self.config(backdoor_port=1234)
|
|
sock = mock.Mock()
|
|
sock.getsockname.return_value = ('127.0.0.1', 1234)
|
|
listen_mock.return_value = sock
|
|
port = eventlet_backdoor.initialize_if_enabled(self.conf)
|
|
self.assertEqual(1234, port)
|
|
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
@mock.patch.object(eventlet, 'listen')
|
|
def test_backdoor_port_inuse(self, listen_mock, spawn_mock):
|
|
self.config(backdoor_port=2345)
|
|
listen_mock.side_effect = socket.error(errno.EADDRINUSE, '')
|
|
self.assertRaises(socket.error,
|
|
eventlet_backdoor.initialize_if_enabled, self.conf)
|
|
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
def test_backdoor_port_range_inuse(self, spawn_mock):
|
|
self.config(backdoor_port='8800:8801')
|
|
port = eventlet_backdoor.initialize_if_enabled(self.conf)
|
|
self.assertEqual(8800, port)
|
|
port = eventlet_backdoor.initialize_if_enabled(self.conf)
|
|
self.assertEqual(8801, port)
|
|
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
@mock.patch.object(eventlet, 'listen')
|
|
def test_backdoor_port_range(self, listen_mock, spawn_mock):
|
|
self.config(backdoor_port='8800:8899')
|
|
sock = mock.Mock()
|
|
sock.getsockname.return_value = ('127.0.0.1', 8800)
|
|
listen_mock.return_value = sock
|
|
port = eventlet_backdoor.initialize_if_enabled(self.conf)
|
|
self.assertEqual(8800, port)
|
|
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
@mock.patch.object(eventlet, 'listen')
|
|
def test_backdoor_port_range_one_inuse(self, listen_mock, spawn_mock):
|
|
self.config(backdoor_port='8800:8900')
|
|
sock = mock.Mock()
|
|
sock.getsockname.return_value = ('127.0.0.1', 8801)
|
|
listen_mock.side_effect = [socket.error(errno.EADDRINUSE, ''), sock]
|
|
port = eventlet_backdoor.initialize_if_enabled(self.conf)
|
|
self.assertEqual(8801, port)
|
|
|
|
@mock.patch.object(eventlet, 'spawn')
|
|
@mock.patch.object(eventlet, 'listen')
|
|
def test_backdoor_port_range_all_inuse(self, listen_mock, spawn_mock):
|
|
self.config(backdoor_port='8800:8899')
|
|
side_effects = []
|
|
for i in range(8800, 8900):
|
|
side_effects.append(socket.error(errno.EADDRINUSE, ''))
|
|
listen_mock.side_effect = side_effects
|
|
self.assertRaises(socket.error,
|
|
eventlet_backdoor.initialize_if_enabled, self.conf)
|
|
|
|
def test_backdoor_port_reverse_range(self):
|
|
self.config(backdoor_port='8888:7777')
|
|
self.assertRaises(eventlet_backdoor.EventletBackdoorConfigValueError,
|
|
eventlet_backdoor.initialize_if_enabled, self.conf)
|
|
|
|
def test_backdoor_port_bad(self):
|
|
self.config(backdoor_port='abc')
|
|
self.assertRaises(eventlet_backdoor.EventletBackdoorConfigValueError,
|
|
eventlet_backdoor.initialize_if_enabled, self.conf)
|