Fixed issues after implement security improvements over authorizationInfo and AuthenticationInfo (d9da718d) · Commits · OCF / capif

tests/libraries/api_publish_service/bodyRequests.py

+93 −16

Original line number	Original line	Diff line number	Diff line
	@@ -3,7 +3,10 @@ def create_service_api_description(api_name="service_1",
	supported_features="0",		supported_features="0",
	vendor_specific_service_api_description=None,		vendor_specific_service_api_description=None,
	vendor_specific_aef_profile=None,		vendor_specific_aef_profile=None,
	api_status=None):		api_status=None,
			security_method="default",
			domain_name=None,
			interface_descriptions=None):
	aef_ids = list()		aef_ids = list()
	if isinstance(aef_id, list):		if isinstance(aef_id, list):
	aef_ids = aef_id		aef_ids = aef_id
	@@ -12,7 +15,32 @@ def create_service_api_description(api_name="service_1",
	print("aef_id parameter is a string")		print("aef_id parameter is a string")
	aef_ids.append(aef_id)		aef_ids.append(aef_id)

	profiles = create_aef_profiles(aef_ids)		security_methods = list()
			if security_method is not None:
			if isinstance(security_method, list):
			print("security_method parameter is a list")
			if len(security_method) > 0:
			if isinstance(security_method[0], list):
			security_methods = security_method
			else:
			security_methods.append(security_method)
			elif isinstance(security_method, str):
			print("security_methods parameter is a string")
			if security_method == "default":
			for idx in range(len(aef_ids)):
			security_methods.append(["OAUTH"])
			security_methods.append([security_method])
			else:
			print(f"security_method is {security_method}")
			print(f"security_methods: {security_methods}")
			else:
			print("security_method parameter is None")

			profiles = create_aef_profiles(
			aef_ids,
			security_methods,
			domain_name,
			interface_descriptions)

	body = {		body = {
	"apiName": api_name,		"apiName": api_name,
	@@ -58,16 +86,32 @@ def create_service_api_description(api_name="service_1",
	return body		return body


	def create_aef_profiles(aef_ids):		def create_aef_profiles(
			aef_ids,
			security_methods,
			domain_name=None,
			interface_descriptions=None):
	profiles = list()		profiles = list()
	index = 1		index = 1
	for aef_id in aef_ids:		for aef_id in aef_ids:
	profiles.append(create_aef_profile(aef_id, "resource_" + str(index)))		security_method = get_value(security_methods, index-1)
			print(f"aef_id: {aef_id}, security_method: {security_method}")
			profiles.append(
			create_aef_profile(
			aef_id,
			"resource_" + str(index),
			security_method,
			domain_name,
			interface_descriptions))
	index = index+1		index = index+1
	return profiles		return profiles


	def create_aef_profile(aef_id, resource_name):		def create_aef_profile(aef_id,
			resource_name,
			security_method=None,
			domain_name=None,
			interface_descriptions=None):
	data = {		data = {
	"aefId": aef_id,		"aefId": aef_id,
	"versions": [		"versions": [
	@@ -90,15 +134,23 @@ def create_aef_profile(aef_id, resource_name):
	],		],
	"protocol": "HTTP_1_1",		"protocol": "HTTP_1_1",
	"dataFormat": "JSON",		"dataFormat": "JSON",
	"securityMethods": ["PSK"],		}
	"interfaceDescriptions": [
			if domain_name is not None:
			data['domainName'] = domain_name
			elif interface_descriptions is not None:
			data['interfaceDescriptions'] = interface_descriptions
			elif domain_name is None and interface_descriptions is None:
			data['interfaceDescriptions'] = [
	{		{
	"ipv4Addr": "string",		"ipv4Addr": "string",
	"port": 65535,		"port": 65535,
	"securityMethods": ["PSK"]		"securityMethods": security_method
	}		}
	]		]
	}
			if security_method is not None:
			data['securityMethods'] = security_method
	return data		return data


	@@ -109,7 +161,10 @@ def create_service_api_description_patch(aef_id=None,
	service_api_category=None,		service_api_category=None,
	api_supp_feats=None,		api_supp_feats=None,
	pub_api_path=None,		pub_api_path=None,
	ccf_id=None):		ccf_id=None,
			security_method=None,
			domain_name=None,
			interface_descriptions=None):
	body = dict()		body = dict()

	# aef profiles		# aef profiles
	@@ -122,8 +177,26 @@ def create_service_api_description_patch(aef_id=None,
	elif isinstance(aef_id, str):		elif isinstance(aef_id, str):
	print("aef_id parameter is a string")		print("aef_id parameter is a string")
	aef_ids.append(aef_id)		aef_ids.append(aef_id)

			security_methods = list()
			if security_method is not None:
			if isinstance(security_method, list):
			print("security_method parameter is a list")
			if len(security_method) > 0:
			if isinstance(security_method[0], list):
			security_methods = security_method
			else:
			security_methods.append(security_method)
			elif isinstance(security_methods, str):
			print("security_methods parameter is a string")
			security_methods.append([security_methods])

	if aef_ids is not None:		if aef_ids is not None:
	profiles = create_aef_profiles(aef_ids)		profiles = create_aef_profiles(
			aef_ids,
			security_methods,
			domain_name,
			interface_descriptions)
	body['aefProfiles'] = profiles		body['aefProfiles'] = profiles

	# description		# description
	@@ -163,3 +236,7 @@ def create_service_api_description_patch(aef_id=None,
	body['apiStatus']['aefIds'] = aef_ids_active		body['apiStatus']['aefIds'] = aef_ids_active

	return body		return body


			def get_value(lst, index):
			return lst[index] if index < len(lst) else None