python Implementation Interface Concurrency Test Script

Commonly used website performance test indicators are: concurrency, response time, throughput, performance counters and so on.

1. Concurrent number

Concurrent number refers to the number of requests that the system can handle at the same time, which also reflects the load capacity of the system.

2. Response time

Response time is one of the most important indexes of a system, and its numerical value directly reflects the speed of the system. Response time is the total time it takes to execute a request from the beginning until the response data is finally received.

3. Throughput

Throughput refers to the number of requests that the system can handle per unit time, reflecting the ability of the system to handle requests, which is the most commonly used performance test index at present.
QPS (queries per second), TPS (transactions per second) are common quantifications of throughput, and HPS (HTTP requests per second).
Several important things related to throughput are concurrency and response time.
The relationship between QPS (TPS), concurrency number and response time is as follows:
QPS (TPS) = Concurrency/Average Response Time

4. Performance counters

Performance counter is a data index describing the performance of server or operating system, such as the number of memory used and process time. It plays the role of "monitoring and analyzing" in performance test, especially plays a very key role in analyzing all scalability and locating new energy bottlenecks.
In Linux, you can use top or uptime commands to see the current system load and resource utilization.
Resource utilization rate: refers to the use of various resources in the system, such as cpu occupancy rate of 68%, memory occupancy rate of 55%, and "actual use of resources/total available amount of resources" is used to form resource utilization rate.

Pressure measurement script (interface for placing orders):

#!/usr/bin/env python
#-*- coding:utf-8 -*-

import requests,time,json,threading,random

class Presstest(object):
  headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.121 Safari/537.36',
    'Content-Type': 'application/json; charset=UTF-8',
  }
  def __init__(self,login_url,press_url,phone="1376193000",password="123456"):
    self.login_url = login_url
    self.press_url = press_url
    self.phone = phone
    self.password = password
    self.session = requests.Session()
    self.session.headers = self.headers

  def login(self):
    ''' Login acquisition session'''
    data = data = {'t': int(time.time() * 1000), 'userName': self.phone, 'passWord': self.password}
    res = self.session.post(self.login_url,data=json.dumps(data))
    XToken = res.json().get('data').get('companyToken')
    self.session.headers['X-Token'] = XToken

  def testinterface(self):
    ''' Pressure measurement interface '''
    self.session.headers['X-UnionId'] = 'of6uw1CUVhP533sQok'
    data = {"id": int(''.join(str(random.choice(range(10))) for _ in range(10))),
        "openId": "oMr0c5LGJjlTc", "addressId": 10, "shipType": "SELF", "totalAmount": 5,
        "receivable": 5, "carts": [
        {"amount": 1, "barcode": "1234567890", "skuId": 1, "spec": "34", "itemAmount": 5, "price": 0,
         "cover": "xxxx-dd.oss-cn-shanghai.aliyuncs.com/dfc91fd067ac464c096c90af33a196a5.png",
         "name": " Sassoon shampoo ", "packingType": " Bottle ", "placeOfOrigin": " Shanghai ", "productId": "310153323435134976",
         "retailPrice": 5, "suitableAge": "1-100"}], "formId": "the formId is a mock one", "comments": "aa"}
    global ERROR_NUM
    try:
      html = self.session.post(self.press_url, data=json.dumps(data))
      if html.json().get('code') != 0:
        print(html.json())
        ERROR_NUM += 1
    except Exception as e:
      print(e)
      ERROR_NUM += 1

  def testonework(self):
    '''1 Second concurrent processing of a single task '''
    i = 0
    while i < ONE_WORKER_NUM:
      i += 1
      self.work()
    time.sleep(LOOP_SLEEP)

  def run(self):
    ''' Concurrent testing using multithreaded processes '''
    t1 = time.time()
    Threads = []

    for i in range(THREAD_NUM):
      t = threading.Thread(target=self.testonework, name="T" + str(i))
      t.setDaemon(True)
      Threads.append(t)

    for t in Threads:
      t.start()
    for t in Threads:
      t.join()
    t2 = time.time()

    print("=============== Pressure measurement result ===================")
    print("URL:", self.press_url)
    print(" Number of tasks :", THREAD_NUM, "*", ONE_WORKER_NUM, "=", THREAD_NUM * ONE_WORKER_NUM)
    print(" Total time spent ( Seconds ):", t2 - t1)
    print(" Time consuming per request ( Seconds ):", (t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM))
    print(" Number of bearer requests per second :", 1 / ((t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM)))
    print(" Number of errors :", ERROR_NUM)


if __name__ == '__main__':
  login_url = 'https://ds.xxxxx.com/sys/sysUser/login'
  press_url = 'https://ds.xxxxx.com/weshop/order/checkout'
  phone = "1376193000"
  password = "123456"
  
  THREAD_NUM = 1     #  Total number of concurrent threads 
  ONE_WORKER_NUM = 5   #  Number of cycles per thread 
  LOOP_SLEEP = 0.1    #  Time interval per request ( Seconds )
  ERROR_NUM = 0      #  Number of errors 
  
  obj = Presstest(login_url=login_url,press_url=press_url,phone=phone,password=password)
  obj.login()
  obj.run()

Output:

=============== Pressure measurement result ===================
URL: https://ds.xxxxx.com/weshop/order/checkout
 Number of tasks : 1 * 5 = 5
 Total time spent ( Seconds ): 1.9810078144073486
 Time consuming per request ( Seconds ): 0.39620156288146974
 Number of bearer requests per second : 2.5239678327547805
 Number of errors : 0