Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Addition of more user defined parameters #48

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Addition of more user defined parameters #48

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9be5f9a
Update availability and continuity tests
Ioannischatzis Mar 19, 2024
783b7bd
Merge branch 'master' of https://github.com/5gasp/CICD_LTR
Ioannischatzis Mar 19, 2024
114fca5
Merge branch 'master' into availability_and_continuity_tests2
Ioannischatzis Mar 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
123 changes: 25 additions & 98 deletions ftp/tests/availability_and_continuity_bandwidth/TCTest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,6 @@ def install_requirements():
# ----------------------- END OF INSTALL REQUIREMENTS ----------------------- #

import requests
import socket
import dns.resolver
import logging
import time

Expand All @@ -38,8 +36,8 @@ def getAuthBearer(username: str, password: str, nef_url: str) -> str:
}
data = {
"grant_type": "",
"username": "[email protected]",
"password": "pass",
"username": username,
"password": password,
"scope": "",
"client_id": "",
"client_secret": ""
Expand All @@ -58,59 +56,6 @@ def getAuthBearer(username: str, password: str, nef_url: str) -> str:
logging.error("Request failed with status code:" + str(response.status_code) + response.content.decode('utf-8'))
return None

def getIP():

# Set the hostname that you want to look up
hostname = "cameraaistr.fidegad.external.ip"

try:
# Use the socket module to get the IP address of the pod's default DNS server
dns_server = socket.gethostbyname("kube-dns.kube-system.svc.cluster.local")
except:
return "127.0.0.1"
# Use the dns.resolver module to perform the nslookup using the pod's default DNS server
resolver = dns.resolver.Resolver(configure=False)
resolver.nameservers = [dns_server]
result = resolver.resolve(hostname)

# Print the IP addresses returned by the nslookup
for ip in result:
return ip

def createNEFMonitoringSubscription(bearer, nef_url, externalId):
streaming_server_ip_address = getIP()
print("Streaming Server IP address by kubernetes: " + str(streaming_server_ip_address))
headers = {
"accept": "application/json",
"Content-Type": "application/json",
"Authorization": "Bearer " + bearer
}
data = {
"externalId": externalId,
"notificationDestination": "http://" + str(streaming_server_ip_address) + ":85/handover_event_hook",
"monitoringType": "LOCATION_REPORTING",
"maximumNumberOfReports": 1000,
"monitorExpireTime": "2025-02-12T09:06:50.811Z",
"maximumDetectionTime": 1000,
"reachabilityType": "DATA"
}

response = requests.post(
nef_url.rstrip('/') + "/nef/api/v1/3gpp-monitoring-event/v1/myNetapp/subscriptions"
, headers=headers
, json=data
)

if response.status_code == 201:
logging.info("Subscription succeeded with status code:" + str(response.status_code) + response.content.decode('utf-8'))
# Do something with the response data
data = response.json()
logging.info(data)
return True
else:
# Handle the error
logging.error("Subscription request failed with status code:" + str(response.status_code) + response.content.decode('utf-8'))
return False

def startMovingUE(bearer, nef_url, supi):
headers = {
Expand Down Expand Up @@ -168,44 +113,10 @@ def stopMovingUE(bearer, nef_url, supi):
logging.info("Stop looping of UE with SUPI " + supi + " failed with status code:" + str(response.status_code) + response.content.decode('utf-8'))
return False

def deleteNEFMonitoringSubscription(bearer, nef_url, externalId):
headers = {
"accept": "application/json",
"Content-Type": "application/json",
"Authorization": "Bearer " + bearer
}
response = requests.get(
nef_url.rstrip('/') + "/nef/api/v1/3gpp-monitoring-event/v1/myNetapp/subscriptions"
, headers=headers
)

if 200 <= response.status_code < 300:
logging.info("Get Subscription succeeded with status code:" + str(response.status_code) + response.content.decode('utf-8'))
if len(response.content) > 0:
# Do something with the response data
data = response.json()
for row in data:
if row["externalId"] == externalId:
delete_response = requests.delete(row["link"], headers=headers)
if delete_response.status_code == 200:
logging.info("Deletion of Subscription request of " + row["externalId"] + " succeeded with status code: " + str(
delete_response.status_code) + delete_response.content.decode('utf-8'))
else:
logging.error("Deletion of Subscription request of " + row["externalId"] + " failed with status code: " + str(
delete_response.status_code) + delete_response.content.decode('utf-8'))
return False
logging.info(data)
else:
logging.info("No active subscriptions found")
return True
else:
# Handle the error
logging.error("Subscription deletion request failed with status code:" + str(response.status_code) + response.content.decode('utf-8'))
return False

def iterateForAUE(nef_url: str, ue: str, timeforacircle: str)-> str :
def iterateForAUE(nef_url: str, nef_username: str, nef_password: str, ue: str, timeforacircle: str, liveness_url: str)-> str :
try:
bearer = getAuthBearer("[email protected]", "pass", nef_url)
bearer = getAuthBearer(nef_username, nef_password, nef_url)
if bearer is not None:
stopMovingUE(bearer, nef_url, "202010000000004")
stopMovingUE(bearer, nef_url, "202010000000005")
Expand All @@ -220,20 +131,34 @@ def iterateForAUE(nef_url: str, ue: str, timeforacircle: str)-> str :
return -1

try:
bearer = getAuthBearer("[email protected]", "pass", nef_url)
bearer = getAuthBearer(nef_username, nef_password, nef_url)
if bearer is not None:
createNEFMonitoringSubscription(bearer, nef_url, "1000" + ue + "@domain.com")
startMovingUE(bearer, nef_url, "20201000000000" + ue)
time.sleep(int(timeforacircle) * 2)
stopMovingUE(bearer, nef_url, "20201000000000" + ue)
deleteNEFMonitoringSubscription(bearer, nef_url, "1000" + ue + "@domain.com")
return "Success"
if check_liveness(liveness_url):
return "Success"
else:
return -1
except ConnectionError:
logging.error("Connection with NEF failed!")
except:
logging.error("Communication with NEF failed!")
return -1

def check_liveness(url):
try:
response = requests.get(url)
# A status code of 200 means OK, which indicates the page is live.
if response.status_code == 200:
print(f"The page at {url} is live. Status Code: {response.status_code}")
return True
else:
print(f"The page at {url} encountered an issue. Status Code: {response.status_code}")
return False
except requests.ConnectionError:
print(f"Failed to connect to {url}")
return False
# -------------------------- UNINSTALL REQUIREMENTS ------------------------- #
def uninstall_requirements():
curr_dir = os.path.dirname(os.path.realpath(__file__))
Expand All @@ -257,5 +182,7 @@ def uninstall_requirements():

if __name__ == '__main__':
nef_url = "http://10.10.10.20:8888"
iterateForAUE(nef_url, "4", "36")
nef_username = "[email protected]"
nef_password = "pass"
iterateForAUE(nef_url, nef_username, nef_password, "4", "36")

11 changes: 10 additions & 1 deletion ftp/tests/availability_and_continuity_bandwidth/availability_and_continuity_bandwidth.robot
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,18 @@ ${availability_and_continuity_bandwidth_UENumber} 4
*** Test Cases ***
Testing for UE${availability_and_continuity_bandwidth_UENumber}
${availability_and_continuity_bandwidth_NEFURL}= Get Environment Variable availability_and_continuity_bandwidth_NEFURL
${availability_and_continuity_bandwidth_NEFUSERNAME}= Get Environment Variable availability_and_continuity_bandwidth_NEFUSERNAME
${availability_and_continuity_bandwidth_NEFPASSWORD}= Get Environment Variable availability_and_continuity_bandwidth_NEFPASSWORD
${availability_and_continuity_bandwidth_LIVENESS}= Get Environment Variable availability_and_continuity_bandwidth_LIVENESS
Should Not Be Empty ${availability_and_continuity_bandwidth_NEFURL}
Log The NEF URL is: ${availability_and_continuity_bandwidth_NEFURL}
${result}= iterateForAUE ${availability_and_continuity_bandwidth_NEFURL} ${availability_and_continuity_bandwidth_UENumber} ${availability_and_continuity_bandwidth_DurationForUE4}
Should Not Be Empty ${availability_and_continuity_bandwidth_NEFUSERNAME}
Log The NEF USERNAME is: ${availability_and_continuity_bandwidth_NEFUSERNAME}
Should Not Be Empty ${availability_and_continuity_bandwidth_NEFPASSWORD}
Log The NEF PASSWORD is: ${availability_and_continuity_bandwidth_NEFPASSWORD}
Should Not Be Empty ${availability_and_continuity_bandwidth_LIVENESS}
Log The LIVENESS ENDPOINT is: ${availability_and_continuity_bandwidth_LIVENESS}
${result}= iterateForAUE ${availability_and_continuity_bandwidth_NEFURL} ${availability_and_continuity_bandwidth_NEFUSERNAME} ${availability_and_continuity_bandwidth_NEFPASSWORD} ${availability_and_continuity_bandwidth_UENumber} ${availability_and_continuity_bandwidth_DurationForUE4} ${availability_and_continuity_bandwidth_LIVENESS}
IF '${result}' != '-1'
Should Be Equal As Strings ${result} Success
ELSE
Expand Down
5 changes: 4 additions & 1 deletion ftp/tests/availability_and_continuity_bandwidth/run_test.bat
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,9 @@
@echo off
REM Set Environment Variables
set availability_and_continuity_bandwidth_NEFURL=http://10.10.10.20:8888
set availability_and_continuity_bandwidth_NEFURL= http://10.10.10.20:8888
set availability_and_continuity_bandwidth_NEFUSERNAME=admin@my-email.com
set availability_and_continuity_bandwidth_NEFPASSWORD=pass
set availability_and_continuity_bandwidth_LIVENESS=http://10.10.10.221:85

REM Run Robot Framework Test
robot availability_and_continuity_bandwidth.robot
4 changes: 4 additions & 0 deletions ftp/tests/availability_and_continuity_bandwidth/run_test.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
#!/bin/bash
export availability_and_continuity_bandwidth_NEFURL=http://10.10.10.20:8888
export availability_and_continuity_bandwidth_NEFUSERNAME=admin@my-email.com
export availability_and_continuity_bandwidth_NEFPASSWORD=pass
export availability_and_continuity_bandwidth_LIVENESS=http://10.10.10.221:85

python3 -m robot availability_and_continuity_bandwidth.robot
123 changes: 25 additions & 98 deletions ftp/tests/availability_and_continuity_communication_loss/TCTest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,6 @@ def install_requirements():
# ----------------------- END OF INSTALL REQUIREMENTS ----------------------- #

import requests
import socket
import dns.resolver
import logging
import time

Expand All @@ -38,8 +36,8 @@ def getAuthBearer(username: str, password: str, nef_url: str) -> str:
}
data = {
"grant_type": "",
"username": "[email protected]",
"password": "pass",
"username": username,
"password": password,
"scope": "",
"client_id": "",
"client_secret": ""
Expand All @@ -58,59 +56,6 @@ def getAuthBearer(username: str, password: str, nef_url: str) -> str:
logging.error("Request failed with status code:" + str(response.status_code) + response.content.decode('utf-8'))
return None

def getIP():

# Set the hostname that you want to look up
hostname = "cameraaistr.fidegad.external.ip"

try:
# Use the socket module to get the IP address of the pod's default DNS server
dns_server = socket.gethostbyname("kube-dns.kube-system.svc.cluster.local")
except:
return "127.0.0.1"
# Use the dns.resolver module to perform the nslookup using the pod's default DNS server
resolver = dns.resolver.Resolver(configure=False)
resolver.nameservers = [dns_server]
result = resolver.resolve(hostname)

# Print the IP addresses returned by the nslookup
for ip in result:
return ip

def createNEFMonitoringSubscription(bearer, nef_url, externalId):
streaming_server_ip_address = getIP()
print("Streaming Server IP address by kubernetes: " + str(streaming_server_ip_address))
headers = {
"accept": "application/json",
"Content-Type": "application/json",
"Authorization": "Bearer " + bearer
}
data = {
"externalId": externalId,
"notificationDestination": "http://" + str(streaming_server_ip_address) + ":85/handover_event_hook",
"monitoringType": "LOCATION_REPORTING",
"maximumNumberOfReports": 1000,
"monitorExpireTime": "2025-02-12T09:06:50.811Z",
"maximumDetectionTime": 1000,
"reachabilityType": "DATA"
}

response = requests.post(
nef_url.rstrip('/') + "/nef/api/v1/3gpp-monitoring-event/v1/myNetapp/subscriptions"
, headers=headers
, json=data
)

if response.status_code == 201:
logging.info("Subscription succeeded with status code:" + str(response.status_code) + response.content.decode('utf-8'))
# Do something with the response data
data = response.json()
logging.info(data)
return True
else:
# Handle the error
logging.error("Subscription request failed with status code:" + str(response.status_code) + response.content.decode('utf-8'))
return False

def startMovingUE(bearer, nef_url, supi):
headers = {
Expand Down Expand Up @@ -168,44 +113,10 @@ def stopMovingUE(bearer, nef_url, supi):
logging.info("Stop looping of UE with SUPI " + supi + " failed with status code:" + str(response.status_code) + response.content.decode('utf-8'))
return False

def deleteNEFMonitoringSubscription(bearer, nef_url, externalId):
headers = {
"accept": "application/json",
"Content-Type": "application/json",
"Authorization": "Bearer " + bearer
}
response = requests.get(
nef_url.rstrip('/') + "/nef/api/v1/3gpp-monitoring-event/v1/myNetapp/subscriptions"
, headers=headers
)

if 200 <= response.status_code < 300:
logging.info("Get Subscription succeeded with status code:" + str(response.status_code) + response.content.decode('utf-8'))
if len(response.content) > 0:
# Do something with the response data
data = response.json()
for row in data:
if row["externalId"] == externalId:
delete_response = requests.delete(row["link"], headers=headers)
if delete_response.status_code == 200:
logging.info("Deletion of Subscription request of " + row["externalId"] + " succeeded with status code: " + str(
delete_response.status_code) + delete_response.content.decode('utf-8'))
else:
logging.error("Deletion of Subscription request of " + row["externalId"] + " failed with status code: " + str(
delete_response.status_code) + delete_response.content.decode('utf-8'))
return False
logging.info(data)
else:
logging.info("No active subscriptions found")
return True
else:
# Handle the error
logging.error("Subscription deletion request failed with status code:" + str(response.status_code) + response.content.decode('utf-8'))
return False

def iterateForAUE(nef_url: str, ue: str, timeforacircle: str)-> str :
def iterateForAUE(nef_url: str, nef_username: str, nef_password: str, ue: str, timeforacircle: str, liveness_url: str)-> str :
try:
bearer = getAuthBearer("[email protected]", "pass", nef_url)
bearer = getAuthBearer(nef_username, nef_password, nef_url)
if bearer is not None:
stopMovingUE(bearer, nef_url, "202010000000004")
stopMovingUE(bearer, nef_url, "202010000000005")
Expand All @@ -220,20 +131,34 @@ def iterateForAUE(nef_url: str, ue: str, timeforacircle: str)-> str :
return -1

try:
bearer = getAuthBearer("[email protected]", "pass", nef_url)
bearer = getAuthBearer(nef_username, nef_password, nef_url)
if bearer is not None:
createNEFMonitoringSubscription(bearer, nef_url, "1000" + ue + "@domain.com")
startMovingUE(bearer, nef_url, "20201000000000" + ue)
time.sleep(int(timeforacircle) * 2)
stopMovingUE(bearer, nef_url, "20201000000000" + ue)
deleteNEFMonitoringSubscription(bearer, nef_url, "1000" + ue + "@domain.com")
return "Success"
if check_liveness(liveness_url):
return "Success"
else:
return -1
except ConnectionError:
logging.error("Connection with NEF failed!")
except:
logging.error("Communication with NEF failed!")
return -1

def check_liveness(url):
try:
response = requests.get(url)
# A status code of 200 means OK, which indicates the page is live.
if response.status_code == 200:
print(f"The page at {url} is live. Status Code: {response.status_code}")
return True
else:
print(f"The page at {url} encountered an issue. Status Code: {response.status_code}")
return False
except requests.ConnectionError:
print(f"Failed to connect to {url}")
return False
# -------------------------- UNINSTALL REQUIREMENTS ------------------------- #
def uninstall_requirements():
curr_dir = os.path.dirname(os.path.realpath(__file__))
Expand All @@ -257,5 +182,7 @@ def uninstall_requirements():

if __name__ == '__main__':
nef_url = "http://10.10.10.20:8888"
iterateForAUE(nef_url, "8", "33")
nef_username = "[email protected]"
nef_password = "pass"
iterateForAUE(nef_url, nef_username, nef_password, "4", "36")

Loading