peterjpxie

import requests

url = "http://httpbin.org/post"

payload = "{\n    \"key1\": 1,\n    \"key2\": \"value2\"\n}"
headers = {
    'Content-Type': "application/json,text/plain",
    'User-Agent': "PostmanRuntime/7.15.0",
    'Accept': "*/*",
    'Cache-Control': "no-cache",

peterjpxie

import requests
import json

def test_post_headers_body_json():
    url = 'https://httpbin.org/post'
    # Additional headers.
    headers = {'Content-Type': 'application/json' } 
    # Body
    payload = {'key1': 1, 'key2': 'value2'}
    # convert dict to json string by json.dumps() for body data. 

peterjpxie

# pytest -sv
test_post_headers_body_json_pprint.py::test_post_headers_body_json
-----------Request----------->
POST https://httpbin.org/post

User-Agent: python-requests/2.22.0
Accept-Encoding: gzip, deflate
Accept: */*
Connection: keep-alive
Content-Length: 39

peterjpxie

import requests
import json

def pretty_print_request(request):
    print( '\n{}\n{}\n\n{}\n\n{}\n'.format(
        '-----------Request----------->',
        request.method + ' ' + request.url,
        '\n'.join('{}: {}'.format(k, v) for k, v in request.headers.items()),
        request.body)
    )

peterjpxie

from flask import Flask

app = Flask(__name__)

@app.route('/json', methods=['POST', 'GET'])
def test_json():
    return '{"code": 1, "message": "Hello, World!" }'
    
# Run in HTTP
app.run(host='127.0.0.1', port='5000')  

peterjpxie

import requests

def test_mock_service():
    url = 'http://127.0.0.1:5000/json'    
    resp = requests.get(url)           
    assert resp.status_code == 200
    assert resp.json()["code"] == 1
    print(resp.text)

peterjpxie

from flask import Flask
import time

app = Flask(__name__)

@app.route('/json', methods=['POST', 'GET'])
def test_json():
    time.sleep(0.2) # simulate delay
    return '{"code": 1, "message": "Hello, World!" }'
    

peterjpxie

# Updated test function in perf_test_mock_service.py
def test_mock_service():
    url = 'http://127.0.0.1:5000/json'    
    resp = requests.get(url)
    if resp.status_code != 200:
        print('Test failed with response status code %s.' % resp.status_code )
        return 'fail', resp.elapsed.total_seconds()
    elif resp.json()["code"] != 1:
        print('Test failed with code %s != 1.' %  resp.json()["code"] )
        return 'fail', resp.elapsed.total_seconds()

peterjpxie

import requests
import queue
import sys
import time

queue_results = queue.Queue()
# def test_mock_service():  - omitted 

def loop_test(loop_wait=0, loop_times=sys.maxsize):
    looped_times = 0    

peterjpxie

import requests
import threading
import queue
import sys
import time

# Global variables
queue_results = queue.Queue()
start_time = 0