Verified Commit 1a830cca authored by Johannes Visintini's avatar Johannes Visintini
Browse files

implement getting requests from a json file

parent 239f9a1a
# Check ohsome API
This small script checks the given ohsome API and performs an request for every endpoint present in the swagger files.
This small script checks the given ohsome API and performs an request for every endpoint present in the swagger files or requests given in a json file.
## Installation
......@@ -9,6 +9,7 @@ This small script checks the given ohsome API and performs an request for every
```
## Usage
Check all endpoints with default values read from the swagger files:
```bash
% python3 check-ohsome-api.py v1
Getting Swagger jsons
......@@ -18,3 +19,35 @@ Result:
Status Amount
200 100.00 %
```
Check requests given in `example-requests.json`:
```bash
% python3 check-ohsome-api.py v1 -r example-requests.json
Running requests
POST 400: https://api.ohsome.org/v1/elements/count
{'time': '2010-01-01/2016-08-01/P2Y2M2D', 'filter': 'highway=residential and type:way', 'note': 'bboxes missing on purpose'}
100%|████████████████████████████████████████████████████████████████| 3/3 [00:02<00:00, 1.19it/s]
Result:
Status Amount
200 66.67 %
400 33.33 %
```
Usage:
```bash
% python3 check-ohsome-api.py -h
usage: check-ohsome-api.py [-h] [-b BASE_URL] [-r REQUESTS_JSON] [-d] api_version
test ohsome API based on available paths in swagger files
positional arguments:
api_version api version path to be tested
optional arguments:
-h, --help show this help message and exit
-b BASE_URL, --base-url BASE_URL
base url
-r REQUESTS_JSON, --requests-json REQUESTS_JSON
use the requests in this file instead of getting them from swagger files
-d, --debug show debug output
```
......@@ -5,55 +5,81 @@ import requests
from collections import Counter
from tqdm import tqdm
parser = argparse.ArgumentParser(description='test ohsome API based on available paths in swagger files')
parser.add_argument('api_version', type=str, help='api version path to be tested')
parser.add_argument('--base-url', type=str, default='https://api.ohsome.org', help='base url')
parser.add_argument('--debug', action='store_true', help='show debug output')
args = parser.parse_args()
BASE_URL = "{}/{}".format(args.base_url, args.api_version)
specs = ["Data Aggregation", "Data Extraction", "Metadata"]
test_requests = []
print("Getting Swagger jsons")
for spec in specs:
swagger_url = "{}/docs?group={}".format(BASE_URL, spec)
if args.debug:
print("Getting {} Swagger json".format(spec))
print("Requesting {}".format(swagger_url))
req = requests.get(swagger_url)
j = json.loads(req.text)
for path in j['paths']:
url = BASE_URL + path
for method in j['paths'][path]:
request_parameters = {}
if 'parameters' in j['paths'][path][method].keys():
for parameter in j['paths'][path][method]['parameters']:
if 'default' in parameter.keys():
request_parameters[parameter['name']] = parameter['default']
test_requests.append({'method': method, 'url': url, 'parameters': request_parameters})
status_codes = []
print("Running requests")
t = tqdm(test_requests)
for request in t:
t.set_description("{} {}".format(request['method'].upper(), request['url']))
if request['method'] == 'get':
r = requests.get(request['url'], params=request['parameters'])
def get_requests_from_swagger_jsons(base_url, specs, args):
test_requests = []
print("Getting Swagger jsons")
for spec in specs:
swagger_url = "{}/docs?group={}".format(base_url, spec)
if args.debug:
print("Getting {} Swagger json".format(spec))
print("Requesting {}".format(swagger_url))
req = requests.get(swagger_url)
j = json.loads(req.text)
for path in j['paths']:
url = base_url + path
for method in j['paths'][path]:
request_parameters = {}
if 'parameters' in j['paths'][path][method].keys():
for parameter in j['paths'][path][method]['parameters']:
if 'default' in parameter.keys():
request_parameters[parameter['name']] = parameter['default']
test_requests.append({'method': method.upper(), 'url': url, 'parameters': request_parameters})
return test_requests
def load_request_json(base_url, args):
test_requests = []
with open(args.requests_json) as fo:
j = json.load(fo)
for req in j['requests']:
method = req['method'].upper()
url = base_url + req['path']
parameters = req['parameters'] if 'parameters' in req.keys() else {}
test_requests.append({'method': method, 'url': url, 'parameters': parameters})
return test_requests
def perform_requests(test_requests, args):
status_codes = []
print("Running requests")
t = tqdm(test_requests)
for request in t:
t.set_description("{} {}".format(request['method'], request['url']))
if request['method'] == 'GET':
r = requests.get(request['url'], params=request['parameters'])
else:
r = requests.post(request['url'], data=request['parameters'])
status_codes.append(r.status_code)
if r.status_code != 200 or args.debug:
tqdm.write('{} {}: {}'.format(request['method'], r.status_code, r.url))
if request['method'] == 'POST':
tqdm.write(' → {}'.format(request['parameters']))
t.set_description('')
return status_codes
def print_results(status_codes, args):
print("Result:")
print("Status Amount")
for value, count in sorted(Counter(status_codes).items()):
print("{:<6} {:6.2f} %".format(value, count / len(status_codes) * 100))
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='test ohsome API based on available paths in swagger files')
parser.add_argument('api_version', type=str, help='api version path to be tested')
parser.add_argument('-b', '--base-url', type=str, default='https://api.ohsome.org', help='base url')
parser.add_argument('-r', '--requests-json', type=str, help='use the requests in this file instead of getting them from swagger files')
parser.add_argument('-d', '--debug', action='store_true', help='show debug output')
args = parser.parse_args()
base_url = "{}/{}".format(args.base_url, args.api_version)
specs = ["Data Aggregation", "Data Extraction", "Metadata"]
if args.requests_json:
test_requests = load_request_json(base_url, args)
else:
r = requests.post(request['url'], data=request['parameters'])
status_codes.append(r.status_code)
if r.status_code != 200 or args.debug:
tqdm.write('{} {}: {}'.format(request['method'].upper(), r.status_code, r.url))
if request['method'] == 'post':
tqdm.write(' → {}'.format(request['parameters']))
t.set_description('')
print("Result:")
print("Status Amount")
for value, count in sorted(Counter(status_codes).items()):
print("{:<6} {:6.2f} %".format(value, count / len(status_codes) * 100))
test_requests = get_requests_from_swagger_jsons(base_url, specs, args)
status_codes = perform_requests(test_requests, args)
print_results(status_codes, args)
{
"requests": [
{
"method": "POST",
"path": "/elements/count",
"parameters": {
"time": "2010-01-01/2016-08-01/P2Y2M2D",
"filter": "highway=residential and type:way",
"note": "bboxes missing on purpose"
}
},
{
"method": "POST",
"path": "/elements/count",
"parameters": {
"bboxes": "8.6128,49.3183,8.7294,49.4376",
"time": "2010-01-01/2016-08-01/P2Y2M2D",
"filter": "highway=residential and type:way"
}
},
{
"method": "GET",
"path": "/metadata"
}
]
}
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment