test_file.py 文件源码

def test_token_refresh_stream_body(self):
        expiration = (datetime.datetime.utcnow() +
                      datetime.timedelta(minutes=15))
        credentials = self._create_test_credentials(expiration=expiration)

        storage = file_module.Storage(FILENAME)
        storage.put(credentials)
        credentials = storage.get()
        new_cred = copy.copy(credentials)
        new_cred.access_token = 'bar'
        storage.put(new_cred)

        valid_access_token = '1/3w'
        token_response = {'access_token': valid_access_token,
                          'expires_in': 3600}
        http = http_mock.HttpMockSequence([
            ({'status': http_client.UNAUTHORIZED},
             b'Initial token expired'),
            ({'status': http_client.UNAUTHORIZED},
             b'Store token expired'),
            ({'status': http_client.OK},
             json.dumps(token_response).encode('utf-8')),
            ({'status': http_client.OK}, 'echo_request_body')
        ])

        body = six.StringIO('streaming body')

        credentials.authorize(http)
        _, content = transport.request(http, 'https://example.com', body=body)
        self.assertEqual(content, 'streaming body')
        self.assertEqual(credentials.access_token, valid_access_token)