tests/itests/voltha/test_voltha_xpon.py

# Copyright 2017-present Open Networking Foundation
#
# 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 tests.itests.voltha.rest_base import RestBase

from google.protobuf.json_format import MessageToDict, ParseDict
import unittest

from voltha.protos import bbf_fiber_base_pb2 as fb
from voltha.protos.device_pb2 import Device
from voltha.protos import bbf_fiber_gemport_body_pb2 as gemport
from voltha.protos import bbf_fiber_tcont_body_pb2 as tcont
from voltha.protos import bbf_fiber_traffic_descriptor_profile_body_pb2 as tdp
from common.utils.consulhelpers import get_endpoint_from_consul
from tests.itests.voltha.xpon_scenario import scenario as xpon_scenario
from tests.itests.test_utils import get_pod_ip
from testconfig import config

'''
These tests use the Ponsim OLT to verify create, update, and delete
functionalities of ChannelgroupConfig, ChannelpartitionConfig,
ChannelpairConfig, ChannelterminationConfig, VOntAni, OntAni, and VEnets
for xPON
The prerequisite for this test are:
 1. voltha ensemble is running
      docker-compose -f compose/docker-compose-system-test.yml up -d
 2. ponsim olt is running with PONSIM-OLT
      sudo -s
      . ./env.sh
      ./ponsim/main.py -v
'''

device_type = 'ponsim_olt'
host_and_port = '172.17.0.1:50060'

#for ordering the test cases
id = 3
LOCAL_CONSUL = "localhost:8500"

orch_env = 'docker-compose'
if 'test_parameters' in config and 'orch_env' in config['test_parameters']:
    orch_env = config['test_parameters']['orch_env']
print 'orchestration-environment: %s' % orch_env

# Retrieve details of the REST entry point
if orch_env == 'k8s-single-node':
    rest_endpoint = get_pod_ip('voltha') + ':8443'
else:
    rest_endpoint = get_endpoint_from_consul(LOCAL_CONSUL, 'voltha-envoy-8443')
# Construct the base_url
BASE_URL = 'https://' + rest_endpoint

class GlobalPreChecks(RestBase):
    base_url = BASE_URL

    # def test_000_get_root(self):
    #     res = self.get('/#!/', expected_content_type='text/html')
    #     self.assertGreaterEqual(res.find('swagger'), 0)

    def test_001_get_health(self):
        res = self.get('/health')
        self.assertEqual(res['state'], 'HEALTHY')

class TestXPon(RestBase):
    base_url = BASE_URL

    def test_002_setup_device(self):
        global device
        device = self.add_device()
        self.verify_device_preprovisioned_state(device['id'])
        self.activate_device(device['id'])

    def _remove_device(self):
        self.deactivate_device(device['id'])
        self.delete_device(device['id'])

    #~~~~~~~~~~~~~~~~~~~~~~ Helper Functions ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    # Create a new simulated device
    def add_device(self):
        return self.post('/api/v1/devices',
                           MessageToDict(Device(
                                type=device_type,
                                host_and_port=host_and_port
                            )),
                         expected_http_code=200)

    def verify_device_preprovisioned_state(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'], 'ponsim_olt')
        self.assertEqual(device['admin_state'], 'PREPROVISIONED')
        self.assertEqual(device['oper_status'], 'UNKNOWN')

    # Active the simulated device.
    # This will trigger the simulation of random alarms
    def activate_device(self, device_id):
        path = '/api/v1/devices/{}'.format(device_id)
        self.post(path + '/enable', expected_http_code=200)
        device = self.get(path)
        self.assertEqual(device['admin_state'], 'ENABLED')

    def deactivate_device(self, device_id):
        path = '/api/v1/devices/{}'.format(device_id)
        self.post(path + '/disable', expected_http_code=200)
        device = self.get(path)
        self.assertEqual(device['admin_state'], 'DISABLED')

    def delete_device(self, device_id):
        path = '/api/v1/devices/{}'.format(device_id)
        self.delete(path + '/delete', expected_http_code=200)
        device = self.get(path, expected_http_code=404)
        self.assertIsNone(device)

    # Add cg, cpair, cpart
    def add(self, type, config, req, name):
        res = self.verify(type)
        prev_len = len(res[config])
        self.post(self.get_path(type, name, ''),
                  MessageToDict(req, preserving_proto_field_name = True),
                  expected_http_code = 200)
        return self.verify(type), prev_len

    # Modify the existing cg, cpair, cpart
    def modify(self, type, req, name):
        self.post(self.get_path(type, name, '/modify'),
                  MessageToDict(req, preserving_proto_field_name = True),
                  expected_http_code = 200)
        return self.verify(type)

    # Delete cg, cpair, cpart
    def remove(self, type, config, name):
        res = self.verify(type)
        prev_len = len(res[config])
        self.delete(self.get_path(type, name, '/delete'),
                    expected_http_code = 200)
        return self.verify(type), prev_len

    # Retrieve the desired item upon Post message
    def verify(self, type):
        if(type == 'channel_terminations'):
            return self.get('/api/v1/devices/{}/{}'.format(device['id'], type))
        return self.get('/api/v1/{}'.format(type))

    def get_path(self, type, name, operation):
        if(type == 'channel_terminations'):
            return '/api/v1/devices/{}/{}/{}{}'.format(device['id'],
                type, name, operation)
        return '/api/v1/{}/{}{}'.format(type, name, operation)

    # Method to check if the result is same as the change requested
    def search(self, req, result):
        dict1 = MessageToDict(req,
                              including_default_value_fields = True,
                              preserving_proto_field_name = True)
        #skip comparison of READ-ONLY fields
        result['id'] = ''
        if isinstance(req, fb.ChannelgroupConfig):
            result['cg_index'] = 0
        elif isinstance(req, tcont.TcontsConfigData):
            if not dict1['alloc_id']:
                result['alloc_id'] = 0
        elif isinstance(req, gemport.GemportsConfigData):
            if not dict1['gemport_id']:
                result['gemport_id'] = 0
        return dict1 == result


#~~~~~~~~~~~~~~ Function to create test cases on the fly ~~~~~~~~~~~~~~~~
def create_dynamic_method(key, value):
    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'},
        'gemport':{'type':'gemports',
                   'config':'gemports_config'},
        'tcont': {'type':'tconts',
                   'config':'tconts_config'},
        'tdp':    {'type':'traffic_descriptor_profiles',
                   'config':'traffic_descriptor_profiles'}
    }

    def _add(self, type, config, req, name):
        result, prev_len = self.add(type, config, req, name)
        self.assertEqual(result[config][prev_len]['name'], name)
        self.assertEqual(len(result[config]), prev_len+1)
        self.assertEqual(self.search(req, result[config][prev_len]), True)

    def _mod(self, type, config, req, name):
        result = self.modify(type, req, name)
        self.assertEqual(self.search(req, result[config][0]), True)

    def _del(self, type, config, req, name):
        result, prev_len = self.remove(type, config, name)
        self.assertEqual(len(result[config]), prev_len-1)

    def _operate(self, obj_action, type_config, req, name):
        if obj_action == 'add':
            _add(self, type_config['type'], type_config['config'], req, name)
        elif obj_action == 'mod':
            _mod(self, type_config['type'], type_config['config'], req, name)
        elif obj_action == 'del':
            _del(self, type_config['type'], type_config['config'], req, name)
    
    def _config(self):
        ParseDict(value['rpc'], value['pb2'])
        return value['pb2']

    def dynamic_test_method(self):
        _obj_action = [val for val in key.split('-')]
        _type_config = obj_type_config[_obj_action[0]]
        _req = _config(self)
        _operate(self, _obj_action[1], _type_config, _req, value['rpc']['name'])

    return dynamic_test_method

#read the set instructions for tests
#dynamically create test cases in desired sequence
for item in xpon_scenario:
    id = id + 1
    if(isinstance(item, dict)):
        for k,v in item.items():
            dynamic_method = create_dynamic_method(k, v)
            dynamic_method.__name__ = 'test_{:3d}_{}'.format(id, k).replace(
                ' ', '0')
            setattr(TestXPon, dynamic_method.__name__, dynamic_method)
            del dynamic_method

if __name__ == '__main__':
    unittest.main()