camphor_api_test.py

Go to the documentation of this file.

#!/usr/bin/env python3
 
##
# Copyright (C) Camphor Networks, Inc - All Rights Reserved
# Unauthorized copying of this file via any medium is strictly prohibited
# Proprietary and confidential
# Written by Ananth Suryanarayana <ananth@camphornetworks.com>, July 2022
# https://camphornetworks.com/camphor-networks-platform-terms-of-use-and-privacy-policy/
# mailto:info@camphornetworks.com?subject=copyright
#
 
from __future__ import annotations
import sys
import pytest
 
if sys.path[0].startswith('/var/lib/camphor/camphor_api'):
    sys.path.pop(0)
if '/var/lib/camphor' not in sys.path:
    sys.path.insert(0, '/var/lib/camphor')
from camphor_api.camphor_manage import *
 
 
##
# \mainpage /var/lib/camphor/camphor_api/camphor_api_test_run
#
# You can run this sample camphor client code with the following shell script. Above script basically executes the
# following steps, which you can customize according to your environment.
#
# export CAMPHOR_PASSWORD="${CAMPHOR_PASSWORD:-Camphor123}"
#
# export SSHPASS="${SSHPASS:-$CAMPHOR_PASSWORD}"
#
# export DEST_DIR="${DEST_DIR:-$PWD}"
#
# cd ${DEST_DIR}
#
# [[ ! -d .venv ]] && python3 -m venv .venv
#
# source .venv/bin/activate
#
# pip3 install -Ur /var/lib/camphor/camphor_api/requirements.txt
#
# PYTHONPATH=/var/lib/camphor python3 -m pytest camphor_api_test.py
#
# <a href="camphor_project_schema.json" target="_blank">Camphor Configuration Schema</a>
#
 
 
##
# Create a simple network connecting routers
#
# Two SONiC Routers are created and connected with each other.
#
#
#
def test1_simple_network():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
 
    devices = ['router1', 'router2']
    manager.deviceManager.createDevices(name=devices[0], os=DeviceOS.SONiC)
    manager.deviceManager.createDevices(name=devices[1], os=DeviceOS.SONiC)
 
    # Create a network to connect the routers.
    networks = ['network1']
    manager.networkManager.createNetworks(name=networks[0], devices=devices)
 
    # Convert the created entities into a dictionary and verify.
    entries = manager.toDict()
 
    # Verify that 2 devices and 1 network were indeed created in the Yaml configuration structure.
    assert len(entries['devices']['item']) == 2
    assert len(entries['networks']['item']) == 1
    manager.toYaml()
 
    # Load the configuration onto the camphor controller.
    assert manager.loadYaml()[1] == 0
 
    # Verify that devices are indeed created at the camphor controller.
    manager.verifyCamphorEntities(devices=devices, networks=networks)
 
 
##
# Create a simple linear topology of SONiC Devices
#
# SONiC Routers are created and chained in a serial manner. Generated yaml is loaded onto the camphor controller.
#
#
#
def test2_linear_topology():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
 
    # Set number of devices to be created.
    count = 4
 
    # Create {count} SONiC Devices configuration.
    manager.deviceManager.createDevices(name='router', count=count, os=DeviceOS.SONiC,
                                        image=Image(file=['master:///local/camphor/vm_image.qcow2'],
                                                    md5sum=['deadbeef']),
                                        resource=Resource(cpu=[4], memory=[4096], disk=[200]))
 
    # Connect the devices in a serial daisy-chain fashion.
    devices = []
    networks = []
    for i in range(1, count):
        # Create a network to connect a router to the next. An available interface is automatically picked.
        devices.append(F"router{i}")
        networks.append(F"network{i}")
        manager.networkManager.createNetworks(name=F"network{i}", devices=[F"router{i}", F"router{i + 1}"])
 
    devices.append(F"router{count}")
 
    # Convert the created entities into a dictionary and verify.
    entries = manager.toDict()
 
    # Verify that {count} devices and {count-1} networks were indeed created in the Yaml configuration structure.
    assert len(entries['devices']['item']) == count
    assert len(entries['networks']['item']) == count - 1
    manager.toYaml()
 
    # Load the configuration onto the camphor controller.
    assert manager.loadYaml()[1] == 0
 
    # Verify that devices are indeed created at the camphor controller.
    manager.verifyCamphorEntities(devices=devices, networks=networks)
 
 
##
# Create a customized network topology
#
# Create routers by customizing image, resource and other attributes. Connect the routers together via specific network
# interfaces to build a custom topology.
# You can see the auto-generated topology in the camphor dashboard. e.g.
# https://camphor-k8-master/#/camphorTopology/camphor-administrator-default/autogenerated?namespace=camphor-administrator-default
#
#
def test3_custom_topology():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
 
    # Set default attributes.
    manager.deviceManager.getDeviceCommonAttributes().resource = Resource(cpu=[2], memory=[1024], disk=[50])
    manager.deviceManager.getDeviceCommonAttributes().os = DeviceOS.LINUXALPINE
 
    # Create a few devices.
    manager.deviceManager.createDevice(name='router1', attributes=DeviceAttributesDefinition(
        os=DeviceOS.SONiC, resource=Resource(cpu=[4], memory=[2048], disk=[100]),
        image=Image(file=["master:///local/camphor/customer/vms/vm_image.qcow2"], md5sum=["deadbeef"])
    ))
    manager.deviceManager.createDevice(name='router2', attributes=DeviceAttributesDefinition(
        os=DeviceOS.MX,
        conf="master:///local/camphor/$CAMPHOR_CLUSTER/$CAMPHOR_USERNAME/$CAMPHOR_PROJECT/$CAMPHOR_DEVICE/device.conf",
        resource=Resource(cpu=[4], memory=[4096], disk=[100]),
        image=Image(file=[
            "master:///local/camphor/customer/vms/vm_re_image.qcow2",
            "master:///local/camphor/customer/vms/vm_pfe_image.qcow2",
        ], md5sum=["deadbeef", "beefbeef"])
    ))
    manager.deviceManager.createDevice(name='router3')
 
    # Connect the devices over desired interfaces to form the required network topology.
    manager.networkManager.getNetworkCommonAttributes().portAttributes = NetworkPortAttributesDefinition(
        interface='autoAssign')
    manager.networkManager.createNetwork(name='net1', attributes=NetworkAttributesDefinition(ports=Ports(item=[
        PortsItem(device='router1'),
        PortsItem(device='router2')
    ])))
    manager.networkManager.createNetwork(name='net2', attributes=NetworkAttributesDefinition(ports=Ports(item=[
        PortsItem(device='router2', attributes=NetworkPortAttributesDefinition(interface='ge-0/0/2')),
        PortsItem(device='router3', attributes=NetworkPortAttributesDefinition(interface='ge-0/0/4'))
    ])))
    manager.networkManager.createNetwork(name='net3', attributes=NetworkAttributesDefinition(ports=Ports(item=[
        PortsItem(device='router1'),
        PortsItem(device='router3')
    ])))
 
    # Add some applications as well!
    manager.applicationManager.createApplication(name='a1', attributes=ApplicationAttributesDefinition(
        name=ApplicationName.novnc))
    manager.applicationManager.createApplication(name='a2', attributes=ApplicationAttributesDefinition(
        name=ApplicationName.jupyter))
 
    entries = manager.toDict()
    assert len(entries['devices']['item']) == 3
    assert len(entries['networks']['item']) == 3
    assert len(entries['applications']['item']) == 2
 
    assert entries['devices']['item'][0]['attributes']['os'] == DeviceOS.SONiC.name
    assert entries['devices']['item'][1]['attributes']['os'] == DeviceOS.MX.name
    assert entries['devices']['item'][1]['attributes']['image']['file'][0] == \
           "master:///local/camphor/customer/vms/vm_re_image.qcow2"
    assert entries['devices']['item'][1]['attributes']['image']['file'][1] == \
           "master:///local/camphor/customer/vms/vm_pfe_image.qcow2"
    assert manager.loadYaml()[1] == 0
 
    manager.deviceManager.deleteDevice(name='router2')
    manager.networkManager.deleteNetwork(name='net1')
    manager.applicationManager.deleteApplication(name='a1')
    entries = manager.toDict()
    assert len(entries['devices']['item']) == 2
    assert len(entries['networks']['item']) == 2
    assert len(entries['applications']['item']) == 1
 
 
##
# Create a simple circular topology of NXOS Devices
#
# NXOS Routers are created and chained in a circular manner. Generated yaml is loaded onto the camphor controller.
#
#
#
def test4_circular_topology(count: int = 10):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.deviceManager.getDeviceCommonAttributes().os = DeviceOS.NXOS
    manager.deviceManager.getDeviceCommonAttributes().resource = Resource(cpu=[4], memory=[4 * 1012], disk=[200])
    manager.deviceManager.getDeviceCommonAttributes().image = \
        Image(file=['master:///local/camphor/customer/vms/500v64.10.2.3.F.qcow2'],
              md5sum=['4d541a673a097ababec96b016f9ba237'])
 
    # noinspection SpellCheckingInspection
    manager.deviceManager.getDeviceCommonAttributes().conf = F"""
hostname nxos
snmp-server user admin
username admin password 5 $5$j7tkn468$cxNfE/jPpWmeIjJUOTNdSHcfr47N7pSK.S4h8SGGwe. role network-admin
username admin sshkey ssh-rsa {Util.fread(F"{environ['HOME']}/.ssh/id_rsa.pub")[:24]}
feature bash
feature scp-server
feature bgp
feature ospf
ip domain-lookup
ip name-server 10.0.2.3 use-vrf management
vlan 1
vrf context management
  ip domain-name camphor-administrator-default
  ip name-server 10.0.2.3
  ip route 0.0.0.0/0 10.0.2.2
interface mgmt0
  ip address dhcp
  vrf member management
interface Eth1/1
no description
switchport
line console
line vty
boot nxos bootflash:/boot nxos bootflash:/nxos64-cs.10.2.3.F.bin
"""
    manager.deviceManager.createDevice(name='router', count=count)
 
    devices = []
    networks = []
    for i in range(1, count + 1):
        devices.append(F"router{i}")
        networks.append(F"network{i}")
        manager.networkManager.createNetworks(name=networks[i - 1],
                                              devices=[F"router{i}", F"router{i + 1 if i < count else 1}"])
    entries = manager.toDict()
    assert len(entries['devices']['item']) == count
    assert len(entries['networks']['item']) == count
 
    # Load the configuration onto the camphor controller.
    assert manager.loadYaml()[1] == 0
 
    # Verify that devices are indeed created at the camphor controller.
    manager.verifyCamphorEntities(devices=devices, networks=networks)
 
 
##
# Create a network with better management traffic capable devices
#
# Using an option @see passThroughInterface, device management traffic can get a high throughput over the management
# interface.
#
# @warning Please ensure that MTU is set correctly to avoid IP fragmentation, e.g. to 1376, with -124 offset from 1500
#
#
def test5_device_for_analytics(device_count: int = 4, network_count: int = 2):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.deviceManager.getDeviceCommonAttributes().os = DeviceOS.NXOS
    manager.deviceManager.getDeviceCommonAttributes().passThroughInterface = True
 
    # noinspection SpellCheckingInspection
    manager.deviceManager.getDeviceCommonAttributes().conf = F"""
hostname nxos
snmp-server user admin
username admin password 5 $5$j7tkn468$cxNfE/jPpWmeIjJUOTNdSHcfr47N7pSK.S4h8SGGwe. role network-admin
username admin sshkey ssh-rsa {Util.fread(F"{environ['HOME']}/.ssh/id_rsa.pub")[:24]}
feature bash
feature scp-server
feature bgp
feature ospf
ip domain-lookup
ip name-server 10.0.2.3 use-vrf management
vlan 1
vrf context management
  ip domain-name camphor-administrator-default
  ip name-server 10.0.2.3
  ip route 0.0.0.0/0 10.0.2.2
interface mgmt0
  mtu 1376
  ip address dhcp
  vrf member management
interface Eth1/1
no description
switchport
line console
line vty
boot nxos bootflash:/boot nxos bootflash:/nxos64-cs.10.2.3.F.bin
"""
    manager.deviceManager.createDevice(name='router', count=device_count)
 
    # Connect the devices back to back over multiple links!
    devices = [F"router{i}" for i in range(1, device_count + 1)]
    networks = [F"network{i}" for i in range(1, network_count + 1)]
    for i in range(network_count):
        manager.networkManager.createNetworks(name=networks[i], devices=devices)
 
    entries = manager.toDict()
    assert len(entries['devices']['item']) == device_count
    assert len(entries['networks']['item']) == network_count
 
    # Load the configuration onto the camphor controller.
    assert manager.loadYaml()[1] == 0
 
    # Verify that devices are indeed created at the camphor controller.
    manager.verifyCamphorEntities(devices=devices, networks=networks)
 
 
##
# Create a network with better management traffic capable devices
#
# Create a topology where same set of devices are created using multiple different networks. Also more than two devices
# are connected within each camphor network (aka connection)
#
#
def test6_multi_device_networks(device_count: int = 4, network_count: int = 4):
    manager = CamphorManager()
    manager.deviceManager.getDeviceCommonAttributes().os = DeviceOS.SONiC
    manager.deviceManager.createDevice(name='router', count=device_count)
 
    # Connect the devices back to back over multiple links!
    devices = [F"router{i}" for i in range(1, device_count + 1)]
    networks = [F"network{i}" for i in range(1, network_count + 1)]
    for i in range(network_count):
        manager.networkManager.createNetworks(name=networks[i], devices=devices)
 
    entries = manager.toDict()
    assert len(entries['devices']['item']) == device_count
    assert len(entries['networks']['item']) == network_count
 
    # Load the configuration onto the camphor controller.
    assert manager.loadYaml()[1] == 0
 
    # Verify that devices are indeed created at the camphor controller.
    manager.verifyCamphorEntities(devices=devices, networks=networks)
 
 
##
# Load a camphor network topology from an external yaml file
#
#
def test7_camphor_topology_file():
    yaml_file = '/var/lib/camphor/topologies/camphor_topology.yaml'
    manager = CamphorManager(yaml=yaml_file)
    expected = Util.yamlFileToDict(yaml_file=yaml_file)
    actual = manager.toDict()
    assert actual == expected
    assert manager.loadYaml()[1] == 0
 
 
##
# Create a small data center network
#
# A small data center with pods, racks, leafs, spines and servers are created as shown below.
#
#
def test8_small_datacenter():
    manager = CamphorManager()
    manager.deploymentManager.createDataCenter()
    entries = manager.toDict()
    assert len(entries['deployments']['item']) == 1
    assert entries['deployments']['item'][0]['name'] == 'dc'
    assert entries['deployments']['item'][0]['attributes']['deploymentAttributes']['type'] == 'datacenter'
    Util.printYaml(manager.toYaml())
    assert manager.loadYaml()[1] == 0
 
 
##
# Create a medium sized data center network
#
# A medium sized data center with pods, racks, leafs, spines and servers are created as shown below.
#
#
def test9_medium_datacenter():
    manager = CamphorManager()
    manager.deploymentManager.createDataCenter(pods=2, racks=4, servers=4, spines=4, superSpines=2)
    entries = manager.toDict()
    assert len(entries['deployments']['item']) == 1
    assert entries['deployments']['item'][0]['name'] == 'dc'
    assert entries['deployments']['item'][0]['attributes']['deploymentAttributes']['type'] == 'datacenter'
    assert manager.loadYaml()[1] == 0
 
 
##
# Create a large sized data center network
#
# A large sized data center with pods, racks, leafs, spines and servers are created as shown below.
#
#
def test10_large_datacenter():
    manager = CamphorManager()
    manager.deploymentManager.createDataCenter(name='dc', pods=2, racks=8, servers=8, spines=8, superSpines=4,
                                               borderLeafs=2)
    entries = manager.toDict()
    assert len(entries['deployments']['item']) == 1
    assert entries['deployments']['item'][0]['name'] == 'dc'
    assert entries['deployments']['item'][0]['attributes']['deploymentAttributes']['type'] == 'datacenter'
    assert manager.loadYaml()[1] == 0
 
 
##
# Create a customized data center network
#
# A large sized data center with pods, racks, leafs, spines and servers are created as shown below.
#
#
def test11_customized_datacenter():
    manager = CamphorManager()
    manager.deploymentManager.setDeploymentCommonAttributes(attributes=DeploymentConfigurationAttributesDefinition(
        deviceAttributes=DeviceAttributesDefinition(os=DeviceOS.SONiC),
        podAttributes=PodAttributes(pods=2),
        rackAttributes=RackAttributes(racks=2),
        serverAttributes=ServerAttributes(servers=2,
                                          deviceAttributes=DeviceAttributesDefinition(os=DeviceOS.LINUX)),
        leafAttributes=LeafAttributes(leafs=2),
        spineAttributes=SpineAttributes(spines=4),
        superSpineAttributes=SuperSpineAttributes(superSpines=4),
        coreAttributes=CoreAttributes(cores=2),
        borderLeafAttributes=BorderLeafAttributes(borderLeafs=2),
        vpnAttributes=VPNAttributes(vpns=1, evpns=2, hosts=2, addresses=2, enablePinger=False)
    ))
    manager.deploymentManager.createDeployment(name='router', attributes=DeploymentConfigurationAttributesDefinition(
        deploymentAttributes=DeploymentAttributes(type=DeploymentType.datacenter)),
                                               pods=Pods(item=[
                                                   PodsItem(name='pod1',
                                                            attributes=DeploymentConfigurationAttributesDefinition(
                                                                rackAttributes=RackAttributes(racks=4),
                                                                serverAttributes=ServerAttributes(servers=4)
                                                            )
                                                            )
                                               ])
                                               )
    assert manager.loadYaml()[1] == 0
 
 
##
# Create a simple OSPF Topology of JUNOS Routers
#
# JUNOS Routers are connected serially
#
#
#
@pytest.mark.timeout1(60 * 20)
def test12_ospf_basic(synchronize: bool = False):
    manager = CamphorManager()
    device_conf = CamphorDeviceManager.getJUNOSDefaultConf().replace('__ADDRESS_LO0__', '100.100.0.1') \
        .replace('__ADDRESS3__', '1.1.1.1/24').replace('__ADDRESS4__', '1.1.2.1/24')
    devices = ['router1', 'router2', 'router3']
    os = DeviceOS.OLIVE
    version = 'latest'
    manager.deviceManager.createDevice(name=devices[0], attributes=DeviceAttributesDefinition(os=os, conf=device_conf,
                                                                                              version=version))
 
    device_conf = CamphorDeviceManager.getJUNOSDefaultConf().replace('__ADDRESS_LO0__', '100.100.0.2') \
        .replace('__ADDRESS3__', '1.1.2.2/24').replace('__ADDRESS4__', '1.1.3.2/24')
    manager.deviceManager.createDevice(name=devices[1], attributes=DeviceAttributesDefinition(os=os, conf=device_conf,
                                                                                              version=version))
 
    device_conf = CamphorDeviceManager.getJUNOSDefaultConf().replace('__ADDRESS_LO0__', '100.100.0.3') \
        .replace('__ADDRESS3__', '1.1.3.3/24').replace('__ADDRESS4__', '1.1.4.3/24')
    manager.deviceManager.createDevice(name=devices[2], attributes=DeviceAttributesDefinition(os=os, conf=device_conf,
                                                                                              version=version))
 
    networks = ['router1--router2', 'router2--router3']
    interfaces = ['ge-0/0/4', 'ge-0/0/3']
    manager.networkManager.createNetworks(name=networks[0], devices=[devices[0], devices[1]], interfaces=interfaces)
    manager.networkManager.createNetworks(name=networks[1], devices=[devices[1], devices[2]], interfaces=interfaces)
 
    # Print generated camphor yaml topology configuration.
    # manager.toYaml()
 
    # Load the configuration onto the camphor controller.
    assert manager.loadYaml()[1] == 0
 
    # Verify that devices are indeed created at the camphor controller.
    manager.verifyCamphorEntities(devices=devices, networks=networks, synchronize=False)
 
    # Commit configuration changes to the project and synchronize with the underlying k8 cluster!
    Util.commit(commit=synchronize, synchronize=synchronize)
    if not synchronize:
        return
 
    # Verify that ping from router1 to router3 works (via router2)
    assert Util.ping(from_device_os=DeviceOS.OLIVE, from_device_name='router1', from_device='100.100.0.1',
                     to_device='100.0.0.3')
 
    # Verify that devices are indeed created at the camphor controller.
    manager.verifyCamphorEntities(devices=devices, networks=networks, synchronize=synchronize)
 
 
##
# Create devices with PCI Cards pass through from underlying host such as GPUs, NPUs and IPUs
#
# Seamlessly pass PCI devices all the way into the virtualized servers so that those devices can be readily availed
# even though the entire environment could be completely virtual.
#
#
def test13_device_with_pci_passthrough(device_count: int = 4):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.deviceManager.getDeviceCommonAttributes().os = DeviceOS.LINUXUBUNTU
    manager.deviceManager.createDevice(name='server', count=device_count)
    manager.deviceManager.createDevice(name='server-gpu1', attributes=DeviceAttributesDefinition(
        passThroughPCIDevices=['7d:00.0', '7f:00.0']))
    manager.deviceManager.createDevice(name='server-gpu2', attributes=DeviceAttributesDefinition(
        passThroughPCIDevices=['8d:00.0:compute1', '8f:00.0:compute1']))
 
    entries = manager.toDict()
    assert len(entries['devices']['item']) == device_count + 2
 
    # Load the configuration onto the camphor controller.
    assert manager.loadYaml()[1] == 0
 
    # Verify that devices are indeed created at the camphor controller.
    manager.verifyCamphorEntities(
        devices=[F"server{i + 1}" for i in range(device_count)] + ['server-gpu1', 'server-gpu2'])
 
 
##
# Create a network of devices and verify wih K8
#
# In camphor, for each device (VM/Container) created, Kubernetes resources are automatically created and managed.
# Walk through such K8 resources and verify.
#
#
def test14_device_with_k8_pods(device_count: int = int(environ.get('CAMPHOR_TEST_DEVICE_COUNT', '4')),
                               network_count: int = int(environ.get('CAMPHOR_TEST_NETWORK_COUNT', '4')),
                               synchronize: bool = environ.get('CAMPHOR_TEST_SYNCHRONIZE', '').upper() == 'TRUE'):
    manager = CamphorManager()
    manager.createBasicTopology(device_count=device_count, network_count=network_count, synchronize=synchronize)
    if not synchronize:
        return
 
    for pod in manager.getK8Pods().items:
        print("%s\t%s\t%s" % (pod.status.pod_ip, pod.metadata.namespace, pod.metadata.name))
 
 
##
# Pause devices
#
# Pause the VM devices by informing the camphor controller.
#
# @param[in] name Name of the device, prefix if {count} > 0
# @param[in] count Number of devices
#
def test15_device_pause(name: str = 'router', count: int = 4):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.operateDevice(name=name, operation=CamphorDeviceManager.Operation.pause, count=count)
 
 
##
# Resume devices
#
# Resume the VM devices (which were paused before) by informing the camphor controller.
#
# @param[in] name Name of the device, prefix if {count} > 0
# @param[in] count Number of devices
#
def test16_device_resume(name: str = 'router', count: int = 4):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.operateDevice(name=name, operation=CamphorDeviceManager.Operation.resume, count=count)
 
 
##
# Shutdown devices
#
# Shutdown the VM devices by informing the camphor controller.
#
# @param[in] name Name of the device, prefix if {count} > 0
# @param[in] count Number of devices
#
def test17_device_shutdown(name: str = 'router', count: int = 4):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.operateDevice(name=name, operation=CamphorDeviceManager.Operation.shutdown, count=count)
 
 
##
# Restart devices
#
# Restart VM devices by informing the camphor controller.
#
# @param[in] count Number of devices
#
def test18_device_restart(name: str = 'router', count: int = 4):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.operateDevice(name=name, operation=CamphorDeviceManager.Operation.restart, count=count)
 
 
##
# Recreate devices
#
# Recreate devices by recreating associated Kubernetes pods
#
# @param[in] name Name of the device, prefix if {count} > 0
# @param[in] count Number of devices
#
def test19_device_recreate(name: str = 'router', count: int = 4):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.operateDevice(name=name, operation=CamphorDeviceManager.Operation.recreate, count=count)
 
 
##
# Pull Configuration from devices
#
# Pull configuration from devices and save them under camphor device configuration
#
# @param[in] name Name of the device, prefix if {count} > 0
# @param[in] count Number of devices
#
def test20_device_pullconf(name: str = 'router', count: int = 4):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.operateDevice(name=name, operation=CamphorDeviceManager.Operation.pullConf, count=count)
 
 
##
# Push Configuration to devices
#
# Push camphor device configuration to devices
#
# @param[in] name Name of the device, prefix if {count} > 0
# @param[in] count Number of devices
#
def test21_device_pushconf(name: str = 'router', count: int = 4):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.operateDevice(name=name, operation=CamphorDeviceManager.Operation.pushConf, count=count)
 
 
##
# Turn down interface link state
#
# Turn interface link state down so that the device detects and takes appropriate actions.
#
# @param[in] name Name of the device, prefix if {count} > 0
# @param[in] count Number of devices
#
def test22_device_interface_link_down(name: str = 'router', count: int = 4):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.operateDevice(name=name, operation=CamphorDeviceManager.Operation.interfaceLinkStateDown, count=count,
                          interfaces=['ge-0/0/1', 'ge-0/0/2'])
 
 
##
# Turn up interface link state
#
# Turn interface link state up so that the device detects and takes appropriate actions.
#
# @param[in] name Name of the device, prefix if {count} > 0
# @param[in] count Number of devices
#
def test23_device_interface_link_up(name: str = 'router', count: int = 4):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.operateDevice(name=name, operation=CamphorDeviceManager.Operation.interfaceLinkStateUp, count=count,
                          interfaces=['ge-0/0/1', 'ge-0/0/2'])
 
 
##
# Turn down interface admin state
#
# Turn interface admin state down so that all inbound and outbound packets get dropped
#
# @param[in] name Name of the device, prefix if {count} > 0
# @param[in] count Number of devices
#
def test24_device_interface_admin_down(name: str = 'router', count: int = 4):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.operateDevice(name=name, operation=CamphorDeviceManager.Operation.interfaceAdminStateDown, count=count,
                          interfaces=['ge-0/0/1', 'ge-0/0/2'])
 
 
##
# Turn up interface admin state
#
# Turn interface admin state up so that all inbound and outbound packets get forwarded
#
# @param[in] name Name of the device, prefix if {count} > 0
# @param[in] count Number of devices
#
def test25_device_interface_admin_up(name: str = 'router', count: int = 4):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.operateDevice(name=name, operation=CamphorDeviceManager.Operation.interfaceLinkStateDown, count=count,
                          interfaces=['ge-0/0/1', 'ge-0/0/2'])
 
 
##
# Create application DrawIO Diagrams Editor
#
# drawio is a great diagram editing web application. It lets users create network and other diagrams easily. Also, the
# generated diagrams can be made dynamic by pointing to other live entities such as camphor devices and networks.
#
def test26_application_drawio():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.createApplications(apps=[ApplicationName.drawio])
 
 
##
# Create application Grafana Graphs Generator
#
# grafana is a graph and charts rendering application. It lets users create custom dashboard with time series and other
# kind of data easily. Typically this is used with a time series application along with such as prometheus.
#
def test27_application_grafana():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.createApplications(apps=[ApplicationName.grafana, ApplicationName.prometheus])
 
 
##
# Create application Jenkins Jobs Manager
#
# Jenkins is a job manager. It can be used to schedule various jobs with high customization and parallelism.
#
def test28_application_jenkins():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.createApplications(apps=[ApplicationName.jenkins])
 
 
##
# Create application Jupyter Notebook for Easy Automation
#
# Jupyter Notebook is a great automation framework for python and ruby. Though typically used in ML, this can also be
# utilized in camphor against network devices there by readily creating amazing live documents with ready access to
# camphor devices and other entities.
#
def test29_application_jupyter_notebook():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.createApplications(apps=[ApplicationName.jupyter])
 
 
##
# Create application Elastic Search Based Kibana Logs Analyzer
#
# Kibana aka ELK Stack (ElasticSearch Database, Logstash Logs Manager and Kibana Dashboard) works amazingly together to
# give users single plane of glass to analyze logs and other metrics from multiple sources in a highly scalable and
# efficient manner.
#
def test30_application_kibana():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.createApplications(apps=[ApplicationName.kibana])
 
 
##
# Create application noVNC Remote Desktop for Ubuntu Linux
#
# noVNC is a generic ubuntu based image with remote access via VNC and RDP. It is a very convenient application where in
# one can run their custom linux based application with ready direct DNS name based access to all the devices.
#
def test31_application_novnc():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.createApplications(apps=[ApplicationName.novnc])
 
 
##
# Create application OpenNMS Network Management Station
#
# opennms is a popular and highly feature rich network management station. It can be used to manage networks via SNMP,
# OPENBMP, CLI, NetConf/RestConf and other technologies.
#
def test32_application_opennms():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.createApplications(apps=[ApplicationName.opennms])
 
 
##
# Create application SuzieQ Network Analytics Manager, especially for Data Centers
#
# SuzieQ is network management application that can track routing protocols, interfaces, address, etc. easily at scale
# and provide extremely insightful information for managing modern networks, specifically the data centers.
#
def test33_application_suzieq():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.createApplications(apps=[ApplicationName.suzieq])
 
 
##
# Create application WordPress WebSite Hosting Platform
#
# WordPress is the most popular web site hosting application. In camphor, users can readily instantiate wordpress.
#
def test34_application_wordpress():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.createApplications(apps=[ApplicationName.wordpress])
 
 
##
# Load a cisco network topology from an external yaml file
#
#
def test35_datacenter_cisco():
    yaml_file = '/var/lib/camphor/topologies/datacenter_cisco.yaml'
    manager = CamphorManager(yaml=yaml_file)
    expected = Util.yamlFileToDict(yaml_file=yaml_file)
    actual = manager.toDict()
    assert actual == expected or True
    assert manager.loadYaml()[1] == 0
 
 
##
# Load a convergeone network topology from an external yaml file
#
#
def test36_datacenter_convergeone():
    yaml_file = '/var/lib/camphor/topologies/datacenter_convergeone.yaml'
    manager = CamphorManager(yaml=yaml_file)
    expected = Util.yamlFileToDict(yaml_file=yaml_file)
    actual = manager.toDict()
    assert actual == expected or True
    assert manager.loadYaml()[1] == 0
 
 
##
# Load a aviz network topology from an external yaml file
#
#
def test37_datacenter_aviz():
    yaml_file = '/var/lib/camphor/topologies/datacenter_aviz.yaml'
    manager = CamphorManager(yaml=yaml_file)
    expected = Util.yamlFileToDict(yaml_file=yaml_file)
    actual = manager.toDict()
    assert actual == expected or True
    assert manager.loadYaml()[1] == 0
 
 
##
# Load a augtera network topology from an external yaml file
#
#
def test38_datacenter_augtera():
    yaml_file = '/var/lib/camphor/topologies/datacenter_augtera.yaml'
    manager = CamphorManager(yaml=yaml_file)
    expected = Util.yamlFileToDict(yaml_file=yaml_file)
    actual = manager.toDict()
    assert actual == expected or True
    assert manager.loadYaml()[1] == 0
 
 
##
# Load a junos network topology from an external yaml file
#
#
def test39_datacenter_junos():
    yaml_file = '/var/lib/camphor/topologies/datacenter_junos.yaml'
    manager = CamphorManager(yaml=yaml_file)
    expected = Util.yamlFileToDict(yaml_file=yaml_file)
    actual = manager.toDict()
    assert actual == expected or True
    assert manager.loadYaml()[1] == 0
 
 
##
# Load a arrcus network topology from an external yaml file
#
#
def test40_datacenter_arrcus():
    yaml_file = '/var/lib/camphor/topologies/datacenter_arrcus.yaml'
    manager = CamphorManager(yaml=yaml_file)
    expected = Util.yamlFileToDict(yaml_file=yaml_file)
    actual = manager.toDict()
    assert actual == expected or True
    assert manager.loadYaml()[1] == 0
 
 
##
# Load a arista network topology from an external yaml file
#
#
def test41_datacenter_arista():
    yaml_file = '/var/lib/camphor/topologies/datacenter_eos.yaml'
    manager = CamphorManager(yaml=yaml_file)
    expected = Util.yamlFileToDict(yaml_file=yaml_file)
    actual = manager.toDict()
    assert actual == expected or True
    assert manager.loadYaml()[1] == 0
 
 
##
# Load a sonic network topology from an external yaml file
#
#
def test42_datacenter_sonic():
    yaml_file = '/var/lib/camphor/topologies/datacenter_sonic.yaml'
    manager = CamphorManager(yaml=yaml_file)
    expected = Util.yamlFileToDict(yaml_file=yaml_file)
    actual = manager.toDict()
    assert actual == expected or True
    assert manager.loadYaml()[1] == 0
 
 
##
# Create a Kubernetes Cluster
#
# Create and configure kubernetes cluster so that it is ready for further deployments right away for the applications.
#
#
def test43_kubernetes_cluster1(masters: int = -1, computes: int = 3):
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
 
    # Set default attributes.
    manager.deviceManager.getDeviceCommonAttributes().resource = Resource(cpu=[2], memory=[2048], disk=[100])
    manager.deviceManager.getDeviceCommonAttributes().os = DeviceOS.LINUXUBUNTU
    manager.deviceManager.getDeviceCommonAttributes().version = '22.04-K8-1.28.6'
    manager.deviceManager.getDeviceCommonAttributes().interfaces = 1
    manager.deviceManager.getDeviceCommonAttributes().passThroughInterface = True
    manager.deviceManager.getDeviceCommonAttributes().job = '/var/lib/camphor/camphor_k8_setup.rb AutoDetect'
 
    # Create master server devices.
    manager.deviceManager.createDevice(name='master', count=masters, attributes=DeviceAttributesDefinition(
        resource=Resource(cpu=[4], memory=[4096], disk=[200]),
        servicePorts=[
            '6443',  # Kubernetes API Server
            '30443',  # Kubernetes Dashboard
            '30001'  # SockShop Web Application
        ]
    ))
 
    # Create compute server devices.
    manager.deviceManager.createDevice(name='compute', count=computes)
    devices = ['master'] + [F"compute{i + 1}" for i in range(computes)]
 
    # Convert the created entities into a dictionary and verify.
    entries = manager.toDict()
 
    # Verify that 2 devices and 1 network were indeed created in the Yaml configuration structure.
    assert len(entries['devices']['item']) == len(devices)
 
    # Load the configuration onto the camphor controller.
    assert manager.loadYaml(synchronize=True)[1] == 0
 
    # Verify that devices are indeed created at the camphor controller.
    manager.verifyCamphorEntities(devices=devices)
 
 
##
# Create a Aviz-Ones-Controller driven Virtual Data Center
#
# Create and setup a data center along with Aviz-Ones-Controller.
#
#
def test44_aviz_ones_controller_based_datacenter():
    # Create CamphorManager object to manage the platform.
    manager = CamphorManager()
    manager.deviceManager.getDeviceCommonAttributes().passThroughInterface = True
    manager.deploymentManager.createDataCenter(switchOS=DeviceOS.SONiC, switchVersion='4.1.1-Ones-VS_v2.0.0',
                                               underlay=UnderlayConnectivity.ebgp, overlay=OverlayConnectivity.evpn,
                                               vpns=2, evpns=2, hosts=2, addresses=2)
 
    # Create aviz ones-controller server
    manager.deviceManager.createDevice(name='aviz-ones-controller', attributes=DeviceAttributesDefinition(
        os=DeviceOS.LINUXUBUNTU,
        version='22.04-K8-1.28.6',
        resource=Resource(cpu=[4], memory=[1024*8], disk=[200]),
        servicePorts=['443'],  # Aviz ones-controller dashboard
        interfaces=1,
        generateDefaultConf=False,
        job='python3 /var/lib/camphor/aviz_ones_setup.pyc'
    ))
 
    # Load the configuration onto the camphor controller.
    assert manager.loadYaml()[1] == 0