Skip to content
Snippets Groups Projects

fix: Change p4_service_handler and test scripts for rel2

Merged famelis requested to merge fix/p4 into release/2.0.1
4 files
+ 60
44
Compare changes
  • Side-by-side
  • Inline
Files
4
@@ -47,7 +47,7 @@ def create_rule_set(endpoint_a, endpoint_b):
}
]
}
)
)
def create_rule_del(endpoint_a, endpoint_b):
return json_config_rule_delete(
@@ -68,7 +68,17 @@ def create_rule_del(endpoint_a, endpoint_b):
}
]
}
)
)
def find_names(uuid_a, uuid_b, device_endpoints):
endpoint_a, endpoint_b = None, None
for endpoint in device_endpoints:
if endpoint.endpoint_id.endpoint_uuid.uuid == uuid_a:
endpoint_a = endpoint.name
elif endpoint.endpoint_id.endpoint_uuid.uuid == uuid_b:
endpoint_b = endpoint.name
return (endpoint_a, endpoint_b)
class P4ServiceHandler(_ServiceHandler):
def __init__(self,
@@ -127,12 +137,21 @@ class P4ServiceHandler(_ServiceHandler):
device = self.__task_executor.get_device(DeviceId(**json_device_id(device_uuid)))
del device.device_config.config_rules[:]
# Find names from uuids
(endpoint_a, endpoint_b) = find_names(matched_endpoint_uuid, endpoint_uuid, device.device_endpoints)
if endpoint_a is None:
LOGGER.exception('Unable to find name of endpoint({:s})'.format(str(matched_endpoint_uuid)))
raise Exception('Unable to find name of endpoint({:s})'.format(str(matched_endpoint_uuid)))
if endpoint_b is None:
LOGGER.exception('Unable to find name of endpoint({:s})'.format(str(endpoint_uuid)))
raise Exception('Unable to find name of endpoint({:s})'.format(str(endpoint_uuid)))
# One way
rule = create_rule_set(matched_endpoint_uuid, endpoint_uuid)
rule = create_rule_set(endpoint_a, endpoint_b)
device.device_config.config_rules.append(ConfigRule(**rule))
# The other way
rule = create_rule_set(endpoint_uuid, matched_endpoint_uuid)
rule = create_rule_set(endpoint_b, endpoint_a)
device.device_config.config_rules.append(ConfigRule(**rule))
self.__task_executor.configure_device(device)
@@ -189,11 +208,20 @@ class P4ServiceHandler(_ServiceHandler):
del device.device_config.config_rules[:]
# Find names from uuids
(endpoint_a, endpoint_b) = find_names(matched_endpoint_uuid, endpoint_uuid, device.device_endpoints)
if endpoint_a is None:
LOGGER.exception('Unable to find name of endpoint({:s})'.format(str(matched_endpoint_uuid)))
raise Exception('Unable to find name of endpoint({:s})'.format(str(matched_endpoint_uuid)))
if endpoint_b is None:
LOGGER.exception('Unable to find name of endpoint({:s})'.format(str(endpoint_uuid)))
raise Exception('Unable to find name of endpoint({:s})'.format(str(endpoint_uuid)))
# One way
rule = create_rule_del(matched_endpoint_uuid, endpoint_uuid)
rule = create_rule_del(endpoint_a, endpoint_b)
device.device_config.config_rules.append(ConfigRule(**rule))
# The other way
rule = create_rule_del(endpoint_uuid, matched_endpoint_uuid)
rule = create_rule_del(endpoint_b, endpoint_a)
device.device_config.config_rules.append(ConfigRule(**rule))
self.__task_executor.configure_device(device)
Loading