test: P4 device driver

from copy import deepcopy

from device.proto.context_pb2 import DeviceDriverEnum, DeviceOperationalStatusEnum

from .Tools import config_rule_set

DEVICE_P4_ID = 0

DEVICE_P4_NAME = 'device:leaf1'

DEVICE_P4_TYPE = 'p4-switch'

DEVICE_P4_ADDRESS = '127.0.0.1'

DEVICE_P4_PORT = '50101'

DEVICE_P4_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_P4]

DEVICE_P4_VENDOR = 'Open Networking Foundation'

DEVICE_P4_HW_VER = 'BMv2 simple_switch'

DEVICE_P4_SW_VER = 'Stratum'

DEVICE_P4_PIPECONF = 'org.onosproject.pipelines.fabric'

DEVICE_P4_WORKERS = 2

DEVICE_P4_GRACE_PERIOD = 60

DEVICE_P4_UUID = {'device_uuid': {'uuid': DEVICE_P4_NAME}}

DEVICE_P4 = {

    'device_id': deepcopy(DEVICE_P4_UUID),

    'device_type': DEVICE_P4_TYPE,

    'device_config': {'config_rules': []},

    'device_operational_status': DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_DISABLED,

    'device_drivers': DEVICE_P4_DRIVERS,

    'device_endpoints': [],

}

DEVICE_P4_CONNECT_RULES = [

    config_rule_set('_connect/address', DEVICE_P4_ADDRESS),

    config_rule_set('_connect/port', DEVICE_P4_PORT),

    config_rule_set('_connect/settings', {

        'id': int(DEVICE_P4_ID),

        'name': DEVICE_P4_NAME,

        'hw-ver': DEVICE_P4_HW_VER,

        'sw-ver': DEVICE_P4_SW_VER,

        'pipeconf': DEVICE_P4_PIPECONF

    }),

]

DEVICE_P4_CONFIG_RULES = [

    config_rule_set('key1', 'value1'),

]

import grpc, logging

from concurrent import futures

from p4.v1 import p4runtime_pb2_grpc

from .Device_P4 import(

    DEVICE_P4_ADDRESS, DEVICE_P4_PORT,

    DEVICE_P4_WORKERS, DEVICE_P4_GRACE_PERIOD)

from .MockP4RuntimeServicerImpl import MockP4RuntimeServicerImpl

LOGGER = logging.getLogger(__name__)

class MockP4RuntimeService:

    def __init__(

            self, address=DEVICE_P4_ADDRESS, port=DEVICE_P4_PORT,

            max_workers=DEVICE_P4_WORKERS,

            grace_period=DEVICE_P4_GRACE_PERIOD):

        self.address = address

        self.port = port

        self.endpoint = '{:s}:{:s}'.format(str(self.address), str(self.port))

        self.max_workers = max_workers

        self.grace_period = grace_period

        self.pool = None

        self.server = None

        self.servicer = None

    def start(self):

        LOGGER.info(

            'Starting P4Runtime service on {:s} with max_workers: {:s})'.format(

                str(self.endpoint), str(self.max_workers)))

        self.pool = futures.ThreadPoolExecutor(max_workers=self.max_workers)

        self.server = grpc.server(self.pool)

        self.servicer = MockP4RuntimeServicerImpl()

        p4runtime_pb2_grpc.add_P4RuntimeServicer_to_server(

            self.servicer, self.server)

        _ = self.server.add_insecure_port(self.endpoint)

        LOGGER.info('Listening on {:s}...'.format(str(self.endpoint)))

        self.server.start()

        LOGGER.debug('P4Runtime service started')

    def stop(self):

        LOGGER.debug(

            'Stopping P4Runtime service (grace period {:s} seconds)...'.format(

                str(self.grace_period)))

        self.server.stop(self.grace_period)

        LOGGER.debug('P4Runtime service stopped')

import queue

from google.rpc import code_pb2

from p4.v1 import p4runtime_pb2, p4runtime_pb2_grpc

from p4.config.v1 import p4info_pb2

class MockP4RuntimeServicerImpl(p4runtime_pb2_grpc.P4RuntimeServicer):

    def __init__(self):

        self.p4info = p4info_pb2.P4Info()

        self.p4runtime_api_version = "1.3.0"

        self.stored_packet_out = queue.Queue()

    def GetForwardingPipelineConfig(self, request, context):

        rep = p4runtime_pb2.GetForwardingPipelineConfigResponse()

        if self.p4info is not None:

            rep.config.p4info.CopyFrom(self.p4info)

        return rep

    def SetForwardingPipelineConfig(self, request, context):

        self.p4info.CopyFrom(request.config.p4info)

        return p4runtime_pb2.SetForwardingPipelineConfigResponse()

    def Write(self, request, context):

        return p4runtime_pb2.WriteResponse()

    def Read(self, request, context):

        yield p4runtime_pb2.ReadResponse()

    def StreamChannel(self, request_iterator, context):

        for req in request_iterator:

            if req.HasField('arbitration'):

                rep = p4runtime_pb2.StreamMessageResponse()

                rep.arbitration.CopyFrom(req.arbitration)

                rep.arbitration.status.code = code_pb2.OK

                yield rep

            elif req.HasField('packet'):

                self.stored_packet_out.put(req)

    def Capabilities(self, request, context):

        rep = p4runtime_pb2.CapabilitiesResponse()

        rep.p4runtime_api_version = self.p4runtime_api_version

        return rep