From b5aba72a9675eca2b04825912e2e149ddefa5ba5 Mon Sep 17 00:00:00 2001
From: rahhal <mrahhal@cttc.es>
Date: Mon, 12 Aug 2024 15:23:16 +0000
Subject: [PATCH] Merged
---
src/nbi/Dockerfile | 13 +
.../nbi_plugins/camara_qod/Resources.py | 509 +++++++++---------
src/nbi/tests/test_camara_qod.py | 178 +++---
3 files changed, 373 insertions(+), 327 deletions(-)
diff --git a/src/nbi/Dockerfile b/src/nbi/Dockerfile
index 1435e9757..e3baa5b56 100644
--- a/src/nbi/Dockerfile
+++ b/src/nbi/Dockerfile
@@ -78,6 +78,7 @@ COPY src/nbi/requirements.in requirements.in
RUN pip-compile --quiet --output-file=requirements.txt requirements.in
RUN python3 -m pip install -r requirements.txt
+
# Add component files into working directory
WORKDIR /var/teraflow
COPY src/nbi/. nbi/
@@ -92,5 +93,17 @@ COPY src/slice/client/. slice/client/
RUN mkdir -p /var/teraflow/tests/tools
COPY src/tests/tools/mock_osm/. tests/tools/mock_osm/
+# Add component sub-folder for qos_profile
+RUN mkdir -p /var/teraflow/qos_profile
+WORKDIR /var/teraflow/qos_profile
+COPY src/qos_profile/requirements.in requirements.in
+RUN pip-compile --quiet --output-file=requirements.txt requirements.in
+RUN python3 -m pip install -r requirements.txt
+
+# Add qos_profile files into working directory
+WORKDIR /var/teraflow
+COPY src/qos_profile/. qos_profile/
+
+
# Start the service
ENTRYPOINT ["python", "-m", "nbi.service"]
diff --git a/src/nbi/service/rest_server/nbi_plugins/camara_qod/Resources.py b/src/nbi/service/rest_server/nbi_plugins/camara_qod/Resources.py
index 093802fbd..15a0205fe 100644
--- a/src/nbi/service/rest_server/nbi_plugins/camara_qod/Resources.py
+++ b/src/nbi/service/rest_server/nbi_plugins/camara_qod/Resources.py
@@ -11,274 +11,273 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-import json
-import profile
+
from flask.json import jsonify
from flask_restful import Resource, request
from enum import Enum
import uuid
+from src.qos_profile.client import QoSProfileClient
+from common.proto.context_pb2 import Empty, Uuid, QoSProfileValueUnitPair, QoSProfileId, QoSProfile
+import json
+import uuid
+from flask.json import jsonify
+from src.qos_profile.client import QoSProfileClient
+from src.context.service.database.QoSProfile import grpc_message_to_qos_table_data
+from src.qos_profile.tests.test_crud import create_qos_profile_from_json
-# Enum for session status
+
class SessionStatus(Enum):
ACTIVE = 'ACTIVE'
INACTIVE = 'INACTIVE'
+#
+#class Rate:
+# def __init__(self, value, unit):
+# self.value = value
+# self.unit = unit
+#
+#class Duration:
+# def __init__(self, value, unit):
+# self.value = value
+# self.unit = unit
+#
+#class ProfileCreate:
+# def __init__(self, name, description, status, targetMinUpstreamRate=None, maxUpstreamRate=None, maxUpstreamBurstRate=None,
+# targetMinDownstreamRate=None, maxDownstreamRate=None, maxDownstreamBurstRate=None, minDuration=None, maxDuration=None,
+# priority=None, packetDelayBudget=None, jitter=None, packetErrorLossRate=None):
+# self.name = name
+# self.description = description
+# self.status = status
+# self.targetMinUpstreamRate = targetMinUpstreamRate
+# self.maxUpstreamRate = maxUpstreamRate
+# self.maxUpstreamBurstRate = maxUpstreamBurstRate
+# self.targetMinDownstreamRate = targetMinDownstreamRate
+# self.maxDownstreamRate = maxDownstreamRate
+# self.maxDownstreamBurstRate = maxDownstreamBurstRate
+# self.minDuration = minDuration
+# self.maxDuration = maxDuration
+# self.priority = priority
+# self.packetDelayBudget = packetDelayBudget
+# self.jitter = jitter
+# self.packetErrorLossRate = packetErrorLossRate
+#
+#class Profile(ProfileCreate):
+# def __init__(self, profile_id, **kwargs):
+# super().__init__(**kwargs)
+# self.profile_id = profile_id
+#
+#profiles = []
-class Rate:
- def __init__(self, value, unit):
- self.value = value
- self.unit = unit
-
-class Duration:
- def __init__(self, value, unit):
- self.value = value
- self.unit = unit
-
-class ProfileCreate:
- def __init__(self, name, description, status, targetMinUpstreamRate=None, maxUpstreamRate=None, maxUpstreamBurstRate=None,
- targetMinDownstreamRate=None, maxDownstreamRate=None, maxDownstreamBurstRate=None, minDuration=None, maxDuration=None,
- priority=None, packetDelayBudget=None, jitter=None, packetErrorLossRate=None):
- self.name = name
- self.description = description
- self.status = status
- self.targetMinUpstreamRate = targetMinUpstreamRate
- self.maxUpstreamRate = maxUpstreamRate
- self.maxUpstreamBurstRate = maxUpstreamBurstRate
- self.targetMinDownstreamRate = targetMinDownstreamRate
- self.maxDownstreamRate = maxDownstreamRate
- self.maxDownstreamBurstRate = maxDownstreamBurstRate
- self.minDuration = minDuration
- self.maxDuration = maxDuration
- self.priority = priority
- self.packetDelayBudget = packetDelayBudget
- self.jitter = jitter
- self.packetErrorLossRate = packetErrorLossRate
-
-class Profile(ProfileCreate):
- def __init__(self, profile_id, **kwargs):
- super().__init__(**kwargs)
- self.profile_id = profile_id
-
-profiles = []
+class _Resource(Resource):
+ def __init__(self) -> None:
+ super().__init__()
+ self.client = QoSProfileClient()
class ProfileList(Resource):
def post(self):
data = request.get_json()
- profile_id = str(uuid.uuid4())
- new_profile = Profile(
- profile_id=profile_id,
- name=data.get('name'),
- description=data.get('description'),
- status=data.get('status', SessionStatus.ACTIVE.value),
- targetMinUpstreamRate=data.get('targetMinUpstreamRate'),
- maxUpstreamRate=data.get('maxUpstreamRate'),
- maxUpstreamBurstRate=data.get('maxUpstreamBurstRate'),
- targetMinDownstreamRate=data.get('targetMinDownstreamRate'),
- maxDownstreamRate=data.get('maxDownstreamRate'),
- maxDownstreamBurstRate=data.get('maxDownstreamBurstRate'),
- minDuration=data.get('minDuration'),
- maxDuration=data.get('maxDuration'),
- priority=data.get('priority'),
- packetDelayBudget=data.get('packetDelayBudget'),
- jitter=data.get('jitter'),
- packetErrorLossRate=data.get('packetErrorLossRate')
- )
- profiles.append(new_profile.__dict__)
- return jsonify(new_profile.__dict__), 201
-
-class ProfileDetail(Resource):
- def get(self, profile_id):
- profile = next((p for p in profiles if p["profile_id"] == profile_id), None)
- if profile:
- return jsonify(profile)
- return {"message": "Profile not found"}, 404
+ qos_profile = create_qos_profile_from_json(data)
+ try:
+ self.client.CreateQoSProfile(qos_profile)
+ return jsonify({"status": "QoS Profile created successfully"}), 201
+ except Exception as e:
+ # Step 4: Handle any exceptions that occur
+ return jsonify({"error": str(e)}), 500
+
+class ProfileDetail(_Resource):
+ def get(self,qos_profile_id):
+ qos_get_profile=self.client.GetQoSProfile(qos_profile_id)
+ qos_table_data = grpc_message_to_qos_table_data(qos_get_profile)
+ return qos_table_data
+
+# def put(self, profile_id):
+# data = request.get_json()
+# profile = next((p for p in profiles if p["profile_id"] == profile_id), None)
+# if profile:
+# for key, value in data.items():
+# if key in profile:
+# profile[key] = value
+# return jsonify(profile)
+# return {"message": "Profile not found"}, 404
+#
+#class Profile_delete_by_name(Resource):
+# def delete(self, name):
+# global profiles
+# profile = next((p for p in profiles if p["name"] == name), None)
+# if profile:
+# profiles = [p for p in profiles if p["name"] != name]
+# return {"message": "Profile deleted successfully"}, 200
+# return {"message": "Profile not found"}, 404
+
+#class AllProfiles(Resource):
+# def get(self):
+# qos_profiles_got = list(qos_profile_client.GetQoSProfiles(Empty()))
- def put(self, profile_id):
- data = request.get_json()
- profile = next((p for p in profiles if p["profile_id"] == profile_id), None)
- if profile:
- for key, value in data.items():
- if key in profile:
- profile[key] = value
- return jsonify(profile)
- return {"message": "Profile not found"}, 404
-class Profile_delete_by_name(Resource):
- def delete(self, name):
- global profiles
- profile = next((p for p in profiles if p["name"] == name), None)
- if profile:
- profiles = [p for p in profiles if p["name"] != name]
- return {"message": "Profile deleted successfully"}, 200
- return {"message": "Profile not found"}, 404
-
-class AllProfiles(Resource):
- def get(self):
- return jsonify(profiles)
-
-class Delete_all_profile(Resource):
- def delete(self):
- global profiles
- profile_count = len(profiles)
- profiles = []
- return {"message": f"Deleted {profile_count} sessions."}, 200
-
-
-class PortRange:
- def __init__(self, from_port, to_port):
- self.from_port = from_port
- self.to_port = to_port
-
-class PortDetails:
- def __init__(self, ranges=None, ports=None):
- self.ranges = ranges or []
- self.ports = ports or []
-
-class Device:
- def __init__(self, phoneNumber, networkAccessIdentifier, publicAddress, publicPort, ipv6Address):
- self.phoneNumber = phoneNumber
- self.networkAccessIdentifier = networkAccessIdentifier
- self.publicAddress = publicAddress
- self.publicPort = publicPort
- self.ipv6Address = ipv6Address
-
-class ApplicationServer:
- def __init__(self, ipv4Address, ipv6Address):
- self.ipv4Address = ipv4Address
- self.ipv6Address = ipv6Address
-
-class Webhook:
- def __init__(self, notificationUrl, notificationAuthToken):
- self.notificationUrl = notificationUrl
- self.notificationAuthToken = notificationAuthToken
-
-class SessionCreate:
- def __init__(self, device, applicationServer, devicePorts, applicationServerPorts, qosProfile, webhook, duration):
- self.device = device
- self.applicationServer = applicationServer
- self.devicePorts = devicePorts
- self.applicationServerPorts = applicationServerPorts
- self.qosProfile = qosProfile
- self.webhook = webhook
- self.duration = duration
-
-sessions = []
-
-class SessionList(Resource):
- def post(self):
- data = request.get_json()
- session_id = str(uuid.uuid4())
- profile_name = data.get('qosProfile')
-
- profile = next((p for p in profiles if p["name"] == profile_name), None)
- if not profile:
- return jsonify({"error": "QoS profile not found"}), 404
-
- device = Device(
- phoneNumber=data['device']['phoneNumber'],
- networkAccessIdentifier=data['device']['networkAccessIdentifier'],
- publicAddress=data['device']['ipv4Address']['publicAddress'],
- publicPort=data['device']['ipv4Address']['publicPort'],
- ipv6Address=data['device']['ipv6Address']
- )
-
- applicationServer = ApplicationServer(
- ipv4Address=data['applicationServer']['ipv4Address'],
- ipv6Address=data['applicationServer']['ipv6Address']
- )
-
- devicePorts = PortDetails(
- ranges=[PortRange(r['from'], r['to']) for r in data['devicePorts']['ranges']],
- ports=data['devicePorts']['ports']
- )
-
- applicationServerPorts = PortDetails(
- ranges=[PortRange(r['from'], r['to']) for r in data['applicationServerPorts']['ranges']],
- ports=data['applicationServerPorts']['ports']
- )
-
- webhook = Webhook(
- notificationUrl=data['webhook']['notificationUrl'],
- notificationAuthToken=data['webhook']['notificationAuthToken']
- )
-
- new_session = SessionCreate(
- device=device,
- applicationServer=applicationServer,
- devicePorts=devicePorts,
- applicationServerPorts=applicationServerPorts,
- qosProfile=profile_name,
- webhook=webhook,
- duration=data['duration']
- )
-
- session_data = {
- "session_id": session_id,
- "device": {
- "phoneNumber": device.phoneNumber,
- "networkAccessIdentifier": device.networkAccessIdentifier,
- "ipv4Address": {
- "publicAddress": device.publicAddress,
- "publicPort": device.publicPort
- },
- "ipv6Address": device.ipv6Address
- },
- "applicationServer": {
- "ipv4Address": applicationServer.ipv4Address,
- "ipv6Address": applicationServer.ipv6Address
- },
- "devicePorts": {
- "ranges": [{"from": r.from_port, "to": r.to_port} for r in devicePorts.ranges],
- "ports": devicePorts.ports
- },
- "applicationServerPorts": {
- "ranges": [{"from": r.from_port, "to": r.to_port} for r in applicationServerPorts.ranges],
- "ports": applicationServerPorts.ports
- },
- "qosProfile": new_session.qosProfile,
- "webhook": {
- "notificationUrl": webhook.notificationUrl,
- "notificationAuthToken": webhook.notificationAuthToken
- },
- "duration": new_session.duration,
- "status": SessionStatus.ACTIVE.value
- }
-
- sessions.append(session_data)
- return jsonify(session_data), 201
-
-class SessionDetail(Resource):
- def get(self, session_id):
- session = next((s for s in sessions if s["session_id"] == session_id), None)
- if session:
- return jsonify(session)
- return {"message": "Session not found"}, 404
-
- def put(self, session_id):
- data = request.get_json()
- session = next((s for s in sessions if s["session_id"] == session_id), None)
- if session:
- for key, value in data.items():
- if key in session:
- session[key] = value
- return jsonify(session)
- return {"message": "Session not found"}, 404
-
- def delete(self, session_id):
- global sessions
- session = next((s for s in sessions if s["session_id"] == session_id), None)
- if session:
- sessions = [s for s in sessions if s["session_id"] != session_id]
- return {"message": "Session deleted successfully"}, 200
- return {"message": "Session not found"}, 404
-
-class AllSessions(Resource):
- def get(self):
- return jsonify(sessions)
-
-class DeleteAllSessions(Resource):
- def delete(self):
- global sessions
- session_count = len(sessions)
- sessions = []
- return {"message": f"Deleted {session_count} sessions."}, 200
+#class Delete_all_profile(Resource):
+# def delete(self):
+# global profiles
+# profile_count = len(profiles)
+# profiles = []
+# return {"message": f"Deleted {profile_count} sessions."}, 200
+
+####session####
+#class PortRange:
+# def __init__(self, from_port, to_port):
+# self.from_port = from_port
+# self.to_port = to_port
+#
+#class PortDetails:
+# def __init__(self, ranges=None, ports=None):
+# self.ranges = ranges or []
+# self.ports = ports or []
+#
+#class Device:
+# def __init__(self, phoneNumber, networkAccessIdentifier, publicAddress, publicPort, ipv6Address):
+# self.phoneNumber = phoneNumber
+# self.networkAccessIdentifier = networkAccessIdentifier
+# self.publicAddress = publicAddress
+# self.publicPort = publicPort
+# self.ipv6Address = ipv6Address
+#
+#class ApplicationServer:
+# def __init__(self, ipv4Address, ipv6Address):
+# self.ipv4Address = ipv4Address
+# self.ipv6Address = ipv6Address
+#
+#class Webhook:
+# def __init__(self, notificationUrl, notificationAuthToken):
+# self.notificationUrl = notificationUrl
+# self.notificationAuthToken = notificationAuthToken
+#
+#class SessionCreate:
+# def __init__(self, device, applicationServer, devicePorts, applicationServerPorts, qosProfile, webhook, duration):
+# self.device = device
+# self.applicationServer = applicationServer
+# self.devicePorts = devicePorts
+# self.applicationServerPorts = applicationServerPorts
+# self.qosProfile = qosProfile
+# self.webhook = webhook
+# self.duration = duration
+#
+#sessions = []
+#
+#class SessionList(Resource):
+# def post(self):
+# data = request.get_json()
+# session_id = str(uuid.uuid4())
+# profile_name = data.get('qosProfile')
+#
+# profile = next((p for p in profiles if p["name"] == profile_name), None)
+# if not profile:
+# return jsonify({"error": "QoS profile not found"}), 404
+#
+# device = Device(
+# phoneNumber=data['device']['phoneNumber'],
+# networkAccessIdentifier=data['device']['networkAccessIdentifier'],
+# publicAddress=data['device']['ipv4Address']['publicAddress'],
+# publicPort=data['device']['ipv4Address']['publicPort'],
+# ipv6Address=data['device']['ipv6Address']
+# )
+#
+# applicationServer = ApplicationServer(
+# ipv4Address=data['applicationServer']['ipv4Address'],
+# ipv6Address=data['applicationServer']['ipv6Address']
+# )
+#
+# devicePorts = PortDetails(
+# ranges=[PortRange(r['from'], r['to']) for r in data['devicePorts']['ranges']],
+# ports=data['devicePorts']['ports']
+# )
+#
+# applicationServerPorts = PortDetails(
+# ranges=[PortRange(r['from'], r['to']) for r in data['applicationServerPorts']['ranges']],
+# ports=data['applicationServerPorts']['ports']
+# )
+#
+# webhook = Webhook(
+# notificationUrl=data['webhook']['notificationUrl'],
+# notificationAuthToken=data['webhook']['notificationAuthToken']
+# )
+#
+# new_session = SessionCreate(
+# device=device,
+# applicationServer=applicationServer,
+# devicePorts=devicePorts,
+# applicationServerPorts=applicationServerPorts,
+# qosProfile=profile_name,
+# webhook=webhook,
+# duration=data['duration']
+# )
+#
+# session_data = {
+# "session_id": session_id,
+# "device": {
+# "phoneNumber": device.phoneNumber,
+# "networkAccessIdentifier": device.networkAccessIdentifier,
+# "ipv4Address": {
+# "publicAddress": device.publicAddress,
+# "publicPort": device.publicPort
+# },
+# "ipv6Address": device.ipv6Address
+# },
+# "applicationServer": {
+# "ipv4Address": applicationServer.ipv4Address,
+# "ipv6Address": applicationServer.ipv6Address
+# },
+# "devicePorts": {
+# "ranges": [{"from": r.from_port, "to": r.to_port} for r in devicePorts.ranges],
+# "ports": devicePorts.ports
+# },
+# "applicationServerPorts": {
+# "ranges": [{"from": r.from_port, "to": r.to_port} for r in applicationServerPorts.ranges],
+# "ports": applicationServerPorts.ports
+# },
+# "qosProfile": new_session.qosProfile,
+# "webhook": {
+# "notificationUrl": webhook.notificationUrl,
+# "notificationAuthToken": webhook.notificationAuthToken
+# },
+# "duration": new_session.duration,
+# "status": SessionStatus.ACTIVE.value
+# }
+#
+# sessions.append(session_data)
+# return jsonify(session_data), 201
+#
+#class SessionDetail(Resource):
+# def get(self, session_id):
+# session = next((s for s in sessions if s["session_id"] == session_id), None)
+# if session:
+# return jsonify(session)
+# return {"message": "Session not found"}, 404
+#
+# def put(self, session_id):
+# data = request.get_json()
+# session = next((s for s in sessions if s["session_id"] == session_id), None)
+# if session:
+# for key, value in data.items():
+# if key in session:
+# session[key] = value
+# return jsonify(session)
+# return {"message": "Session not found"}, 404
+#
+# def delete(self, session_id):
+# global sessions
+# session = next((s for s in sessions if s["session_id"] == session_id), None)
+# if session:
+# sessions = [s for s in sessions if s["session_id"] != session_id]
+# return {"message": "Session deleted successfully"}, 200
+# return {"message": "Session not found"}, 404
+#
+#class AllSessions(Resource):
+# def get(self):
+# return jsonify(sessions)
+#
+#class DeleteAllSessions(Resource):
+# def delete(self):
+# global sessions
+# session_count = len(sessions)
+# sessions = []
+# return {"message": f"Deleted {session_count} sessions."}, 200
+#
\ No newline at end of file
diff --git a/src/nbi/tests/test_camara_qod.py b/src/nbi/tests/test_camara_qod.py
index f89436b2e..f23ff84d8 100644
--- a/src/nbi/tests/test_camara_qod.py
+++ b/src/nbi/tests/test_camara_qod.py
@@ -1,81 +1,115 @@
+from urllib import response
import requests
BASE_URL = 'http://10.1.7.197/camara/qod/v0'
def test_create_profile():
response = requests.post(f'{BASE_URL}/profiles', json={
- "name": "Test Profile",
- "description": "This is a test profile",
- "status": "ACTIVE",
- "targetMinUpstreamRate": {"value": 1000, "unit": "Kbps"},
- "maxUpstreamRate": {"value": 5000, "unit": "Kbps"},
- "maxUpstreamBurstRate": {"value": 10000, "unit": "Kbps"},
- "targetMinDownstreamRate": {"value": 2000, "unit": "Kbps"},
- "maxDownstreamRate": {"value": 10000, "unit": "Kbps"},
- "maxDownstreamBurstRate": {"value": 20000, "unit": "Kbps"},
- "minDuration": {"value": 1, "unit": "hours"},
- "maxDuration": {"value": 24, "unit": "hours"},
- "priority": 1,
- "packetDelayBudget": {"value": 100, "unit": "ms"},
- "jitter": {"value": 10, "unit": "ms"},
- "packetErrorLossRate": 0.01
- })
+"qos_profile_id": "f00406f5-8e36-4abc-a0ec-b871c7f062b7",
+"name": "QCI_1_voice",
+"description": "QoS profile for video streaming",
+"status": "ACTIVE",
+"targetMinUpstreamRate": {
+ "value": 10,
+ "unit": "bps"
+},
+"maxUpstreamRate": {
+ "value": 10,
+ "unit": "bps"
+},
+"maxUpstreamBurstRate": {
+ "value": 10,
+ "unit": "bps"
+},
+"targetMinDownstreamRate": {
+ "value": 10,
+ "unit": "bps"
+},
+"maxDownstreamRate": {
+ "value": 10,
+ "unit": "bps"
+},
+"maxDownstreamBurstRate": {
+ "value": 10,
+ "unit": "bps"
+},
+"minDuration": {
+ "value": 12,
+ "unit": "Minutes"
+},
+"maxDuration": {
+ "value": 12,
+ "unit": "Minutes"
+},
+"priority": 20,
+"packetDelayBudget": {
+ "value": 12,
+ "unit": "Minutes"
+},
+"jitter": {
+ "value": 12,
+ "unit": "Minutes"
+},
+"packetErrorLossRate": 3
+}
+)
+def get_profile():
+ response=requests.get(f'{BASE_URL}/profile/ProfileDetail')
+#def test_delete_profile_by_name():
+# response = requests.delete(f'{BASE_URL}/profiles/delete_by_name/Test Profile')
-
-def test_update_profile():
- response = requests.put(f'{BASE_URL}/profiles/270a9c75-6a4e-452e-9dc5-804fc0204dcb', json={
- "description": "AM UPDATING THIS ",
- "name":"test2"
- })
-
-
-def test_delete_profile_by_name():
- response = requests.delete(f'{BASE_URL}/profiles/delete_by_name/Test Profile')
-
-def test_create_session():
- response = requests.post(f'{BASE_URL}/sessions', json={
- "device": {
- "phoneNumber": "1234567890",
- "networkAccessIdentifier": "nai",
- "ipv4Address": {"publicAddress": "84.125.93.10", "publicPort": 59765},
- "ipv6Address": "2001:db8:85a3:8d3:1319:8a2e:370:7344"
- },
- "applicationServer": {
- "ipv4Address": "192.168.0.1/24",
- "ipv6Address": "2001:db8:85a3:8d3:1319:8a2e:370:7344"
- },
- "devicePorts": {
- "ranges": [{"from": 5010, "to": 5020}],
- "ports": [5060, 5070]
- },
- "applicationServerPorts": {
- "ranges": [{"from": 6010, "to": 6020}],
- "ports": [6060, 6070]
- },
- "qosProfile": "Test Profile",
- "webhook": {
- "notificationUrl": "https://application-server.com",
- "notificationAuthToken": "c8974e592c2fa383d4a3960714"
- },
- "duration": {"value": 1, "unit": "hours"}
- })
-
-
-def test_update_session():
- response = requests.put(f'{BASE_URL}/sessions/88c51c03-13bf-4542-b3f2-43a9edccfc95', json={
- "duration": {"value": 32, "unit": "hours"}
- })
-
-def test_delete_session():
- response = requests.delete(f'{BASE_URL}/sessions/e1b53005-ffba-4e66-b7fe-0c5e022d8d7d')
-
-#def test_delete_all_sessions():
- #BASE_URL = 'http://10.1.7.197/camara/qod/v0'
- #DELETE_ALL_URL = f"{BASE_URL}/sessions/delete_all"
- #response = requests.delete(DELETE_ALL_URL)
-#def test_delete_all_profiles():
- #BASE_URL = 'http://10.1.7.197/camara/qod/v0'
- #DELETE_ALL_URL_Profile = f"{BASE_URL}/profiles/delete_all"
- #response = requests.delete(DELETE_ALL_URL_Profile)
\ No newline at end of file
+#def test_update_profile():
+# response = requests.put(f'{BASE_URL}/profiles/270a9c75-6a4e-452e-9dc5-804fc0204dcb', json={
+# "description": "AM UPDATING THIS ",
+# "name":"test2"
+# })
+#
+#
+#
+#def test_create_session():
+# response = requests.post(f'{BASE_URL}/sessions', json={
+# "device": {
+# "phoneNumber": "1234567890",
+# "networkAccessIdentifier": "nai",
+# "ipv4Address": {"publicAddress": "84.125.93.10", "publicPort": 59765},
+# "ipv6Address": "2001:db8:85a3:8d3:1319:8a2e:370:7344"
+# },
+# "applicationServer": {
+# "ipv4Address": "192.168.0.1/24",
+# "ipv6Address": "2001:db8:85a3:8d3:1319:8a2e:370:7344"
+# },
+# "devicePorts": {
+# "ranges": [{"from": 5010, "to": 5020}],
+# "ports": [5060, 5070]
+# },
+# "applicationServerPorts": {
+# "ranges": [{"from": 6010, "to": 6020}],
+# "ports": [6060, 6070]
+# },
+# "qosProfile": "Test Profile",
+# "webhook": {
+# "notificationUrl": "https://application-server.com",
+# "notificationAuthToken": "c8974e592c2fa383d4a3960714"
+# },
+# "duration": {"value": 1, "unit": "hours"}
+# })
+#
+#
+#def test_update_session():
+# response = requests.put(f'{BASE_URL}/sessions/88c51c03-13bf-4542-b3f2-43a9edccfc95', json={
+# "duration": {"value": 32, "unit": "hours"}
+# })
+#
+#def test_delete_session():
+# response = requests.delete(f'{BASE_URL}/sessions/e1b53005-ffba-4e66-b7fe-0c5e022d8d7d')
+#
+##def test_delete_all_sessions():
+# #BASE_URL = 'http://10.1.7.197/camara/qod/v0'
+# #DELETE_ALL_URL = f"{BASE_URL}/sessions/delete_all"
+# #response = requests.delete(DELETE_ALL_URL)
+##def test_delete_all_profiles():
+# #BASE_URL = 'http://10.1.7.197/camara/qod/v0'
+# #DELETE_ALL_URL_Profile = f"{BASE_URL}/profiles/delete_all"
+# #response = requests.delete(DELETE_ALL_URL_Profile)
\ No newline at end of file
--
GitLab