Skip to content

Commit df7e5b1

Browse files
capnmav77capnmav77
committed
Added support for Admin Users for App services and App Endpoints
- added new test scripts for Admin users - modified api_base script to support Admin Users Change-Id: I74f9216d1e4e322a114a77b1d675c64b5bc64457 Reviewed-on: https://review.couchbase.org/c/TAF/+/227101 Tested-by: Build Bot <build@couchbase.com> Reviewed-by: VIPUL BHARDWAJ <vipul.bhardwaj@couchbase.com>
1 parent bdae760 commit df7e5b1

8 files changed

Lines changed: 2733 additions & 0 deletions

‎pytests/Capella/RestAPIv4/AppService/add_app_service_admin_users.py‎

Lines changed: 498 additions & 0 deletions
Large diffs are not rendered by default.

‎pytests/Capella/RestAPIv4/AppService/delete_app_service_admin_users.py‎

Lines changed: 361 additions & 0 deletions
Large diffs are not rendered by default.

‎pytests/Capella/RestAPIv4/AppService/get_app_service_admin_users.py‎

Lines changed: 408 additions & 0 deletions
Large diffs are not rendered by default.

‎pytests/Capella/RestAPIv4/AppService/get_app_service_certificates.py‎

Lines changed: 313 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,313 @@
1+
"""
2+
Created on April 17, 2025
3+
4+
@author: Created using cbRAT cbModule by ramesh
5+
"""
6+
7+
from pytests.Capella.RestAPIv4.AppService.get_app_service import GetAppService
8+
9+
10+
class GetCertificates(GetAppService):
11+
12+
def setUp(self, nomenclature="AppService_GET"):
13+
GetAppService.setUp(self, nomenclature)
14+
15+
def tearDown(self):
16+
super(GetCertificates, self).tearDown()
17+
18+
def test_api_path(self):
19+
testcases = [
20+
{
21+
"description": "Send call with valid path params"
22+
}, {
23+
"description": "Replace api version in URI",
24+
"url": "/v3/organizations/{}/projects/{}/clusters/{}/appservices/{}/certificates",
25+
"expected_status_code": 404,
26+
"expected_error": {
27+
"errorType": "RouteNotFound",
28+
"message": "Not found"
29+
}
30+
}, {
31+
"description": "Replace the last path param name in URI",
32+
"url": "/v4/organizations/{}/projects/{}/clusters/{}/appservices/{}/certificate",
33+
"expected_status_code": 404,
34+
"expected_error": "404 page not found"
35+
}, {
36+
"description": "Add an invalid segment to the URI",
37+
"url": "/v4/organizations/{}/projects/{}/clusters/{}/appservices/{}/certificates/certificate",
38+
"expected_status_code": 404,
39+
"expected_error": "404 page not found"
40+
}, {
41+
"description": "Call API with non-hex organizationId",
42+
"invalid_organizationId": self.replace_last_character(
43+
self.organisation_id, non_hex=True),
44+
"expected_status_code": 400,
45+
"expected_error": {
46+
"code": 1000,
47+
"hint": "Check if you have provided a valid URL and all "
48+
"the required params are present in the request "
49+
"body.",
50+
"httpStatusCode": 400,
51+
"message": "The server cannot or will not process the "
52+
"request due to something that is perceived to "
53+
"be a client error."
54+
}
55+
}, {
56+
"description": "Call API with non-hex projectId",
57+
"invalid_projectId": self.replace_last_character(
58+
self.project_id, non_hex=True),
59+
"expected_status_code": 400,
60+
"expected_error": {
61+
"code": 1000,
62+
"hint": "Check if you have provided a valid URL and all "
63+
"the required params are present in the request "
64+
"body.",
65+
"httpStatusCode": 400,
66+
"message": "The server cannot or will not process the "
67+
"request due to something that is perceived to "
68+
"be a client error."
69+
}
70+
}, {
71+
"description": "Call API with non-hex clusterId",
72+
"invalid_clusterId": self.replace_last_character(
73+
self.cluster_id, non_hex=True),
74+
"expected_status_code": 400,
75+
"expected_error": {
76+
"code": 1000,
77+
"hint": "Check if you have provided a valid URL and all "
78+
"the required params are present in the request "
79+
"body.",
80+
"httpStatusCode": 400,
81+
"message": "The server cannot or will not process the "
82+
"request due to something that is perceived to "
83+
"be a client error."
84+
}
85+
}, {
86+
"description": "Call API with non-hex appServiceId",
87+
"invalid_appServiceId": self.replace_last_character(
88+
self.app_service_id, non_hex=True),
89+
"expected_status_code": 400,
90+
"expected_error": {
91+
"code": 1000,
92+
"hint": "Check if you have provided a valid URL and all "
93+
"the required params are present in the request "
94+
"body.",
95+
"httpStatusCode": 400,
96+
"message": "The server cannot or will not process the "
97+
"request due to something that is perceived to "
98+
"be a client error."
99+
}
100+
}
101+
]
102+
failures = list()
103+
for testcase in testcases:
104+
self.log.info("Executing test: {}".format(testcase["description"]))
105+
organization = self.organisation_id
106+
project = self.project_id
107+
cluster = self.cluster_id
108+
appService = self.app_service_id
109+
110+
if "url" in testcase:
111+
self.capellaAPI.cluster_ops_apis.certificates_endpoint = \
112+
testcase["url"]
113+
if "invalid_organizationId" in testcase:
114+
organization = testcase["invalid_organizationId"]
115+
elif "invalid_projectId" in testcase:
116+
project = testcase["invalid_projectId"]
117+
elif "invalid_clusterId" in testcase:
118+
cluster = testcase["invalid_clusterId"]
119+
elif "invalid_appServiceId" in testcase:
120+
appService = testcase["invalid_appServiceId"]
121+
122+
result = self.capellaAPI.cluster_ops_apis.fetch_app_service_certificate_info(
123+
organization, project, cluster, appService)
124+
if result.status_code == 429:
125+
self.handle_rate_limit(int(result.headers["Retry-After"]))
126+
result = self.capellaAPI.cluster_ops_apis.fetch_app_service_certificate_info(
127+
organization, project, cluster, appService)
128+
self.capellaAPI.cluster_ops_apis.certificates_endpoint = \
129+
"/v4/organizations/{}/projects/{}/clusters/{}/appservices/{}/certificates"
130+
self.validate_testcase(result, [200], testcase, failures)
131+
132+
if failures:
133+
for fail in failures:
134+
self.log.warning(fail)
135+
self.fail("{} tests FAILED out of {} TOTAL tests"
136+
.format(len(failures), len(testcases)))
137+
138+
def test_authorization(self):
139+
failures = list()
140+
for testcase in self.v4_RBAC_injection_init([
141+
"organizationOwner", "projectOwner", "projectManager",
142+
"projectViewer", "projectDataReaderWriter", "projectDataReader"
143+
]):
144+
self.log.info("Executing test: {}".format(testcase["description"]))
145+
header = dict()
146+
self.auth_test_setup(testcase, failures, header,
147+
self.project_id, self.other_project_id)
148+
result = self.capellaAPI.cluster_ops_apis.fetch_app_service_certificate_info(
149+
self.organisation_id, self.project_id, self.cluster_id,
150+
self.app_service_id,
151+
header)
152+
if result.status_code == 429:
153+
self.handle_rate_limit(int(result.headers["Retry-After"]))
154+
result = self.capellaAPI.cluster_ops_apis.fetch_app_service_certificate_info(
155+
self.organisation_id, self.project_id, self.cluster_id,
156+
self.app_service_id,
157+
header)
158+
self.validate_testcase(result, [200], testcase, failures)
159+
160+
if failures:
161+
for fail in failures:
162+
self.log.warning(fail)
163+
self.fail("{} tests FAILED.".format(len(failures)))
164+
165+
def test_query_parameters(self):
166+
self.log.debug(
167+
"Correct Params - organization ID: {}, project ID: {}, "
168+
"cluster ID: {}, appService ID: {}".format(
169+
self.organisation_id, self.project_id, self.cluster_id,
170+
self.app_service_id))
171+
testcases = 0
172+
failures = list()
173+
for combination in self.create_path_combinations(
174+
self.organisation_id, self.project_id, self.cluster_id,
175+
self.app_service_id):
176+
testcases += 1
177+
testcase = {
178+
"description": "organization ID: {}, project ID: {}, "
179+
"cluster ID: {}, appService ID: {}"
180+
.format(str(combination[0]), str(combination[1]),
181+
str(combination[2]), str(combination[3])),
182+
"organizationID": combination[0],
183+
"projectID": combination[1],
184+
"clusterID": combination[2],
185+
"appServiceID": combination[3]
186+
}
187+
if not (combination[0] == self.organisation_id and
188+
combination[1] == self.project_id and
189+
combination[2] == self.cluster_id and
190+
combination[3] == self.app_service_id):
191+
if (combination[0] == "" or combination[1] == "" or
192+
combination[2] == "" or combination[3] == ""):
193+
testcase["expected_status_code"] = 404
194+
testcase["expected_error"] = "404 page not found"
195+
elif any(variable in [
196+
int, bool, float, list, tuple, set, type(None)] for
197+
variable in [
198+
type(combination[0]), type(combination[1]),
199+
type(combination[2]), type(combination[3])]):
200+
testcase["expected_status_code"] = 400
201+
testcase["expected_error"] = {
202+
"code": 1000,
203+
"hint": "Check if you have provided a valid URL and "
204+
"all the required params are present in the "
205+
"request body.",
206+
"httpStatusCode": 400,
207+
"message": "The server cannot or will not process the "
208+
"request due to something that is "
209+
"perceived to be a client error."
210+
}
211+
elif combination[0] != self.organisation_id:
212+
testcase["expected_status_code"] = 403
213+
testcase["expected_error"] = {
214+
"code": 1002,
215+
"hint": "Your access to the requested resource is "
216+
"denied. Please make sure you have the "
217+
"necessary permissions to access the "
218+
"resource.",
219+
"httpStatusCode": 403,
220+
"message": "Access Denied."
221+
}
222+
elif combination[3] != self.bucket_id and not \
223+
isinstance(combination[3], type(None)):
224+
testcase["expected_status_code"] = 400
225+
testcase["expected_error"] = {
226+
"code": 400,
227+
"hint": "Please review your request and ensure "
228+
"that all required parameters are "
229+
"correctly provided.",
230+
"message": "BucketID is invalid.",
231+
"httpStatusCode": 400
232+
}
233+
elif combination[2] != self.cluster_id:
234+
testcase["expected_status_code"] = 404
235+
testcase["expected_error"] = {
236+
"code": 4025,
237+
"hint": "The requested cluster details could not be "
238+
"found or fetched. Please ensure that the "
239+
"correct cluster ID is provided.",
240+
"message": "Unable to fetch the cluster details.",
241+
"httpStatusCode": 404
242+
}
243+
elif combination[1] != self.project_id:
244+
testcase["expected_status_code"] = 422
245+
testcase["expected_error"] = {
246+
"code": 4031,
247+
"hint": "Please provide a valid projectId.",
248+
"httpStatusCode": 422,
249+
"message": "Unable to process the request. The "
250+
"provided projectId {} is not valid for "
251+
"the cluster {}."
252+
.format(combination[1], combination[2])
253+
}
254+
elif isinstance(combination[3], type(None)):
255+
testcase["expected_status_code"] = 404
256+
testcase["expected_error"] = {
257+
"code": 6008,
258+
"hint": "The requested bucket does not exist. Please "
259+
"ensure that the correct bucket ID is "
260+
"provided.",
261+
"httpStatusCode": 404,
262+
"message": "Unable to find the specified bucket."
263+
}
264+
else:
265+
testcase["expected_status_code"] = 400
266+
testcase["expected_error"] = {
267+
"code": 400,
268+
"hint": "Please review your request and ensure that "
269+
"all required parameters are correctly "
270+
"provided.",
271+
"message": "BucketID is invalid.",
272+
"httpStatusCode": 400
273+
}
274+
self.log.info("Executing test: {}".format(testcase["description"]))
275+
if "param" in testcase:
276+
kwarg = {testcase["param"]: testcase["paramValue"]}
277+
else:
278+
kwarg = dict()
279+
280+
result = self.capellaAPI.cluster_ops_apis.fetch_app_service_certificate_info(self.organisation_id, self.project_id, self.cluster_id,
281+
self.app_service_id,
282+
**kwarg)
283+
if result.status_code == 429:
284+
self.handle_rate_limit(int(result.headers["Retry-After"]))
285+
result = self.capellaAPI.cluster_ops_apis.fetch_app_service_certificate_info(self.organisation_id, self.project_id, self.cluster_id,
286+
self.app_service_id,
287+
**kwarg)
288+
self.validate_testcase(result, [200], testcase, failures)
289+
290+
if failures:
291+
for fail in failures:
292+
self.log.warning(fail)
293+
self.fail("{} tests FAILED out of {} TOTAL tests"
294+
.format(len(failures), testcases))
295+
296+
def test_multiple_requests_using_API_keys_with_same_role_which_has_access(
297+
self):
298+
api_func_list = [[
299+
self.capellaAPI.cluster_ops_apis.fetch_app_service_certificate_info, (
300+
self.organisation_id, self.project_id, self.cluster_id,
301+
self.app_service_id
302+
)
303+
]]
304+
self.throttle_test(api_func_list)
305+
306+
def test_multiple_requests_using_API_keys_with_diff_role(self):
307+
api_func_list = [[
308+
self.capellaAPI.cluster_ops_apis.fetch_app_service_certificate_info, (
309+
self.organisation_id, self.project_id, self.cluster_id,
310+
self.app_service_id
311+
)
312+
]]
313+
self.throttle_test(api_func_list, True, self.project_id)

0 commit comments

Comments
 (0)