def test_network_interfaces_update(self, fake_interfaces):
record = dict(fake_resources.NETWORK_INTERFACE1.items())
payload = {'name': 'New'}
db_data = payload.copy()
record.update(payload)
fake_interfaces.return_value = record
resp = self.put('/v1/network-interfaces/1', data=payload)
self.assertEqual(resp.json['name'], db_data['name'])
self.assertEqual(200, resp.status_code)
self.assertEqual(
resp.json['ip_address'],
fake_resources.NETWORK_INTERFACE1.ip_address
)
fake_interfaces.assert_called_once_with(mock.ANY, '1', db_data)
python类ANY的实例源码
def test_send_mail(self):
"""
send_mail should send mail and mark as sent if everything goes well.
"""
job = self.create_job(state='queued')
with mailman_mock() as mailman:
job.send_mail('token')
mailman.send_draft.assert_called_with(
job.message_id_int, mock.ANY)
mailman.mark_as_sent.assert_called_with(job.message_id_int,
job.sent_mail_rfc_id,
'testmail')
mailman.quit.assert_called_with()
job = job.key.get()
self.assertEquals(job.state, 'done')
self.assertTrue(job.sent_mail_rfc_id)
def test_create_without_instance_name(self):
self.mock_object(
self.fake_driver.resource, 'create_instances',
mock.Mock(return_value=mock.Mock))
self.fake_driver.create(
'fake_image_id',
'fake_flavor_id',
'fake_net_id'
)
self.fake_driver.resource.create_instances. \
assert_called_once_with(
ImageId='fake_image_id',
MinCount=1,
MaxCount=1,
InstanceType='fake_flavor_id',
SubnetId='fake_net_id',
IamInstanceProfile={
'Arn': '',
'Name': mock.ANY
}
)
def test_create_without_instance_name(self):
self.mock_object(
self.fake_driver.client.servers, 'create',
mock.Mock(return_value=mock.Mock))
# NOTE: in fact: mock.Mock is novaclient.v2.servers.Server
self.fake_driver.create(
'fake_image_id',
'fake_flavor_id',
'fake_net_id'
)
self.fake_driver.client.servers.create. \
assert_called_once_with(
name=mock.ANY,
image='fake_image_id',
flavor='fake_flavor_id',
nics=[{'net-id': 'fake_net_id'}]
)
def test_get_orders(requests_post, logger):
client = build_client()
output = client.get_orders()
assert output == test_body
requests_post.assert_called_with(
url=build_url('/open_orders'),
json={
'book': test_book,
'key': test_key,
'nonce': test_nonce,
'signature': mock.ANY
}
)
logger.debug.assert_called_with(
"[client: test_client_id] get user's open orders for btc_usd")
with pytest.raises(InvalidOrderBookError):
client.get_orders(book='invalid_book')
def test_proxy_info(http_mock, resp_mock):
http = Mock()
http.request.return_value = (Mock(), Mock())
http_mock.return_value = http
Connection.set_proxy_info(
'example.com',
8080,
proxy_type=PROXY_TYPE_SOCKS5,
)
make_request("GET", "http://httpbin.org/get")
http_mock.assert_called_with(timeout=None, ca_certs=ANY, proxy_info=ANY)
http.request.assert_called_with("http://httpbin.org/get", "GET",
body=None, headers=None)
proxy_info = http_mock.call_args[1]['proxy_info']
assert_equal(proxy_info.proxy_host, 'example.com')
assert_equal(proxy_info.proxy_port, 8080)
assert_equal(proxy_info.proxy_type, PROXY_TYPE_SOCKS5)
def testPassThrough(self, mock_request):
mock_response = Mock()
mock_response.ok = True,
mock_response.content = json.dumps({'key': 'value'})
mock_request.return_value = mock_response
assert_equal(self.r.timeout, sentinel.timeout)
assert_equal((mock_response, {'key': 'value'}), self.r.request('GET', base_uri))
mock_request.assert_called_once_with(
'GET',
base_uri + '.json',
headers=ANY,
timeout=sentinel.timeout,
auth=ANY
)
def test_add(self, mock_check_conn, mock_makedirs, mock_configparser,
mock_open):
config = mock_configparser.return_value
params = copy.copy(self.add_params)
self.manager.add(**params)
expected_calls = [mock.call('VirtualBMC', i, self.add_params[i])
for i in self.add_params]
self.assertEqual(sorted(expected_calls),
sorted(config.set.call_args_list))
config.add_section.assert_called_once_with('VirtualBMC')
config.write.assert_called_once_with(mock.ANY)
mock_check_conn.assert_called_once_with(
self.add_params['libvirt_uri'], self.add_params['domain_name'],
sasl_username=self.add_params['libvirt_sasl_username'],
sasl_password=self.add_params['libvirt_sasl_password'])
mock_makedirs.assert_called_once_with(
os.path.join(_CONFIG_PATH, self.add_params['domain_name']))
mock_configparser.assert_called_once_with()
def test_add_with_port_as_int(self, mock_check_conn, mock_makedirs,
mock_configparser, mock_open):
config = mock_configparser.return_value
params = copy.copy(self.add_params)
params['port'] = int(params['port'])
self.manager.add(**params)
expected_calls = [mock.call('VirtualBMC', i, self.add_params[i])
for i in self.add_params]
self.assertEqual(sorted(expected_calls),
sorted(config.set.call_args_list))
config.add_section.assert_called_once_with('VirtualBMC')
config.write.assert_called_once_with(mock.ANY)
mock_check_conn.assert_called_once_with(
self.add_params['libvirt_uri'], self.add_params['domain_name'],
sasl_username=self.add_params['libvirt_sasl_username'],
sasl_password=self.add_params['libvirt_sasl_password'])
mock_makedirs.assert_called_once_with(
os.path.join(_CONFIG_PATH, self.add_params['domain_name']))
mock_configparser.assert_called_once_with()
def test_get_all_capsules(self, mock_container_get_by_uuid,
mock_capsule_list,
mock_container_show):
test_container = utils.get_test_container()
test_container_obj = objects.Container(self.context,
**test_container)
mock_container_get_by_uuid.return_value = test_container_obj
test_capsule = utils.get_test_capsule()
test_capsule_obj = objects.Capsule(self.context, **test_capsule)
mock_capsule_list.return_value = [test_capsule_obj]
mock_container_show.return_value = test_container_obj
response = self.app.get('/capsules/')
mock_capsule_list.assert_called_once_with(mock.ANY,
1000, None, 'id', 'asc',
filters=None)
context = mock_capsule_list.call_args[0][0]
self.assertIs(False, context.all_tenants)
self.assertEqual(200, response.status_int)
actual_capsules = response.json['capsules']
self.assertEqual(1, len(actual_capsules))
self.assertEqual(test_capsule['uuid'],
actual_capsules[0].get('uuid'))
def test_get_all_hosts(self, mock_host_list, mock_policy):
mock_policy.return_value = True
test_host = utils.get_test_compute_node()
numat = numa.NUMATopology._from_dict(test_host['numa_topology'])
test_host['numa_topology'] = numat
hosts = [objects.ComputeNode(self.context, **test_host)]
mock_host_list.return_value = hosts
response = self.get('/v1/hosts')
mock_host_list.assert_called_once_with(mock.ANY,
1000, None, 'hostname', 'asc',
filters=None)
self.assertEqual(200, response.status_int)
actual_hosts = response.json['hosts']
self.assertEqual(1, len(actual_hosts))
self.assertEqual(test_host['uuid'],
actual_hosts[0].get('uuid'))
def test_get_all_containers(self, mock_container_list,
mock_container_show):
test_container = utils.get_test_container()
containers = [objects.Container(self.context, **test_container)]
mock_container_list.return_value = containers
mock_container_show.return_value = containers[0]
response = self.get('/v1/containers/')
mock_container_list.assert_called_once_with(mock.ANY,
1000, None, 'id', 'asc',
filters=None)
context = mock_container_list.call_args[0][0]
self.assertIs(False, context.all_tenants)
self.assertEqual(200, response.status_int)
actual_containers = response.json['containers']
self.assertEqual(1, len(actual_containers))
self.assertEqual(test_container['uuid'],
actual_containers[0].get('uuid'))
def test_get_all_containers_all_tenants(self, mock_container_list,
mock_container_show, mock_policy):
mock_policy.return_value = True
test_container = utils.get_test_container()
containers = [objects.Container(self.context, **test_container)]
mock_container_list.return_value = containers
mock_container_show.return_value = containers[0]
response = self.get('/v1/containers/?all_tenants=1')
mock_container_list.assert_called_once_with(mock.ANY,
1000, None, 'id', 'asc',
filters=None)
context = mock_container_list.call_args[0][0]
self.assertIs(True, context.all_tenants)
self.assertEqual(200, response.status_int)
actual_containers = response.json['containers']
self.assertEqual(1, len(actual_containers))
self.assertEqual(test_container['uuid'],
actual_containers[0].get('uuid'))
def test_get_one_by_uuid_all_tenants(self, mock_container_get_by_uuid,
mock_container_show, mock_policy):
mock_policy.return_value = True
test_container = utils.get_test_container()
test_container_obj = objects.Container(self.context, **test_container)
mock_container_get_by_uuid.return_value = test_container_obj
mock_container_show.return_value = test_container_obj
response = self.get('/v1/containers/%s/?all_tenants=1' %
test_container['uuid'])
mock_container_get_by_uuid.assert_called_once_with(
mock.ANY,
test_container['uuid'])
context = mock_container_get_by_uuid.call_args[0][0]
self.assertIs(True, context.all_tenants)
self.assertEqual(200, response.status_int)
self.assertEqual(test_container['uuid'],
response.json['uuid'])
def _action_test(self, container, action, ident_field,
mock_container_action, status_code, query_param=''):
test_container_obj = objects.Container(self.context, **container)
ident = container.get(ident_field)
get_by_ident_loc = 'zun.objects.Container.get_by_%s' % ident_field
with patch(get_by_ident_loc) as mock_get_by_indent:
mock_get_by_indent.return_value = test_container_obj
response = self.post('/v1/containers/%s/%s/?%s' %
(ident, action, query_param))
self.assertEqual(status_code, response.status_int)
# Only PUT should work, others like GET should fail
self.assertRaises(AppError, self.get,
('/v1/containers/%s/%s/' %
(ident, action)))
if query_param:
value = query_param.split('=')[1]
mock_container_action.assert_called_once_with(
mock.ANY, test_container_obj, value)
else:
mock_container_action.assert_called_once_with(
mock.ANY, test_container_obj)
def test_delete_container_by_uuid_all_tenants(self, mock_get_by_uuid,
mock_container_delete,
mock_validate, mock_policy):
mock_policy.return_value = True
test_container = utils.get_test_container()
test_container_obj = objects.Container(self.context, **test_container)
mock_get_by_uuid.return_value = test_container_obj
container_uuid = test_container.get('uuid')
response = self.delete('/v1/containers/%s/?all_tenants=1' %
container_uuid)
self.assertEqual(204, response.status_int)
mock_container_delete.assert_called_once_with(
mock.ANY, test_container_obj, False)
context = mock_container_delete.call_args[0][0]
self.assertIs(True, context.all_tenants)
def test_kill_container_by_uuid(self,
mock_get_by_uuid, mock_container_kill,
mock_validate):
test_container_obj = objects.Container(self.context,
**utils.get_test_container())
mock_container_kill.return_value = test_container_obj
test_container = utils.get_test_container()
test_container_obj = objects.Container(self.context, **test_container)
mock_get_by_uuid.return_value = test_container_obj
container_uuid = test_container.get('uuid')
url = '/v1/containers/%s/%s/' % (container_uuid, 'kill')
cmd = {'signal': '9'}
response = self.post(url, cmd)
self.assertEqual(202, response.status_int)
mock_container_kill.assert_called_once_with(
mock.ANY, test_container_obj, cmd['signal'])
def test_resize_container_by_uuid(self,
mock_get_by_uuid,
mock_container_resize,
mock_validate):
test_container_obj = objects.Container(self.context,
**utils.get_test_container())
mock_container_resize.return_value = test_container_obj
test_container = utils.get_test_container()
test_container_obj = objects.Container(self.context, **test_container)
mock_get_by_uuid.return_value = test_container_obj
container_name = test_container.get('name')
url = '/v1/containers/%s/%s/' % (container_name, 'resize')
cmd = {'h': '100', 'w': '100'}
response = self.post(url, cmd)
self.assertEqual(200, response.status_int)
mock_container_resize.assert_called_once_with(
mock.ANY, test_container_obj, cmd['h'], cmd['w'])
def test_get_archive_by_uuid(self,
mock_get_by_uuid,
container_get_archive,
mock_validate):
container_get_archive.return_value = ("", "")
test_container = utils.get_test_container()
test_container_obj = objects.Container(self.context, **test_container)
mock_get_by_uuid.return_value = test_container_obj
container_uuid = test_container.get('uuid')
url = '/v1/containers/%s/%s/' % (container_uuid, 'get_archive')
cmd = {'path': '/home/1.txt'}
response = self.get(url, cmd)
self.assertEqual(200, response.status_int)
container_get_archive.assert_called_once_with(
mock.ANY, test_container_obj, cmd['path'])
