openstack/neutron
Code Issues Proposed changes

Merge "use test tools from neutron-lib"

Browse Source
This commit is contained in:
Zuul 2019-06-05 22:48:17 +00:00 committed by Gerrit Code Review
commit 6002df86fb
15 changed files with 46 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
neutron/tests/base.py
View File

@ -30,6 +30,7 @@ import mock
from neutron_lib.callbacks import manager as registry_manager
from neutron_lib.db import api as db_api
from neutron_lib import fixture
from neutron_lib.tests import tools as lib_test_tools
from neutron_lib.tests.unit import fake_notifier
from oslo_concurrency.fixture import lockutils
from oslo_config import cfg
@ -246,7 +247,7 @@ class DietTestCase(base.BaseTestCase):
self.addOnException(self.check_for_systemexit)
self.orig_pid = os.getpid()
tools.reset_random_seed()
lib_test_tools.reset_random_seed()
def addOnException(self, handler):

2
neutron/tests/fullstack/base.py
View File

@ -16,6 +16,7 @@ from concurrent import futures
import os
import netaddr
from neutron_lib.tests import tools
from oslo_config import cfg
from oslo_log import log as logging
@ -27,7 +28,6 @@ from neutron.tests.common import helpers
from neutron.tests.common import machine_fixtures
from neutron.tests.common import net_helpers
from neutron.tests.fullstack.resources import client as client_resource
from neutron.tests import tools
from neutron.tests.unit import testlib_api

46
neutron/tests/tools.py
View File

@ -14,17 +14,13 @@
# under the License.
import datetime
import os
import random
import time
import warnings
import fixtures
import mock
import netaddr
from neutron_lib import constants
from neutron_lib.services.logapi import constants as log_const
from neutron_lib.tests import tools
from neutron_lib.utils import helpers
from neutron_lib.utils import net
from oslo_utils import netutils
@ -57,28 +53,6 @@ class WarningsFixture(fixtures.Fixture):
"once", category=wtype, module='^neutron\\.')
class OpenFixture(fixtures.Fixture):
"""Mock access to a specific file while preserving open for others."""
def __init__(self, filepath, contents=''):
self.path = filepath
self.contents = contents
def _setUp(self):
self.mock_open = mock.mock_open(read_data=self.contents)
self._orig_open = open
def replacement_open(name, *args, **kwargs):
if name == self.path:
return self.mock_open(name, *args, **kwargs)
return self._orig_open(name, *args, **kwargs)
self._patch = mock.patch('six.moves.builtins.open',
new=replacement_open)
self._patch.start()
self.addCleanup(self._patch.stop)
class SafeCleanupFixture(fixtures.Fixture):
"""Catch errors in daughter fixture cleanup."""
@ -248,16 +222,6 @@ def get_random_ip_version():
return random.choice(constants.IP_ALLOWED_VERSIONS)
def get_random_EUI():
return netaddr.EUI(
net.get_random_mac(['fe', '16', '3e', '00', '00', '00'])
)
def get_random_ip_network(version=4):
return netaddr.IPNetwork(tools.get_random_cidr(version=version))
def get_random_ip_address(version=4):
if version == 4:
ip_string = '10.%d.%d.%d' % (random.randint(3, 254),
@ -311,16 +275,6 @@ def get_random_network_segment_range_network_type():
constants.TYPE_GENEVE])
def reset_random_seed():
# reset random seed to make sure other processes extracting values from RNG
# don't get the same results (useful especially when you then use the
# random values to allocate system resources from global pool, like ports
# to listen). Use both current time and pid to make sure no tests started
# at the same time get the same values from RNG
seed = time.time() + os.getpid()
random.seed(seed)
def get_random_ipv6_mode():
return random.choice(constants.IPV6_MODES)

4
neutron/tests/unit/agent/l3/test_ha_router.py
View File

@ -17,13 +17,13 @@ import signal
import mock
from neutron_lib import constants as n_consts
from neutron_lib import fixture as lib_fixtures
from oslo_utils import uuidutils
from neutron.agent.l3 import ha_router
from neutron.agent.l3 import router_info
from neutron.tests import base
from neutron.tests.common import l3_test_common
from neutron.tests import tools
_uuid = uuidutils.generate_uuid
@ -131,7 +131,7 @@ class TestBasicRouterOperations(base.BaseTestCase):
ri.keepalived_manager.get_full_config_file_path.return_value = (
'ha_state')
self.mock_open = self.useFixture(
tools.OpenFixture('ha_state', read_return)).mock_open
lib_fixtures.OpenFixture('ha_state', read_return)).mock_open
self.assertEqual(expected, ri.ha_state)
def test_ha_state_master(self):

10
neutron/tests/unit/agent/linux/test_daemon.py
View File

@ -20,11 +20,11 @@ import sys
import mock
from neutron_lib import exceptions
from neutron_lib import fixture as lib_fixtures
import testtools
from neutron.agent.linux import daemon
from neutron.tests import base
from neutron.tests import tools
FAKE_FD = 8
@ -194,7 +194,7 @@ class TestPidfile(base.BaseTestCase):
def test_is_running(self):
mock_open = self.useFixture(
tools.OpenFixture('/proc/34/cmdline', 'python')).mock_open
lib_fixtures.OpenFixture('/proc/34/cmdline', 'python')).mock_open
p = daemon.Pidfile('thefile', 'python')
with mock.patch.object(p, 'read') as read:
@ -205,7 +205,8 @@ class TestPidfile(base.BaseTestCase):
def test_is_running_uuid_true(self):
mock_open = self.useFixture(
tools.OpenFixture('/proc/34/cmdline', 'python 1234')).mock_open
lib_fixtures.OpenFixture(
'/proc/34/cmdline', 'python 1234')).mock_open
p = daemon.Pidfile('thefile', 'python', uuid='1234')
with mock.patch.object(p, 'read') as read:
@ -216,7 +217,8 @@ class TestPidfile(base.BaseTestCase):
def test_is_running_uuid_false(self):
mock_open = self.useFixture(
tools.OpenFixture('/proc/34/cmdline', 'python 1234')).mock_open
lib_fixtures.OpenFixture(
'/proc/34/cmdline', 'python 1234')).mock_open
p = daemon.Pidfile('thefile', 'python', uuid='6789')
with mock.patch.object(p, 'read') as read:

16
neutron/tests/unit/agent/linux/test_dhcp.py
View File

@ -20,6 +20,7 @@ import netaddr
from neutron_lib.api.definitions import extra_dhcp_opt as edo_ext
from neutron_lib import constants
from neutron_lib import exceptions
from neutron_lib import fixture as lib_fixtures
from oslo_config import cfg
import oslo_messaging
from oslo_utils import fileutils
@ -31,7 +32,6 @@ from neutron.conf.agent import common as config
from neutron.conf.agent import dhcp as dhcp_config
from neutron.conf import common as base_config
from neutron.tests import base
from neutron.tests import tools
class FakeIPAllocation(object):
@ -1226,7 +1226,7 @@ class TestDhcpLocalProcess(TestBase):
def test_get_interface_name(self):
net = FakeDualNetwork()
path = '/dhcp/%s/interface' % net.id
self.useFixture(tools.OpenFixture(path, 'tap0'))
self.useFixture(lib_fixtures.OpenFixture(path, 'tap0'))
lp = LocalChild(self.conf, net)
self.assertEqual(lp.interface_name, 'tap0')
@ -2005,7 +2005,7 @@ class TestDnsmasq(TestBase):
def test_reload_allocations_no_interface(self):
net = FakeDualNetwork()
ipath = '/dhcp/%s/interface' % net.id
self.useFixture(tools.OpenFixture(ipath))
self.useFixture(lib_fixtures.OpenFixture(ipath))
test_pm = mock.Mock()
dm = self._get_dnsmasq(net, test_pm)
dm.reload_allocations()
@ -2019,8 +2019,8 @@ class TestDnsmasq(TestBase):
net = FakeDualNetwork()
hpath = '/dhcp/%s/host' % net.id
ipath = '/dhcp/%s/interface' % net.id
self.useFixture(tools.OpenFixture(hpath))
self.useFixture(tools.OpenFixture(ipath, 'tapdancingmice'))
self.useFixture(lib_fixtures.OpenFixture(hpath))
self.useFixture(lib_fixtures.OpenFixture(ipath, 'tapdancingmice'))
test_pm = mock.Mock()
dm = self._get_dnsmasq(net, test_pm)
dm.reload_allocations()
@ -2437,7 +2437,7 @@ class TestDnsmasq(TestBase):
lines = ["00:00:80:aa:bb:cc,inst-name,192.168.0.1",
"00:00:80:aa:bb:cc,inst-name,[fdca:3ba5:a17a::1]"]
mock_open = self.useFixture(
tools.OpenFixture(filename, '\n'.join(lines))).mock_open
lib_fixtures.OpenFixture(filename, '\n'.join(lines))).mock_open
dnsmasq = self._get_dnsmasq(FakeDualNetwork())
leases = dnsmasq._read_hosts_file_leases(filename)
@ -2452,7 +2452,7 @@ class TestDnsmasq(TestBase):
"00:00:80:aa:bb:cc,id:client2,inst-name,"
"[fdca:3ba5:a17a::1]"]
mock_open = self.useFixture(
tools.OpenFixture(filename, '\n'.join(lines))).mock_open
lib_fixtures.OpenFixture(filename, '\n'.join(lines))).mock_open
dnsmasq = self._get_dnsmasq(FakeDualNetwork())
leases = dnsmasq._read_hosts_file_leases(filename)
@ -2497,7 +2497,7 @@ class TestDnsmasq(TestBase):
lines.append(bad_line)
mock_open = self.useFixture(
tools.OpenFixture(filename, '\n'.join(lines))).mock_open
lib_fixtures.OpenFixture(filename, '\n'.join(lines))).mock_open
dnsmasq = self._get_dnsmasq(FakeDualNetwork())
with mock.patch('os.path.exists', return_value=True), \

6
neutron/tests/unit/agent/linux/test_external_process.py
View File

@ -15,13 +15,13 @@
import os.path
import mock
from neutron_lib import fixture as lib_fixtures
from oslo_config import cfg
from oslo_utils import fileutils
import psutil
from neutron.agent.linux import external_process as ep
from neutron.tests import base
from neutron.tests import tools
TEST_UUID = 'test-uuid'
@ -293,12 +293,12 @@ class TestProcessManager(base.BaseTestCase):
self.assertEqual(retval, '/var/path/uuid.pid')
def test_pid(self):
self.useFixture(tools.OpenFixture('/var/path/uuid.pid', '5'))
self.useFixture(lib_fixtures.OpenFixture('/var/path/uuid.pid', '5'))
manager = ep.ProcessManager(self.conf, 'uuid')
self.assertEqual(manager.pid, 5)
def test_pid_no_an_int(self):
self.useFixture(tools.OpenFixture('/var/path/uuid.pid', 'foo'))
self.useFixture(lib_fixtures.OpenFixture('/var/path/uuid.pid', 'foo'))
manager = ep.ProcessManager(self.conf, 'uuid')
self.assertIsNone(manager.pid)

6
neutron/tests/unit/agent/linux/test_utils.py
View File

@ -21,13 +21,13 @@ import six
import testtools
from neutron_lib import exceptions
from neutron_lib import fixture as lib_fixtures
from oslo_config import cfg
import oslo_i18n
from neutron.agent.linux import utils
from neutron.tests import base
from neutron.tests.common import helpers
from neutron.tests import tools
_marker = object()
@ -306,7 +306,7 @@ class TestGetCmdlineFromPid(base.BaseTestCase):
def _test_cmdline(self, process, expected_cmd):
self.process_is_running_mock.return_value = True
mock_open = self.useFixture(
tools.OpenFixture('/proc/%s/cmdline' % self.pid, process)
lib_fixtures.OpenFixture('/proc/%s/cmdline' % self.pid, process)
).mock_open
cmdline = utils.get_cmdline_from_pid(self.pid)
mock_open.assert_called_once_with('/proc/%s/cmdline' % self.pid, 'r')
@ -330,7 +330,7 @@ class TestGetCmdlineFromPid(base.BaseTestCase):
def test_no_process_running(self):
self.process_is_running_mock.return_value = False
mock_open = self.useFixture(
tools.OpenFixture('/proc/%s/cmdline' % self.pid)
lib_fixtures.OpenFixture('/proc/%s/cmdline' % self.pid)
).mock_open
cmdline = utils.get_cmdline_from_pid(self.pid)
mock_open.assert_not_called()

4
neutron/tests/unit/agent/metadata/test_driver.py
View File

@ -17,6 +17,7 @@ import os
import mock
from neutron_lib import constants
from neutron_lib import fixture as lib_fixtures
from oslo_config import cfg
from oslo_utils import uuidutils
@ -29,7 +30,6 @@ from neutron.conf.agent.l3 import config as l3_config
from neutron.conf.agent.l3 import ha as ha_conf
from neutron.conf.agent.metadata import config as meta_conf
from neutron.tests import base
from neutron.tests import tools
from neutron.tests.unit.agent.linux import test_utils
_uuid = uuidutils.generate_uuid
@ -134,7 +134,7 @@ class TestMetadataDriverProcess(base.BaseTestCase):
agent.conf.state_path),
"%s.conf" % router_id)
mock_open = self.useFixture(
tools.OpenFixture(cfg_file)).mock_open
lib_fixtures.OpenFixture(cfg_file)).mock_open
agent.metadata_driver.spawn_monitored_metadata_proxy(
agent.process_monitor,
router_ns,

12
neutron/tests/unit/common/test_ipv6_utils.py
View File

@ -17,10 +17,10 @@ import collections
import mock
from neutron_lib import constants
from neutron_lib import fixture as lib_fixtures
from neutron.common import ipv6_utils
from neutron.tests import base
from neutron.tests import tools
class TestIsEnabledAndBindByDefault(base.BaseTestCase):
@ -37,24 +37,26 @@ class TestIsEnabledAndBindByDefault(base.BaseTestCase):
self.proc_path = '/proc/sys/net/ipv6/conf/default/disable_ipv6'
def test_enabled(self):
self.useFixture(tools.OpenFixture(self.proc_path, '0'))
self.useFixture(lib_fixtures.OpenFixture(self.proc_path, '0'))
enabled = ipv6_utils.is_enabled_and_bind_by_default()
self.assertTrue(enabled)
def test_disabled(self):
self.useFixture(tools.OpenFixture(self.proc_path, '1'))
self.useFixture(lib_fixtures.OpenFixture(self.proc_path, '1'))
enabled = ipv6_utils.is_enabled_and_bind_by_default()
self.assertFalse(enabled)
def test_disabled_non_exists(self):
mo = self.useFixture(tools.OpenFixture(self.proc_path, '1')).mock_open
mo = self.useFixture(
lib_fixtures.OpenFixture(self.proc_path, '1')).mock_open
self.mock_exists.return_value = False
enabled = ipv6_utils.is_enabled_and_bind_by_default()
self.assertFalse(enabled)
self.assertFalse(mo.called)
def test_memoize(self):
mo = self.useFixture(tools.OpenFixture(self.proc_path, '0')).mock_open
mo = self.useFixture(
lib_fixtures.OpenFixture(self.proc_path, '0')).mock_open
ipv6_utils.is_enabled_and_bind_by_default()
enabled = ipv6_utils.is_enabled_and_bind_by_default()
self.assertTrue(enabled)

2
neutron/tests/unit/db/test_dvr_mac_db.py
View File

@ -24,11 +24,11 @@ from neutron_lib import exceptions as lib_exc
from neutron_lib.exceptions import dvr as dvr_exc
from neutron_lib import fixture
from neutron_lib.plugins import directory
from neutron_lib.tests import tools
from neutron_lib.utils import net
from neutron.db import dvr_mac_db
from neutron.objects import router
from neutron.tests import tools
from neutron.tests.unit.plugins.ml2 import test_plugin

9
neutron/tests/unit/db/test_migration.py
View File

@ -25,6 +25,7 @@ from alembic.operations import ops as alembic_ops
from alembic import script as alembic_script
import fixtures
import mock
from neutron_lib import fixture as lib_fixtures
from neutron_lib.utils import helpers
from oslo_utils import fileutils
import pkg_resources
@ -409,10 +410,10 @@ class TestCli(base.BaseTestCase):
heads[1]: FakeRevision(labels=cli.EXPAND_BRANCH)}
fc.return_value.get_revision.side_effect = revs.__getitem__
mock_open_con = self.useFixture(
tools.OpenFixture(cli._get_contract_head_file_path(
lib_fixtures.OpenFixture(cli._get_contract_head_file_path(
fake_config), contract_head + '\n')).mock_open
mock_open_ex = self.useFixture(
tools.OpenFixture(cli._get_expand_head_file_path(
lib_fixtures.OpenFixture(cli._get_expand_head_file_path(
fake_config), expand_head + '\n')).mock_open
if contract_head in heads and expand_head in heads:
@ -452,10 +453,10 @@ class TestCli(base.BaseTestCase):
def test_update_head_files_success(self, *mocks):
heads = ['a', 'b']
mock_open_con = self.useFixture(
tools.OpenFixture(cli._get_contract_head_file_path(
lib_fixtures.OpenFixture(cli._get_contract_head_file_path(
self.configs[0]))).mock_open
mock_open_ex = self.useFixture(
tools.OpenFixture(cli._get_expand_head_file_path(
lib_fixtures.OpenFixture(cli._get_expand_head_file_path(
self.configs[0]))).mock_open
with mock.patch('alembic.script.ScriptDirectory.from_config') as fc:
fc.return_value.get_heads.return_value = heads

7
neutron/tests/unit/objects/test_base.py
View File

@ -25,6 +25,7 @@ from neutron_lib.db import model_query
from neutron_lib import exceptions as n_exc
from neutron_lib.objects import exceptions as o_exc
from neutron_lib.objects import utils as obj_utils
from neutron_lib.tests import tools as lib_test_tools
from neutron_lib.utils import helpers
from oslo_db import exception as obj_exc
from oslo_db.sqlalchemy import utils as db_utils
@ -452,7 +453,7 @@ def get_random_dscp_mark():
def get_list_of_random_networks(num=10):
for i in range(5):
res = [tools.get_random_ip_network() for i in range(num)]
res = [lib_test_tools.get_random_ip_network() for i in range(num)]
# make sure there are no duplicates
if len(set(res)) == num:
return res
@ -511,13 +512,13 @@ FIELD_TYPE_VALUE_GENERATOR_MAP = {
common_types.FlowDirectionEnumField: tools.get_random_flow_direction,
common_types.HARouterEnumField: tools.get_random_ha_states,
common_types.IpamAllocationStatusEnumField: tools.get_random_ipam_status,
common_types.IPNetworkField: tools.get_random_ip_network,
common_types.IPNetworkField: lib_test_tools.get_random_ip_network,
common_types.IPNetworkPrefixLenField: tools.get_random_prefixlen,
common_types.IPV6ModeEnumField: tools.get_random_ipv6_mode,
common_types.IPVersionEnumField: tools.get_random_ip_version,
common_types.IpProtocolEnumField: tools.get_random_ip_protocol,
common_types.ListOfIPNetworksField: get_list_of_random_networks,
common_types.MACAddressField: tools.get_random_EUI,
common_types.MACAddressField: lib_test_tools.get_random_EUI,
common_types.NetworkSegmentRangeNetworkTypeEnumField:
tools.get_random_network_segment_range_network_type,
common_types.PortBindingStatusEnumField:

2
neutron/tests/unit/objects/test_common_types.py
View File

@ -16,12 +16,12 @@ import itertools
from neutron_lib import constants as const
from neutron_lib.db import constants as db_const
from neutron_lib.tests import tools
from neutron_lib.utils import net
from oslo_serialization import jsonutils
from neutron.objects import common_types
from neutron.tests import base as test_base
from neutron.tests import tools
class TestField(object):

2
neutron/tests/unit/objects/test_ports.py
View File

@ -12,6 +12,7 @@
import mock
from neutron_lib import constants
from neutron_lib.tests import tools
from oslo_utils import uuidutils
import testscenarios
@ -20,7 +21,6 @@ from neutron.objects import network
from neutron.objects import ports
from neutron.objects.qos import binding
from neutron.objects.qos import policy
from neutron.tests import tools
from neutron.tests.unit.objects import test_base as obj_test_base
from neutron.tests.unit import testlib_api