import conductor.api.controllers.v1.plans
import conductor.common.music.api
import conductor.data.plugins.service_controller.sdnc
-
-
+from conductor.common.utils import cipherUtils
LOG = log.getLogger(__name__)
config.set_override('username', secret_dict['aai']['username'], 'aai')
config.set_override('password', secret_dict['aai']['password'], 'aai')
config.set_override('username', secret_dict['conductor_api']['username'], 'conductor_api')
- config.set_override('password', secret_dict['conductor_api']['password'], 'conductor_api')
+ config.set_override('password', decrypt_pass(secret_dict['conductor_api']['password']), 'conductor_api')
config.set_override('aafuser', secret_dict['music_api']['aafuser'], 'music_api')
config.set_override('aafpass', secret_dict['music_api']['aafpass'], 'music_api')
config.set_override('aafns', secret_dict['music_api']['aafns'], 'music_api')
config.set_override('aaf_conductor_user', secret_dict['aaf_api']['aaf_conductor_user'], 'aaf_api')
+def decrypt_pass(passwd):
+ if passwd == '' or passwd == 'NA':
+ return passwd
+ else:
+ return cipherUtils.AESCipher.get_instance().decrypt(passwd)
+
+
def delete_secrets():
""" This is intended to delete the secrets for a clean initialization for
testing Application. Actual deployment will have a preload script.
#
# -------------------------------------------------------------------------
# Copyright (c) 2015-2017 AT&T Intellectual Property
+# Copyright (C) 2020 Wipro Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
VERSIONS = ["2016-11-01", "2017-10-10", "2018-02-01"]
LOCATION_KEYS = ['latitude', 'longitude', 'host_name', 'clli_code']
INVENTORY_PROVIDERS = ['aai']
-INVENTORY_TYPES = ['cloud', 'service', 'transport', 'vfmodule']
+INVENTORY_TYPES = ['cloud', 'service', 'transport', 'vfmodule', 'nssi']
DEFAULT_INVENTORY_PROVIDER = INVENTORY_PROVIDERS[0]
CANDIDATE_KEYS = ['candidate_id', 'cost', 'inventory_type', 'location_id',
'location_type']
'split': True,
'required': ['evaluate'],
},
+ 'threshold': {
+ 'split': True,
+ 'required': ['attribute', 'threshold', 'operator'],
+ 'optional': ['unit']
+ },
'distance_between_demands': {
'required': ['distance'],
'thresholds': {
raise TranslatorException(
"Optimization goal 'minimize' must "
"contain a single function of 'sum'.")
-
operands = optimization_copy['minimize']['sum']
if type(operands) is not list:
# or len(operands) != 2:
#
# -------------------------------------------------------------------------
# Copyright (c) 2015-2017 AT&T Intellectual Property
+# Copyright (C) 2020 Wipro Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# add candidate to demand candidates
resolved_demands[name].append(candidate)
+ elif inventory_type == 'nssi':
+ if filtering_attributes and model_invariant_id:
+ resolved_demands[name] = self.get_nssi_candidates(filtering_attributes,
+ model_invariant_id, model_version_id,
+ service_role, candidate_uniqueness)
+
else:
LOG.error("Unknown inventory_type "
" {}".format(inventory_type))
directives = None
return directives
+ def get_nssi_candidates(self, filtering_attributes, model_invariant_id, model_version_id, service_role,
+ candidate_uniqueness):
+ raw_path = ('nodes/service-instances' +
+ '?model-invariant-id={}'.format(model_invariant_id) +
+ ('&model-version-id={}'.format(model_version_id) if model_version_id else '') +
+ ('&service-role={}'.format(service_role) if service_role else '') +
+ '&depth=2')
+
+ path = self._aai_versioned_path(raw_path)
+ aai_response = self._request('get', path, data=None)
+
+ if aai_response is None or aai_response.status_code != 200:
+ return None
+ return self.filter_nssi_candidates(aai_response.json(), filtering_attributes, candidate_uniqueness)
+
+ def filter_nssi_candidates(self, response_body, filtering_attributes, candidate_uniqueness):
+
+ candidates = list()
+ if filtering_attributes and response_body is not None:
+ nssi_instances = response_body.get("service-instance", [])
+
+ for nssi_instance in nssi_instances:
+
+ inventory_attributes = dict()
+ inventory_attributes["orchestration-status"] = nssi_instance.get('orchestration-status')
+ inventory_attributes["service-role"] = nssi_instance.get('service-role')
+
+ if self.match_inventory_attributes(filtering_attributes, inventory_attributes,
+ nssi_instance.get('service-instance-id')):
+
+ properties = list()
+ relationships = nssi_instance['relationship-list']['relationship']
+ for relationship in relationships:
+ if relationship['related-to'] == 'service-instance':
+ properties = relationship['related-to-property']
+
+ nsi_name = None
+ if properties:
+ for prop in properties:
+ if prop['property-key'] == 'service-instance.service-instance-name':
+ nsi_name = prop['property-value']
+
+ slice_profiles = nssi_instance.get('slice-profiles').get('slice-profile')
+ slice_profile = min(slice_profiles, key=lambda x: x['latency'])
+
+ candidate = dict()
+ candidate['candidate_id'] = nssi_instance.get('service-instance-id')
+ candidate['instance_name'] = nssi_instance.get('service-instance-name')
+ candidate['cost'] = self.conf.data.nssi_candidate_cost
+ candidate['candidate_type'] = 'nssi'
+ candidate['inventory_type'] = 'nssi'
+ candidate['inventory_provider'] = 'aai'
+ candidate['domain'] = nssi_instance.get('environment-context')
+ candidate['latency'] = slice_profile.get('latency')
+ candidate['max_number_of_ues'] = slice_profile.get('max-number-of-UEs')
+ candidate['coverage_area_ta_list'] = slice_profile.get('coverage-area-TA-list')
+ candidate['ue_mobility_level'] = slice_profile.get('ue-mobility-level')
+ candidate['resource_sharing_level'] = slice_profile.get('resource-sharing-level')
+ candidate['exp_data_rate_ul'] = slice_profile.get('exp-data-rate-UL')
+ candidate['exp_data_rate_dl'] = slice_profile.get('exp-data-rate-DL')
+ candidate['area_traffic_cap_ul'] = slice_profile.get('area-traffic-cap-UL')
+ candidate['area_traffic_cap_dl'] = slice_profile.get('area-traffic-cap-DL')
+ candidate['activity_factor'] = slice_profile.get('activity-factor')
+ candidate['e2e_latency'] = slice_profile.get('e2e-latency')
+ candidate['jitter'] = slice_profile.get('jitter')
+ candidate['survival_time'] = slice_profile.get('survival-time')
+ candidate['exp_data_rate'] = slice_profile.get('exp-data-rate')
+ candidate['payload_size'] = slice_profile.get('payload-size')
+ candidate['traffic_density'] = slice_profile.get('traffic-density')
+ candidate['conn_density'] = slice_profile.get('conn-density')
+ candidate['reliability'] = slice_profile.get('reliability')
+ candidate['service_area_dimension'] = slice_profile.get('service-area-dimension')
+ candidate['cs_availability'] = slice_profile.get('cs-availability')
+ candidate['uniqueness'] = candidate_uniqueness
+ if nsi_name:
+ candidate['nsi_name'] = nsi_name
+ candidates.append(candidate)
+
+ return candidates
#
# -------------------------------------------------------------------------
# Copyright (c) 2015-2017 AT&T Intellectual Property
+# Copyright (C) 2020 Wipro Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
default=2.0),
cfg.FloatOpt('service_candidate_cost',
default=1.0),
+ cfg.FloatOpt('nssi_candidate_cost',
+ default=1.0),
]
CONF.register_opts(DATA_OPTS, group='data')
--- /dev/null
+#
+# -------------------------------------------------------------------------
+# Copyright (C) 2020 Wipro Limited.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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.
+#
+# -------------------------------------------------------------------------
+#
+
+from conductor.i18n import _LI
+from conductor.solver.optimizer.constraints import constraint
+from oslo_log import log
+
+LOG = log.getLogger(__name__)
+
+
+class Threshold(constraint.Constraint):
+
+ OPERATIONS = {'gte': lambda x, y: x >= y,
+ 'lte': lambda x, y: x <= y,
+ 'gt': lambda x, y: x > y,
+ 'lt': lambda x, y: x < y,
+ 'eq': lambda x, y: x == y
+ }
+
+ def __init__(self, _name, _type, _demand_list, _priority=0,
+ _properties=None):
+ constraint.Constraint.__init__(
+ self, _name, _type, _demand_list, _priority)
+ self.attribute = _properties.get('attribute')
+ self.operation = self.OPERATIONS.get(_properties.get('operator'))
+ self.threshold = _properties.get('threshold')
+
+ def solve(self, _decision_path, _candidate_list, _request):
+
+ filtered_candidates = list()
+ demand_name = _decision_path.current_demand.name
+
+ LOG.info(_LI("Solving constraint type '{}' for demand - [{}]").format(
+ self.constraint_type, demand_name))
+
+ for candidate in _candidate_list:
+ attribute_value = candidate.get(self.attribute)
+ if self.operation(attribute_value, self.threshold):
+ filtered_candidates.append(candidate)
+
+ return filtered_candidates
+
+
+
+
+
#
# -------------------------------------------------------------------------
# Copyright (c) 2015-2017 AT&T Intellectual Property
+# Copyright (C) 2020 Wipro Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
msg = "--- demand = {}, chosen resource = {} at {}"
for demand_name in _best_path.decisions:
resource = _best_path.decisions[demand_name]
- LOG.debug(msg.format(demand_name, resource["candidate_id"],
- resource["location_id"]))
+ if 'location_id' in resource:
+ LOG.debug(msg.format(demand_name, resource["candidate_id"],
+ resource["location_id"]))
msg = "--- total value of decision = {}"
LOG.debug(msg.format(_best_path.total_value))
#
# -------------------------------------------------------------------------
# Copyright (c) 2015-2017 AT&T Intellectual Property
+# Copyright (C) 2020 Wipro Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
import service as service_constraint
from conductor.solver.optimizer.constraints import vim_fit
from conductor.solver.optimizer.constraints import zone
+from conductor.solver.optimizer.constraints import threshold
from conductor.solver.request import demand
from conductor.solver.request import objective
from conductor.solver.request.functions import aic_version
_properties=c_property)
self.constraints[my_attribute_constraint.name] = \
my_attribute_constraint
+ elif constraint_type == "threshold":
+ c_property = constraint_info.get("properties")
+ my_threshold_constraint = \
+ threshold.Threshold(constraint_id,
+ constraint_type,
+ constraint_demands,
+ _properties=c_property)
+ self.constraints[my_threshold_constraint.name] = my_threshold_constraint
elif constraint_type == "hpa":
LOG.debug("Creating constraint - {}".format(constraint_type))
c_property = constraint_info.get("properties")
constraint.rank = 7
elif constraint.constraint_type == "region_fit":
constraint.rank = 8
- else:
+ elif constraint.constraint_type == "threshold":
constraint.rank = 9
+ else:
+ constraint.rank = 10
def attr_sort(self, attrs=['rank']):
# this helper for sorting the rank
#
# -------------------------------------------------------------------------
# Copyright (c) 2015-2017 AT&T Intellectual Property
+# Copyright (C) 2020 Wipro Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
p.message = message
# Metrics to Prometheus
- m_svc_name = p.template['parameters'].get('service_name', 'N/A')
+ m_svc_name = p.template.get('parameters', {}).get('service_name', 'N/A')
PC.VNF_FAILURE.labels('ONAP', m_svc_name).inc()
while 'FAILURE' in _is_success:
'aic_version': resource.get("cloud_region_version")},
}
+ if rec["candidate"]["inventory_type"] == "nssi":
+ rec["candidate"] = resource
+
if resource.get('vim-id'):
rec["candidate"]['vim-id'] = resource.get('vim-id')
#
# -------------------------------------------------------------------------
# Copyright (c) 2018 AT&T Intellectual Property
+# Copyright (C) 2020 Wipro Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
'/aai/v14/cloud-infrastructure/complexes/complex/DLLSTX233','get_complex_DLLSTX233',
'/aai/v14/cloud-infrastructure/cloud-regions/cloud-region/HPA-cloud/cloud-region-1/flavors/', 'get_flavors_region_1',
'/aai/v14/cloud-infrastructure/cloud-regions/cloud-region/HPA-cloud/cloud-region-2/flavors/', 'get_flavors_region_2',
+ '/aai/v14/nodes/service-instances', 'get_nssi',
)
return json.dumps(json_data)
+class get_nssi:
+ def GET(self):
+ print("------------------------------------------------------")
+ replyfile = "get_nssi_response.json"
+ # replyToAaiGet (web, replydir, replyfile)
+ fullreply = replydir + replyfile
+ trid = web.ctx.env.get('X_TRANSACTIONID', '111111')
+ # print ("X-TransactionId : {}".format(trid))
+ print("this is the context : {}".format(web.ctx.fullpath))
+ with open(fullreply) as json_file:
+ json_data = json.load(json_file)
+ print(json_data)
+
+ web.header('Content-Type', 'application/json')
+ web.header('X-TransactionId', trid)
+ return json.dumps(json_data)
+
if __name__ == "__main__":
app = web.application(urls, globals())
--- /dev/null
+{"service-instance": [{
+ "service-instance-id": "1a636c4d-5e76-427e-bfd6-241a947224b0",
+ "service-instance-name": "nssi_test_0211",
+ "service-type": "embb",
+ "service-role": "nssi",
+ "environment-context": "cn",
+ "model-invariant-id": "21d57d4b-52ad-4d3c-a798-248b5bb9124a",
+ "model-version-id": "bfba363e-e39c-4bd9-a9d5-1371c28f4d22",
+ "resource-version": "1581418601616",
+ "orchestration-status": "active",
+ "relationship-list": {
+ "relationship": [
+ {
+ "related-to": "service-instance",
+ "relationship-label": "org.onap.relationships.inventory.ComposedOf",
+ "related-link": "/aai/v16/business/customers/customer/5GCustomer/service-subscriptions/service-subscription/5G/service-instances/service-instance/4115d3c8-dd59-45d6-b09d-e756dee9b518",
+ "relationship-data": [
+ {
+ "relationship-key": "customer.global-customer-id",
+ "relationship-value": "5GCustomer"
+ },
+ {
+ "relationship-key": "service-subscription.service-type",
+ "relationship-value": "5G"
+ },
+ {
+ "relationship-key": "service-instance.service-instance-id",
+ "relationship-value": "4115d3c8-dd59-45d6-b09d-e756dee9b518"
+ }
+ ],
+ "related-to-property": [
+ {
+ "property-key": "service-instance.service-instance-name",
+ "property-value": "nsi_test_0211"
+ }
+ ]
+ }
+ ]
+ },
+ "slice-profiles": {
+ "slice-profile": [
+ {
+ "profile-id": "cdad9f49-4201-4e3a-aac1-b0f27902c299",
+ "latency": 20,
+ "max-number-of-UEs": 0,
+ "coverage-area-TA-list": "[{\"province\":\"??\",\"city\":\"???\",\"county\":\"???\",\"street\":\"?????\"}]",
+ "ue-mobility-level": "stationary",
+ "resource-sharing-level": "0",
+ "exp-data-rate-UL": 100,
+ "exp-data-rate-DL": 100,
+ "activity-factor": 0,
+ "e2e-latency": 0,
+ "jitter": 0,
+ "survival-time": 0,
+ "exp-data-rate": 0,
+ "payload-size": 0,
+ "traffic-density": 0,
+ "conn-density": 0,
+ "reliability": 99.999,
+ "resource-version": "1581418602494"
+ }
+ ]
+ }
+}]}
--- /dev/null
+[
+ {
+ "exp_data_rate":0,
+ "conn_density":0,
+ "coverage_area_ta_list":"[{\"province\":\"??\",\"city\":\"???\",\"county\":\"???\",\"street\":\"?????\"}]",
+ "activity_factor":0,
+ "cs_availability":null,
+ "candidate_id":"1a636c4d-5e76-427e-bfd6-241a947224b0",
+ "area_traffic_cap_dl":null,
+ "latency":20,
+ "service_area_dimension":null,
+ "domain":"cn",
+ "e2e_latency":0,
+ "area_traffic_cap_ul":null,
+ "inventory_provider":"aai",
+ "exp_data_rate_ul":100,
+ "max_number_of_ues":0,
+ "ue_mobility_level":"stationary",
+ "candidate_type":"nssi",
+ "traffic_density":0,
+ "payload_size":0,
+ "exp_data_rate_dl":100,
+ "jitter":0,
+ "survival_time":0,
+ "resource_sharing_level":"0",
+ "inventory_type":"nssi",
+ "reliability":null,
+ "cost":1.0,
+ "nsi_name": "nsi_test_0211",
+ "instance_name": "nssi_test_0211",
+ "uniqueness": "true"
+ }
+]
\ No newline at end of file
--- /dev/null
+{"service-instance": [{
+ "service-instance-id": "1a636c4d-5e76-427e-bfd6-241a947224b0",
+ "service-instance-name": "nssi_test_0211",
+ "service-type": "embb",
+ "service-role": "nssi",
+ "environment-context": "cn",
+ "model-invariant-id": "21d57d4b-52ad-4d3c-a798-248b5bb9124a",
+ "model-version-id": "bfba363e-e39c-4bd9-a9d5-1371c28f4d22",
+ "resource-version": "1581418601616",
+ "orchestration-status": "active",
+ "relationship-list": {
+ "relationship": [
+ {
+ "related-to": "service-instance",
+ "relationship-label": "org.onap.relationships.inventory.ComposedOf",
+ "related-link": "/aai/v16/business/customers/customer/5GCustomer/service-subscriptions/service-subscription/5G/service-instances/service-instance/4115d3c8-dd59-45d6-b09d-e756dee9b518",
+ "relationship-data": [
+ {
+ "relationship-key": "customer.global-customer-id",
+ "relationship-value": "5GCustomer"
+ },
+ {
+ "relationship-key": "service-subscription.service-type",
+ "relationship-value": "5G"
+ },
+ {
+ "relationship-key": "service-instance.service-instance-id",
+ "relationship-value": "4115d3c8-dd59-45d6-b09d-e756dee9b518"
+ }
+ ],
+ "related-to-property": [
+ {
+ "property-key": "service-instance.service-instance-name",
+ "property-value": "nsi_test_0211"
+ }
+ ]
+ }
+ ]
+ },
+ "slice-profiles": {
+ "slice-profile": [
+ {
+ "profile-id": "cdad9f49-4201-4e3a-aac1-b0f27902c299",
+ "latency": 20,
+ "max-number-of-UEs": 0,
+ "coverage-area-TA-list": "[{\"province\":\"??\",\"city\":\"???\",\"county\":\"???\",\"street\":\"?????\"}]",
+ "ue-mobility-level": "stationary",
+ "resource-sharing-level": "0",
+ "exp-data-rate-UL": 100,
+ "exp-data-rate-DL": 100,
+ "activity-factor": 0,
+ "e2e-latency": 0,
+ "jitter": 0,
+ "survival-time": 0,
+ "exp-data-rate": 0,
+ "payload-size": 0,
+ "traffic-density": 0,
+ "conn-density": 0,
+ "resource-version": "1581418602494"
+ }
+ ]
+ }
+}]}
#
# -------------------------------------------------------------------------
# Copyright (c) 2015-2017 AT&T Intellectual Property
+# Copyright (C) 2020 Wipro Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
self.assertEqual(None, self.aai_ep.match_hpa(candidate_json['candidate_list'][1],
feature_json[5]))
+ def test_get_nssi_candidates(self):
+ nssi_response_file = './conductor/tests/unit/data/plugins/inventory_provider/nssi_response.json'
+ nssi_response = json.loads(open(nssi_response_file).read())
+ nssi_candidates_file = './conductor/tests/unit/data/plugins/inventory_provider/nssi_candidate.json'
+ nssi_candidates = json.loads(open(nssi_candidates_file).read())
+
+ service_role = 'nssi'
+ model_invariant_id = '21d57d4b-52ad-4d3c-a798-248b5bb9124a'
+ model_version_id = 'bfba363e-e39c-4bd9-a9d5-1371c28f4d22'
+ orchestration_status = 'active'
+ filtering_attributes = dict()
+ filtering_attributes['orchestration-status'] = orchestration_status
+ filtering_attributes['service-role'] = service_role
+ filtering_attributes['model-invariant-id'] = model_invariant_id
+ filtering_attributes['model-version-id'] = model_version_id
+
+ self.assertEqual(nssi_candidates, self.aai_ep.filter_nssi_candidates(nssi_response, filtering_attributes, "true"))
+
+ nssi_response['service-instance'][0]['orchestration-status'] = 'deactivated'
+
+ self.assertEqual([], self.aai_ep.filter_nssi_candidates(nssi_response, filtering_attributes, "true"))
+
+ nssi_response['service-instance'][0]['service-role'] = 'service'
+
+ self.assertEqual([], self.aai_ep.filter_nssi_candidates(nssi_response, filtering_attributes, "true"))
+
+ self.assertEqual([], self.aai_ep.filter_nssi_candidates(None, filtering_attributes, "true"))
+
+ self.assertEqual([], self.aai_ep.filter_nssi_candidates(None, None, "true"))
+
+ self.assertEqual([], self.aai_ep.filter_nssi_candidates(nssi_response, None, "true"))
--- /dev/null
+#
+# -------------------------------------------------------------------------
+# Copyright (C) 2020 Wipro Limited.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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.
+#
+# -------------------------------------------------------------------------
+#
+
+
+class NotImplementedError(NotImplementedError):
+ # FIXME(jd) This is used by WSME to return a correct HTTP code. We should
+ # not expose it here but wrap our methods in the API to convert it to a
+ # proper HTTP error.
+ code = 501
--- /dev/null
+#
+# -------------------------------------------------------------------------
+# Copyright (C) 2020 Wipro Limited.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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 unittest
+from conductor.solver.optimizer.constraints.threshold import Threshold
+from conductor.solver.optimizer.decision_path import DecisionPath
+from conductor.solver.request.demand import Demand
+
+
+class TestThreshold(unittest.TestCase):
+
+ def test_solve(self):
+
+ candidates_file = './conductor/tests/unit/data/plugins/inventory_provider/nssi_candidate.json'
+ candidates = json.loads(open(candidates_file).read())
+
+ properties = {'attribute': 'latency', 'threshold': 30, 'operator': 'lte'}
+
+ threshold_obj = Threshold('urllc_threshold', 'threshold', ['URLLC'], _priority=0, _properties=properties)
+
+ decision_path = DecisionPath()
+ decision_path.current_demand = Demand('URLLC')
+
+ self.assertEqual(candidates, threshold_obj.solve(decision_path, candidates, None))
+
+ properties = {'attribute': 'latency', 'threshold': 10, 'operator': 'lte'}
+
+ threshold_obj = Threshold('urllc_threshold', 'threshold', ['URLLC'], _priority=0, _properties=properties)
+
+ self.assertEqual([], threshold_obj.solve(decision_path, candidates, None))
+
+ properties = {'attribute': 'exp_data_rate_ul', 'threshold': 70, 'operator': 'gte'}
+
+ threshold_obj = Threshold('urllc_threshold', 'threshold', ['URLLC'], _priority=0, _properties=properties)
+
+ self.assertEqual(candidates, threshold_obj.solve(decision_path, candidates, None))
+
+ properties = {'attribute': 'exp_data_rate_ul', 'threshold': 120, 'operator': 'gte'}
+
+ threshold_obj = Threshold('urllc_threshold', 'threshold', ['URLLC'], _priority=0, _properties=properties)
+
+ self.assertEqual([], threshold_obj.solve(decision_path, candidates, None))
+
+
+if __name__ == "__main__":
+ unittest.main()
Code Review / optf / has.git / commitdiff