-
Notifications
You must be signed in to change notification settings - Fork 5
/
test-api.py
executable file
·192 lines (172 loc) · 6.43 KB
/
test-api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
#! /usr/bin/env python3.5
import random
import string
from api import API
from endpoint import Endpoint
from restClient import RestClient
from mock_servers.simple_endpoint import EndpointHandler
from application import Application
from User import User
class APITest(object):
def __init__(self, debug=True):
self.debub = debug
self.client = RestClient()
# TODO: Move this to map to make it extensible
self.store_client = RestClient("store")
def populateAPIs(self, count=10):
"""
"apiDefinition": "{ \"paths\": { \"/*\": { \"get\": { \"description\": \"Global Get\" } } }, \"swagger\": \"2.0\" }",
"cacheTimeout": 10,
"transport": ["https"],
"policies": ["Unlimited"],
"visibility": "PUBLIC",
"businessInformation": {},
"corsConfiguration": {}
:param count:
:return:
"""
template = {
"name": None,
"context": None,
"version": "1.0.0",
"service_url": "http://sample.knnect.com/api/endpoint",
"description": "",
"wsdlUri": "http://www.webservicex.com/globalweather.asmx?wsdl",
"isDefaultVersion": True,
"responseCaching": "Disabled",
"cacheTimeout": 300,
"destinationStatsEnabled": "Disabled",
"businessInformation": {
"businessOwner": "Kasun Thennakoon",
"businessOwnerEmail": "[email protected]",
"technicalOwner": "Knnect",
"technicalOwnerEmail": "[email protected]"
},
"tags": [
"test_tag1",
"fintech"
],
"transport": [
"http",
"https"
],
"visibility": "PUBLIC",
"securityScheme": [
"string"
],
"scopes": [
"string"
],
}
policies = ["Gold", "Silver", "Bronze"]
apis = []
for index in range(count):
random_char = random.choice(
string.ascii_lowercase) + random.choice(string.ascii_lowercase)
name_context = "{}_sample_{}".format(random_char, index)
template['name'] = name_context
template['description'] = "A Sample API for testing APIM 3.0 " + name_context
template['context'] = "/" + name_context
api = API(**template)
api.set_rest_client(self.client)
api.save()
api.set_policies(policies)
apis.append(api)
return apis
def populate_users(self, count=10):
for index in range(count):
random_char = random.choice(
string.ascii_lowercase) + random.choice(string.ascii_lowercase)
userName = "{}user{}".format(random_char, index)
user = User(userName)
print("DEBUG: {name} user created with password {name}123".format(
name=user.userName))
user.save()
def publishAPIs(self, apis, state="Published"):
"""
:param apis: List of APIs or single API object returned from API.createAPI method
:param state: State string of the api need to be hange to
:return:
"""
if type(apis) is list:
for api in apis:
api.change_lc(state)
else:
api = apis
api.change_lc(state)
def delete_all(self):
API.delete_all(self.client)
def delete_all_users(self):
User.delete_all()
def delete_all_endpoints(self):
Endpoint.delete_all(self.client)
def delete_all_applications(self):
Application.delete_all(self.store_client)
def populate_global_endpoints(self, count=20):
endpoint_types = ['basic', 'digest']
endpoints = []
for index in range(count):
random_char = random.choice(
string.ascii_lowercase) + random.choice(string.ascii_lowercase)
name = "{}_sample_endpoint_{}".format(random_char, index)
service_url = "https://sample.knnect.com/api/endpoint/" + name
max_tps = random.randint(1, 1000)
endpoint = Endpoint(name, "http", service_url, max_tps)
rand_endpoint = endpoint_types[random.randint(0, 1)]
endpoint.set_security(rand_endpoint, name, "sample_password")
endpoint.set_rest_client(self.client)
endpoint.save()
endpoints.append(endpoint)
return endpoints
def create_application(self, count=2):
"""
"apiDefinition": "{ \"paths\": { \"/*\": { \"get\": { \"description\": \"Global Get\" } } }, \"swagger\": \"2.0\" }",
"cacheTimeout": 10,
"transport": ["https"],
"policies": ["Unlimited"],
"visibility": "PUBLIC",
"businessInformation": {},
"corsConfiguration": {}
:param count:
:return:
"""
template = {
"name": None,
"description": None,
"throttlingTier": "50PerMin"
}
apps = []
for index in range(count):
random_char = random.choice(
string.ascii_lowercase) + random.choice(string.ascii_lowercase)
random_name = "{}_sample_app_{}".format(random_char, index)
template['name'] = random_name
template['description'] = "Sample description for " + random_name
app = Application(**template)
app.set_rest_client(self.store_client)
app.save()
apps.append(app)
return apps
def main():
tester = APITest()
print("INFO: Deleting existing APIs ...")
tester.delete_all()
print("INFO: Deleting existing Endpoints ...")
tester.delete_all_endpoints()
print("INFO: Creating new APIs ...")
apis = tester.populateAPIs(12)
api = API.get(apis[0].id)
print("INFO: Creating new Global endpoints ...")
endpoints = tester.populate_global_endpoints(2)
endpoint = Endpoint.get(endpoints[0].id)
print("INFO: Publishing newly created APIs ...")
tester.publishAPIs(apis)
print("INFO: Creating new Store Application ...")
tester.delete_all_applications()
applications = tester.create_application(9)
print("INFO: Populating users ...")
tester.delete_all_users()
tester.populate_users(10)
# EndpointHandler.run()
if __name__ == '__main__':
main()