| from random import randint |
| from time import time, sleep |
| |
| from google.protobuf.json_format import MessageToDict, ParseDict |
| from unittest import main, skip |
| from voltha.protos.device_pb2 import Device |
| from tests.itests.voltha.rest_base import RestBase |
| from common.utils.consulhelpers import get_endpoint_from_consul, \ |
| get_all_instances_of_service |
| from common.utils.consulhelpers import verify_all_services_healthy |
| from tests.itests.docutests.test_utils import \ |
| run_command_to_completion_with_raw_stdout, \ |
| run_command_to_completion_with_stdout_in_list |
| from voltha.protos.voltha_pb2 import AlarmFilter |
| from google.protobuf.empty_pb2 import Empty |
| import grpc |
| from voltha.protos import third_party |
| from voltha.protos import voltha_pb2 |
| from voltha.core.flow_decomposer import * |
| from voltha.protos.openflow_13_pb2 import FlowTableUpdate |
| from voltha.protos import bbf_fiber_base_pb2 as fb |
| from tests.itests.voltha.test_voltha_xpon import scenario as xpon_scenario |
| |
| LOCAL_CONSUL = "localhost:8500" |
| DOCKER_COMPOSE_FILE = "compose/docker-compose-system-test.yml" |
| |
| command_defs = dict( |
| docker_ps="docker ps", |
| docker_compose_start_all="docker-compose -f {} up -d " |
| .format(DOCKER_COMPOSE_FILE), |
| docker_stop_and_remove_all_containers="docker-compose -f {} down" |
| .format(DOCKER_COMPOSE_FILE), |
| docker_compose_start_voltha="docker-compose -f {} up -d voltha " |
| .format(DOCKER_COMPOSE_FILE), |
| docker_compose_stop_voltha="docker-compose -f {} stop voltha" |
| .format(DOCKER_COMPOSE_FILE), |
| docker_compose_remove_voltha="docker-compose -f {} rm -f voltha" |
| .format(DOCKER_COMPOSE_FILE), |
| docker_compose_scale_voltha="docker-compose -f {} scale " |
| "voltha=".format(DOCKER_COMPOSE_FILE), |
| kafka_topics="kafkacat -b {} -L", |
| kafka_alarms="kafkacat -o end -b {} -C -t voltha.alarms -c 2", |
| kafka_kpis="kafkacat -o end -b {} -C -t voltha.kpis -c 5" |
| ) |
| |
| obj_type_config = { |
| 'cg': {'type':'channel_groups', |
| 'config':'channelgroup_config'}, |
| 'cpart': {'type':'channel_partitions', |
| 'config':'channelpartition_config'}, |
| 'cpair': {'type':'channel_pairs', |
| 'config':'channelpair_config'}, |
| 'cterm': {'type':'channel_terminations', |
| 'config':'channeltermination_config'}, |
| 'vontani':{'type':'v_ont_anis', |
| 'config':'v_ontani_config'}, |
| 'ontani': {'type':'ont_anis', |
| 'config':'ontani_config'}, |
| 'venet': {'type':'v_enets', |
| 'config':'v_enet_config'} |
| } |
| |
| |
| class DispatcherTest(RestBase): |
| def setUp(self): |
| self.grpc_channels = dict() |
| |
| t0 = [time()] |
| |
| def pt(self, msg=''): |
| t1 = time() |
| print '%20.8f ms - %s' % (1000 * (t1 - DispatcherTest.t0[0]), |
| msg) |
| DispatcherTest.t0[0] = t1 |
| |
| def wait_till(self, msg, predicate, interval=0.1, timeout=5.0): |
| deadline = time() + timeout |
| while time() < deadline: |
| if predicate(): |
| return |
| sleep(interval) |
| self.fail('Timed out while waiting for condition: {}'.format(msg)) |
| |
| def get_channel(self, voltha_grpc): |
| if voltha_grpc not in self.grpc_channels: |
| self.grpc_channels[voltha_grpc] = grpc.insecure_channel( |
| voltha_grpc) |
| return self.grpc_channels[voltha_grpc] |
| |
| def test_01_global_rest_apis(self): |
| # Start the voltha ensemble with a single voltha instance |
| self._stop_and_remove_all_containers() |
| sleep(5) # A small wait for the system to settle down |
| self.start_all_containers() |
| self.set_rest_endpoint() |
| self.set_kafka_endpoint() |
| |
| # self._get_root_rest() |
| self._get_schema_rest() |
| self._get_health_rest() |
| self._get_voltha_rest() |
| self._list_voltha_instances_rest() |
| self._get_voltha_instance_rest() |
| olt_id = self._add_olt_device_rest() |
| self._verify_device_preprovisioned_state_rest(olt_id) |
| self._activate_device_rest(olt_id) |
| ldev_id = self._wait_for_logical_device_rest(olt_id) |
| ldevices = self._list_logical_devices_rest() |
| logical_device_id = ldevices['items'][0]['id'] |
| self._get_logical_device_rest(logical_device_id) |
| self._list_logical_device_ports_rest(logical_device_id) |
| self._list_and_update_logical_device_flows_rest(logical_device_id) |
| self._list_and_update_logical_device_flow_groups_rest( |
| logical_device_id) |
| devices = self._list_devices_rest() |
| device_id = devices['items'][0]['id'] |
| self._get_device_rest(device_id) |
| self._list_device_ports_rest(device_id) |
| self._list_device_flows_rest(device_id) |
| self._list_device_flow_groups_rest(device_id) |
| self._get_images_rest(device_id) |
| self._self_test_rest(device_id) |
| dtypes = self._list_device_types_rest() |
| self._get_device_type_rest(dtypes['items'][0]['id']) |
| alarm_filter = self._create_device_filter_rest(olt_id) |
| self._remove_device_filter_rest(alarm_filter['id']) |
| #for xPON objects |
| for item in xpon_scenario: |
| for key,value in item.items(): |
| try: |
| _device_id = None |
| _obj_action = [val for val in key.split('-')] |
| _type_config = obj_type_config[_obj_action[0]] |
| if _obj_action[0] == "cterm": |
| _device_id = olt_id |
| if _obj_action[1] == "mod": |
| continue |
| elif _obj_action[1] == "add": |
| _xpon_obj = self._create_xpon_object_rest(_type_config, |
| value, |
| _device_id) |
| elif _obj_action[1] == "del": |
| self._delete_xpon_object_rest(_type_config, |
| value, |
| _device_id) |
| except Exception, e: |
| print 'An error occurred', e |
| continue |
| |
| # TODO: PM APIs test |
| |
| @skip('Test fails due to environment configuration. Need to investigate. Refer to VOL-427') |
| def test_02_cross_instances_dispatch(self): |
| |
| def prompt(input_func, text): |
| val = input_func(text) |
| return val |
| |
| def prompt_for_return(text): |
| return raw_input(text) |
| |
| # Start the voltha ensemble with a single voltha instance |
| self._stop_and_remove_all_containers() |
| sleep(5) # A small wait for the system to settle down |
| self.start_all_containers() |
| self.set_rest_endpoint() |
| self.set_kafka_endpoint() |
| |
| # Scale voltha to 3 instances and setup the voltha grpc assigments |
| self._scale_voltha(3) |
| sleep(10) # A small wait for the system to settle down |
| voltha_instances = get_all_instances_of_service(LOCAL_CONSUL, |
| 'voltha-grpc') |
| self.assertEqual(len(voltha_instances), 3) |
| self.ponsim_voltha_stub_local = voltha_pb2.VolthaLocalServiceStub( |
| self.get_channel(self._get_grpc_address(voltha_instances[2]))) |
| self.ponsim_voltha_stub_global = voltha_pb2.VolthaGlobalServiceStub( |
| self.get_channel(self._get_grpc_address(voltha_instances[2]))) |
| |
| self.simulated_voltha_stub_local = voltha_pb2.VolthaLocalServiceStub( |
| self.get_channel(self._get_grpc_address(voltha_instances[1]))) |
| self.simulated_voltha_stub_global = voltha_pb2.VolthaGlobalServiceStub( |
| self.get_channel(self._get_grpc_address(voltha_instances[1]))) |
| |
| self.empty_voltha_stub_local = voltha_pb2.VolthaLocalServiceStub( |
| self.get_channel(self._get_grpc_address(voltha_instances[0]))) |
| self.empty_voltha_stub_global = voltha_pb2.VolthaGlobalServiceStub( |
| self.get_channel(self._get_grpc_address(voltha_instances[0]))) |
| |
| # Prompt the user to start ponsim |
| # Get the user to start PONSIM as root |
| prompt(prompt_for_return, |
| '\nStart PONSIM as root in another window ...') |
| |
| prompt(prompt_for_return, |
| '\nEnsure port forwarding is set on ponmgnt ...') |
| |
| # Test 1: |
| # A. Get the list of adapters using a global stub |
| # B. Get the list of adapters using a local stub |
| # C. Verify that the two lists are the same |
| adapters_g = self._get_adapters_grpc(self.ponsim_voltha_stub_global) |
| adapters_l = self._get_adapters_grpc(self.empty_voltha_stub_local) |
| assert adapters_g == adapters_l |
| |
| # Test 2: |
| # A. Provision a pomsim olt using the ponsim_voltha_stub |
| # B. Enable the posim olt using the simulated_voltha_stub |
| # C. Wait for onu discovery using the empty_voltha_stub |
| ponsim_olt = self._provision_ponsim_olt_grpc( |
| self.ponsim_voltha_stub_local) |
| ponsim_logical_device_id = self._enable_device_grpc( |
| self.simulated_voltha_stub_global, |
| ponsim_olt.id) |
| self._wait_for_onu_discovery_grpc(self.empty_voltha_stub_global, |
| ponsim_olt.id, |
| count=4) |
| # Test 3: |
| # A. Provision a simulated olt using the simulated_voltha_stub |
| # B. Enable the simulated olt using the ponsim_voltha_stub |
| # C. Wait for onu discovery using the empty_voltha_stub |
| simulated_olt = self._provision_simulated_olt_grpc( |
| self.simulated_voltha_stub_local) |
| simulated_logical_device_id = self._enable_device_grpc( |
| self.ponsim_voltha_stub_global, simulated_olt.id) |
| self._wait_for_onu_discovery_grpc(self.empty_voltha_stub_global, |
| simulated_olt.id, count=4) |
| |
| # Test 4: |
| # Verify that we have at least 8 devices created using the global |
| # REST and also via direct grpc in the empty stub |
| devices_via_rest = self._list_devices_rest(8)['items'] |
| devices_via_global_grpc = self._get_devices_grpc( |
| self.empty_voltha_stub_global) |
| assert len(devices_via_rest) == len(devices_via_global_grpc) |
| |
| # Test 5: |
| # A. Create 2 Alarms filters using REST |
| # B. Ensure it is present across all instances |
| # C. Ensure when requesting the alarm filters we do not get |
| # duplicate results |
| alarm_filter1 = self._create_device_filter_rest(ponsim_olt.id) |
| alarm_filter2 = self._create_device_filter_rest(simulated_olt.id) |
| global_filters = self._get_alarm_filters_rest() |
| filter = self._get_alarm_filters_grpc(self.simulated_voltha_stub_local) |
| assert global_filters == MessageToDict(filter) |
| filter = self._get_alarm_filters_grpc(self.ponsim_voltha_stub_local) |
| assert global_filters == MessageToDict(filter) |
| filter = self._get_alarm_filters_grpc(self.empty_voltha_stub_local) |
| assert global_filters == MessageToDict(filter) |
| filter = self._get_alarm_filters_grpc(self.empty_voltha_stub_global) |
| assert global_filters == MessageToDict(filter) |
| |
| # Test 6: |
| # A. Delete an alarm filter |
| # B. Ensure that filter is deleted from all instances |
| self._remove_device_filter_rest(alarm_filter1['id']) |
| previous_filters = global_filters |
| global_filters = self._get_alarm_filters_rest() |
| assert global_filters != previous_filters |
| filter = self._get_alarm_filters_grpc(self.simulated_voltha_stub_local) |
| assert global_filters == MessageToDict(filter) |
| filter = self._get_alarm_filters_grpc(self.ponsim_voltha_stub_local) |
| assert global_filters == MessageToDict(filter) |
| filter = self._get_alarm_filters_grpc(self.empty_voltha_stub_local) |
| assert global_filters == MessageToDict(filter) |
| filter = self._get_alarm_filters_grpc(self.empty_voltha_stub_global) |
| assert global_filters == MessageToDict(filter) |
| |
| # Test 7: |
| # A. Simulate EAPOL install on ponsim instance using grpc |
| # B. Validate the flows using global REST |
| # C. Retrieve the flows from global grpc using empty voltha instance |
| self._install_eapol_flow_grpc(self.ponsim_voltha_stub_local, |
| ponsim_logical_device_id) |
| self._verify_olt_eapol_flow_rest(ponsim_olt.id) |
| res = self._get_olt_flows_grpc(self.empty_voltha_stub_global, |
| ponsim_logical_device_id) |
| |
| # Test 8: |
| # A. Create xPON objects instance using REST |
| # B. Ensuring that Channeltermination is present on specific instances |
| # C. Ensuring that other xPON objects are present in all instances |
| for item in xpon_scenario: |
| for key,value in item.items(): |
| _obj_action = [val for val in key.split('-')] |
| _type_config = obj_type_config[_obj_action[0]] |
| if _obj_action[1] == "mod": |
| continue |
| if _obj_action[0] == "cterm": |
| if _obj_action[1] == "add": |
| #Ponsim OLT |
| self._create_xpon_object_rest(_type_config, |
| value, |
| ponsim_olt.id) |
| self._verify_xpon_object_on_device( |
| _type_config, |
| self.ponsim_voltha_stub_global, |
| ponsim_olt.id) |
| self._delete_xpon_object_rest(_type_config, |
| value, |
| ponsim_olt.id) |
| #Simulated OLT |
| self._create_xpon_object_rest(_type_config, |
| value, |
| simulated_olt.id) |
| self._verify_xpon_object_on_device( |
| _type_config, |
| self.simulated_voltha_stub_global, |
| simulated_olt.id) |
| self._delete_xpon_object_rest(_type_config, |
| value, |
| simulated_olt.id) |
| elif _obj_action[1] == "del": |
| continue |
| else: |
| if _obj_action[1] == "add": |
| self._create_xpon_object_rest(_type_config, value) |
| #Checking with Ponsim OLT |
| self._verify_xpon_object_on_device( |
| _type_config, |
| self.ponsim_voltha_stub_global) |
| #Checking with empty instance |
| self._verify_xpon_object_on_device( |
| _type_config, |
| self.empty_voltha_stub_global) |
| #Checking with Simulated OLT |
| self._verify_xpon_object_on_device( |
| _type_config, |
| self.simulated_voltha_stub_global) |
| elif _obj_action[1] == "del": |
| self._delete_xpon_object_rest(_type_config, value) |
| |
| # TODO: More tests to be added as new features are added |
| |
| |
| def _get_grpc_address(self, voltha_instance): |
| address = '{}:{}'.format(voltha_instance['ServiceAddress'], |
| voltha_instance['ServicePort']) |
| return address |
| |
| def _stop_and_remove_all_containers(self): |
| # check if there are any running containers first |
| cmd = command_defs['docker_ps'] |
| out, err, rc = run_command_to_completion_with_stdout_in_list(cmd) |
| self.assertEqual(rc, 0) |
| if len(out) > 1: # not counting docker ps header |
| cmd = command_defs['docker_stop_and_remove_all_containers'] |
| out, err, rc = run_command_to_completion_with_raw_stdout(cmd) |
| self.assertEqual(rc, 0) |
| |
| def start_all_containers(self): |
| t0 = time() |
| |
| # start all the containers |
| self.pt("Starting all containers ...") |
| cmd = command_defs['docker_compose_start_all'] |
| out, err, rc = run_command_to_completion_with_raw_stdout(cmd) |
| self.assertEqual(rc, 0) |
| |
| self.pt("Waiting for voltha and chameleon containers to be ready ...") |
| self.wait_till('voltha services HEALTHY', |
| lambda: verify_all_services_healthy( |
| LOCAL_CONSUL, service_name='voltha-grpc') == True, |
| timeout=10) |
| self.wait_till('chameleon services HEALTHY', |
| lambda: verify_all_services_healthy( |
| LOCAL_CONSUL, |
| service_name='chameleon-rest') == True, |
| timeout=10) |
| |
| # Chameleon takes some time to compile the protos and make them |
| # available. So let's wait 10 seconds |
| sleep(10) |
| |
| def set_rest_endpoint(self): |
| self.rest_endpoint = get_endpoint_from_consul(LOCAL_CONSUL, |
| 'envoy-8443') |
| self.base_url = 'https://' + self.rest_endpoint |
| |
| def set_kafka_endpoint(self): |
| self.kafka_endpoint = get_endpoint_from_consul(LOCAL_CONSUL, 'kafka') |
| |
| def _scale_voltha(self, scale=2): |
| self.pt("Scaling voltha ...") |
| cmd = command_defs['docker_compose_scale_voltha'] + str(scale) |
| out, err, rc = run_command_to_completion_with_raw_stdout(cmd) |
| self.assertEqual(rc, 0) |
| |
| def _get_root_rest(self): |
| res = self.get('/', expected_content_type='text/html') |
| self.assertGreaterEqual(res.find('swagger'), 0) |
| |
| def _get_schema_rest(self): |
| res = self.get('/schema') |
| self.assertEqual(set(res.keys()), |
| {'protos', 'yang_from', 'swagger_from'}) |
| |
| def _get_health_rest(self): |
| res = self.get('/health') |
| self.assertEqual(res['state'], 'HEALTHY') |
| |
| # ~~~~~~~~~~~~~~~~~~~~~ TOP LEVEL VOLTHA OPERATIONS ~~~~~~~~~~~~~~~~~~~~~~~ |
| |
| def _get_voltha_rest(self): |
| res = self.get('/api/v1') |
| self.assertEqual(res['version'], '0.9.0') |
| |
| def _list_voltha_instances_rest(self): |
| res = self.get('/api/v1/instances') |
| self.assertEqual(len(res['items']), 1) |
| |
| def _get_voltha_instance_rest(self): |
| res = self.get('/api/v1/instances') |
| voltha_id = res['items'][0] |
| res = self.get('/api/v1/instances/{}'.format(voltha_id)) |
| self.assertEqual(res['version'], '0.9.0') |
| |
| def _add_olt_device_rest(self, grpc=None): |
| device = Device( |
| type='simulated_olt', |
| mac_address='00:00:00:00:00:01' |
| ) |
| device = self.post('/api/v1/devices', MessageToDict(device), |
| expected_http_code=200) |
| return device['id'] |
| |
| def _provision_simulated_olt_grpc(self, stub): |
| device = Device( |
| type='simulated_olt', |
| mac_address='00:00:00:00:00:01' |
| ) |
| device = stub.CreateDevice(device) |
| return device |
| |
| def _provision_ponsim_olt_grpc(self, stub): |
| device = Device( |
| type='ponsim_olt', |
| host_and_port='172.17.0.1:50060' |
| ) |
| device = stub.CreateDevice(device) |
| return device |
| |
| def _enable_device_grpc(self, stub, device_id): |
| logical_device_id = None |
| try: |
| stub.EnableDevice(voltha_pb2.ID(id=device_id)) |
| while True: |
| device = stub.GetDevice(voltha_pb2.ID(id=device_id)) |
| # If this is an OLT then acquire logical device id |
| if device.oper_status == voltha_pb2.OperStatus.ACTIVE: |
| if device.type.endswith('_olt'): |
| assert device.parent_id |
| logical_device_id = device.parent_id |
| self.pt('success (logical device id = {})'.format( |
| logical_device_id)) |
| else: |
| self.pt('success (device id = {})'.format(device.id)) |
| break |
| self.pt('waiting for device to be enabled...') |
| sleep(.5) |
| except Exception, e: |
| self.pt('Error enabling {}. Error:{}'.format(device_id, e)) |
| return logical_device_id |
| |
| def _delete_device_grpc(self, stub, device_id): |
| try: |
| stub.DeleteDevice(voltha_pb2.ID(id=device_id)) |
| while True: |
| device = stub.GetDevice(voltha_pb2.ID(id=device_id)) |
| assert not device |
| except Exception, e: |
| self.pt('deleting device {}. Error:{}'.format(device_id, e)) |
| |
| def _get_devices_grpc(self, stub): |
| res = stub.ListDevices(Empty()) |
| return res.items |
| |
| def _get_adapters_grpc(self, stub): |
| res = stub.ListAdapters(Empty()) |
| return res.items |
| |
| def _find_onus_grpc(self, stub, olt_id): |
| devices = self._get_devices_grpc(stub) |
| return [ |
| d for d in devices |
| if d.parent_id == olt_id |
| ] |
| |
| def _wait_for_onu_discovery_grpc(self, stub, olt_id, count=4): |
| # shortly after we shall see the discovery of four new onus, linked to |
| # the olt device |
| self.wait_till( |
| 'find ONUs linked to the olt device', |
| lambda: len(self._find_onus_grpc(stub, olt_id)) >= count, |
| 2 |
| ) |
| # verify that they are properly set |
| onus = self._find_onus_grpc(stub, olt_id) |
| for onu in onus: |
| self.assertEqual(onu.admin_state, 3) # ENABLED |
| self.assertEqual(onu.oper_status, 4) # ACTIVE |
| |
| return [onu.id for onu in onus] |
| |
| def _verify_device_preprovisioned_state_rest(self, olt_id): |
| # we also check that so far what we read back is same as what we get |
| # back on create |
| device = self.get('/api/v1/devices/{}'.format(olt_id)) |
| self.assertNotEqual(device['id'], '') |
| self.assertEqual(device['adapter'], 'simulated_olt') |
| self.assertEqual(device['admin_state'], 'PREPROVISIONED') |
| self.assertEqual(device['oper_status'], 'UNKNOWN') |
| |
| def _activate_device_rest(self, olt_id): |
| path = '/api/v1/devices/{}'.format(olt_id) |
| self.post(path + '/enable', expected_http_code=200) |
| device = self.get(path) |
| self.assertEqual(device['admin_state'], 'ENABLED') |
| |
| self.wait_till( |
| 'admin state moves to ACTIVATING or ACTIVE', |
| lambda: self.get(path)['oper_status'] in ('ACTIVATING', 'ACTIVE'), |
| timeout=0.5) |
| |
| # eventually, it shall move to active state and by then we shall have |
| # device details filled, connect_state set, and device ports created |
| self.wait_till( |
| 'admin state ACTIVE', |
| lambda: self.get(path)['oper_status'] == 'ACTIVE', |
| timeout=0.5) |
| device = self.get(path) |
| images = device['images'] |
| image = images['image'] |
| image_1 = image[0] |
| version = image_1['version'] |
| self.assertNotEqual(version, '') |
| self.assertEqual(device['connect_status'], 'REACHABLE') |
| |
| ports = self.get(path + '/ports')['items'] |
| self.assertEqual(len(ports), 2) |
| |
| def _wait_for_logical_device_rest(self, olt_id): |
| # we shall find the logical device id from the parent_id of the olt |
| # (root) device |
| device = self.get( |
| '/api/v1/devices/{}'.format(olt_id)) |
| self.assertNotEqual(device['parent_id'], '') |
| logical_device = self.get( |
| '/api/v1/logical_devices/{}'.format(device['parent_id'])) |
| |
| # the logical device shall be linked back to the hard device, |
| # its ports too |
| self.assertEqual(logical_device['root_device_id'], device['id']) |
| |
| logical_ports = self.get( |
| '/api/v1/logical_devices/{}/ports'.format( |
| logical_device['id']) |
| )['items'] |
| self.assertGreaterEqual(len(logical_ports), 1) |
| logical_port = logical_ports[0] |
| self.assertEqual(logical_port['id'], 'nni') |
| self.assertEqual(logical_port['ofp_port']['name'], 'nni') |
| self.assertEqual(logical_port['ofp_port']['port_no'], 129) |
| self.assertEqual(logical_port['device_id'], device['id']) |
| self.assertEqual(logical_port['device_port_no'], 2) |
| return logical_device['id'] |
| |
| def _list_logical_devices_rest(self): |
| res = self.get('/api/v1/logical_devices') |
| self.assertGreaterEqual(len(res['items']), 1) |
| return res |
| |
| def _get_logical_device_rest(self, id): |
| res = self.get('/api/v1/logical_devices/{}'.format(id)) |
| self.assertIsNotNone(res['datapath_id']) |
| |
| def _list_logical_device_ports_rest(self, id): |
| res = self.get('/api/v1/logical_devices/{}/ports'.format(id)) |
| self.assertGreaterEqual(len(res['items']), 1) |
| |
| def _list_and_update_logical_device_flows_rest(self, id): |
| |
| # retrieve flow list |
| res = self.get('/api/v1/logical_devices/{}/flows'.format(id)) |
| len_before = len(res['items']) |
| |
| # add some flows |
| req = ofp.FlowTableUpdate( |
| id=id, |
| flow_mod=mk_simple_flow_mod( |
| cookie=randint(1, 10000000000), |
| priority=len_before, |
| match_fields=[ |
| in_port(129) |
| ], |
| actions=[ |
| output(1) |
| ] |
| ) |
| ) |
| res = self.post('/api/v1/logical_devices/{}/flows'.format(id), |
| MessageToDict(req, preserving_proto_field_name=True), |
| expected_http_code=200) |
| # TODO check some stuff on res |
| |
| res = self.get('/api/v1/logical_devices/{}/flows'.format(id)) |
| len_after = len(res['items']) |
| self.assertGreater(len_after, len_before) |
| |
| def _list_and_update_logical_device_flow_groups_rest(self, id): |
| |
| # retrieve flow list |
| res = self.get('/api/v1/logical_devices/{}/flow_groups'.format(id)) |
| len_before = len(res['items']) |
| |
| # add some flows |
| req = ofp.FlowGroupTableUpdate( |
| id=id, |
| group_mod=ofp.ofp_group_mod( |
| command=ofp.OFPGC_ADD, |
| type=ofp.OFPGT_ALL, |
| group_id=len_before + 1, |
| buckets=[ |
| ofp.ofp_bucket( |
| actions=[ |
| ofp.ofp_action( |
| type=ofp.OFPAT_OUTPUT, |
| output=ofp.ofp_action_output( |
| port=1 |
| ) |
| ) |
| ] |
| ) |
| ] |
| ) |
| ) |
| res = self.post('/api/v1/logical_devices/{}/flow_groups'.format(id), |
| MessageToDict(req, preserving_proto_field_name=True), |
| expected_http_code=200) |
| # TODO check some stuff on res |
| |
| res = self.get('/api/v1/logical_devices/{}/flow_groups'.format(id)) |
| len_after = len(res['items']) |
| self.assertGreater(len_after, len_before) |
| |
| def _list_devices_rest(self, count=2): |
| res = self.get('/api/v1/devices') |
| self.assertGreaterEqual(len(res['items']), count) |
| return res |
| |
| def _get_device_rest(self, id): |
| res = self.get('/api/v1/devices/{}'.format(id)) |
| # TODO test result |
| |
| def _list_device_ports_rest(self, id, count=2): |
| res = self.get('/api/v1/devices/{}/ports'.format(id)) |
| self.assertGreaterEqual(len(res['items']), count) |
| |
| def _list_device_flows_rest(self, id, count=1): |
| # pump some flows into the logical device |
| res = self.get('/api/v1/devices/{}/flows'.format(id)) |
| self.assertGreaterEqual(len(res['items']), count) |
| |
| def _list_device_flow_groups_rest(self, id, count=0): |
| res = self.get('/api/v1/devices/{}/flow_groups'.format(id)) |
| self.assertGreaterEqual(len(res['items']), count) |
| |
| def _list_device_types_rest(self, count=2): |
| res = self.get('/api/v1/device_types') |
| self.assertGreaterEqual(len(res['items']), count) |
| return res |
| |
| def _get_device_type_rest(self, dtype): |
| res = self.get('/api/v1/device_types/{}'.format(dtype)) |
| self.assertIsNotNone(res) |
| # TODO test the result |
| |
| def _list_device_groups(self): |
| pass |
| # res = self.get('/api/v1/device_groups') |
| # self.assertGreaterEqual(len(res['items']), 1) |
| |
| def _get_device_group(self): |
| pass |
| # res = self.get('/api/v1/device_groups/1') |
| # # TODO test the result |
| |
| def _get_images_rest(self, id): |
| res = self.get('/api/v1/devices/{}/images'.format(id)) |
| self.assertIsNotNone(res) |
| |
| def _self_test_rest(self, id): |
| res = self.post('/api/v1/devices/{}/self_test'.format(id), |
| expected_http_code=200) |
| self.assertIsNotNone(res) |
| |
| def _create_device_filter_rest(self, device_id): |
| rules = list() |
| rule = dict() |
| |
| # Create a filter with a single rule |
| rule['key'] = 'device_id' |
| rule['value'] = device_id |
| rules.append(rule) |
| |
| alarm_filter = AlarmFilter(rules=rules) |
| alarm_filter = self.post('/api/v1/alarm_filters', |
| MessageToDict(alarm_filter), |
| expected_http_code=200) |
| self.assertIsNotNone(alarm_filter) |
| return alarm_filter |
| |
| def _remove_device_filter_rest(self, alarm_filter_id): |
| path = '/api/v1/alarm_filters/{}'.format(alarm_filter_id) |
| self.delete(path, expected_http_code=200) |
| alarm_filter = self.get(path, expected_http_code=200, grpc_status=5) |
| self.assertIsNone(alarm_filter) |
| |
| def _get_alarm_filter_grpc(self, stub, alarm_filter_id): |
| res = stub.GetAlarmFilter(voltha_pb2.ID(id=alarm_filter_id)) |
| return res |
| |
| def _get_alarm_filters_grpc(self, stub): |
| res = stub.ListAlarmFilters(Empty()) |
| return res |
| |
| def _get_alarm_filters_rest(self): |
| res = self.get('/api/v1/alarm_filters') |
| return res |
| |
| def _install_eapol_flow_grpc(self, stub, logical_device_id): |
| """ |
| Install an EAPOL flow on the given logical device. If device is not |
| given, it will be applied to logical device of the last pre-provisioned |
| OLT device. |
| """ |
| |
| # gather NNI and UNI port IDs |
| nni_port_no, unis = self._get_logical_ports(stub, logical_device_id) |
| |
| # construct and push flow rule |
| for uni_port_no, _ in unis: |
| update = FlowTableUpdate( |
| id=logical_device_id, |
| flow_mod=mk_simple_flow_mod( |
| priority=2000, |
| match_fields=[in_port(uni_port_no), eth_type(0x888e)], |
| actions=[ |
| # push_vlan(0x8100), |
| # set_field(vlan_vid(4096 + 4000)), |
| output(ofp.OFPP_CONTROLLER) |
| ] |
| ) |
| ) |
| res = stub.UpdateLogicalDeviceFlowTable(update) |
| self.pt('success for uni {} ({})'.format(uni_port_no, res)) |
| |
| def _get_logical_ports(self, stub, logical_device_id): |
| """ |
| Return the NNI port number and the first usable UNI port of logical |
| device, and the vlan associated with the latter. |
| """ |
| ports = stub.ListLogicalDevicePorts( |
| voltha_pb2.ID(id=logical_device_id)).items |
| nni = None |
| unis = [] |
| for port in ports: |
| if port.root_port: |
| assert nni is None, "There shall be only one root port" |
| nni = port.ofp_port.port_no |
| else: |
| uni = port.ofp_port.port_no |
| uni_device = self._get_device_grpc(stub, port.device_id) |
| vlan = uni_device.vlan |
| unis.append((uni, vlan)) |
| |
| assert nni is not None, "No NNI port found" |
| assert unis, "Not a single UNI?" |
| |
| return nni, unis |
| |
| def _get_device_grpc(self, stub, device_id, depth=0): |
| res = stub.GetDevice(voltha_pb2.ID(id=device_id), |
| metadata=(('get-depth', str(depth)),)) |
| return res |
| |
| def _verify_olt_eapol_flow_rest(self, logical_device_id): |
| flows = self.get('/api/v1/devices/{}/flows'.format(logical_device_id))[ |
| 'items'] |
| self.assertEqual(len(flows), 2) |
| flow = flows[1] |
| self.assertEqual(flow['table_id'], 0) |
| self.assertEqual(flow['priority'], 2000) |
| |
| |
| def _get_olt_flows_grpc(self, stub, logical_device_id): |
| res = stub.ListLogicalDeviceFlows(voltha_pb2.ID(id=logical_device_id)) |
| return res |
| |
| #For xPON objects |
| def _get_path(self, type, name, operation, device_id=None): |
| if(type == 'channel_terminations'): |
| return '/api/v1/devices/{}/{}/{}{}'.format(device_id, type, name, |
| operation) |
| return '/api/v1/{}/{}{}'.format(type, name, operation) |
| |
| def _get_xpon_object_rest(self, obj_type, device_id=None): |
| if obj_type["type"] == "channel_terminations": |
| res = self.get('/api/v1/devices/{}/{}'.format(device_id, |
| obj_type["type"])) |
| else: |
| res = self.get('/api/v1/{}'.format(obj_type["type"])) |
| return res |
| |
| def _get_xpon_object_grpc(self, stub, obj_type, device_id=None): |
| if obj_type["type"] == "channel_groups": |
| res = stub.GetAllChannelgroupConfig(Empty()) |
| elif obj_type["type"] == "channel_partitions": |
| res = stub.GetAllChannelpartitionConfig(Empty()) |
| elif obj_type["type"] == "channel_pairs": |
| res = stub.GetAllChannelpairConfig(Empty()) |
| elif obj_type["type"] == "channel_terminations": |
| res = stub.GetAllChannelterminationConfig( |
| voltha_pb2.ID(id=device_id)) |
| elif obj_type["type"] == "v_ont_anis": |
| res = stub.GetAllVOntaniConfig(Empty()) |
| elif obj_type["type"] == "ont_anis": |
| res = stub.GetAllOntaniConfig(Empty()) |
| elif obj_type["type"] == "v_enets": |
| res = stub.GetAllVEnetConfig(Empty()) |
| return res |
| |
| def _create_xpon_object_rest(self, obj_type, value, device_id=None): |
| ParseDict(value['rpc'], value['pb2']) |
| request = value['pb2'] |
| self.post(self._get_path(obj_type["type"], value['rpc']['name'], "", |
| device_id), |
| MessageToDict(request, preserving_proto_field_name = True), |
| expected_http_code = 200) |
| return request |
| |
| def _delete_xpon_object_rest(self, obj_type, value, device_id=None): |
| self.delete(self._get_path(obj_type["type"], value['rpc']['name'], |
| "/delete", device_id), expected_http_code = 200) |
| |
| def _verify_xpon_object_on_device(self, type_config, stub, device_id=None): |
| global_xpon_obj = self._get_xpon_object_rest(type_config, device_id) |
| xpon_obj = self._get_xpon_object_grpc(stub, type_config, device_id) |
| assert global_xpon_obj == MessageToDict(xpon_obj, |
| including_default_value_fields = True, |
| preserving_proto_field_name = True) |
| |
| if __name__ == '__main__': |
| main() |