open-traffic-generator / models-convergence Goto Github PK

import snappi_convergence


@pytest.fixture(scope='session')
def capi():
    api = snappi_convergence.api(location='localhost:443', ext='ixnetwork')
    yield api
    if getattr(api, 'assistant', None) is not None:
        api.assistant.Session.remove()


def bgp_convergence_config(utils, capi):

    config = capi.config()

    tx, rx1, rx2 = (
        config.ports
        .port(name='tx', location=utils.settings.ports[0])
        .port(name='rx1', location=utils.settings.ports[1])
        .port(name='rx2', location=utils.settings.ports[2])
    )

    config.options.port_options.location_preemption = True
    ly = config.layer1.layer1()[-1]
    ly.name = 'ly'
    ly.port_names = [tx.name, rx1.name, rx2.name]
    ly.ieee_media_defaults = False
    ly.auto_negotiate = False
    ly.speed = utils.settings.speed

    tx_device, rx1_device, rx2_device = (
        config.devices
        .device(name="tx_device", container_name=tx.name)
        .device(name="rx1_device", container_name=rx1.name)
        .device(name="rx2_device", container_name=rx2.name)
    )

    # tx_device config
    tx_eth = tx_device.ethernet
    tx_eth.name = "tx_eth"
    tx_ipv4 = tx_eth.ipv4
    tx_ipv4.name = "tx_ipv4"
    tx_ipv4.address = "21.1.1.2"
    tx_ipv4.prefix = "24"
    tx_ipv4.gateway = "21.1.1.1"

    # rx1_device config
    rx1_eth = rx1_device.ethernet
    rx1_eth.name = "rx1_eth"
    rx1_ipv4 = rx1_eth.ipv4
    rx1_ipv4.name = "rx1_ipv4"
    rx1_ipv4.address = "22.1.1.2"
    rx1_ipv4.prefix = "24"
    rx1_ipv4.gateway = "22.1.1.1"
    rx1_bgpv4 = rx1_ipv4.bgpv4
    rx1_bgpv4.name = "rx1_bgpv4"
    rx1_bgpv4.as_type = "ebgp"
    rx1_bgpv4.dut_address = "22.1.1.1"
    rx1_bgpv4.as_number = "65200"
    rx1_rr = rx1_bgpv4.bgpv4_routes.bgpv4route(name="rx1_rr")[-1]
    rx1_rr.addresses.bgpv4routeaddress(count=1000,
                                       address='200.1.0.1',
                                       prefix=32)

    # rx2_device config
    rx2_eth = rx2_device.ethernet
    rx2_eth.name = "rx2_eth"
    rx2_ipv4 = rx2_eth.ipv4
    rx2_ipv4.name = "rx2_ipv4"
    rx2_ipv4.address = "23.1.1.2"
    rx2_ipv4.prefix = "24"
    rx2_ipv4.gateway = "23.1.1.1"
    rx2_bgpv4 = rx2_ipv4.bgpv4
    rx2_bgpv4.name = "rx2_bgp"
    rx2_bgpv4.as_type = "ebgp"
    rx2_bgpv4.dut_address = "23.1.1.1"
    rx2_bgpv4.as_number = "65200"

    rx2_rr = rx2_bgpv4.bgpv4_routes.bgpv4route(name="rx2_rr")[-1]
    rx2_rr.addresses.bgpv4routeaddress(count=1000,
                                       address='200.1.0.1',
                                       prefix=32)

    # flow config
    flow = config.flows.flow(name='convergence_test')[-1]
    flow.tx_rx.device.tx_names = [tx_device.name]
    flow.tx_rx.device.rx_names = [rx1_rr.name, rx2_rr.name]

    flow.size.fixed = "1024"
    flow.rate.percentage = "50"
    flow.metrics.enable = True

    return config


PRIMARY_ROUTES_NAME = 'rx1_rr'
SECONDARY_ROUTES_NAME = 'rx2_rr'
PRIMARY_PORT_NAME = 'rx1'
# maximum convergence 3s
MAX_CON = 3000000


def test_bgp_cp_dp_convergence(utils, capi, bgp_convergence_config):
    """
    5. set advanced metric settings & start traffic
    6. Shut down receive primary port
    7. Wait for traffic to converge
    8. Obtain cp/dp convergence and validate against expected
    """
    # convergence config
    bgp_convergence_config.rx_rate_threshold.threshold = 90
    bgp_convergence_config.convergence_event = (
        bgp_convergence_config.convergence_event.LINK_DOWN)

    capi.set_convergence(bgp_convergence_config)

    # Start traffic
    # TODO: Keeping it as set_transmit_state but need to update
    # as per the model
    ts = capi.transmit_state()
    ts.state = ts.START
    capi.set_transmit_state(ts)

    # Wait for traffic to start
    utils.wait_for(
        lambda: is_traffic_started(capi), 'traffic to start and reach 50%'
    )

    # Shut down the primary port
    link_state = capi.link_state()
    link_state.state = link_state.DOWN
    link_state.port_names = [PRIMARY_PORT_NAME]
    capi.set_convergence_state(link_state)

    # Wait for traffic to converge
    utils.wait_for(
        lambda: is_traffic_converged(capi), 'traffic to converge'
    )

    # get convergence metrics
    request = capi.metrics_request()
    convergence_metrics = capi.get_convergence(request)

    for metrics in convergence_metrics:
        assert metrics.control_plane_data_plane_convergence_us < MAX_CON


def is_traffic_started(capi):
    """
    Returns true if traffic in start state
    """
    flow_stats = get_flow_stats(capi)
    return all([int(fs.frames_rx_rate) == int(fs.frames_tx_rate) / 2
               for fs in flow_stats])


def is_traffic_converged(api):
    """
    Returns true if traffic in stop state
    """
    flow_stats = get_flow_stats(api)
    return all([int(fs.frames_tx_rate) == int(fs.frames_rx_rate)
               for fs in flow_stats if fs.port_rx == 'rx2'])


def get_flow_stats(capi):
    # TODO: Keeping it as get_flow_metrics but need to update as per the model
    request = capi.metrics_request()
    request.flow_names = []
    return capi.get_flow_metrics(request)```