def test_auth(self):
        token_client_v3 = token_client.V3TokenClient('fake_url')
        response = fake_http.fake_http_response(
            None, status=201,
        )
        body = {'access': {'token': 'fake_token'}}

        with mock.patch.object(token_client_v3, 'post') as post_mock:
            post_mock.return_value = response, body
            resp = token_client_v3.auth(username='fake_user',
                                        password='fake_pass')

        self.assertIsInstance(resp, rest_client.ResponseBody)
        req_dict = json.dumps({
            'auth': {
                'identity': {
                    'methods': ['password'],
                    'password': {
                        'user': {
                            'name': 'fake_user',
                            'password': 'fake_pass',
                        }
                    }
                },
            }
        }, sort_keys=True)
        post_mock.assert_called_once_with('fake_url/auth/tokens',
                                          body=req_dict)

def _fake_v3_response(self, uri, method="GET", body=None, headers=None,
                      redirections=5, connection_type=None):
    fake_headers = {
        "x-subject-token": TOKEN
    }
    return (fake_http.fake_http_response(fake_headers, status=201),
            json.dumps(IDENTITY_V3_RESPONSE))

 def raw_request(*args, **kwargs):
     self.assertIn("X-OpenStack-Nova-API-Version", kwargs["headers"])
     self.assertEqual("2.2", kwargs["headers"]["X-OpenStack-Nova-API-Version"])
     return (
         fake_http.fake_http_response(headers={self.client.api_microversion_header_name: "2.2"}, status=200),
         "",
     )

 def test_no_microverion_header_in_response(self, mock_request):
     response = fake_http.fake_http_response(
         headers={},
     )
     mock_request.return_value = response, ''
     self.assertRaises(exceptions.InvalidHTTPResponseHeader,
                       self.client.get, 'fake_url')

 def raw_request(*args, **kwargs):
     self.assertIn('X-OpenStack-Nova-API-Version', kwargs['headers'])
     self.assertEqual('2.2',
                      kwargs['headers']['X-OpenStack-Nova-API-Version'])
     return (fake_http.fake_http_response(
         headers={self.client.api_microversion_header_name: '2.2'},
         status=200), '')

 def create_response(self, body, to_utf=False, status=200, headers=None):
     json_body = {}
     if body:
         json_body = json.dumps(body)
         if to_utf:
             json_body = json_body.encode('utf-8')
     resp = fake_http.fake_http_response(headers, status=status), json_body
     return resp

 def test_date_time_or_null_format(self):
     instance = None
     resp = fake_http.fake_http_response('', status=200)
     body = {'date-time': instance}
     rest_client.RestClient.validate_response(self.date_time_schema[1],
                                              resp, body)
     self.assertRaises(exceptions.InvalidHTTPResponseBody,
                       rest_client.RestClient.validate_response,
                       self.date_time_schema[0], resp, body)

 def test_valid_date_time_format(self):
     valid_instances = ['2016-10-02T10:00:00-05:00',
                        '2016-10-02T10:00:00+09:00',
                        '2016-10-02T15:00:00Z',
                        '2016-10-02T15:00:00.05Z']
     resp = fake_http.fake_http_response('', status=200)
     for instance in valid_instances:
         body = {'date-time': instance}
         for schema in self.date_time_schema:
             rest_client.RestClient.validate_response(schema, resp, body)

def _fake_auth_failure_response():
    # the response body isn't really used in this case, but lets send it anyway
    # to have a safe check in some future change on the rest client.
    body = {
        "unauthorized": {
            "message": "Unauthorized",
            "code": "401"
        }
    }
    return fake_http.fake_http_response({}, status=401), json.dumps(body)

 def _test_is_resource_deleted(self, flavor_id, is_deleted=True,
                               bytes_body=False):
     body = json.dumps({'flavors': [TestFlavorsClient.FAKE_FLAVOR]})
     if bytes_body:
         body = body.encode('utf-8')
     response = fake_http.fake_http_response({}, status=200), body
     self.useFixture(fixtures.MockPatch(
         'tempest.lib.common.rest_client.RestClient.get',
         return_value=response))
     self.assertEqual(is_deleted,
                      self.client.is_resource_deleted(flavor_id))

    def test_request_with_str_body(self):
        token_client_v2 = token_client.TokenClient('fake_url')
        response = fake_http.fake_http_response(
            None, status=200,
        )
        body = str('{"access": {"token": "fake_token"}}')

        with mock.patch.object(token_client_v2, 'raw_request') as mock_raw_r:
            mock_raw_r.return_value = response, body
            resp, body = token_client_v2.request('GET', 'fake_uri')
        self.assertIsInstance(body, dict)

    def test_create_request_token(self):
        mock_resp = self._mock_token_response(self.FAKE_CREATE_REQUEST_TOKEN)
        resp = fake_http.fake_http_response(None, status=201), mock_resp
        self.useFixture(fixtures.MockPatch(
            'tempest.lib.common.rest_client.RestClient.post',
            return_value=resp))

        resp = self.client.create_request_token(
            consumer_key='12345',
            consumer_secret='23456',
            project_id='c8f58432c6f00162f04d3250f')
        self.assertEqual(self.FAKE_CREATE_REQUEST_TOKEN, resp)

    def test_update_resource(self, mock_put):
        response = fake_http.fake_http_response(headers=None, status=200)
        mock_put.return_value = response, '{"baz": "qux"}'

        put_data = {'foo': 'bar'}
        resp = self.client.update_resource('/fake_url', put_data)

        self.assertEqual({'status': '200'}, resp.response)
        self.assertEqual("qux", resp["baz"])
        mock_put.assert_called_once_with('v2.0/fake_url', '{"foo": "bar"}')
        self.mock_expected_success.assert_called_once_with(
            200, 200)

    def test_update_resource_expect_different_values(self, mock_put):
        response = fake_http.fake_http_response(headers=None, status=201)
        mock_put.return_value = response, '{}'

        put_data = {'foo': 'bar'}
        resp = self.client.update_resource('/fake_url', put_data,
                                           expect_response_code=201,
                                           expect_empty_body=True)

        self.assertEqual({'status': '201'}, resp.response)
        self._assert_empty(resp)
        mock_put.assert_called_once_with('v2.0/fake_url', '{"foo": "bar"}')
        self.mock_expected_success.assert_called_once_with(
            201, 201)

    def test_create_access_token(self):
        mock_resp = self._mock_token_response(self.FAKE_CREATE_ACCESS_TOKEN)
        req_secret = self.FAKE_CREATE_REQUEST_TOKEN['oauth_token_secret']
        resp = fake_http.fake_http_response(None, status=201), mock_resp
        self.useFixture(fixtures.MockPatch(
            'tempest.lib.common.rest_client.RestClient.post',
            return_value=resp))

        resp = self.client.create_access_token(
            consumer_key='12345',
            consumer_secret='23456',
            request_key=self.FAKE_CREATE_REQUEST_TOKEN['oauth_token'],
            request_secret=req_secret,
            oauth_verifier='8171')
        self.assertEqual(self.FAKE_CREATE_ACCESS_TOKEN, resp)

    def set_data(self, r_code, enc=None, r_body=None, absolute_limit=True):
        if enc is None:
            enc = self.c_type
        resp_dict = {'status': r_code, 'content-type': enc}
        resp_body = {'resp_body': 'fake_resp_body'}

        if absolute_limit is False:
            resp_dict.update({'retry-after': 120})
            resp_body.update({'overLimit': {'message': 'fake_message'}})
        resp = fake_http.fake_http_response(headers=resp_dict,
                                            status=int(r_code),
                                            body=json.dumps(resp_body))
        data = {
            "resp": resp,
            "resp_body": json.dumps(resp_body)
        }
        if r_body is not None:
            data.update({"resp_body": r_body})
        return data

 def test_invalid_date_time_format(self):
     invalid_instances = ['2016-10-02 T10:00:00-05:00',
                          '2016-10-02T 15:00:00',
                          '2016-10-02T15:00:00.05 Z',
                          '2016-10-02:15:00:00.05Z',
                          'T15:00:00.05Z',
                          '2016:10:02T15:00:00',
                          '2016-10-02T15-00-00',
                          '2016-10-02T15.05Z',
                          '09MAR2015 11:15',
                          '13 Oct 2015 05:55:36 GMT',
                          '']
     resp = fake_http.fake_http_response('', status=200)
     for instance in invalid_instances:
         body = {'date-time': instance}
         for schema in self.date_time_schema:
             self.assertRaises(exceptions.InvalidHTTPResponseBody,
                               rest_client.RestClient.validate_response,
                               schema, resp, body)

    def test_auth_with_project_id_and_domain_id(self):
        token_client_v3 = token_client.V3TokenClient('fake_url')
        response = fake_http.fake_http_response(
            None, status=201,
        )
        body = {'access': {'token': 'fake_token'}}

        with mock.patch.object(token_client_v3, 'post') as post_mock:
            post_mock.return_value = response, body
            resp = token_client_v3.auth(
                username='fake_user', password='fake_pass',
                project_id='fcac2a055a294e4c82d0a9c21c620eb4',
                user_domain_id='14f4a9a99973404d8c20ba1d2af163ff',
                project_domain_id='291f63ae9ac54ee292ca09e5f72d9676')

        self.assertIsInstance(resp, rest_client.ResponseBody)
        req_dict = json.dumps({
            'auth': {
                'identity': {
                    'methods': ['password'],
                    'password': {
                        'user': {
                            'name': 'fake_user',
                            'password': 'fake_pass',
                            'domain': {
                                'id': '14f4a9a99973404d8c20ba1d2af163ff'
                            }
                        }
                    }
                },
                'scope': {
                    'project': {
                        'id': 'fcac2a055a294e4c82d0a9c21c620eb4',
                        'domain': {
                            'id': '291f63ae9ac54ee292ca09e5f72d9676'
                        }
                    }
                }
            }
        }, sort_keys=True)
        post_mock.assert_called_once_with('fake_url/auth/tokens',
                                          body=req_dict)

    def test_auth(self):
        token_client_v2 = token_client.TokenClient('fake_url')
        response = fake_http.fake_http_response(
            None, status=200,
        )
        body = {'access': {'token': 'fake_token'}}

        with mock.patch.object(token_client_v2, 'post') as post_mock:
            post_mock.return_value = response, body
            resp = token_client_v2.auth('fake_user', 'fake_pass')

        self.assertIsInstance(resp, rest_client.ResponseBody)
        req_dict = json.dumps({
            'auth': {
                'passwordCredentials': {
                    'username': 'fake_user',
                    'password': 'fake_pass',
                },
            }
        }, sort_keys=True)
        post_mock.assert_called_once_with('fake_url/tokens',
                                          body=req_dict)

    def test_network_cleanup(self, MockRestClient):
        def side_effect(**args):
            return {"security_groups": [{"tenant_id": args['tenant_id'],
                                         "name": args['name'],
                                         "description": args['name'],
                                         "security_group_rules": [],
                                         "id": "sg-%s" % args['tenant_id']}]}
        creds = dynamic_creds.DynamicCredentialProvider(
            neutron_available=True,
            project_network_cidr='10.100.0.0/16', project_network_mask_bits=28,
            **self.fixed_params)
        # Create primary tenant and network
        self._mock_assign_user_role()
        self._mock_list_role()
        self._mock_user_create('1234', 'fake_prim_user')
        self._mock_tenant_create('1234', 'fake_prim_tenant')
        self._mock_network_create(creds, '1234', 'fake_net')
        self._mock_subnet_create(creds, '1234', 'fake_subnet')
        self._mock_router_create('1234', 'fake_router')
        router_interface_mock = self.patch(
            'tempest.lib.services.network.routers_client.RoutersClient.'
            'add_router_interface')
        creds.get_primary_creds()
        router_interface_mock.assert_called_once_with('1234', subnet_id='1234')
        router_interface_mock.reset_mock()
        # Create alternate tenant and network
        self._mock_user_create('12345', 'fake_alt_user')
        self._mock_tenant_create('12345', 'fake_alt_tenant')
        self._mock_network_create(creds, '12345', 'fake_alt_net')
        self._mock_subnet_create(creds, '12345', 'fake_alt_subnet')
        self._mock_router_create('12345', 'fake_alt_router')
        creds.get_alt_creds()
        router_interface_mock.assert_called_once_with('12345',
                                                      subnet_id='12345')
        router_interface_mock.reset_mock()
        # Create admin tenant and networks
        self._mock_user_create('123456', 'fake_admin_user')
        self._mock_tenant_create('123456', 'fake_admin_tenant')
        self._mock_network_create(creds, '123456', 'fake_admin_net')
        self._mock_subnet_create(creds, '123456', 'fake_admin_subnet')
        self._mock_router_create('123456', 'fake_admin_router')
        self._mock_list_roles('123456', 'admin')
        creds.get_admin_creds()
        self.patchobject(self.users_client.UsersClient, 'delete_user')
        self.patchobject(self.tenants_client_class, self.delete_tenant)
        net = mock.patch.object(creds.networks_admin_client, 'delete_network')
        net_mock = net.start()
        subnet = mock.patch.object(creds.subnets_admin_client, 'delete_subnet')
        subnet_mock = subnet.start()
        router = mock.patch.object(creds.routers_admin_client, 'delete_router')
        router_mock = router.start()
        remove_router_interface_mock = self.patch(
            'tempest.lib.services.network.routers_client.RoutersClient.'
            'remove_router_interface')
        return_values = ({'status': 200}, {'ports': []})
        port_list_mock = mock.patch.object(creds.ports_admin_client,
                                           'list_ports',
                                           return_value=return_values)

        port_list_mock.start()
        secgroup_list_mock = mock.patch.object(
            creds.security_groups_admin_client,
            'list_security_groups',
            side_effect=side_effect)
        secgroup_list_mock.start()

        return_values = fake_http.fake_http_response({}, status=204), ''
        remove_secgroup_mock = self.patch(
            'tempest.lib.services.network.security_groups_client.'
            'SecurityGroupsClient.delete', return_value=return_values)
        creds.clear_creds()
        # Verify default security group delete
        calls = remove_secgroup_mock.mock_calls
        self.assertEqual(len(calls), 3)
        args = map(lambda x: x[1][0], calls)
        args = list(args)
        self.assertIn('v2.0/security-groups/sg-1234', args)
        self.assertIn('v2.0/security-groups/sg-12345', args)
        self.assertIn('v2.0/security-groups/sg-123456', args)
        # Verify remove router interface calls
        calls = remove_router_interface_mock.mock_calls
        self.assertEqual(len(calls), 3)
        args = map(lambda x: (x[1][0], x[2]), calls)
        args = list(args)
        self.assertIn(('1234', {'subnet_id': '1234'}), args)
        self.assertIn(('12345', {'subnet_id': '12345'}), args)
        self.assertIn(('123456', {'subnet_id': '123456'}), args)
        # Verify network delete calls
        calls = net_mock.mock_calls
        self.assertEqual(len(calls), 3)
        args = map(lambda x: x[1][0], calls)
        args = list(args)
        self.assertIn('1234', args)
        self.assertIn('12345', args)
        self.assertIn('123456', args)
        # Verify subnet delete calls
        calls = subnet_mock.mock_calls
        self.assertEqual(len(calls), 3)
        args = map(lambda x: x[1][0], calls)
        args = list(args)
#.........这里部分代码省略.........