Newer
Older
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): # pylint: disable=redefined-outer-name
results = device_driverapi_emulated.SetConfig(DEVICE_CONFIG_IF1)
LOGGER.info('results:\n{:s}'.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{:s}'.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): # pylint: disable=redefined-outer-name
stored_config = device_driverapi_emulated.GetConfig()
LOGGER.info('stored_config:\n{:s}'.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
stored_config = device_driverapi_emulated.GetConfig([PATH_IF.format('IF2')])
LOGGER.info('stored_config:\n{:s}'.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): # pylint: disable=redefined-outer-name
results = device_driverapi_emulated.DeleteConfig([(PATH_ADDRIPV4.format('IF2', 0, '10.2.0.1'), '')])
LOGGER.info('results:\n{:s}'.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{:s}'.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): # pylint: disable=redefined-outer-name
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{:s}'.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{:s}'.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: {:s}'.format(str(sample)))
timestamp,resource_key,resource_value = sample
samples.append((timestamp, resource_key, resource_value))
LOGGER.info('samples:\n{:s}'.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{:s}'.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{:s}'.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