test_unitary_driverapi.py

import copy, logging, math, pytest, time
from device.service.drivers.emulated.EmulatedDriver import EmulatedDriver

LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)

PATH_IF = '/interfaces/interface[name="{}"]'
PATH_SUBIF = PATH_IF + '/subinterfaces/subinterface[index="{}"]'
PATH_ADDRIPV4 = PATH_SUBIF + '/ipv4/address[ip="{}"]'

DEVICE_CONFIG_IF1 = []
DEVICE_CONFIG_IF1.append((PATH_IF      .format('IF1'               ) + '/config/name',           "IF1"     ))
DEVICE_CONFIG_IF1.append((PATH_IF      .format('IF1'               ) + '/config/enabled',        True      ))
DEVICE_CONFIG_IF1.append((PATH_SUBIF   .format('IF1', 0            ) + '/config/index',          0         ))
DEVICE_CONFIG_IF1.append((PATH_ADDRIPV4.format('IF1', 0, '10.1.0.1') + '/config/ip',             "10.1.0.1"))
DEVICE_CONFIG_IF1.append((PATH_ADDRIPV4.format('IF1', 0, '10.1.0.1') + '/config/prefix_length',  24        ))

DEVICE_CONFIG_IF2 = []
DEVICE_CONFIG_IF2.append((PATH_IF      .format('IF2'               ) + '/config/name',           "IF2"     ))
DEVICE_CONFIG_IF2.append((PATH_IF      .format('IF2'               ) + '/config/enabled',        True      ))
DEVICE_CONFIG_IF2.append((PATH_SUBIF   .format('IF2', 0            ) + '/config/index',          0         ))
DEVICE_CONFIG_IF2.append((PATH_ADDRIPV4.format('IF2', 0, '10.2.0.1') + '/config/ip',             "10.2.0.1"))
DEVICE_CONFIG_IF2.append((PATH_ADDRIPV4.format('IF2', 0, '10.2.0.1') + '/config/prefix_length',  24        ))

PATH_IF_TX_PKTS = PATH_IF + 'state/tx_packets_per_second'
PATH_IF_RX_PKTS = PATH_IF + 'state/rx_packets_per_second'

DEVICE_STATE_IF1_TX_PKTS = PATH_IF_TX_PKTS.format('IF1')
DEVICE_STATE_IF1_RX_PKTS = PATH_IF_RX_PKTS.format('IF1')
DEVICE_STATE_IF2_TX_PKTS = PATH_IF_TX_PKTS.format('IF2')
DEVICE_STATE_IF2_RX_PKTS = PATH_IF_RX_PKTS.format('IF2')

@pytest.fixture(scope='session')
def device_driverapi_emulated():
    _driver = EmulatedDriver('127.0.0.1', 0)
    _driver.Connect()
    yield _driver
    _driver.Disconnect()

def test_device_driverapi_emulated_setconfig(device_driverapi_emulated : EmulatedDriver):
    # should work
    results = device_driverapi_emulated.SetConfig(DEVICE_CONFIG_IF1)
    LOGGER.info('results:\n{}'.format('\n'.join(map(str, results))))
    assert len(results) == len(DEVICE_CONFIG_IF1)
    for result in results: assert isinstance(result, bool) and result

    results = device_driverapi_emulated.SetConfig(DEVICE_CONFIG_IF2)
    LOGGER.info('results:\n{}'.format('\n'.join(map(str, results))))
    assert len(results) == len(DEVICE_CONFIG_IF2)
    for result in results: assert isinstance(result, bool) and result

def test_device_driverapi_emulated_getconfig(device_driverapi_emulated : EmulatedDriver):
    stored_config = device_driverapi_emulated.GetConfig()
    LOGGER.info('stored_config:\n{}'.format('\n'.join(map(str, stored_config))))
    assert len(stored_config) == len(DEVICE_CONFIG_IF1) + len(DEVICE_CONFIG_IF2)
    for config_row in stored_config: assert (config_row in DEVICE_CONFIG_IF1) or (config_row in DEVICE_CONFIG_IF2)
    for config_row in DEVICE_CONFIG_IF1: assert config_row in stored_config
    for config_row in DEVICE_CONFIG_IF2: assert config_row in stored_config

    # should work
    stored_config = device_driverapi_emulated.GetConfig([PATH_IF.format('IF2')])
    LOGGER.info('stored_config:\n{}'.format('\n'.join(map(str, stored_config))))
    assert len(stored_config) == 1
    stored_config = stored_config[0]
    LOGGER.info('stored_config[0]:\n{}'.format('\n'.join(map(str, stored_config))))
    assert len(stored_config) == len(DEVICE_CONFIG_IF2)
    for config_row in stored_config: assert config_row in DEVICE_CONFIG_IF2
    for config_row in DEVICE_CONFIG_IF2: assert config_row in stored_config

def test_device_driverapi_emulated_deleteconfig(device_driverapi_emulated : EmulatedDriver):
    # should work
    results = device_driverapi_emulated.DeleteConfig([PATH_ADDRIPV4.format('IF2', 0, '10.2.0.1')])
    LOGGER.info('results:\n{}'.format('\n'.join(map(str, results))))
    assert (len(results) == 1) and isinstance(results[0], bool) and results[0]

    stored_config = device_driverapi_emulated.GetConfig()
    LOGGER.info('stored_config:\n{}'.format('\n'.join(map(str, stored_config))))

    device_config_if2 = list(filter(lambda row: '10.2.0.1' not in row[0], copy.deepcopy(DEVICE_CONFIG_IF2)))
    assert len(stored_config) == len(DEVICE_CONFIG_IF1) + len(device_config_if2)
    for config_row in stored_config: assert (config_row in DEVICE_CONFIG_IF1) or (config_row in device_config_if2)
    for config_row in DEVICE_CONFIG_IF1: assert config_row in stored_config
    for config_row in device_config_if2: assert config_row in stored_config

def test_device_driverapi_emulated_subscriptions(device_driverapi_emulated : EmulatedDriver):
    # should work
    duration = 10.0
    interval = 1.5
    results = device_driverapi_emulated.SubscribeState([
        (DEVICE_STATE_IF1_TX_PKTS, duration, interval),
        (DEVICE_STATE_IF1_RX_PKTS, duration, interval),
        (DEVICE_STATE_IF2_TX_PKTS, duration, interval),
        (DEVICE_STATE_IF2_RX_PKTS, duration, interval),
    ])
    LOGGER.info('results:\n{}'.format('\n'.join(map(str, results))))
    assert len(results) == 4
    for result in results: assert isinstance(result, bool) and result

    stored_config = device_driverapi_emulated.GetConfig()
    LOGGER.info('stored_config:\n{}'.format('\n'.join(map(str, stored_config))))

    time.sleep(duration + 1.0) # let time to generate samples, plus 1 second extra time

    samples = []
    for sample in device_driverapi_emulated.GetState(blocking=False):
        LOGGER.info('sample: {}'.format(str(sample)))
        timestamp,resource_key,resource_value = sample
        samples.append((timestamp, resource_key, resource_value))
    LOGGER.info('samples:\n{}'.format('\n'.join(map(str, samples))))
    assert len(samples) == 4 * (math.floor(duration/interval) + 1)

    results = device_driverapi_emulated.UnsubscribeState([
        (DEVICE_STATE_IF1_TX_PKTS, 10.0, 1.5),
        (DEVICE_STATE_IF1_RX_PKTS, 10.0, 1.5),
        (DEVICE_STATE_IF2_TX_PKTS, 10.0, 1.5),
        (DEVICE_STATE_IF2_RX_PKTS, 10.0, 1.5),
    ])
    LOGGER.info('results:\n{}'.format('\n'.join(map(str, results))))
    assert len(results) == 4
    for result in results: assert isinstance(result, bool) and result

    stored_config = device_driverapi_emulated.GetConfig()
    LOGGER.info('stored_config:\n{}'.format('\n'.join(map(str, stored_config))))
    device_config_if2 = list(filter(lambda row: '10.2.0.1' not in row[0], copy.deepcopy(DEVICE_CONFIG_IF2)))
    assert len(stored_config) == len(DEVICE_CONFIG_IF1) + len(device_config_if2)
    for config_row in stored_config: assert (config_row in DEVICE_CONFIG_IF1) or (config_row in device_config_if2)
    for config_row in DEVICE_CONFIG_IF1: assert config_row in stored_config
    for config_row in device_config_if2: assert config_row in stored_config