我们从Python开源项目中，提取了以下41个代码示例，用于说明如何使用mock.sentinel()。
def test_http_methods(self, mock_create_http_method): s = mock.sentinel schema = ResourceSchema({}, self.mock_handle_http) for method in ('delete', 'get', 'post', 'put'): mock_create_http_method.reset_mock() fn = getattr(schema, 'http_{}'.format(method)) result = fn(s.logic, s.request, s.response, s.params, s.required, title=s.title) self.assertEqual(result, s.result) self.assertEqual( mock_create_http_method.call_args_list, [mock.call(schema, s.logic, method.upper(), request=s.request, response=s.response, params=s.params, required=s.required, title=s.title, before=None, after=None, allowed_exceptions=None, omit_args=None)])
def test_get_annotation(self): """Tests that get_annotation works for badly decorated functions.""" def decorator(fn): def wrapper(): fn() return wrapper mock_logic = mock.Mock() mock_logic._schema_annotation = mock.sentinel.logic_annotation wrapper = decorator(mock_logic) self.assertEqual(ResourceSchemaAnnotation.get_annotation(mock_logic), mock.sentinel.logic_annotation) self.assertEqual(ResourceSchemaAnnotation.get_annotation(wrapper), mock.sentinel.logic_annotation) wrapper._schema_annotation = mock.sentinel.wrapper_annotation self.assertEqual(ResourceSchemaAnnotation.get_annotation(wrapper), mock.sentinel.wrapper_annotation) delattr(wrapper, '_schema_annotation') delattr(mock_logic, '_schema_annotation') self.assertIsNone(ResourceSchemaAnnotation.get_annotation(wrapper))
def _get_all_hosts(self): host_values = [ ('host1', 'node1', {'instances': { 'member1': mock.sentinel, 'instance13': mock.sentinel }}), ('host2', 'node2', {'instances': { 'member2': mock.sentinel, 'member3': mock.sentinel, 'member4': mock.sentinel, 'member5': mock.sentinel, 'instance14': mock.sentinel }}), ('host3', 'node3', {'instances': { 'instance15': mock.sentinel }}), ('host4', 'node4', {'instances': { 'member6': mock.sentinel, 'member7': mock.sentinel, 'instance16': mock.sentinel }})] return [fakes.FakeHostState(host, node, values) for host, node, values in host_values]
def _test_create_quota_usage_if_missing_created(self, per_project_quotas): # Tests that the QuotaUsage is created. user_usages = {} if per_project_quotas: resource = sqlalchemy_api.PER_PROJECT_QUOTAS[0] else: resource = 'fake-resource' project_id = 'fake-project' user_id = 'fake_user' session = mock.sentinel quota_usage = mock.sentinel with mock.patch.object(sqlalchemy_api, '_quota_usage_create', return_value=quota_usage) as quc: self.assertTrue(sqlalchemy_api._create_quota_usage_if_missing( user_usages, resource, None, project_id, user_id, session)) self.assertEqual(quota_usage, user_usages[resource]) # Now test if the QuotaUsage was created with a user_id or not. if per_project_quotas: quc.assert_called_once_with( project_id, None, resource, 0, 0, None, session) else: quc.assert_called_once_with( project_id, user_id, resource, 0, 0, None, session)
def test_compute_node_get_model(self, mock_model_query): class FakeFiltered(object): def first(self): return mock.sentinel.first fake_filtered_cn = FakeFiltered() class FakeModelQuery(object): def filter_by(self, id): return fake_filtered_cn mock_model_query.return_value = FakeModelQuery() result = sqlalchemy_api.compute_node_get_model(self.ctxt, self.item["id"]) self.assertEqual(result, mock.sentinel.first) mock_model_query.assert_called_once_with(self.ctxt, models.ComputeNode)
def test_get_domain_info_with_more_return(self, mock_get_domain): instance = objects.Instance(**self.test_instance) dom_mock = mock.MagicMock() dom_mock.info.return_value = [ 1, 2048, 737, 8, 12345, 888888 ] dom_mock.ID.return_value = mock.sentinel.instance_id mock_get_domain.return_value = dom_mock drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False) info = drvr.get_info(instance) self.assertEqual(1, info.state) self.assertEqual(2048, info.max_mem_kb) self.assertEqual(737, info.mem_kb) self.assertEqual(8, info.num_cpu) self.assertEqual(12345, info.cpu_time_ns) self.assertEqual(mock.sentinel.instance_id, info.id) dom_mock.info.assert_called_once_with() dom_mock.ID.assert_called_once_with() mock_get_domain.assert_called_once_with(instance)
def test_redirect(self): url = "url" decorated = requires(B, get=AAA, post=AA)(undecorated_view) self.assertEqual(decorated._gate.login_url, settings.LOGIN_URL) decorated = requires(B, login_url=url, get=AAA, post=AA)( undecorated_view) self.assertEqual(decorated._gate.login_url, url) with patch('django.contrib.auth.views.redirect_to_login', return_value=sentinel) as do_redirect: request = self.mock_get_request() redirect = decorated(request) do_redirect.assert_called_once_with(request.get_full_path(), url) self.assertIs(redirect, sentinel)
def setUp(self): self.mock_handle_http = mock.Mock(autospec=True, spec=handle_http, return_value=mock.sentinel.result)
def test_create_request_schema(self): schema = ResourceSchema({'definitions': mock.sentinel.definitions}, self.mock_handle_http) request_schema = schema._create_request_schema(params=('a', 'b', 'c'), required=('b', 'c')) self.assertEqual(request_schema, { 'additionalProperties': True, 'definitions': mock.sentinel.definitions, 'properties': {'a': {'$ref': '#/definitions/a'}, 'b': {'$ref': '#/definitions/b'}, 'c': {'$ref': '#/definitions/c'}}, 'required': ('b', 'c'), 'type': 'object', })
def test_init(self): s = mock.sentinel annotation = ResourceSchemaAnnotation( s.logic, 'POST', s.schema, s.request_schema, s.response_schema) self.assertEqual(annotation.logic, s.logic) self.assertEqual(annotation.http_method, 'POST') self.assertEqual(annotation.schema, s.schema) self.assertEqual(annotation.request_schema, s.request_schema) self.assertEqual(annotation.response_schema, s.response_schema) self.assertEqual(annotation.title, 'Create')
def test_init(self): s = mock.sentinel router = Router(s.schema_dir, s.resource_schema_class, s.default_base_handler) self.assertEqual(s.schema_dir, router.schema_dir) self.assertEqual(s.resource_schema_class, router.resource_schema_class) self.assertEqual(s.default_base_handler, router.default_base_handler) # ensure default_base_handler is `object` if not passed router = Router(s.schema_dir, s.resource_schema_class) self.assertEqual(object, router.default_base_handler)
def testSplitVariablesGCD(self, group_variables_mock): group = {'a': [([-1], [0])], 'b': [([-2], [1])], 'c': [([0, 6], [2, 5, 6, 7, 8])], 'd': [([1], [3, 4, 5, 6, 7])], 'ret': [([3, 8], [9])]} group_variables_mock.return_value = group dataflow.split_variables( mock.sentinel, [0, 1, 2, 3, 4], mock.sentinel, mock.sentinel)
def test_msg_str_is_xml_msg(self): self.get_error_msg.return_value = sentinel.err res = exceptions.get_api_exception_message('a message', 401, True) assert_equal(res, sentinel.err)
def setup(self): self.p_Session = patch.object(generic, 'Session') Session = self.p_Session.start() self.p_requests = patch.object(generic, 'requests') rquests = self.p_requests.start() Session.return_value = rquests self.response = MagicMock() self.response.status_code = 200 self.response.text = sentinel.response_text self.request = rquests.request self.request.return_value = self.response self.client = python_kemptech_api.generic.HttpClient() self.client.endpoint = 'ep/'
def test_file_parameter_set(self): open_ = mock_open(read_data='myData') with patch.object(generic, "open", open_, create=True): # as my_open: self.client._do_request('GET','MyCommand', parameters=sentinel.params, file='my_filename') args = self.request.call_args # check positional arguments assert_equal(args[0], ('GET', 'ep/MyCommand?')) # check kwargs kw = args[1] assert_equal(kw['params'], sentinel.params) assert_in('data', kw)
def test_400_status_code(self): self.response.status_code = 400 res = self.client._do_request('GET','MyCommand') assert_equal(res, sentinel.response_text)
def test_with_index(self): with patch.object(LoadMaster, 'build_virtual_service') as build_virtual_service: with patch.object(models, 'get_data'): with patch.object(LoadMaster, '_get'): build_virtual_service.return_value = sentinel.vs res = self.lm.get_virtual_service(index=12) assert_equal(res, sentinel.vs)
def test_get_data_ok(): with patch.object(api_xml, '_get_xml_field') as _get_xml_field: _get_xml_field.return_value = {'Data': sentinel.data} res = api_xml.get_data('anyxml') assert_equal(sentinel.data, res) # Broken by performance imporovements, need to fix later #def test_get_data_no_Data_key(): # with patch.object(api_xml, '_get_xml_field') as _get_xml_field: # _get_xml_field.return_value = {'junk': 'anything'} # res = api_xml.get_data('anyxml') # assert_equal(res, {})
def test_get_success_msg(): with patch.object(api_xml, '_get_xml_field') as _get_xml_field: _get_xml_field.return_value = {'Success': sentinel.data} res = api_xml.get_success_msg('anyxml') assert_equal("{'Success': sentinel.data}", res)
def test_get_error_msg(): with patch.object(api_xml, '_get_xml_field') as _get_xml_field: _get_xml_field.return_value = {'Error': sentinel.data} res = api_xml.get_error_msg('anyxml') assert_equal("{'Error': sentinel.data}", res)
def test_get_data_field(): with patch.object(api_xml, '_get_xml_field') as _get_xml_field: _get_xml_field.return_value = {'Data': sentinel.data} res = api_xml.get_data_field('any_xml', 'any_field') assert_equal({'Data': sentinel.data}, res)
def test_with_index_ok(self): with patch.object(VirtualService, 'build_real_server') as build_real_server: with patch.object(objects, 'get_data'): with patch.object(VirtualService, '_get'): self.vs.index = self build_real_server.return_value = sentinel.rs res = self.vs.get_real_server('1.1.1.1', 80) assert_equal(res, sentinel.rs)
def test_without_index_ok(self): with patch.object(VirtualService, 'build_real_server') as build_real_server: with patch.object(objects, 'get_data'): with patch.object(VirtualService, '_get'): self.vs.index = None build_real_server.return_value = sentinel.rs res = self.vs.get_real_server('1.1.1.1', 80) assert_equal(res, sentinel.rs)
def test_without_index_invalid_port(self): with patch.object(VirtualService, 'build_real_server') as build_real_server: with patch.object(objects, 'get_data'): with patch.object(VirtualService, '_get'): self.vs.index = None build_real_server.return_value = sentinel.rs with assert_raises(ValidationError): self.vs.get_real_server('1.1.1.1.', 'junk')
def test_route_token_required(self): request = requestMock(b"/", headers={b'Authorization': [str(mock.sentinel)]}) returned = self.app.execute_endpoint("token_required_endpoint", request) self.assertEqual(returned, self.userinfo) self.auth.decode_header_value.assert_called_once_with(str(mock.sentinel))
def test_process(self): m = CPUs().process(sentinel, self.results, Mock()).maps[0] for i in ('cacheSizeL1', 'cacheSizeL2', 'cacheSizeL3', 'cacheSpeedL2', 'cacheSpeedL3', 'clockspeed', 'cores', 'extspeed', 'socket', 'threads'): self.assertEquals(getattr(m, i), 1) self.assertEquals(m.description, 'Description') self.assertEquals(m.id, 'DeviceID') self.assertEquals(m.title, 'Name')
def test_commit_raises_exception_on_timeout(self): transaction = impl_idl.OvsVsctlTransaction(mock.sentinel, mock.Mock(), 1) with testtools.ExpectedException(exceptions.TimeoutException): transaction.commit()
def test_post_commit_does_not_raise_exception(self): with mock.patch.object(impl_idl.OvsVsctlTransaction, "do_post_commit", side_effect=Exception): transaction = impl_idl.OvsVsctlTransaction(mock.sentinel, mock.Mock(), 0) transaction.post_commit(mock.Mock())
def test_instance_get_all_by_filters_paginated_allows_deleted_marker( self, mock_paginate, mock_fill, mock_get): ctxt = mock.MagicMock() ctxt.elevated.return_value = mock.sentinel.elevated sqlalchemy_api.instance_get_all_by_filters_sort(ctxt, {}, marker='foo') mock_get.assert_called_once_with(mock.sentinel.elevated, 'foo') ctxt.elevated.assert_called_once_with(read_deleted='yes')
def test_create_quota_usage_if_missing_not_created(self): # Tests that QuotaUsage isn't created if it's already in user_usages. resource = 'fake-resource' project_id = 'fake-project' user_id = 'fake_user' session = mock.sentinel quota_usage = mock.sentinel user_usages = {resource: quota_usage} with mock.patch.object(sqlalchemy_api, '_quota_usage_create') as quc: self.assertFalse(sqlalchemy_api._create_quota_usage_if_missing( user_usages, resource, None, project_id, user_id, session)) self.assertFalse(quc.called)
def test_migrate_disk_and_power_off_resize_copy_disk_info(self, mock_disk_info, mock_shared, mock_path, mock_destroy, mock_copy, mock_execuate): instance = self._create_instance() disk_info = self._disk_info() disk_info_text = jsonutils.loads(disk_info) instance_base = os.path.dirname(disk_info_text[0]['path']) flavor = {'root_gb': 10, 'ephemeral_gb': 25} flavor_obj = objects.Flavor(**flavor) mock_disk_info.return_value = disk_info mock_path.return_value = instance_base mock_shared.return_value = False src_disk_info_path = os.path.join(instance_base + '_resize', 'disk.info') with mock.patch.object(os.path, 'exists', autospec=True) \ as mock_exists: # disk.info exists on the source mock_exists.side_effect = \ lambda path: path == src_disk_info_path self.drvr.migrate_disk_and_power_off(context.get_admin_context(), instance, mock.sentinel, flavor_obj, None) self.assertTrue(mock_exists.called) dst_disk_info_path = os.path.join(instance_base, 'disk.info') mock_copy.assert_any_call(src_disk_info_path, dst_disk_info_path, host=mock.sentinel, on_execute=mock.ANY, on_completion=mock.ANY)
def test_get_by_uuid(self, mock_db_get): obj = objects.InstanceGroup.get_by_uuid(mock.sentinel.ctx, _DB_UUID) mock_db_get.assert_called_once_with(mock.sentinel.ctx, _DB_UUID) self.assertEqual(_INST_GROUP_DB['members'], obj.members) self.assertEqual(_INST_GROUP_DB['policies'], obj.policies) self.assertEqual(_DB_UUID, obj.uuid) self.assertEqual(_INST_GROUP_DB['project_id'], obj.project_id) self.assertEqual(_INST_GROUP_DB['user_id'], obj.user_id) self.assertEqual(_INST_GROUP_DB['name'], obj.name)
def test_get_by_instance_uuid(self, mock_db_get): objects.InstanceGroup.get_by_instance_uuid( mock.sentinel.ctx, mock.sentinel.instance_uuid) mock_db_get.assert_called_once_with( mock.sentinel.ctx, mock.sentinel.instance_uuid)
def test_refresh(self, mock_db_get): changed_group = copy.deepcopy(_INST_GROUP_DB) changed_group['name'] = 'new_name' mock_db_get.side_effect = [_INST_GROUP_DB, changed_group] obj = objects.InstanceGroup.get_by_uuid(mock.sentinel.ctx, _DB_UUID) self.assertEqual(_INST_GROUP_DB['name'], obj.name) obj.refresh() self.assertEqual('new_name', obj.name) self.assertEqual(set([]), obj.obj_what_changed())
def test_save_without_hosts(self, mock_db_get, mock_db_update, mock_notify): mock_db_get.side_effect = [_INST_GROUP_DB, _INST_GROUP_DB] obj = objects.InstanceGroup.get_by_uuid(mock.sentinel.ctx, _DB_UUID) obj.hosts = ['fake-host1'] self.assertRaises(exception.InstanceGroupSaveException, obj.save) # make sure that we can save by removing hosts from what is updated obj.obj_reset_changes(['hosts']) obj.save() # since hosts was the only update, there is no actual call self.assertFalse(mock_db_update.called) self.assertFalse(mock_notify.called)
def test_create(self, mock_db_create, mock_notify): obj = objects.InstanceGroup(context=mock.sentinel.ctx) obj.uuid = _DB_UUID obj.name = _INST_GROUP_DB['name'] obj.user_id = _INST_GROUP_DB['user_id'] obj.project_id = _INST_GROUP_DB['project_id'] obj.members = _INST_GROUP_DB['members'] obj.policies = _INST_GROUP_DB['policies'] obj.updated_at = _TS_NOW obj.created_at = _TS_NOW obj.deleted_at = None obj.deleted = False obj.create() mock_db_create.assert_called_once_with( mock.sentinel.ctx, {'uuid': _DB_UUID, 'name': _INST_GROUP_DB['name'], 'user_id': _INST_GROUP_DB['user_id'], 'project_id': _INST_GROUP_DB['project_id'], 'created_at': _TS_NOW, 'updated_at': _TS_NOW, 'deleted_at': None, 'deleted': False, }, members=_INST_GROUP_DB['members'], policies=_INST_GROUP_DB['policies']) mock_notify.assert_called_once_with( mock.sentinel.ctx, "create", {'uuid': _DB_UUID, 'name': _INST_GROUP_DB['name'], 'user_id': _INST_GROUP_DB['user_id'], 'project_id': _INST_GROUP_DB['project_id'], 'created_at': _TS_NOW, 'updated_at': _TS_NOW, 'deleted_at': None, 'deleted': False, 'members': _INST_GROUP_DB['members'], 'policies': _INST_GROUP_DB['policies'], 'server_group_id': _DB_UUID}) self.assertRaises(exception.ObjectActionError, obj.create)
def test_destroy(self, mock_db_delete, mock_notify): obj = objects.InstanceGroup(context=mock.sentinel.ctx) obj.uuid = _DB_UUID obj.destroy() mock_db_delete.assert_called_once_with(mock.sentinel.ctx, _DB_UUID) mock_notify.assert_called_once_with(mock.sentinel.ctx, "delete", {'server_group_id': _DB_UUID})
def test_add_members(self, mock_members_add_db, mock_notify): mock_members_add_db.return_value = [mock.sentinel.members] members = objects.InstanceGroup.add_members(mock.sentinel.ctx, _DB_UUID, mock.sentinel.members) self.assertEqual([mock.sentinel.members], members) mock_members_add_db.assert_called_once_with( mock.sentinel.ctx, _DB_UUID, mock.sentinel.members) mock_notify.assert_called_once_with( mock.sentinel.ctx, "addmember", {'instance_uuids': mock.sentinel.members, 'server_group_id': _DB_UUID})
def test_count_members_by_user(self, mock_get_db, mock_il_get): mock_il_get.return_value = [mock.ANY] obj = objects.InstanceGroup.get_by_uuid(mock.sentinel.ctx, _DB_UUID) expected_filters = { 'uuid': ['instance_id1', 'instance_id2'], 'user_id': 'fake_user', 'deleted': False } self.assertEqual(1, obj.count_members_by_user('fake_user')) mock_il_get.assert_called_once_with(mock.sentinel.ctx, filters=expected_filters)
