grequests并发之小试牛刀

第一章 使用背景

题主为某供应链金融公司的测试人员，系统有额度开单、链单转让、链单融资功能，这些功能点在用户串行请求情况下均不会出现金额溢出（如额度 < 所有使用该额度开出的链单金额总和）。目前的测试需求是，测试用户并发情况下，是否出现金额溢出。
python的requests模块只能实现用户串行请求，那么有没有满足我的测试需求的第三方模块呢？
在网络上搜索一番，的确已有，就是grequests模块！传为K神(Kenneth Reitz)所作，结合requests、gevent模块实现，Python友対之赞誉有加。

第二章 小试牛刀

2.1 了解功法

使用python命令pip install grequests安装：

pip install grequests

查看已安装的grequests模块：

Name: grequests
Version: 0.6.0
Summary: Requests + Gevent
Home-page: https://github.com/kennethreitz/grequests
Author: Kenneth Reitz
Author-email: [e-mail]
License: BSD
Location: d:\python3\lib\site-packages
Requires: gevent, requests
Required-by:

根据已经尝鲜的朋友介绍，我要解决的测试需求需要用到grequests.request、grequests.map方法，当然grequests也提供grequests.get、grequests.post等，大家可根据喜好使用。
使用前，先使用help查看这2个函数的内功心法。可以得知，grequests.request和requests.request的修炼功法完全一致，而且是异步的，请看：

Help on function request in module grequests:

request(method, url, **kwargs)
    # synonym

grequests.map的修炼功法呢？请看：

Help on function map in module grequests:

map(requests, stream=False, size=None, exception_handler=None, gtimeout=None)
    Concurrently converts a list of Requests to Responses.
    
    :param requests: a collection of Request objects.
    :param stream: If True, the content will not be downloaded immediately.
    :param size: Specifies the number of requests to make at a time. If None, no throttling occurs.
    :param exception_handler: Callback function, called when exception occured. Params: Request, Exception
    :param gtimeout: Gevent joinall timeout in seconds. (Note: unrelated to requests timeout)

2.2 解决需求

先测试并发额度开单，代码组织如下：
1.初始化一个空的请求数组；
2.for循环下，根据接口定义，使用同一额度并全额开单（额度全部使用完），使用grequests.request产生异步请求对象（grequests.AsyncRequest object），并加入到请求数组；
3.使用grequests.map产生并发请求，并把所有响应对象赋值给变量result；
4.获取result中所有响应成功的标志（如success），预期响应成功的数目等于1，即只有一个全额开单成功。
核心代码如下：

import pytest
import grequests

class TestConcurrentOutput():
	"""并发开单测试"""

	def test_output_chains_concurrently(self, user_login, get_host):
		host = 'http://127.0.0.1:8000/api'
		url = host +'/bill/output/apply'
		method = 'POST'
		params = {
			'headers': settings.headers['json'],
			'cookies': user_login['bn_hxqy']
			'data': json.dumps(payload)
		}
		try:
			# 步骤1:定义任务列表task
			task = list()
			# 步骤2:构建请求，加入任务列表task
			for k in range(0, 3):
				req = grequests.request(method, url, **params)
				task.append(req)
			# 步骤3:产生并发请求，并把所有响应对象赋值给result
			result = grequests.map(task)
			# 打印响应内容
			print(result[0].json(), result[1].json(), result[2].json())
			# 步骤4:收集所有响应的"是否成功"，检查所有响应中只有一个成功
			success_info = [item.json()['success'] for item in result]
			assert success_info.count(True) == 1
		except Exception as e:
			raise e

if __name__ == '__main__':
	pytest.main(['-s', '--cache-clear', __file__])

第三章 再探究竟

下期再说，拜拜！