Enhance UETAG test cases (58f7e6d2) · Commits · MEC - Multi-access Edge Computing / MEC TTCN-3 Test Suite

simu/data/transport_info.csv

+1 −1

transportId,name,description,type,protocol,version,endpoint,security,implSpecificInfo

TransId12345,REST,REST API,REST_HTTP,HTTP,2.0,{},{"oAuth2Info":{"grantTypes":["OAUTH2_CLIENT_CREDENTIALS"],"tokenEndpoint": "/meMp1/security/TokenEndPoint"}},{}

TransId12345,REST,REST API,REST_HTTP,HTTP,2.0,{},security0,{}

simu/mec_database.py

+100 −3

Original line number	Diff line number	Diff line
		@@ -15,6 +15,9 @@ class mec_database:
		__cells__ = None
		__associated__ = None
		__subscribions__ = None
		__transport_info__ = None
		__security__ = None
		__traffic_rules__ = None

		def __init__(self):
		""" Default ctor: Open Pandas database. """
		@@ -26,6 +29,9 @@ class mec_database:
		self.__cells__ = pd.DataFrame(pd.read_csv(self.__path__ + 'cells.csv', encoding = 'utf-8', index_col = 'cellId'))
		self.__associated__ = pd.DataFrame(pd.read_csv(self.__path__ + 'associated.csv', encoding = 'utf-8', index_col = 'cellId'))
		self.__subscriptions__ = pd.DataFrame(pd.read_csv(self.__path__ + 'subscriptions.csv', encoding = 'utf-8', index_col = 'subscription_id'))
		self.__transport_info__ = pd.DataFrame(pd.read_csv(self.__path__ + 'transport_info.csv', encoding = 'utf-8', index_col = 'transportId'))
		self.__security__ = pd.DataFrame(pd.read_csv(self.__path__ + 'security.csv', encoding = 'utf-8', index_col = 'id'))
		self.__traffic_rules__ = pd.DataFrame(pd.read_csv(self.__path__ + 'traffic_rules.csv', encoding = 'utf-8', index_col = 'trafficRuleId'))
		# self.dump()
		# End of __init__

		@@ -38,7 +44,10 @@ class mec_database:
		print("LocationInfo: ", self.__location_info__.head())
		print("Cells: ", self.__cells__.head())
		print("Associated: ", self.__associated__.head())
		print("subscriptions: ", self.__subscribions__.head())
		print("subscriptions: ", self.__subscriptions__.head())
		print("transport_info: ", self.__transport_info__.head())
		print("security: ", self.__security__.head())
		print("traffic_rules: ", self.__traffic_rules__.head())
		# End of dump

		def __to_accessPoint__(self, p_access_point, p_location):
		@@ -106,6 +115,44 @@ class mec_database:
		return resp
		# End of __to_subscriptions__

		def __to_transport_info__(self, p_transport_info):
		"""
		Build a TransportInfo Json message.
		Keyword arguments:
		-- p_transport_info: TransportInfo description
		"""
		print(">>> __to_transport_info__: ", p_transport_info)
		# print("__to_transport_info__: security id: ", p_transport_info[7])
		resp = "{{\t\"id\": \"TransId12345\",\"name\": \"REST\",\"description\": \"REST API\",\"type\": \"REST_HTTP\",\"protocol\": \"HTTP\",\"version\": \"2.0\",\"endpoint\": {},\"security\": {\"oAuth2Info\": {\"grantTypes\": [\"OAUTH2_CLIENT_CREDENTIALS\"],\"tokenEndpoint\": \"/meMp1/security/TokenEndPoint\"}\"},\"implSpecificInfo\": {}}"
		# resp = None
		# try:
		# security = self.__security__.loc[self.__security__['id'] == p_transport_info[7]]
		# print("__to_transport_info__: security= ", security)
		#
		# resp = "{\t\"id\": \"" + p_transport_info[0] + "\",\t\"name\": \"" + p_transport_info[1] + "\",\t\"description\": \"" + p_transport_info[2] + "\",\t\"type\": \"" + p_transport_info[3] + "\",\t\"protocol\": \"" + p_transport_info[4] + "\",\t\"version\": \"" + str(p_transport_info[5]) + "\",\t\"endpoint\": \"" + p_transport_info[6] + "\",\t\"security\": \"" + security + "\",\t\"implSpecificInfo\": \"" + p_transport_info[8] + "\"}"
		# print("__to_transport_info__: ", resp)
		# except KeyError:
		# print("__to_transport_info__: No security entry")
		return resp
		# End of method __to_transport_info__

		def __to_traffic_rule__(self, p_traffic_rule):
		"""
		Build a TrafficRule Json message.
		Keyword arguments:
		-- p_traffic_rule: TrafficRule description
		"""
		print(">>> __to_traffic_rule__: ", p_traffic_rule)
		resp = "{\"trafficRuleId\": \"TrafficRule1\",\"filterType\": \"FLOW\",\"priority\": 1,\"trafficFilter\": [{\"srcAddress\": [\"192.168.1.1\"],\"dstAddress\": [\"192.168.1.1\"],\"srcPort\": [\"8080\"],\"dstPort\": [\"8080\"],\"protocol\": [ \"?\"],\"token\": [\"?\"],\"srcTunnelAddress\": [\"?\"],\"tgtTunnelAddress\": [\"?\"],\"srcTunnelPort\": [\"?\"],\"dstTunnelPort\": [\"?\"],\"qCI\": 1,\"dSCP\": 0,\"tC\": 1}],\"action\": \"DROP\",\"dstInterface\": {\"interfaceType\": \"TUNNEL\",\"tunnelInfo\": {\"tunnelType\": \"GTP_U\",\"tunnelDstAddress\": \"?\",\"tunnelSrcAddress\": \"?\"},\"srcMacAddress\": \"02-00-00-00-00-00\",\"dstMacAddress\": \"02-00-00-00-00-00\",\"dstIpAddress\":\"192.0.2.0\"},\"state\":\"ACTIVE\"}"
		# resp = None
		# try:
		# resp = "{\t\"id\": \"" + p_transport_info[0] + "\",\t\"name\": \"" + p_transport_info[1] + "\",\t\"description\": \"" + p_transport_info[2] + "\",\t\"type\": \"" + p_transport_info[3] + "\",\t\"protocol\": \"" + p_transport_info[4] + "\",\t\"version\": \"" + str(p_transport_info[5]) + "\",\t\"endpoint\": \"" + p_transport_info[6] + "\",\t\"security\": \"" + security + "\",\t\"implSpecificInfo\": \"" + p_transport_info[8] + "\"}"
		# print("__to_transport_info__: ", resp)
		# except KeyError:
		# print("__to_transport_info__: No security entry")
		return resp
		# End of method __to_transport_info__

		def getSubscriberList(self, p_uri):
		""" Build a SubscriberList Json message. """
		print(">>> getSubscriberList")
		@@ -145,7 +192,6 @@ class mec_database:
		print(">>> getAccessPointList")
		resp = None
		try:

		r = self.__access_point_list__.loc[self.__access_point_list__['zoneId'] == p_zone_id]
		print("getAccessPointList: r= ", r)
		resp = "{\"accessPointList\": {\t\"zoneId\": \"" + p_zone_id + "\",\t\"accessPoint\": [\t"
		@@ -166,6 +212,57 @@ class mec_database:
		return "{\"SubscriptionLinkList\": {\"_links\": {\"self\": \"http://example.com" + p_uri + "\"},\"subscription\": [{\"href\": \"http://meAppClient.example.com/rni/v1/notifications/cell_change/77777\",\"subscriptionType\": \"CELL_CHANGE\"},{\"href\": \"http://meAppClient.example.com/rni/v1/notifications/MeasTa/77777\",\"subscriptionType\": \"MEAS_TIMING_ADVANCE\"}]}}"
		# End of method getSubscriptionLinkList

		def getMp1TransportInfoList(self):
		print(">>> getMp1TransportInfoList")
		resp = None
		try:
		resp = "{\"transportInfoList\": [\t"
		for r in self.__transport_info__.itertuples():
		t = tuple(r)
		resp += self.__to_transport_info__(t) + ","
		# En of 'for' statement
		resp = resp[: len(resp) - 1]
		resp += "\t]}"
		print("getMp1TransportInfoList: ", resp)
		except KeyError:
		print("getMp1TransportInfoList: No transportInfo list")
		return resp
		# End of method getMp1TransportInfo

		def getMp1TrafficRulesList(self):
		print(">>> getMp1TrafficRulesList")
		resp = None
		try:
		resp = "{\"TrafficRulesList\": [\t"
		for r in self.__traffic_rules__.itertuples():
		t = tuple(r)
		resp += self.__to_traffic_rule__(t) + ","
		# En of 'for' statement
		resp = resp[: len(resp) - 1]
		resp += "\t]}"
		print("getMp1TrafficRulesList: ", resp)
		except KeyError:
		print("getMp1TrafficRulesList: No TrafficRules list")
		return resp
		# End of method getMp1TrafficRules

		def getMp1TrafficRule(self):
		print(">>> getMp1TrafficRule")
		resp = None
		try:
		resp = "{\"TrafficRule\": [\t"
		for r in self.__traffic_rule__.itertuples():
		t = tuple(r)
		resp += self.__to_traffic_rule__(t) + ","
		# En of 'for' statement
		resp = resp[: len(resp) - 1]
		resp += "\t]}"
		print("getMp1TrafficRule: ", resp)
		except KeyError:
		print("getMp1TrafficRule: No TrafficRules list")
		return resp
		# End of method getMp1TrafficRules

		def getSubscriberFromAddress(self, p_ue_address):
		print(">>> getSubscriberFromAddress: ", p_ue_address)
		result = None

simu/mec_http_server.py

+89 −0

Original line number	Diff line number	Diff line
		@@ -129,6 +129,71 @@ class myHandler(http.server.BaseHTTPRequestHandler):
		return
		# End of do_POST

		# Handler for the PUT requests
		def do_PUT(self):
		print(">>> do_PUT: ", self.path)
		self.protocol_version = self.request_version
		# Check HTTP end_headers
		if self.__check_http_headers__() == False:
		resp = "{\"problemDetails\": {\t\"type\": \"Bad Request\",\t\"title\": \"N/A\",\t\"status\": 400,\t\"detail\": \"Wrong headers\",\t\"instance\": \"N/A\"}}"
		self.protocol_version = self.request_version
		self.send_response(400, 'Bad Request')
		self.send_header('Host', self.headers.get('Host'))
		self.send_header('Authorization', self.headers.get('Authorization'))
		self.send_header('Connection', 'keep-alive')
		self.send_header('Content-Type', 'application/problem+json')
		l = len(resp)
		self.send_header('Content-Length', str(l))
		self.end_headers()
		self.wfile.write(bytes(resp, 'utf-8'))
		return

		resp, content_type = self.__process__()
		print('do_PUT: resp= ', resp, ', Content_Type= ', content_type)
		if (resp == None):
		self.send_response(404, 'Not Found')
		self.send_header('Host', self.headers.get('Host')) # send_header() shall be after send_response()
		self.send_header('Authorization', self.headers.get('Authorization'))
		self.send_header('Connection', 'keep-alive')
		self.send_header('Content-Type', content_type)
		self.send_header('Content-Length', '0')
		self.end_headers()
		else:
		if resp.find('Not Found') != -1:
		self.send_response(404, 'Not Found')
		resp = ""
		elif resp.find('Forbidden') != -1:
		self.send_response(403, 'Forbidden')
		resp = ""
		elif resp.find('Bad Request') != -1:
		self.send_response(400, 'Bad Request')
		resp = ""
		elif resp.find('Precondition Failed') != -1:
		self.send_response(412, 'Precondition Failed')
		resp = "{\"problemDetails\": {\t\"type\": \"Precondition Failed\",\t\"title\": \"N/A\",\t\"status\": 412,\t\"detail\": \"Wrong preconditions\",\t\"instance\": \"N/A\"}}"
		content_type = 'application/problem+json'
		elif resp.find('userTrackingSubscription') != -1:
		self.send_response(201, 'Created')
		resp = ""
		else:
		self.send_response(200, 'OK')
		self.send_header('Host', self.headers.get('Host')) # send_header() shall be after send_response()
		self.send_header('Authorization', self.headers.get('Authorization'))
		self.send_header('Connection', 'keep-alive')
		self.send_header('Content-Type', content_type)
		l = len(resp)
		self.send_header('Content-Length', str(l))
		#self.end_headers() # Unable to bufferize wfile, so calling end_headers() headers and body are sent in 2 different TCP packets
		self._headers_buffer.append(b"\r\n")
		self._headers_buffer.append(bytes(resp, 'utf-8'))
		self.wfile.write(b"".join(self._headers_buffer))
		self._headers_buffer = []
		#self.wfile.write(bytes(resp, 'utf-8'))
		#self.wfile.flush()
		print("<<< do_PUT")
		return
		# End of do_POST

		# Handler for the DELETE requests
		def do_DELETE(self):
		print('>>> do_DELETE: ', self.path)
		@@ -217,6 +282,8 @@ class myHandler(http.server.BaseHTTPRequestHandler):
		return self.__process__rnis__api__(s)
		elif s[2] == 'bwm' and ((s[3] == 'v1') or (s[3] == 'v2')):
		return self.__process__bwm__api__(s)
		elif s[2] == 'mec_app_support' and ((s[3] == 'v1') or (s[3] == 'v2')):
		return self.__process__mp1__api__(s)
		else:
		return None, 'application/json'
		# End of __process__
		@@ -367,6 +434,28 @@ class myHandler(http.server.BaseHTTPRequestHandler):
		return resp, content_type
		# End of __process__bwm__api__

		# Process the MP1 request
		def __process__mp1__api__(self, p_split):
		print (">>> __process__mp1__api__: ", p_split)
		resp = None
		content_type = 'application/json'
		if p_split[4].startswith('transports'): # E.g. /mp1/v1/transports
		resp = self.__db__.getMp1TransportInfoList()
		elif p_split[4].startswith('applications'):
		if p_split[6].startswith('dns_rules'): # E.g. /mp1/v1/applications/appInstId01/dns_rules
		pass
		elif p_split[6].startswith('traffic_rules'): # E.g. GET/PUT /mp1/v1/applications/appInstId01/traffic_rules...
		if length(p_split) == 7: # E.g. GET /mp1/v1/applications/appInstId01/traffic_rules
		resp = self.__db__.getMp1TrafficRulesList()
		else: # E.g. GET/PUT /mp1/v1/applications/appInstId01/traffic_rules/{trafficRuleId}
		resp = self.__db__.getMp1TrafficRule(p_split[7])
		if (resp == None):
		resp = "{\"problemDetails\": {\t\"type\": \"Not supported\",\t\"title\": \"Mp1\",\t\"status\": 400,\t\"detail\": \"Wrong parameters\",\t\"instance\": \"string\"}}"
		content_type = 'application/problem+json'
		print ("<<< __process__mp1__api__: ", resp, ", ", content_type)
		return resp, content_type
		# End of __process__bwm__api__

		# Decode a Json HTTP body message
		def __decode__json__body__(self, p_body):
		print(">>> __decode__json__body__: ", p_body)

simu/mec_simulator.py

+2 −2

Original line number	Diff line number	Diff line
		@@ -8,9 +8,9 @@ import sys, os, time
		import msvcrt

		# MEC simulator IPv4 address
		ADDRESS_BINDING = '172.28.4.87'
		#ADDRESS_BINDING = '172.28.4.87'
		#ADDRESS_BINDING = '192.168.1.21'
		#ADDRESS_BINDING = '192.168.0.17'
		ADDRESS_BINDING = '192.168.56.1'
		#ADDRESS_BINDING = '127.0.0.1'
		# MEC simulator IPv4 listening port
		PORT_NUMBER = 8081