我们从Python开源项目中，提取了以下49个代码示例，用于说明如何使用fixtures.MonkeyPatch()。
def setUp(self): """Run before each test method to initialize test environment.""" super(TestCase, self).setUp() test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0) try: test_timeout = int(test_timeout) except ValueError: # If timeout value is invalid do not set a timeout. test_timeout = 0 if test_timeout > 0: self.useFixture(fixtures.Timeout(test_timeout, gentle=True)) self.useFixture(fixtures.NestedTempfile()) self.useFixture(fixtures.TempHomeDir()) environ_enabled = (lambda var_name: strutils.bool_from_string(os.environ.get(var_name))) if environ_enabled('OS_STDOUT_CAPTURE'): stdout = self.useFixture(fixtures.StringStream('stdout')).stream self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout)) if environ_enabled('OS_STDERR_CAPTURE'): stderr = self.useFixture(fixtures.StringStream('stderr')).stream self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr)) self.start = timeutils.utcnow()
def setUp(self): """Run before each test method to initialize test environment.""" super(TestCase, self).setUp() test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0) try: test_timeout = int(test_timeout) except ValueError: # If timeout value is invalid do not set a timeout. test_timeout = 0 if test_timeout > 0: self.useFixture(fixtures.Timeout(test_timeout, gentle=True)) self.useFixture(fixtures.NestedTempfile()) self.useFixture(fixtures.TempHomeDir()) if os.environ.get('OS_STDOUT_CAPTURE') in _TRUE_VALUES: stdout = self.useFixture(fixtures.StringStream('stdout')).stream self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout)) if os.environ.get('OS_STDERR_CAPTURE') in _TRUE_VALUES: stderr = self.useFixture(fixtures.StringStream('stderr')).stream self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr)) self.log_fixture = self.useFixture(fixtures.FakeLogger())
def test_get_auth_provider_keystone_v3(self): # check if method returns KeystoneV3AuthProvider # make isinstance return True mockIsInstance = mock.Mock(return_value=True) self.useFixture(MonkeyPatch('config_tempest.config_tempest.isinstance', mockIsInstance)) mock_function = mock.Mock() # mock V3Provider, if other provider is called, it fails func2mock = 'config_tempest.config_tempest.auth.KeystoneV3AuthProvider' self.useFixture(MonkeyPatch(func2mock, mock_function)) resp = self.client.get_auth_provider(self.conf, "") self.assertEqual(resp, mock_function()) # check parameters of returned function self.client.get_auth_provider(self.conf, "") mock_function.assert_called_with('', 'http://172.16.52.151:5000/v3', 'true', None)
def test_init_manager_as_admin_using_new_auth(self): self.conf = self._get_alt_conf("v2.0", "v3") self.client = self._get_clients(self.conf) mock_function = mock.Mock(return_value={"id": "my_fake_id"}) func2mock = ('config_tempest.config_tempest.ProjectsClient' '.get_project_by_name') self.useFixture(MonkeyPatch(func2mock, mock_function)) self._get_clients(self.conf, admin=True) # check if admin credentials were set admin_tenant = self.conf.get("auth", "admin_project_name") admin_password = self.conf.get("auth", "admin_password") self.assertEqual(self.conf.get("auth", "admin_username"), "admin") self.assertEqual(admin_tenant, "adminTenant") self.assertEqual(admin_password, "adminPass") # check if admin tenant id was set admin_tenant_id = self.conf.get("identity", "admin_tenant_id") self.assertEqual(admin_tenant_id, "my_fake_id") use_dynamic_creds_bool = self.conf.get("auth", "use_dynamic_credentials") self.assertEqual(use_dynamic_creds_bool, "True")
def _setUp(self): def get_local_auth(token): return FakeSession(token, self.local_auths[token]) def get_sp_auth(sp, token, project): return FakeSession(self.sp_auths[(sp, token, project)], project) def get_projects_at_sp(sp, token): if sp in self.sp_projects: return self.sp_projects[sp] else: return [] self.local_auths = {} self.sp_auths = {} self.sp_projects = {} self.useFixture(fixtures.MonkeyPatch( 'mixmatch.auth.get_sp_auth', get_sp_auth)) self.useFixture(fixtures.MonkeyPatch( 'mixmatch.auth.get_local_auth', get_local_auth)) self.useFixture(fixtures.MonkeyPatch( 'mixmatch.auth.get_projects_at_sp', get_projects_at_sp))
def setUp(self): super(ShellCommandTest, self).setUp() def get_auth_endpoint(bound_self, args): return ('test', {}) self.useFixture(fixtures.MonkeyPatch( 'karborclient.shell.KarborShell._get_endpoint_and_kwargs', get_auth_endpoint)) self.client = mock.MagicMock() # To prevent log descriptors from being closed during # shell tests set a custom StreamHandler self.logger = log.getLogger(None).logger self.logger.level = logging.DEBUG self.color_handler = handlers.ColorHandler(sys.stdout) self.logger.addHandler(self.color_handler)
def setUp(self): super(GetSocketTestCase, self).setUp() self.useFixture(fixtures.MonkeyPatch( "glare.common.wsgi.get_bind_addr", lambda x: ('192.168.0.13', 1234))) addr_info_list = [(2, 1, 6, '', ('192.168.0.13', 80)), (2, 2, 17, '', ('192.168.0.13', 80)), (2, 3, 0, '', ('192.168.0.13', 80))] self.useFixture(fixtures.MonkeyPatch( "glare.common.wsgi.socket.getaddrinfo", lambda *x: addr_info_list)) self.useFixture(fixtures.MonkeyPatch( "glare.common.wsgi.time.time", mock.Mock(side_effect=[0, 1, 5, 10, 20, 35]))) self.useFixture(fixtures.MonkeyPatch( "glare.common.wsgi.utils.validate_key_cert", lambda *x: None)) wsgi.CONF.cert_file = '/etc/ssl/cert' wsgi.CONF.key_file = '/etc/ssl/key' wsgi.CONF.ca_file = '/etc/ssl/ca_cert' wsgi.CONF.tcp_keepidle = 600
def test_correct_configure_socket(self): mock_socket = mock.Mock() self.useFixture(fixtures.MonkeyPatch( 'glare.common.wsgi.ssl.wrap_socket', mock_socket)) self.useFixture(fixtures.MonkeyPatch( 'glare.common.wsgi.eventlet.listen', lambda *x, **y: mock_socket)) server = wsgi.Server() server.default_port = 1234 server.configure_socket() self.assertIn(mock.call.setsockopt( socket.SOL_SOCKET, socket.SO_REUSEADDR, 1), mock_socket.mock_calls) self.assertIn(mock.call.setsockopt( socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1), mock_socket.mock_calls) if hasattr(socket, 'TCP_KEEPIDLE'): self.assertIn(mock.call().setsockopt( socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, wsgi.CONF.tcp_keepidle), mock_socket.mock_calls)
def setUp(self): super(RealTimeServersTest, self).setUp() # Replace libvirt with fakelibvirt self.useFixture(fixtures.MonkeyPatch( 'nova.virt.libvirt.driver.libvirt_utils', fake_libvirt_utils)) self.useFixture(fixtures.MonkeyPatch( 'nova.virt.libvirt.driver.libvirt', fakelibvirt)) self.useFixture(fixtures.MonkeyPatch( 'nova.virt.libvirt.host.libvirt', fakelibvirt)) self.useFixture(fixtures.MonkeyPatch( 'nova.virt.libvirt.guest.libvirt', fakelibvirt)) self.useFixture(fakelibvirt.FakeLibvirtFixture()) self.flags(sysinfo_serial='none', group='libvirt')
def setUp(self): super(NUMAServersTest, self).setUp() # Replace libvirt with fakelibvirt self.useFixture(fixtures.MonkeyPatch( 'nova.virt.libvirt.driver.libvirt_utils', fake_libvirt_utils)) self.useFixture(fixtures.MonkeyPatch( 'nova.virt.libvirt.driver.libvirt', fakelibvirt)) self.useFixture(fixtures.MonkeyPatch( 'nova.virt.libvirt.host.libvirt', fakelibvirt)) self.useFixture(fixtures.MonkeyPatch( 'nova.virt.libvirt.guest.libvirt', fakelibvirt)) self.useFixture(fakelibvirt.FakeLibvirtFixture())
def test_ipv6_supported(self): self.assertIn(test_utils.is_ipv6_supported(), (False, True)) def fake_open(path): raise IOError def fake_socket_fail(x, y): e = socket.error() e.errno = errno.EAFNOSUPPORT raise e def fake_socket_ok(x, y): return tempfile.TemporaryFile() with fixtures.MonkeyPatch('socket.socket', fake_socket_fail): self.assertFalse(test_utils.is_ipv6_supported()) with fixtures.MonkeyPatch('socket.socket', fake_socket_ok): with fixtures.MonkeyPatch('sys.platform', 'windows'): self.assertTrue(test_utils.is_ipv6_supported()) with fixtures.MonkeyPatch('sys.platform', 'linux2'): with fixtures.MonkeyPatch('six.moves.builtins.open', fake_open): self.assertFalse(test_utils.is_ipv6_supported())
def setUp(self): super(LdapDNSTestCase, self).setUp() self.useFixture(fixtures.MonkeyPatch( 'nova.network.ldapdns.ldap', fake_ldap)) dns_class = 'nova.network.ldapdns.LdapDNS' self.driver = importutils.import_object(dns_class) attrs = {'objectClass': ['domainrelatedobject', 'dnsdomain', 'domain', 'dcobject', 'top'], 'associateddomain': ['root'], 'dc': ['root']} self.driver.lobj.add_s("ou=hosts,dc=example,dc=org", attrs.items()) self.driver.create_domain(domain1) self.driver.create_domain(domain2)
def setUp(self): super(_BaseSnapshotTests, self).setUp() self.flags(snapshots_directory='./', group='libvirt') self.context = context.get_admin_context() self.useFixture(fixtures.MonkeyPatch( 'nova.virt.libvirt.driver.libvirt_utils', fake_libvirt_utils)) self.useFixture(fixtures.MonkeyPatch( 'nova.virt.libvirt.imagebackend.libvirt_utils', fake_libvirt_utils)) self.image_service = nova.tests.unit.image.fake.stub_out_image_service( self) self.mock_update_task_state = mock.Mock() test_instance = _create_test_instance() self.instance_ref = objects.Instance(**test_instance) self.instance_ref.info_cache = objects.InstanceInfoCache( network_info=None)
def test_get_dev(self): tempdir = self.useFixture(fixtures.TempDir()).path l = loop.LoopMount(self.file, tempdir) self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', _fake_trycmd_losetup_works)) self.useFixture(fixtures.MonkeyPatch('nova.utils.execute', _fake_noop)) # No error logged, device consumed self.assertTrue(l.get_dev()) self.assertTrue(l.linked) self.assertEqual('', l.error) self.assertEqual('/dev/loop0', l.device) # Free l.unget_dev() self.assertFalse(l.linked) self.assertEqual('', l.error) self.assertIsNone(l.device)
def test_inner_get_dev_fails(self): tempdir = self.useFixture(fixtures.TempDir()).path l = loop.LoopMount(self.file, tempdir) self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', _fake_trycmd_losetup_fails)) # No error logged, device consumed self.assertFalse(l._inner_get_dev()) self.assertFalse(l.linked) self.assertNotEqual('', l.error) self.assertIsNone(l.device) # Free l.unget_dev() self.assertFalse(l.linked) self.assertIsNone(l.device)
def test_nbd_not_loaded(self): tempdir = self.useFixture(fixtures.TempDir()).path n = nbd.NbdMount(self.file, tempdir) # Fake out os.path.exists def fake_exists(path): if path.startswith('/sys/block/nbd'): return False return ORIG_EXISTS(path) self.useFixture(fixtures.MonkeyPatch('os.path.exists', fake_exists)) # This should fail, as we don't have the module "loaded" # TODO(mikal): work out how to force english as the gettext language # so that the error check always passes self.assertIsNone(n._allocate_nbd()) self.assertEqual('nbd unavailable: module not loaded', n.error)
def test_inner_get_dev_works(self): tempdir = self.useFixture(fixtures.TempDir()).path n = nbd.NbdMount(self.file, tempdir) self.useFixture(fixtures.MonkeyPatch('random.shuffle', _fake_noop)) self.useFixture(fixtures.MonkeyPatch('os.path.exists', self.fake_exists_one)) self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', self.fake_trycmd_creates_pid)) self.useFixture(fixtures.MonkeyPatch('nova.utils.execute', _fake_noop)) # No error logged, device consumed self.assertTrue(n._inner_get_dev()) self.assertTrue(n.linked) self.assertEqual('', n.error) self.assertEqual('/dev/nbd0', n.device) # Free n.unget_dev() self.assertFalse(n.linked) self.assertEqual('', n.error) self.assertIsNone(n.device)
def test_get_dev_timeout(self): # Always fail to get a device def fake_get_dev_fails(self): return False self.stubs.Set(nbd.NbdMount, '_inner_get_dev', fake_get_dev_fails) tempdir = self.useFixture(fixtures.TempDir()).path n = nbd.NbdMount(self.file, tempdir) self.useFixture(fixtures.MonkeyPatch('random.shuffle', _fake_noop)) self.useFixture(fixtures.MonkeyPatch('time.sleep', _fake_noop)) self.useFixture(fixtures.MonkeyPatch('nova.utils.execute', _fake_noop)) self.useFixture(fixtures.MonkeyPatch('os.path.exists', self.fake_exists_one)) self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', self.fake_trycmd_creates_pid)) self.useFixture(fixtures.MonkeyPatch(('nova.virt.disk.mount.api.' 'MAX_DEVICE_WAIT'), -10)) # No error logged, device consumed self.assertFalse(n.get_dev())
def test_do_mount_need_to_specify_fs_type(self): # NOTE(mikal): Bug 1094373 saw a regression where we failed to # communicate a failed mount properly. def fake_trycmd(*args, **kwargs): return '', 'broken' self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', fake_trycmd)) imgfile = tempfile.NamedTemporaryFile() self.addCleanup(imgfile.close) tempdir = self.useFixture(fixtures.TempDir()).path mount = nbd.NbdMount(imgfile.name, tempdir) def fake_returns_true(*args, **kwargs): return True mount.get_dev = fake_returns_true mount.map_dev = fake_returns_true self.assertFalse(mount.do_mount())
def setUp(self): super(ServersControllerTriggerCrashDumpTest, self).setUp() self.instance = fakes.stub_instance_obj(None, vm_state=vm_states.ACTIVE) def fake_get(ctrl, ctxt, uuid): if uuid != FAKE_UUID: raise webob.exc.HTTPNotFound(explanation='fakeout') return self.instance self.useFixture( fixtures.MonkeyPatch('nova.api.openstack.compute.servers.' 'ServersController._get_instance', fake_get)) self.req = fakes.HTTPRequest.blank('/servers/%s/action' % FAKE_UUID) self.req.api_version_request =\ api_version_request.APIVersionRequest('2.17') self.body = dict(trigger_crash_dump=None)
def _test_archive_deleted_rows(self, mock_db_archive, verbose=False): self.useFixture(fixtures.MonkeyPatch('sys.stdout', StringIO())) self.commands.archive_deleted_rows(20, verbose=verbose) mock_db_archive.assert_called_once_with(20) output = sys.stdout.getvalue() if verbose: expected = '''\ +-----------+-------------------------+ | Table | Number of Rows Archived | +-----------+-------------------------+ | consoles | 5 | | instances | 10 | +-----------+-------------------------+ ''' self.assertEqual(expected, output) else: self.assertEqual(0, len(output))
def setUp(self): super(PoisonFunctions, self).setUp() # The nova libvirt driver starts an event thread which only # causes trouble in tests. Make sure that if tests don't # properly patch it the test explodes. # explicit import because MonkeyPatch doesn't magic import # correctly if we are patching a method on a class in a # module. import nova.virt.libvirt.host # noqa def evloop(*args, **kwargs): import sys warnings.warn("Forgot to disable libvirt event thread") sys.exit(1) self.useFixture(fixtures.MonkeyPatch( 'nova.virt.libvirt.host.Host._init_events', evloop))
def setUp(self): super(TestCase, self).setUp() stdout = self.useFixture(fixtures.StringStream('stdout')).stream self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout)) stderr = self.useFixture(fixtures.StringStream('stderr')).stream self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))
def stub_out(self, old, new): """Replace a function for the duration of the test. Use the monkey patch fixture to replace a function for the duration of a test. Useful when you want to provide fake methods instead of mocks during testing. This should be used instead of self.stubs.Set (which is based on mox) going forward. """ self.useFixture(fixtures.MonkeyPatch(old, new))
def stub_out(self, old, new): """Replace a function for the duration of the test. Use the monkey patch fixture to replace a function for the duration of a test. Useful when you want to provide fake methods instead of mocks during testing. This should be used instead of self.stubs.Set (which is based on mox) going forward. """ self.useFixture(fixtures.MonkeyPatch(old, new)) # Useful assertions
def setUp(self): super(TestQuery, self).setUp() self.useFixture(fixtures.MonkeyPatch( 'pecan.response', mock.MagicMock())) self.useFixture(fixtures.MockPatch('panko.api.controllers.v2.events' '._build_rbac_query_filters', return_value={'t_filter': [], 'admin_proj': None}))
def captured_output(streamname): stream = StringIO() patch = fixtures.MonkeyPatch('sys.%s' % streamname, stream) with patch: yield stream
def setUp(self): super(BuildSphinxTest, self).setUp() self.useFixture(fixtures.MonkeyPatch( "sphinx.setup_command.BuildDoc.run", lambda self: None)) from distutils import dist self.distr = dist.Distribution() self.distr.packages = ("fake_package",) self.distr.command_options["build_sphinx"] = { "source_dir": ["a", "."]} pkg_fixture = fixtures.PythonPackage( "fake_package", [("fake_module.py", b""), ("another_fake_module_for_testing.py", b""), ("fake_private_module.py", b"")]) self.useFixture(pkg_fixture) self.useFixture(base.DiveDir(pkg_fixture.base)) self.distr.command_options["pbr"] = {} if hasattr(self, "excludes"): self.distr.command_options["pbr"]["autodoc_exclude_modules"] = ( 'setup.cfg', "fake_package.fake_private_module\n" "fake_package.another_fake_*\n" "fake_package.unknown_module") if self.has_opt: options = self.distr.command_options["pbr"] options["autodoc_index_modules"] = ('setup.cfg', self.autodoc)
def make_env(self, exclude=None): env = dict((k, v) for k, v in self.auth_env.items() if k != exclude) self.useFixture(fixtures.MonkeyPatch('os.environ', env))
def setUp(self): testtools.TestCase.setUp(self) if (os.environ.get("OS_STDOUT_CAPTURE") == "True" or os.environ.get("OS_STDOUT_CAPTURE") == "1"): stdout = self.useFixture(fixtures.StringStream("stdout")).stream self.useFixture(fixtures.MonkeyPatch("sys.stdout", stdout)) if (os.environ.get("OS_STDERR_CAPTURE") == "True" or os.environ.get("OS_STDERR_CAPTURE") == "1"): stderr = self.useFixture(fixtures.StringStream("stderr")).stream self.useFixture(fixtures.MonkeyPatch("sys.stderr", stderr))
def _test_discover(self, url, data, function2mock, mock_ret_val, expected_resp): provider = self.FakeAuthProvider(url, data) mocked_function = Mock() mocked_function.return_value = mock_ret_val self.useFixture(MonkeyPatch(function2mock, mocked_function)) resp = api.discover(provider, "RegionOne") self.assertEqual(resp, expected_resp)
def test_get_identity_v3_extensions(self): expected_resp = ['OS-INHERIT', 'OS-OAUTH1', 'OS-SIMPLE-CERT', 'OS-EP-FILTER'] fake_resp = self.FakeRequestResponse() mocked_requests = Mock() mocked_requests.return_value = fake_resp self.useFixture(MonkeyPatch('requests.get', mocked_requests)) resp = api.get_identity_v3_extensions(self.FAKE_URL + "v3") self.assertItemsEqual(resp, expected_resp)
def _mock_list_images(self, return_value, image_name, expected_resp): mock_function = mock.Mock(return_value=return_value) func2mock = self.CLIENT_MOCK + '.list_images' self.useFixture(MonkeyPatch(func2mock, mock_function)) resp = tool._find_image(client=self.client, image_id=None, image_name=image_name) self.assertEqual(resp, expected_resp)
def test_find_image_by_id(self): expected_resp = {"id": "001", "status": "active", "name": "ImageName"} mock_function = mock.Mock(return_value=expected_resp) func2mock = self.CLIENT_MOCK + '.show_image' self.useFixture(MonkeyPatch(func2mock, mock_function)) resp = tool._find_image(client=self.client, image_id="001", image_name="cirros") self.assertEqual(resp, expected_resp)
def test_get_credentials_v2(self): mock_function = mock.Mock() function2mock = 'config_tempest.config_tempest.auth.get_credentials' self.useFixture(MonkeyPatch(function2mock, mock_function)) self.client.get_credentials(self.conf, "name", "Tname", "pass") mock_function.assert_called_with( auth_url=None, fill_in=False, identity_version='v2', disable_ssl_certificate_validation='true', ca_certs=None, password='pass', tenant_name='Tname', username='name')
def test_get_auth_provider_keystone_v2(self): # check if method returns correct method - KeystoneV2AuthProvider mock_function = mock.Mock() # mock V2Provider, if other provider is called, it fails func2mock = 'config_tempest.config_tempest.auth.KeystoneV2AuthProvider' self.useFixture(MonkeyPatch(func2mock, mock_function)) resp = self.client.get_auth_provider(self.conf, "") self.assertEqual(resp, mock_function()) # check parameters of returned function self.client.get_auth_provider(self.conf, "") mock_function.assert_called_with('', 'http://172.16.52.151:5000/v2.0', 'true', None)
def test_init_manager_as_admin(self): mock_function = mock.Mock(return_value={"id": "my_fake_id"}) func2mock = ('config_tempest.config_tempest.ProjectsClient.' 'get_project_by_name') self.useFixture(MonkeyPatch(func2mock, mock_function)) self._get_clients(self.conf, admin=True) # check if admin credentials were set admin_tenant = self.conf.get("identity", "admin_tenant_name") admin_password = self.conf.get("identity", "admin_password") self.assertEqual(self.conf.get("identity", "admin_username"), "admin") self.assertEqual(admin_tenant, "adminTenant") self.assertEqual(admin_password, "adminPass") # check if admin tenant id was set admin_tenant_id = self.conf.get("identity", "admin_tenant_id") self.assertEqual(admin_tenant_id, "my_fake_id")
def _override_setup(self, mock_args): cloud_args = { 'username': 'cloud_user', 'password': 'cloud_pass', 'project_name': 'cloud_project' } mock_function = mock.Mock(return_value=cloud_args) func2mock = 'os_client_config.cloud_config.CloudConfig.config.get' self.useFixture(MonkeyPatch(func2mock, mock_function)) mock_function = mock.Mock(return_value={"id": "my_fake_id"}) func2mock = ('config_tempest.config_tempest.ProjectsClient.' 'get_project_by_name') self.useFixture(MonkeyPatch(func2mock, mock_function))
def test_init_manager_client_config_get_default(self, mock_args): mock_function = mock.Mock(return_value={}) func2mock = 'os_client_config.cloud_config.CloudConfig.config.get' self.useFixture(MonkeyPatch(func2mock, mock_function)) manager = tool.ClientManager(self.conf, admin=False) # cloud_args is empty => check if default credentials were used self._check_credentials(manager, self.conf.get('identity', 'username'), self.conf.get('identity', 'password'), self.conf.get('identity', 'tenant_name'))
def _mock_get_identity_v3_extensions(self): mock_function = mock.Mock(return_value=['FAKE-EXTENSIONS']) func2mock = 'config_tempest.api_discovery.get_identity_v3_extensions' self.useFixture(MonkeyPatch(func2mock, mock_function))
def test_check_volume_backup_service(self): client = self._get_clients(self.conf).volume_service CLIENT_MOCK = ('tempest.lib.services.volume.v2.' 'services_client.ServicesClient') func2mock = '.list_services' mock_function = mock.Mock(return_value={'services': []}) self.useFixture(MonkeyPatch(CLIENT_MOCK + func2mock, mock_function)) tool.check_volume_backup_service(client, self.conf, self.FAKE_SERVICES) self.assertEqual(self.conf.get('volume-feature-enabled', 'backup'), 'False')
def test_check_ceilometer_service(self): client = self._get_clients(self.conf).service_client CLIENT_MOCK = ('tempest.lib.services.identity.v3.' 'services_client.ServicesClient') func2mock = '.list_services' mock_function = mock.Mock(return_value={'services': [ {'name': 'ceilometer', 'enabled': True, 'type': 'metering'}]}) self.useFixture(MonkeyPatch(CLIENT_MOCK + func2mock, mock_function)) tool.check_ceilometer_service(client, self.conf, self.FAKE_SERVICES) self.assertEqual(self.conf.get('service_available', 'ceilometer'), 'True')
def test_configure_horizon(self): mock_function = mock.Mock(return_value=True) self.useFixture(MonkeyPatch('urllib2.urlopen', mock_function)) tool.configure_horizon(self.conf) self.assertEqual(self.conf.get('service_available', 'horizon'), "True") self.assertEqual(self.conf.get('dashboard', 'dashboard_url'), "http://172.16.52.151/dashboard/") self.assertEqual(self.conf.get('dashboard', 'login_url'), "http://172.16.52.151/dashboard/auth/login/")
def _mock_find_or_create_flavor(self, return_value, func2mock, flavor_name, expected_resp, allow_creation=False, flavor_id=None): mock_function = mock.Mock(return_value=return_value) self.useFixture(MonkeyPatch(self.CLIENT_MOCK + func2mock, mock_function)) resp = tool.find_or_create_flavor(self.client, flavor_id=flavor_id, flavor_name=flavor_name, allow_creation=allow_creation) self.assertEqual(resp, expected_resp)
