本文整理汇总了Python中mock.NonCallableMagicMock类的典型用法代码示例。如果您正苦于以下问题:Python NonCallableMagicMock类的具体用法?Python NonCallableMagicMock怎么用?Python NonCallableMagicMock使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了NonCallableMagicMock类的11个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_get_listed_users
def test_get_listed_users(self):
usr1 = models.CassandraUser(self._get_random_name(1025))
usr2 = models.CassandraUser(self._get_random_name(1025))
usr3 = models.CassandraUser(self._get_random_name(1025))
db1 = models.CassandraSchema('db1')
db2 = models.CassandraSchema('db2')
usr1.databases.append(db1.serialize())
usr3.databases.append(db2.serialize())
rv_1 = NonCallableMagicMock()
rv_1.configure_mock(name=usr1.name, super=False)
rv_2 = NonCallableMagicMock()
rv_2.configure_mock(name=usr2.name, super=False)
rv_3 = NonCallableMagicMock()
rv_3.configure_mock(name=usr3.name, super=True)
with patch.object(self.conn, 'execute', return_value=iter(
[rv_1, rv_2, rv_3])):
with patch.object(self.admin, '_get_acl',
return_value={usr1.name: {db1.name: {'SELECT'},
db2.name: {}},
usr3.name: {db2.name: {'SELECT'}}}
):
usrs = self.manager.list_users(self.context)
self.conn.execute.assert_has_calls([
call(self.__LIST_USR_FORMAT),
], any_order=True)
self.assertIn(usr1.serialize(), usrs[0])
self.assertIn(usr2.serialize(), usrs[0])
self.assertIn(usr3.serialize(), usrs[0])
开发者ID:gongwayne,项目名称:Openstack,代码行数:30,代码来源:test_cassandra_manager.py
示例2: setUp
def setUp(self):
super(DesignateBackendTestCase, self).setUp()
self.zone = objects.Zone(
id='e2bed4dc-9d01-11e4-89d3-123b93f75cba',
name='example.com.',
email='[email protected]',
)
self.target = {
'id': '4588652b-50e7-46b9-b688-a9bad40a873e',
'type': 'designate',
'masters': [
{'host': '192.0.2.1', 'port': 53},
],
'options': [
{'key': 'username', 'value': 'user'},
{'key': 'password', 'value': 'secret'},
{'key': 'project_name', 'value': 'project'},
{'key': 'project_zone_name', 'value': 'project_zone'},
{'key': 'user_zone_name', 'value': 'user_zone'},
],
}
self.backend = impl_designate.DesignateBackend(
objects.PoolTarget.from_dict(self.target)
)
# Mock client
self.client = NonCallableMagicMock()
zones = NonCallableMagicMock(spec_set=[
'create', 'delete'])
self.client.configure_mock(zones=zones)
开发者ID:openstack,项目名称:designate,代码行数:33,代码来源:test_designate.py
示例3: setUp
def setUp(self):
super(DesignateBackendTest, self).setUp()
opts = RoObject(
username="user",
password="secret",
project_name="project",
project_domain_name="project_domain",
user_domain_name="user_domain",
)
self.target = RoObject(
{
"id": "4588652b-50e7-46b9-b688-a9bad40a873e",
"type": "dyndns",
"masters": [RoObject({"host": "192.0.2.1", "port": 53})],
"options": opts,
}
)
# Backends blow up when trying to self.admin_context = ... due to
# policy not being initialized
self.admin_context = Mock()
get_context_patcher = patch("designate.context.DesignateContext.get_admin_context")
get_context = get_context_patcher.start()
get_context.return_value = self.admin_context
self.backend = impl_designate.DesignateBackend(self.target)
# Mock client
self.client = NonCallableMagicMock()
zones = NonCallableMagicMock(spec_set=["create", "delete"])
self.client.configure_mock(zones=zones)
开发者ID:ramsateesh,项目名称:designate,代码行数:31,代码来源:test_designate.py
示例4: setUp
def setUp(self):
super(DesignateBackendTest, self).setUp()
opts = RoObject(
username='user',
password='secret',
project_name='project',
project_zone_name='project_zone',
user_zone_name='user_zone'
)
self.target = RoObject({
'id': '4588652b-50e7-46b9-b688-a9bad40a873e',
'type': 'dyndns',
'masters': [RoObject({'host': '192.0.2.1', 'port': 53})],
'options': opts
})
# Backends blow up when trying to self.admin_context = ... due to
# policy not being initialized
self.admin_context = Mock()
get_context_patcher = patch(
'designate.context.DesignateContext.get_admin_context')
get_context = get_context_patcher.start()
get_context.return_value = self.admin_context
self.backend = impl_designate.DesignateBackend(self.target)
# Mock client
self.client = NonCallableMagicMock()
zones = NonCallableMagicMock(spec_set=[
'create', 'delete'])
self.client.configure_mock(zones=zones)
开发者ID:TimSimmons,项目名称:designate,代码行数:31,代码来源:test_designate.py
示例5: test_failed_wait_for_connection
def test_failed_wait_for_connection(self):
host = Mock(spec=Host, address='ip1')
session = self.make_session()
conn = NonCallableMagicMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=False)
session.cluster.connection_factory.return_value = conn
pool = HostConnectionPool(host, HostDistance.LOCAL, session)
session.cluster.connection_factory.assert_called_once_with(host.address)
pool.borrow_connection(timeout=0.01)
self.assertEqual(1, conn.in_flight)
conn.in_flight = MAX_STREAM_PER_CONNECTION
# we're already at the max number of requests for this connection,
# so we this should fail
self.assertRaises(NoConnectionsAvailable, pool.borrow_connection, 0)
开发者ID:EverythingMe,项目名称:python-driver,代码行数:17,代码来源:test_host_connection_pool.py
示例6: test_return_closed_connection
def test_return_closed_connection(self):
host = Mock(spec=Host, address='ip1')
session = self.make_session()
conn = NonCallableMagicMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=True)
session.cluster.connection_factory.return_value = conn
pool = HostConnectionPool(host, HostDistance.LOCAL, session)
session.cluster.connection_factory.assert_called_once_with(host.address)
pool.borrow_connection(timeout=0.01)
conn.is_closed = True
session.cluster.signal_connection_failure.return_value = False
pool.return_connection(conn)
# a new creation should be scheduled
session.submit.assert_called_once()
self.assertFalse(pool.is_shutdown)
开发者ID:EverythingMe,项目名称:python-driver,代码行数:17,代码来源:test_host_connection_pool.py
示例7: test_spawn_when_at_max
def test_spawn_when_at_max(self):
host = Mock(spec=Host, address='ip1')
session = self.make_session()
conn = NonCallableMagicMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=False)
session.cluster.connection_factory.return_value = conn
# core conns = 1, max conns = 2
session.cluster.get_max_connections_per_host.return_value = 2
pool = HostConnectionPool(host, HostDistance.LOCAL, session)
session.cluster.connection_factory.assert_called_once_with(host.address)
pool.borrow_connection(timeout=0.01)
self.assertEqual(1, conn.in_flight)
# make this conn full
conn.in_flight = MAX_STREAM_PER_CONNECTION
# we don't care about making this borrow_connection call succeed for the
# purposes of this test, as long as it results in a new connection
# creation being scheduled
self.assertRaises(NoConnectionsAvailable, pool.borrow_connection, 0)
session.submit.assert_called_once_with(pool._create_new_connection)
开发者ID:EverythingMe,项目名称:python-driver,代码行数:23,代码来源:test_host_connection_pool.py
示例8: DesignateBackendTest
class DesignateBackendTest(oslotest.base.BaseTestCase):
def setUp(self):
super(DesignateBackendTest, self).setUp()
opts = RoObject(
username="user",
password="secret",
project_name="project",
project_domain_name="project_domain",
user_domain_name="user_domain",
)
self.target = RoObject(
{
"id": "4588652b-50e7-46b9-b688-a9bad40a873e",
"type": "dyndns",
"masters": [RoObject({"host": "192.0.2.1", "port": 53})],
"options": opts,
}
)
# Backends blow up when trying to self.admin_context = ... due to
# policy not being initialized
self.admin_context = Mock()
get_context_patcher = patch("designate.context.DesignateContext.get_admin_context")
get_context = get_context_patcher.start()
get_context.return_value = self.admin_context
self.backend = impl_designate.DesignateBackend(self.target)
# Mock client
self.client = NonCallableMagicMock()
zones = NonCallableMagicMock(spec_set=["create", "delete"])
self.client.configure_mock(zones=zones)
def test_create_domain(self):
zone = create_zone()
masters = ["%(host)s:%(port)s" % self.target.masters[0]]
with patch.object(self.backend, "_get_client", return_value=self.client):
self.backend.create_domain(self.admin_context, zone)
self.client.zones.create.assert_called_once_with(zone.name, "SECONDARY", masters=masters)
def test_delete_domain(self):
zone = create_zone()
with patch.object(self.backend, "_get_client", return_value=self.client):
self.backend.delete_domain(self.admin_context, zone)
self.client.zones.delete.assert_called_once_with(zone.name)
def test_delete_domain_notfound(self):
zone = create_zone()
self.client.delete.side_effect = exceptions.NotFound
with patch.object(self.backend, "_get_client", return_value=self.client):
self.backend.delete_domain(self.admin_context, zone)
self.client.zones.delete.assert_called_once_with(zone.name)
def test_delete_domain_exc(self):
class Exc(Exception):
pass
zone = create_zone()
self.client.zones.delete.side_effect = Exc()
with testtools.ExpectedException(Exc):
with patch.object(self.backend, "_get_client", return_value=self.client):
self.backend.delete_domain(self.admin_context, zone)
self.client.zones.delete.assert_called_once_with(zone.name)
开发者ID:ramsateesh,项目名称:designate,代码行数:63,代码来源:test_designate.py
示例9: DesignateBackendTestCase
class DesignateBackendTestCase(tests.TestCase):
def setUp(self):
super(DesignateBackendTestCase, self).setUp()
self.zone = objects.Zone(
id='e2bed4dc-9d01-11e4-89d3-123b93f75cba',
name='example.com.',
email='[email protected]',
)
self.target = {
'id': '4588652b-50e7-46b9-b688-a9bad40a873e',
'type': 'designate',
'masters': [
{'host': '192.0.2.1', 'port': 53},
],
'options': [
{'key': 'username', 'value': 'user'},
{'key': 'password', 'value': 'secret'},
{'key': 'project_name', 'value': 'project'},
{'key': 'project_zone_name', 'value': 'project_zone'},
{'key': 'user_zone_name', 'value': 'user_zone'},
],
}
self.backend = impl_designate.DesignateBackend(
objects.PoolTarget.from_dict(self.target)
)
# Mock client
self.client = NonCallableMagicMock()
zones = NonCallableMagicMock(spec_set=[
'create', 'delete'])
self.client.configure_mock(zones=zones)
def test_create_zone(self):
masters = ["%(host)s:%(port)s" % self.target['masters'][0]]
with patch.object(self.backend, '_get_client',
return_value=self.client):
self.backend.create_zone(self.admin_context, self.zone)
self.client.zones.create.assert_called_once_with(
self.zone.name, 'SECONDARY', masters=masters)
def test_delete_zone(self):
with patch.object(self.backend, '_get_client',
return_value=self.client):
self.backend.delete_zone(self.admin_context, self.zone)
self.client.zones.delete.assert_called_once_with(self.zone.name)
def test_delete_zone_notfound(self):
self.client.delete.side_effect = exceptions.NotFound
with patch.object(self.backend, '_get_client',
return_value=self.client):
self.backend.delete_zone(self.admin_context, self.zone)
self.client.zones.delete.assert_called_once_with(self.zone.name)
def test_delete_zone_exc(self):
self.client.zones.delete.side_effect = Exception
with testtools.ExpectedException(Exception):
with patch.object(self.backend, '_get_client',
return_value=self.client):
self.backend.delete_zone(self.admin_context, self.zone)
self.client.zones.delete.assert_called_once_with(self.zone.name)
开发者ID:openstack,项目名称:designate,代码行数:63,代码来源:test_designate.py
示例10: DesignateBackendTest
class DesignateBackendTest(oslotest.base.BaseTestCase):
def setUp(self):
super(DesignateBackendTest, self).setUp()
opts = RoObject(
username='user',
password='secret',
project_name='project',
project_zone_name='project_zone',
user_zone_name='user_zone'
)
self.target = RoObject({
'id': '4588652b-50e7-46b9-b688-a9bad40a873e',
'type': 'dyndns',
'masters': [RoObject({'host': '192.0.2.1', 'port': 53})],
'options': opts
})
# Backends blow up when trying to self.admin_context = ... due to
# policy not being initialized
self.admin_context = Mock()
get_context_patcher = patch(
'designate.context.DesignateContext.get_admin_context')
get_context = get_context_patcher.start()
get_context.return_value = self.admin_context
self.backend = impl_designate.DesignateBackend(self.target)
# Mock client
self.client = NonCallableMagicMock()
zones = NonCallableMagicMock(spec_set=[
'create', 'delete'])
self.client.configure_mock(zones=zones)
def test_create_zone(self):
zone = create_zone()
masters = ["%(host)s:%(port)s" % self.target.masters[0]]
with patch.object(
self.backend, '_get_client', return_value=self.client):
self.backend.create_zone(self.admin_context, zone)
self.client.zones.create.assert_called_once_with(
zone.name, 'SECONDARY', masters=masters)
def test_delete_zone(self):
zone = create_zone()
with patch.object(
self.backend, '_get_client', return_value=self.client):
self.backend.delete_zone(self.admin_context, zone)
self.client.zones.delete.assert_called_once_with(zone.name)
def test_delete_zone_notfound(self):
zone = create_zone()
self.client.delete.side_effect = exceptions.NotFound
with patch.object(
self.backend, '_get_client', return_value=self.client):
self.backend.delete_zone(self.admin_context, zone)
self.client.zones.delete.assert_called_once_with(zone.name)
def test_delete_zone_exc(self):
class Exc(Exception):
pass
zone = create_zone()
self.client.zones.delete.side_effect = Exc()
with testtools.ExpectedException(Exc):
with patch.object(
self.backend, '_get_client', return_value=self.client):
self.backend.delete_zone(self.admin_context, zone)
self.client.zones.delete.assert_called_once_with(zone.name)
开发者ID:TimSimmons,项目名称:designate,代码行数:68,代码来源:test_designate.py
示例11: test_volume_page_search
def test_volume_page_search(self, mocksolr_interface, mockpaginator, mockrepo):
mockobj = NonCallableMock()
mockobj.pid = 'vol:1'
mockobj.title = 'Lecoq, the detective'
mockobj.date = ['1801']
mockrepo.return_value.get_object.return_value = mockobj
mocksolr = mocksolr_interface.return_value
# simulate sunburnt's fluid interface
mocksolr.query.return_value = mocksolr.query
for method in ['query', 'facet_by', 'sort_by', 'field_limit', 'highlight',
'exclude', 'filter', 'join', 'paginate', 'results_as']:
getattr(mocksolr.query, method).return_value = mocksolr.query
# set up mock results for collection query and facet counts
solr_result = NonCallableMagicMock(spec_set=['__iter__', 'facet_counts'])
# *only* mock iter, to avoid weirdness with django templates & callables
solr_results = [
SolrPage(**{'pid': 'page:1', 'page_order': '1', 'score': 0.5,
'solr_highlights': {'page_text': ['snippet with search term']},
'identifier': ['http://testpid.co/ark:/1234/11/']}),
SolrPage(**{'pid': 'page:233', 'page_order': '123', 'score': 0.02,
'solr_highlights': {'page_text': ['sample text result from content']},
'identifier': ['http://testpid.co/ark:/1234/22/']}),
]
solr_result.__iter__.return_value = solr_results
mocksolr.query.__iter__.return_value = iter(solr_result)
mocksolr.count.return_value = 2
mockpage = NonCallableMock()
mockpaginator.return_value.page.return_value = mockpage
results = NonCallableMagicMock(spec=['__iter__', 'facet_counts', 'highlighting'])
results.__iter__.return_value = iter(solr_result)
mockpage.object_list = results
mockpage.has_other_pages = False
mockpage.paginator.count = 2
mockpage.paginator.page_range = [1]
# patch in highlighting - apparent change in sunburnt behavior
results.highlighting = {
'page:1': {'page_text': ['snippet with search term']},
'page:233': {'page_text': ['sample text result from content']}
}
vol_url = reverse('books:volume', kwargs={'pid': mockobj.pid})
response = self.client.get(vol_url, {'keyword': 'determine'})
self.assertEqual(response.templates[0].name,
views.VolumeDetail.search_template_name,
'volume search template should be used for valid search submission')
for page in iter(solr_result):
self.assertContains(response,
reverse('books:page-image', kwargs={'vol_pid': mockobj.pid,
'pid': page.pid, 'mode': 'mini-thumbnail'}),
msg_prefix='search results should include mini page thumbnail url')
self.assertContains(response, "Page %(page_order)s" % page,
msg_prefix='search results should include page number')
self.assertContains(response, page['score'],
msg_prefix='search results should display page relevance score')
self.assertContains(response, reverse('books:page',
kwargs={'vol_pid': mockobj.pid, 'pid': page['pid']}),
msg_prefix='search results should link to full page view')
self.assertContains(response, '... %s ...' % page['solr_highlights']['page_text'][0],
msg_prefix='solr snippets should display when available')
# ajax request
with patch('readux.books.views.VolumeDetail.get_context_data') as mock_ctx:
results = NonCallableMagicMock(spec=['__iter__', 'facet_counts', 'highlighting'])
results.__iter__.return_value = iter(solr_result)
results.highlighting = {
solr_results[0].pid: {
'page_text': 'sample highlighting snippet'
},
solr_results[1].pid: {
'page_text': 'another highlighting snippet'
}
}
mockpage = NonCallableMagicMock(spec=['__iter__'])
mockpage.object_list = results
mock_ctx.return_value = {
'pages': mockpage,
}
response = self.client.get(vol_url, {'keyword': 'determine'},
HTTP_X_REQUESTED_WITH='XMLHttpRequest')
self.assertEqual('application/json', response['content-type'])
data = json.loads(response.content)
for idx in range(len(data)):
self.assertEqual(solr_results[idx].pid, data[idx]['pid'])
self.assertEqual('p. %s' % solr_results[idx]['page_order'],
data[idx]['label'])
self.assertEqual(reverse('books:page-image', kwargs={'vol_pid': mockobj.pid,
'pid': solr_results[idx].pid, 'mode': 'mini-thumbnail'}),
data[idx]['thumbnail'])
self.assertEqual(results.highlighting[solr_results[idx].pid]['page_text'],
data[idx]['highlights'])
开发者ID:emory-libraries,项目名称:readux,代码行数:96,代码来源:views.py
注:本文中的mock.NonCallableMagicMock类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论