Skip to content
Snippets Groups Projects
Commit b3cb1899 authored by piscione's avatar piscione
Browse files

MEC013 v3.1.1 UETESTNOT Draft TCs available

parent a205dae2
No related branches found
No related tags found
1 merge request!3Merge of TT027 branch into master
Expand all Hide whitespace changes
Inline Side-by-side
Showing
with 909 additions and 0 deletions
MEC013/SRV/UETESTNOT/PlatUeTestSubscription.robot 0 → 100644
+ 101
− 0
View file @ b3cb1899
*** Settings ***
Documentation
... A test suite for validating UE Test Notification (UETESTNOT) operations.
Resource ../../../GenericKeywords.robot
Resource ../../../pics.txt
Resource environment/variables.txt
Library REST ${SCHEMA}://${HOST}:${PORT} ssl_verify=false
Library OperatingSystem
Library String
Library libraries/Server.py
Default Tags TC_MEC_SRV_UELOCSUB
*** Test Cases ***
TC_MEC_MEC013_SRV_UETESTNOT_001_OK
[Documentation]
... Check that the IUT provides a test notification when requested by a MEC Application
...
... ETSI GS MEC 013 3.1.1 Clause 5.3.4
... ETSI GS MEC 013 3.1.1 Clause 6.3.4
... ETSI GS MEC 013 3.1.1 Clause 6.4.3
... ETSI GS MEC 013 3.1.1 Clause 7.5.3.4
[Tags] PIC_MEC_PLAT PIC_SERVICES
[Setup] Create an UserLocationEvent subscription and get sub id UserLocationEventSubscription
Spawn Notification Server TestNotification
Validate Json TestNotification.schema.json ${payload_notification}
[TearDown] Remove an UserLocationEvent subscription ${SUB_ID}
TC_MEC_MEC013_SRV_UETESTNOT_002_OK
[Documentation]
... Check that the IUT terminates notifications after time expiration
...
... ETSI GS MEC 013 3.1.1 Clause 5.3.11
... ETSI GS MEC 013 3.1.1 Clause 6.3.8
... ETSI GS MEC 013 3.1.1 Clause 6.4.8
... ETSI GS MEC 013 3.1.1 Clause 7.16.3.4
[Tags] PIC_MEC_PLAT PIC_SERVICES
[Setup] Create an UserArea subscription and get sub id UserAreaSubscription
Spawn Notification Server UserAreaNotification
Validate Json UserAreaNotification.schema.json ${payload_notification}
[TearDown] Remove an UserArea subscription ${SUB_ID}
*** Keywords ***
Create an UserLocationEvent subscription and get sub id
[Arguments] ${content}
Set Headers {"Accept":"application/json"}
Set Headers {"Authorization":"${TOKEN}"}
Set Headers {"Authorization":"${TOKEN}"}
${file}= Catenate SEPARATOR= jsons/ ${content} .json
${body}= Get File ${file}
Post ${apiRoot}/${apiName}/${apiVersion}/subscriptions/users ${body}
${output}= Output response
Set Suite Variable ${response} ${output}
${elements} = Split String ${response['headers']['Location']} /
Set Suite Variable ${SUB_ID} ${elements[3]}
Create an UserArea subscription and get sub id
[Arguments] ${content}
Set Headers {"Accept":"application/json"}
Set Headers {"Authorization":"${TOKEN}"}
Set Headers {"Authorization":"${TOKEN}"}
${file}= Catenate SEPARATOR= jsons/ ${content} .json
${body}= Get File ${file}
Post ${apiRoot}/${apiName}/${apiVersion}/subscriptions/area ${body}
${output}= Output response
Set Suite Variable ${response} ${output}
${elements} = Split String ${response['headers']['Location']} /
Set Suite Variable ${SUB_ID} ${elements[3]}
Remove an UserLocationEvent subscription
[Arguments] ${subscription_id}
Set Headers {"Accept":"application/json"}
Set Headers {"Authorization":"${TOKEN}"}
Set Headers {"Authorization":"${TOKEN}"}
Delete ${apiRoot}/${apiName}/${apiVersion}/subscriptions/users/${subscription_id}
${output}= Output response
Set Suite Variable ${response} ${output}
Remove an UserArea subscription
[Arguments] ${subscription_id}
Set Headers {"Accept":"application/json"}
Set Headers {"Authorization":"${TOKEN}"}
Set Headers {"Authorization":"${TOKEN}"}
Delete ${apiRoot}/${apiName}/${apiVersion}/subscriptions/area/${subscription_id}
${output}= Output response
Set Suite Variable ${response} ${output}
Spawn Notification Server
[Arguments] ${payload_notification}
${output} Spawn Web Server ${NOTIFICATION_SERVER_IP} ${NOTIFICATION_SERVER_PORT} ${NOTIFICATION_SERVER_TIMEOUT} ${NOTIFICATION_SERVER_HTTP_METHOD} ${NOTIFICATION_SERVER_URI} ${payload_notification}
${length} = Get Length ${output}
Set Suite Variable ${payload_notification} ${output}
Run Keyword If ${length} == 0 Skip
\ No newline at end of file
MEC013/SRV/UETESTNOT/README.md 0 → 100644
+ 0
− 0
View file @ b3cb1899
MEC013/SRV/UETESTNOT/environment/variables.txt 0 → 100644
+ 21
− 0
View file @ b3cb1899
*** Variables ***
# Generic variables
${SCHEMA} http
${HOST} 127.0.0.1
${PORT} 8081
${response} {}
${TOKEN} Basic YWxhZGRpbjpvcGVuc2VzYW1l
${apiRoot}
${apiName} location
${apiVersion} v3
${SUB_ID}
${NON_EXISTENT_SUBSCRIPTION_ID} NON_EXISTENT_SUBSCRIPTION_ID
##Notification Server variables
${NOTIFICATION_SERVER_IP} 127.0.0.1
${NOTIFICATION_SERVER_PORT} 8888
${NOTIFICATION_SERVER_HTTP_METHOD} POST
${NOTIFICATION_SERVER_URI} /callback_url
${NOTIFICATION_SERVER_TIMEOUT} 5
MEC013/SRV/UETESTNOT/jsons/TestNotification.json 0 → 100644
+ 8
− 0
View file @ b3cb1899
{
"_links": {
"self": {
"href": "https://someurl.com/subscriptions/users/subscriptionId"
}
},
"notificationType": "TestNotification"
}
\ No newline at end of file
MEC013/SRV/UETESTNOT/jsons/UserAreaNotification.json 0 → 100644
+ 12
− 0
View file @ b3cb1899
{
"notificationType": "UserAreaNotification",
"callbackReference":"127.0.0.1:8888/callback_uri",
"address": "http://someuri.com/123",
"civicInfo": {
"country": "A1"
},
"userLocationEvent": "ENTERING_AREA_EVENT",
"_links": {
"subscription": "someuri"
}
}
\ No newline at end of file
MEC013/SRV/UETESTNOT/jsons/UserAreaSubscription.json 0 → 100644
+ 29
− 0
View file @ b3cb1899
{
"subscriptionType": "userAreaSubscription",
"clientCorrelator": "0123",
"callbackReference": "http://my.callback.com/area_notification/some-id",
"_links": {
"self": {
"href": "http://meAppServer.example.com/location/v2/subscriptions/area/subscription123"
}
},
"areaDefine": {
"shape": 1,
"points": [
{
"latitude": -80.86302,
"longitude": 41.277306
}
],
"radius": null
},
"addressList": [
"acr:10.0.0.1",
"acr:10.0.0.2"
],
"trackingAccuracy": 10.99,
"expiryDeadline": {
"seconds": 1973507343,
"nanoSeconds\"": 0
}
}
\ No newline at end of file
MEC013/SRV/UETESTNOT/jsons/UserLocationEventSubscription.json 0 → 100644
+ 7
− 0
View file @ b3cb1899
{
"subscriptionType": "UserLocationEventSubscription",
"clientCorrelator":"someCorrelator",
"address": "acr:10.0.0.1",
"callbackReference": "http://10.3.0.0.8:8888",
"requestTestNotification":"true"
}
\ No newline at end of file
MEC013/SRV/UETESTNOT/libraries/Server.py 0 → 100644
+ 149
− 0
View file @ b3cb1899
#!/usr/bin/python3
from http.server import BaseHTTPRequestHandler, HTTPServer
import json, os
import logging
# Library version
__version__ = '0.0.1'
def import_notification_json(subscription_type):
notification_type = subscription_type.split("Subscription")[0]
file_path = "./jsons/"+notification_type+".json"
logging.info(file_path)
logging.info(os.listdir())
try:
with open(file_path, 'r') as json_file:
# Load the JSON data
data = json.load(json_file)
logging.info(data)
return data
except FileNotFoundError:
logging.error(f"Error: File not found at {file_path}")
class Server ( object ):
ROBOT_LIBRARY_VERSION = '0.0.1'
def spawn_web_server (self, host="127.0.0.1", port=8080, timeout=15, method="POST", endpoint="/callback_url", resp_body=None):
class GET_Server(BaseHTTPRequestHandler):
def __call__(self, *args, **kwargs):
"""Handle a request."""
super().__init__(*args, **kwargs)
def __init__(self, endpoint, resp_body):
self.resp_body = resp_body
self.endpoint = endpoint
def do_GET(self):
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
if self.path == self.endpoint:
self.wfile.write(json.dumps(self.resp_body).encode(encoding='utf_8'))
else:
self.wfile.write(json.dumps("wrong endpoint").encode(encoding='utf_8'))
class POST_Server(BaseHTTPRequestHandler):
def __call__(self, *args, **kwargs):
"""Handle a request."""
super().__init__(*args, **kwargs)
def __init__(self, endpoint, resp_body):
self.resp_body = resp_body
self.endpoint = endpoint
self.req_body = None
def do_POST(self):
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
#if self.path == self.endpoint:
# self.wfile.write(json.dumps(self.resp_body).encode(encoding='utf_8'))
#else:
# self.wfile.write(json.dumps("wrong endpoint").encode(encoding='utf_8'))
content_len = int(self.headers.get('Content-Length'))
post_body = self.rfile.read(content_len)
self.req_body=post_body
def get_req_body(self):
return self.req_body
def get_resp_body(self):
return self.resp_body
class PUT_Server(BaseHTTPRequestHandler):
def __call__(self, *args, **kwargs):
"""Handle a request."""
super().__init__(*args, **kwargs)
def __init__(self, endpoint, resp_body):
self.resp_body = resp_body
self.endpoint = endpoint
def do_PUT(self):
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
if self.path == self.endpoint:
self.wfile.write(json.dumps(self.resp_body).encode(encoding='utf_8'))
else:
self.wfile.write(json.dumps("wrong endpoint").encode(encoding='utf_8'))
class DELETE_Server(BaseHTTPRequestHandler):
def __call__(self, *args, **kwargs):
"""Handle a request."""
super().__init__(*args, **kwargs)
def __init__(self, endpoint, resp_body):
self.resp_body = resp_body
self.endpoint = endpoint
def do_DELETE(self):
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
if self.path == self.endpoint:
self.wfile.write(json.dumps(self.resp_body).encode(encoding='utf_8'))
else:
self.wfile.write(json.dumps("wrong endpoint").encode(encoding='utf_8'))
if method == "GET":
self.handler = GET_Server(endpoint, resp_body)
elif method == "POST":
self.handler = POST_Server(endpoint, resp_body)
elif method == "PUT":
self.handler = PUT_Server(endpoint, resp_body)
elif method == "DELETE":
self.handler = DELETE_Server(endpoint, resp_body)
else:
logging.info("Error, unknown endpoint")
exit(1)
self.app = HTTPServer((host, int(port)), self.handler)
self.app.timeout = int(timeout)
self.app.handle_request()
self.app.server_close()
logging.info(self.handler.get_resp_body())
## If a notification is received, then is returned. Otherwise an empty dictionary.
if(self.handler.get_req_body()!=None):
return json.loads(self.handler.get_req_body().decode("windows-1252"))
##OLD mechanism commented: if no response is received, then read the Notification JSON file and return it.
notification_json= import_notification_json(self.handler.get_resp_body())
#return notification_json
return {}
MEC013/SRV/UETESTNOT/schemas/TestNotification.schema.json 0 → 100644
+ 43
− 0
View file @ b3cb1899
{
"properties": {
"_links": {
"description": "Hyperlink related to the resource. This shall be only included in the HTTP responses and in HTTP PUT requests.",
"properties": {
"self": {
"properties": {
"href": {
"description": "URI referring to a resource.",
"format": "uri",
"type": "string",
"x-etsi-mec-cardinality": "1",
"x-etsi-mec-origin-type": "Uri"
}
},
"required": [
"href"
],
"type": "object",
"x-etsi-ref": "6.7.2"
}
},
"required": [
"self"
],
"type": "object",
"x-etsi-mec-cardinality": "0..1",
"x-etsi-mec-origin-type": "Structure (inlined)"
},
"notificationType": {
"description": "Shall be set to \"TestNotification\".",
"type": "string",
"x-etsi-mec-cardinality": "1",
"x-etsi-mec-origin-type": "String"
}
},
"required": [
"notificationType",
"_links"
],
"type": "object",
"x-etsi-ref": "6.4.3"
}
\ No newline at end of file
MEC013/SRV/UETESTNOT/schemas/UserAreaNotification.schema.json 0 → 100644
+ 539
− 0
View file @ b3cb1899
This diff is collapsed.
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