test/unit/extensions/omci/test_me_frame.py

#
# Copyright 2017 the original author or authors.
#
# 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 __future__ import absolute_import
from unittest import TestCase, main
from nose.tools import assert_raises
from pyvoltha.adapters.extensions.omci.me_frame import *
from pyvoltha.adapters.extensions.omci.omci_me import *
from pyvoltha.adapters.extensions.omci.omci import *
import codecs


def hexify(frame):
    """Return a hexadecimal string encoding of input buffer"""
    return codecs.encode(bytes(frame),'hex')

class TestSelectMeFrameGeneration(TestCase):

    def assertGeneratedFrameEquals(self, frame, ref):
        assert isinstance(frame, Packet)
        serialized_hexified_frame = hexify(frame).upper()
        ref = ref.upper()
        if serialized_hexified_frame != ref:
            self.fail('Mismatch:\nReference:\n{}\nGenerated (bad):\n{}'.format(
                ref, serialized_hexified_frame
            ))

    def test_mib_reset_message_serialization(self):
        ref = b'00004F0A000200000000000000000000' \
              b'00000000000000000000000000000000' \
              b'000000000000000000000028'
        frame = OntDataFrame().mib_reset()
        self.assertGeneratedFrameEquals(frame, ref)

    def test_create_gal_ethernet_profile(self):
        ref = b'0000440A011000010030000000000000' \
              b'00000000000000000000000000000000' \
              b'000000000000000000000028'
        frame = GalEthernetProfileFrame(1, max_gem_payload_size=48).create()
        self.assertGeneratedFrameEquals(frame, ref)

    def test_set_tcont_1(self):
        ref = b'0000480A010680008000040000000000' \
              b'00000000000000000000000000000000' \
              b'000000000000000000000028'

        frame = TcontFrame(0x8000, alloc_id=0x400).set()
        self.assertGeneratedFrameEquals(frame, ref)

    def test_set_tcont_2(self):
        ref = b'0000480A010680018000040100000000' \
              b'00000000000000000000000000000000' \
              b'000000000000000000000028'

        frame = TcontFrame(0x8001, alloc_id=0x401).set()
        self.assertGeneratedFrameEquals(frame, ref)

    def test_create_8021p_mapper_service_profile(self):
        ref = b'0000440A00828000ffffffffffffffff' \
              b'ffffffffffffffffffff000000000000' \
              b'000000000000000000000028'
        frame = Ieee8021pMapperServiceProfileFrame(0x8000).create()
        self.assertGeneratedFrameEquals(frame, ref)

    def test_create_mac_bridge_service_profile(self):
        ref = b'0000440A002D02010001008000140002' \
              b'000f0001000000000000000000000000' \
              b'000000000000000000000028'
        data = dict(
            spanning_tree_ind=False,
            learning_ind=True,
            priority=0x8000,
            max_age=20 * 256,
            hello_time=2 * 256,
            forward_delay=15 * 256,
            unknown_mac_address_discard=True
        )
        frame = MacBridgeServiceProfileFrame(0x201, attributes=data).create()
        self.assertGeneratedFrameEquals(frame, ref)

    def test_create_gem_port_network_ctp(self):
        ref = b'0000440A010C01000400800003010000' \
              b'00000000000000000000000000000000' \
              b'000000000000000000000028'

        data = dict(
            port_id=0x400,
            tcont_pointer=0x8000,
            direction=3,
            traffic_management_pointer_upstream=0x100
        )
        frame = GemPortNetworkCtpFrame(0x100, attributes=data).create()
        self.assertGeneratedFrameEquals(frame, ref)

        # Also test direction as a string parameter
        frame = GemPortNetworkCtpFrame(0x100, port_id=0x400,
                                       tcont_id=0x8000,
                                       direction='bi-directional',
                                       upstream_tm=0x100).create()
        self.assertGeneratedFrameEquals(frame, ref)

    def test_create_gem_inteworking_tp(self):
        ref = b'0000440A010A80010100058000000000' \
              b'01000000000000000000000000000000' \
              b'000000000000000000000028'
        frame = GemInterworkingTpFrame(0x8001,
                                       gem_port_network_ctp_pointer=0x100,
                                       interworking_option=5,
                                       service_profile_pointer=0x8000,
                                       interworking_tp_pointer=0x0,
                                       gal_profile_pointer=0x1).create()

        self.assertGeneratedFrameEquals(frame, ref)

    def test_set_8021p_mapper_service_profile(self):
        ref = b'0000480A008280007F80800100000000' \
              b'00000000000000000000000000000000' \
              b'000000000000000000000028'
        ptrs = [0x8001, 0, 0, 0, 0, 0, 0, 0]
        frame = Ieee8021pMapperServiceProfileFrame(0x8000,
                                                   interwork_tp_pointers=ptrs).set()

        self.assertGeneratedFrameEquals(frame, ref)

        ptrs = [0x8001, 0]
        frame = Ieee8021pMapperServiceProfileFrame(0x8000,
                                                   interwork_tp_pointers=ptrs).set()

        self.assertGeneratedFrameEquals(frame, ref)

    def test_create_mac_bridge_port_configuration_data(self):
        ref = b'0000440A002F21010201020380000000' \
              b'00000000000000000000000000000000' \
              b'000000000000000000000028'

        frame = MacBridgePortConfigurationDataFrame(0x2101,
                                                    bridge_id_pointer=0x201,
                                                    port_num=2,
                                                    tp_type=3,
                                                    tp_pointer=0x8000).create()
        self.assertGeneratedFrameEquals(frame, ref)

    def test_create_vlan_tagging_filter_data(self):
        ref = b'0000440A005421010400000000000000' \
              b'00000000000000000000000000000000' \
              b'100100000000000000000028'
        frame = VlanTaggingFilterDataFrame(0x2101,
                                           vlan_tcis=[0x400],
                                           forward_operation=0x10).create()
        self.assertGeneratedFrameEquals(frame, ref)

    def test_create_extended_vlan_tagging_operation_configuration_data(self):
        ref = b'0000440A00AB02020A04010000000000' \
              b'00000000000000000000000000000000' \
              b'000000000000000000000028'
        data = dict(
            association_type=10,
            associated_me_pointer=0x401
        )
        frame = \
            ExtendedVlanTaggingOperationConfigurationDataFrame(0x202,
                                                               attributes=data)\
                .create()

        self.assertGeneratedFrameEquals(frame, ref)

    def test_set_extended_vlan_tagging_operation_configuration_data(self):
        ref = b'0000480A00AB02023800810081000000' \
              b'00000000000000000000000000000000' \
              b'000000000000000000000028'
        data = dict(
            input_tpid=0x8100,
            output_tpid=0x8100,
            downstream_mode=0,  # inverse of upstream
        )
        frame = \
            ExtendedVlanTaggingOperationConfigurationDataFrame(0x202,
                                                               attributes=data)\
                .set()

        self.assertGeneratedFrameEquals(frame, ref)

    def test_set_extended_vlan_tagging_1(self):
        ref = b'0000480A00AB02020400f00000008200' \
              b'5000402f000000082004000000000000' \
              b'000000000000000000000028'
        data = dict(
            received_frame_vlan_tagging_operation_table=\
                VlanTaggingOperation(
                    filter_outer_priority=15,
                    filter_inner_priority=8,
                    filter_inner_vid=1024,
                    filter_inner_tpid_de=5,
                    filter_ether_type=0,
                    treatment_tags_to_remove=1,
                    pad3=2,
                    treatment_outer_priority=15,
                    treatment_inner_priority=8,
                    treatment_inner_vid=1024,
                    treatment_inner_tpid_de=4
                )
        )
        frame = \
            ExtendedVlanTaggingOperationConfigurationDataFrame(0x202,
                                                               attributes=data)\
                .set()

        self.assertGeneratedFrameEquals(frame, ref)

    def test_set_extended_vlan_tagging_2(self):
        ref = b'0000480A00AB02020400F00000008200' \
              b'd000402f00000008200c000000000000' \
              b'000000000000000000000028'
        data = dict(
            received_frame_vlan_tagging_operation_table=
                VlanTaggingOperation(
                    filter_outer_priority=15,
                    filter_inner_priority=8,
                    filter_inner_vid=1025,
                    filter_inner_tpid_de=5,
                    filter_ether_type=0,
                    treatment_tags_to_remove=1,
                    pad3=2,
                    treatment_outer_priority=15,
                    treatment_inner_priority=8,
                    treatment_inner_vid=1025,
                    treatment_inner_tpid_de=4
                )
        )

        frame = \
            ExtendedVlanTaggingOperationConfigurationDataFrame(0x202,
                                                               attributes=data)\
                .set()

        self.assertGeneratedFrameEquals(frame, ref)

    def test_create_mac_bridge_port_configuration_data2(self):
        ref = b'0000440A002F02010201010b04010000' \
              b'00000000000000000000000000000000' \
              b'000000000000000000000028'
        data = dict(
            bridge_id_pointer=0x201,
            encapsulation_methods=0,
            port_num=1,
            port_priority=0,
            port_path_cost=0,
            port_spanning_tree_in=0,
            lan_fcs_ind=0,
            tp_type=11,
            tp_pointer=0x401,
            mac_learning_depth=0
        )
        frame = MacBridgePortConfigurationDataFrame(0x201,
                                                    attributes=data).create()

        self.assertGeneratedFrameEquals(frame, ref)

    def test_set_pptp_ethernet_uni_frame(self):
        ref = b'0000480A000B020109000005EE000000' \
              b'00000000000000000000000000000000' \
              b'000000000000000000000028'
        data = dict(
            administrative_state=0,  # 0 - Unlock
            max_frame_size=1518      # two-octet field
        )
        frame = PptpEthernetUniFrame(0x201,
                                     attributes=data).set()

        self.assertGeneratedFrameEquals(frame, ref)

    def test_constraint_errors(self):
        self.assertTrue(True)  # TODO Also test some attribute constraint failures

    def test_mib_upload_next(self):
        # Test for VOL-649 error. SCAPY was only originally coded for a 'get'
        # action (8-bit MIB Data Sync value) but MIB Upload Next commands have
        # a 16-bit field.
        #
        # 255 and less always worked
        OntDataFrame(sequence_number=0).mib_upload_next()
        OntDataFrame(sequence_number=255).mib_upload_next()
        # But not 256+
        OntDataFrame(sequence_number=256).mib_upload_next()
        OntDataFrame(sequence_number=1000).mib_upload_next()
        OntDataFrame(sequence_number=0xFFFE).mib_upload_next()

        # Also test the optional arguments for the other actions
        OntDataFrame().get()
        OntDataFrame(mib_data_sync=4).set()
        OntDataFrame().mib_reset()
        OntDataFrame().mib_upload()
        # OntDataFrame(ignore_arc=True).get_all_alarms()        Not yet coded
        # OntDataFrame(ignore_arc=False).get_all_alarms()       Not yet coded

        # Range/type checks
        assert_raises(ValueError, OntDataFrame, mib_data_sync=-1)
        assert_raises(ValueError, OntDataFrame, mib_data_sync=256)
        assert_raises(ValueError, OntDataFrame, sequence_number=-1)
        assert_raises(ValueError, OntDataFrame, sequence_number=0x10000)
        assert_raises(TypeError, OntDataFrame, ignore_arc=123)


if __name__ == '__main__':
    main()