virl2_client.models package

This package contains the VIRL2 client library models for labs, nodes, interfaces and links. It also contains classes for node and image definition and helper classes for automation and authentication.

class virl2_client.models.GroupManagement(session: httpx.Client)

Bases: object

__init__(session: httpx.Client) → None

create_group(name: str, description: str = '', members: list[str] | None = None, labs: list[dict[str, str]] | None = None) → dict

Creates a group.

Parameters:

• name – group name

• description – group description

• members – group members

• labs – group labs

Returns:

created group object

delete_group(group_id: str) → None

Deletes a group.

Parameters:

group_id – group uuid4

Returns:

None

get_group(group_id: str) → dict

Gets info for the specified group.

Parameters:

group_id – group uuid4

Returns:

group object

group_id(group_name: str) → str

Get unique user uuid4.

Parameters:

group_name – group name

Returns:

group unique identifier

group_labs(group_id: str) → list[str]

Get the list of labs that are associated with this group.

Parameters:

group_id – group uuid4

Returns:

list of labs associated with this group

group_members(group_id: str) → list[str]

Gets group members.

Parameters:

group_id – group uuid4

Returns:

list of users associated with this group

groups() → list

Get the list of available groups.

Returns:

list of group objects

update_group(group_id: str, name: str | None = None, description: str | None = None, members: list[str] | None = None, labs: list[dict[str, str]] | None = None) → dict

Updates a group.

Parameters:

• group_id – group uuid4

• name – new group name

• description – group description

• members – group members

• labs – group labs

Returns:

updated group object

property base_url: str

class virl2_client.models.Interface(iid: str, node: Node, label: str, slot: int | None, iface_type: str = 'physical')

Bases: object

__init__(iid: str, node: Node, label: str, slot: int | None, iface_type: str = 'physical') → None

A VIRL2 network interface, part of a node.

Parameters:

• iid – interface ID

• node – node object

• label – the label of the interface

• slot – the slot of the interface

• iface_type – the type of the interface, defaults to “physical”

as_dict() → dict[str, str]

bring_up() → None

degree()

get_link_to(other_interface: Interface) → Link | None

Returns the link between this interface and another.

Parameters:

other_interface – the other interface

Returns:

A Link

is_connected()

links()

peer_interfaces()

peer_nodes()

remove_on_server() → None

shutdown() → None

property connected: bool

Whether the interface is connected to a link.

property discovered_ipv4: str | None

property discovered_ipv6: str | None

property discovered_mac_address: str | None

property is_physical

property lab_base_url: str

property link: Link | None

Is link if connected, otherwise None.

property peer_interface: Interface | None

property peer_node: Node | None

property physical: bool

Whether the interface is physical.

property readbytes: int

property readpackets: int

property state: str | None

property writebytes: int

property writepackets: int

class virl2_client.models.Lab(title: str | None, lab_id: str, session: httpx.Client, username: str, password: str, auto_sync=True, auto_sync_interval=1.0, wait=True, wait_max_iterations=500, wait_time=5, hostname: str | None = None, resource_pool_manager: ResourcePoolManagement | None = None)

Bases: object

__init__(title: str | None, lab_id: str, session: httpx.Client, username: str, password: str, auto_sync=True, auto_sync_interval=1.0, wait=True, wait_max_iterations=500, wait_time=5, hostname: str | None = None, resource_pool_manager: ResourcePoolManagement | None = None) → None

A VIRL2 lab network topology. Contains nodes, links and interfaces.

Parameters:

• title – Name / title of the lab

• lab_id – A lab ID

• session – httpx Client session

• username – Username of the user to authenticate

• password – Password of the user to authenticate

• auto_sync – Should local changes sync to the server automatically

• auto_sync_interval – Interval to auto sync in seconds

• wait – Wait for convergence on backend

• wait_max_iterations – Maximum number of tries or calls for convergence

• wait_time – Time to sleep between calls for convergence on backend

• hostname – Force hostname/ip and port for pyATS console terminal server

• resource_pool_manager – a ResourcePoolManagement object shared with parent ClientLibrary

add_node_local(node_id: str, label: str, node_definition: str, image_definition: str | None, configuration: str | None, x: int, y: int, ram: int | None = None, cpus: int | None = None, cpu_limit: int | None = None, data_volume: int | None = None, boot_disk_size: int | None = None, hide_links: bool = False, tags: list | None = None, compute_id: str | None = None, resource_pool: ResourcePool | None = None) → Node

Helper function to add a node to the client library.

build_configurations() → None

Build basic configurations for all nodes in the lab which do not already have a configuration and also do support configuration building.

cleanup_pyats_connections() → None

Closes and cleans up connection that pyATS might still hold.

connect_two_nodes(node1: Node, node2: Node) → Link

Convenience method to connect two nodes within a lab.

Parameters:

• node1 – the first node object

• node2 – the second node object

Returns:

the created link

create_interface(node: Node, slot: int | None = None, wait: bool | None = None) → Interface

Create an interface in the next available slot, or, if a slot is specified, in all available slots up to and including the specified slot.

Parameters:

• node – The node on which the interface is created

• slot – (optional)

• wait – Wait for convergence (if left at default, the lab wait property takes precedence)

Returns:

The newly created interface

create_interface_local(iface_id: str, label: str, node: Node, slot: int | None, iface_type: str = 'physical') → Interface

Helper function to create an interface in the client library.

create_link(i1: Interface, i2: Interface, wait: bool | None = None) → Link

Creates a link between two interfaces

Parameters:

• i1 – the first interface object

• i2 – the second interface object

• wait – Wait for convergence (if left at default, the lab wait property takes precedence)

Returns:

the created link

create_link_local(i1: Interface, i2: Interface, link_id: str, label: str | None = None) → Link

Helper function to create a link in the client library.

create_node(label: str, node_definition: str, x: int = 0, y: int = 0, wait: bool | None = None, populate_interfaces: bool = False, **kwargs) → Node

Creates a node in the lab with the given parameters.

Parameters:

• label – Label

• node_definition – Node definition to use

• x – x coordinate

• y – y coordinate

• wait – Wait for convergence (if left at default, the lab wait property takes precedence)

• populate_interfaces – automatically create pre-defined number of interfaces on node creation

Returns:

a Node object

details() → dict[str, str | list | int]

Returns the lab details (including state) of the lab.

Returns:

The detailed lab state

download() → str

Download the lab from the server in YAML format.

Returns:

The lab in YAML format

find_nodes_by_tag(tag: str) → list[Node]

Returns the nodes identified by the given tag.

Parameters:

tag – tag of the nodes to be returned

Returns:

a list of nodes

get_interface_by_id(interface_id: str) → Interface

Returns the interface identified by the ID.

Parameters:

interface_id – ID of the interface to be returned

Returns:

An Interface object

Raises:

InterfaceNotFound – If interface not found

get_link_by_id(link_id: str) → Link

Returns the link identified by the ID.

Parameters:

link_id – ID of the interface to be returned

Returns:

A Link object

Raises:

LinkNotFound – If interface not found

static get_link_by_interfaces(iface1: Interface, iface2: Interface) → Link | None

DEPRECATED

Returns the link identified by two interface objects.

Deprecated. Use Interface.get_link_to to get link.

Parameters:

• iface1 – the first interface

• iface2 – the second interface

Returns:

the link between the interfaces

Raises:

LinkNotFound – If no such link exists

static get_link_by_nodes(node1: Node, node2: Node) → Link

DEPRECATED

Returns ONE of the links identified by two node objects.

Deprecated. Use Node.get_link_to to get one link or Node.get_links_to to get all links.

Parameters:

• node1 – the first node

• node2 – the second node

Returns:

one of links between the nodes

Raises:

LinkNotFound – If no such link exists

get_node_by_id(node_id: str) → Node

Returns the node identified by the ID.

Parameters:

node_id – ID of the node to be returned

Returns:

A Node object

Raises:

NodeNotFound – If node not found

get_node_by_label(label: str) → Node

Returns the node identified by the label.

Parameters:

label – label of the node to be returned

Returns:

A Node object

Raises:

NodeNotFound – If node not found

get_pyats_testbed(hostname: str | None = None) → str

Return lab’s pyATS YAML testbed. Example usage:

from pyats.topology import loader

lab = client_library.join_existing_lab("lab_1")
testbed_yaml = lab.get_pyats_testbed()

testbed = loader.load(io.StringIO(testbed_yaml))

# wait for lab to be ready
lab.wait_until_lab_converged()

aetest.main(testbed=testbed)

Parameters:

hostname – Force hostname/ip and port for console terminal server

Returns:

The pyATS testbed for the lab in YAML format

has_converged() → bool

import_lab(topology: dict) → None

interfaces() → list[Interface]

Returns the list of interfaces in the lab.

Returns:

A list of Interface objects

is_active() → bool

Returns whether the lab is started.

Returns:

Whether the lab is started

links() → list[Link]

Returns the list of links in the lab.

Returns:

A list of Link objects

need_to_wait(local_wait: bool | None) → bool

nodes() → list[Node]

Returns the list of nodes in the lab.

Returns:

A list of Node objects

remove() → None

Removes the lab from the server. The lab has to be stopped and wiped for this to work.

Use carefully, it removes current lab:

lab.remove()

remove_interface(iface: Interface, wait: bool | None = None) → None

Removes an interface from the lab.

Parameters:

• iface – the interface

• wait – Wait for convergence (if left at default, the lab wait property takes precedence)

remove_link(link: Link, wait: bool | None = None) → None

Removes a link from the lab.

Parameters:

• link – the link

• wait – Wait for convergence (if left at default, the lab wait property takes precedence)

remove_node(node: Node, wait: bool | None = None) → None

Removes a node from the lab.

Parameters:

• node – the node

• wait – Wait for convergence (if left at default, the lab wait property takes precedence)

remove_nodes(wait: bool | None = None) → None

Remove all nodes from the lab.

Parameters:

wait – Wait for convergence (if left at default, the lab wait property takes precedence)

start(wait: bool | None = None) → None

Start all the nodes and links in the lab.

Parameters:

wait (bool) – Wait for convergence (if left at default, the lab wait property takes precedence)

state() → str

Returns the state of the lab.

Returns:

The state as text

stop(wait: bool | None = None) → None

Stops all the nodes and links in the lab.

Parameters:

wait – Wait for convergence (if left at default, the lab wait property takes precedence)

sync(topology_only=True, with_node_configurations: bool | None = None, exclude_configurations=False) → None

Synchronize current lab applying the changes that were done in UI or in another ClientLibrary session:

lab.sync()

Parameters:

• topology_only – do not sync statistics and IP addresses

• with_node_configurations – DEPRECATED, does the opposite of what is expected: disables syncing node configuration when True

• exclude_configurations – disable syncing node configuration

sync_events() → None

sync_l3_addresses_if_outdated() → None

sync_layer3_addresses() → None

Syncs all layer 3 IP addresses from the backend server.

sync_operational() → None

sync_operational_if_outdated() → None

sync_pyats() → None

sync_states() → None

Sync all the states of the various elements with the back end server.

sync_states_if_outdated() → None

sync_statistics() → None

Retrieve the simulation statistic data from the back end server.

sync_statistics_if_outdated() → None

sync_topology_if_outdated() → None

update_lab(topology: dict, exclude_configurations: bool) → None

update_lab_groups(group_list: list[dict[str, str]]) → list[dict[str, str]]

Modifies lab / group association

Parameters:

group_list – list of objects consisting of group id and permission

Returns:

updated objects consisting of group id and permission

wait_until_lab_converged(max_iterations: int | None = None, wait_time: int | None = None) → None

Wait until lab converges.

wipe(wait: bool | None = None) → None

Wipe all the nodes and links in the lab.

Parameters:

wait – Wait for convergence (if left at default, the lab wait property takes precedence)

property description: str

Returns the description of the lab.

Returns:

The lab name

property groups: list[dict[str, str]]

Returns the groups this lab is associated with.

Returns:

associated groups

property id: str

Returns the ID of the lab (a 6 digit hex string).

Returns:

The Lab ID

property lab_base_url: str

property notes: str

Returns the notes of the lab.

Returns:

The lab name

property owner: str

Returns the owner of the lab.

Returns:

A username

property resource_pools: list[ResourcePool]

property session: httpx.Client

Returns the API client session object.

Returns:

The session object

property statistics: dict

Returns some statistics about the lab.

Returns:

A dictionary with stats of the lab

property title: str | None

Returns the title of the lab.

Returns:

The lab name

class virl2_client.models.Licensing(session: httpx.Client)

Bases: object

__init__(session: httpx.Client) → None

cancel_reservation() → bool

Cancel reservation request without completing it.

complete_reservation(authorization_code: str) → str

Complete reservation by installing authorization code from SSMS.

delete_reservation_confirmation_code() → bool

Remove the confirmation code.

delete_reservation_return_code() → bool

Remove the return code.

deregister() → int

Request deregistration from the current SSMS.

disable_reservation_mode() → None

Disable reservation mode in unregistered agent.

discard_reservation(data: str) → str

Discard a reservation authorization code for an already cancelled reservation request.

enable_reservation_mode() → None

Enable reservation mode in unregistered agent.

features() → list[dict[str, str | int]]

Get current licensing features.

get_certificate() → str | None

Get the currently installed licensing public certificate.

get_reservation_confirmation_code() → str

Get the confirmation code.

get_reservation_return_code() → str

Get the return code.

install_certificate(cert: str) → bool

Set up a licensing public certificate for internal deployment of an unregistered product instance.

register(token: str, reregister=False) → bool

Setup licensing registration.

register_renew() → bool

Request a renewal of licensing registration against current SSMS.

register_wait(token: str, reregister=False) → bool

Setup licensing registrations and wait for registration status to be COMPLETED and authorization status to be IN_COMPLIANCE.

release_reservation() → str

Return a completed reservation.

remove_certificate() → bool

Clear any licensing public certificate for internal deployment of an unregistered product instance.

renew_authorization() → bool

Renew licensing authorization with the backend.

request_reservation() → str

Initiate reservation by generating request code and message to the user.

reservation_mode(data: bool) → None

Enable or disable reservation mode in unregistered agent.

set_default_transport() → bool

Setup licensing transport configuration to default values.

set_product_license(product_license: str) → bool

Setup a product license.

set_transport(ssms: str, proxy_server: str | None = None, proxy_port: int | None = None) → bool

Setup licensing transport configuration.

status() → dict[str, Any]

Get current licensing configuration and status.

tech_support() → str

Get current licensing tech support.

update_features(features: dict[str, int] | list[dict[str, int]]) → None

Update licensing feature’s explicit count in reservation mode.

wait_for_status(what: str, *target_status: str) → None

Repeatedly check licensing registration or authorization status, until status matches one of the expected statuses or timeout is reached. :param what: “registration”, “authorization” or other status in licensing API. :param target_status: One or more expected statuses. :raises RuntimeError: When timeout is reached.

property base_url: str

max_wait = 30

wait_interval = 1.5

class virl2_client.models.Link(lab: Lab, lid: str, iface_a: Interface, iface_b: Interface, label: str | None = None)

Bases: object

__init__(lab: Lab, lid: str, iface_a: Interface, iface_b: Interface, label: str | None = None) → None

A VIRL2 network link between two nodes, connecting to two interfaces on these nodes.

Parameters:

• lab – the lab object

• lid – the link ID

• iface_a – the first interface of the link

• iface_b – the second interface of the link

• label – the link label

as_dict() → dict[str, str]

get_condition() → dict | None

Retrieves the current condition on this link. If there is no link condition specified, None is returned.

Returns:

the applied link condition

has_converged() → bool

remove_condition() → None

Removes link conditioning. If there’s no condition applied then this is a no-op for the controller.

remove_on_server() → None

set_condition(bandwidth: int, latency: int, jitter: int, loss: float) → None

Applies conditioning to this link.

Parameters:

• bandwidth (int) – desired bandwidth, 0-10000000 kbps

• latency (int) – desired latency, 0-10000 ms

• jitter (int) – desired jitter, 0-10000 ms

• loss (float) – desired loss, 0-100%

set_condition_by_name(name: str) → None

A convenience function to provide some commonly used link condition settings for various link types.

Inspired by: https://github.com/tylertreat/comcast

Name	Latency (ms)	Bandwidth	Loss (%)
gprs	500	50 kbps	2.0
edge	300	250 kbps	1.5
3g	250	750 kbps	1.5
dialup	185	40 kbps	2.0
dsl1	70	2 mbps	2.0
dsl2	40	8 mbps	0.5
wifi	10	30 mbps	0.1
wan1	80	256 kbps	0.2
wan2	80	100 mbps	0.2
satellite	1500	1 mbps	0.2

Parameters:

name – the predefined condition name as outlined in the table above

Raises:

ValueError – if the given name isn’t known

start(wait: bool | None = None) → None

stop(wait: bool | None = None) → None

wait_until_converged(max_iterations: int | None = None, wait_time: int | None = None) → None

property base_url: str

property interfaces: tuple[Interface, Interface]

property lab_base_url: str

property label: str | None

property node_a: Node

property node_b: Node

property nodes: tuple[Node, Node]

Return nodes this link connects.

property readbytes: int

property readpackets: int

property state: str | None

property writebytes: int

property writepackets: int

class virl2_client.models.Node(lab: Lab, nid: str, label: str, node_definition: str, image_definition: str | None, configuration: str | None, x: int, y: int, ram: int | None, cpus: int | None, cpu_limit: int | None, data_volume: int | None, boot_disk_size: int | None, hide_links: bool, tags: list[str], resource_pool: str | None)

Bases: object

__init__(lab: Lab, nid: str, label: str, node_definition: str, image_definition: str | None, configuration: str | None, x: int, y: int, ram: int | None, cpus: int | None, cpu_limit: int | None, data_volume: int | None, boot_disk_size: int | None, hide_links: bool, tags: list[str], resource_pool: str | None) → None

A VIRL2 Node object. Typically, a virtual machine representing a router, switch or server.

Parameters:

• lab – the Lab this node belongs to

• nid – the Node ID

• label – node label

• node_definition – The node definition of this node

• image_definition – The image definition of this node

• configuration – The initial configuration of this node

• x – X coordinate on topology canvas

• y – Y coordinate on topology canvas

• ram – memory of node in MiB (if applicable)

• cpus – Amount of CPUs in this node (if applicable)

• cpu_limit – CPU limit (default at 100%)

• data_volume – Size in GiB of 2nd HDD (if > 0)

• boot_disk_size – Size in GiB of boot disk (will expand to this size)

• hide_links – Whether node’s links should be hidden in UI visualization

• tags – List of tags

• resource_pool – A resource pool ID if the node is in a resource pool

add_tag(tag: str) → None

console_key() → str

console_logs(console_id: int, lines: int | None = None) → dict

create_interface(slot: int | None = None, wait: bool = False) → Interface

Create an interface in the specified slot or, if no slot is given, in the next available slot.

Parameters:

• slot – (optional)

• wait – Wait for the creation

Returns:

The newly created interface

Return type:

models.Interface

degree() → int

extract_configuration() → None

get_interface_by_label(label: str) → Interface

get_interface_by_slot(slot: int) → Interface

get_link_to(other_node: Node) → Link | None

Returns one link between this node and another.

Parameters:

other_node – the other node

Returns:

a link, if one exists

get_links_to(other_node: Node) → list[Link]

Returns all links between this node and another.

Parameters:

other_node – the other node

Returns:

a list of links

has_converged() → bool

interfaces() → list[Interface]

is_active() → bool

is_booted() → bool

links() → list[Link]

map_l3_addresses_to_interfaces(mapping: dict[str, dict[str, str]]) → None

next_available_interface() → Interface | None

Returns the next available physical interface on this node.

Note: On XR 9000v, the first two physical interfaces are marked as “do not use”… Only the third physical interface can be used to connect to other nodes!

Returns:

an interface or None, if all existing ones are connected

Return type:

models.Interface

peer_interfaces() → list[Interface]

peer_nodes() → list[Node]

physical_interfaces() → list[Interface]

remove_on_server() → None

remove_tag(tag: str) → None

run_pyats_command(command: str) → str

Run a pyATS command in exec mode.

Parameters:

command – the command (like “show version”)

Returns:

the output from the device

run_pyats_config_command(command: str) → str

Run a pyATS command in config mode.

Parameters:

command – the command (like “interface gi0”)

Returns:

the output from the device

start(wait=False) → None

stop(wait=False) → None

sync_layer3_addresses() → None

Acquire all L3 addresses from the controller. For this to work, the device has to be attached to the external network in bridge mode and must run DHCP to acquire an IP address.

sync_operational(response: dict[str, Any] = None)

tags() → list[str]

Returns the tags set on this node

update(node_data: dict[str, Any], exclude_configurations: bool, push_to_server: bool = False) → None

vnc_key() → str

wait_until_converged(max_iterations: int | None = None, wait_time: int | None = None) → None

wipe(wait=False) → None

property boot_disk_size: int

property compute_id

property config: str | None

property configuration: str | None

property cpu_limit: int

property cpu_usage: int | float

property cpus: int

property data_volume: int

property disk_read: int

property disk_write: int

property hide_links: bool

property image_definition: str | None

property lab_base_url: str

property label: str

property node_definition: str

property ram: int

property resource_pool: str

property state: str | None

property x: int

property y: int

class virl2_client.models.NodeImageDefinitions(session: httpx.Client)

Bases: object

__init__(session: httpx.Client) → None

VIRL2 Definition classes to specify a node VM and associated disk images.

Parameters:

session – httpx Client session

download_image_definition(definition_id: str) → str

Example:

client_library.definitions.download_image_definition("iosv-158-3")

Parameters:

definition_id – the image definition ID

Returns:

the image definition as YAML

download_image_file_list() → list[str]

download_node_definition(definition_id: str) → str

Returns the node definition for a given definition ID

Example:

client_library.definitions.download_node_definition("iosv")

Parameters:

definition_id – the node definition ID

Returns:

the node definition as YAML

image_definitions() → list[dict]

Returns all image definitions.

Returns:

list of image definitions

image_definitions_for_node_definition(definition_id: str) → list[dict]

Returns all image definitions for a given node definition

example:

client_library.definitions.image_definitions_for_node_definition("iosv")

Parameters:

definition_id – node definition id

Returns:

list of image definition objects

node_definitions() → list[dict]

Returns all node definitions.

Returns:

list of node definitions

remove_dropfolder_image(filename: str) → str

Returns:

“Success”

remove_image_definition(definition_id: str) → None

Removes the image definition with the given ID.

Example:

client_library.definitions.remove_image_definition("iosv-158-3-custom")

Parameters:

definition_id – the image definition ID to remove

remove_node_definition(definition_id: str) → None

Removes the node definition with the given ID.

Example:

client_library.definitions.remove_node_definition("iosv-custom")

Parameters:

definition_id – the definition ID to delete

upload_image_definition(body, json: bool = False) → str

Upload new image definition.

Parameters:

• body – image definition (yaml or json)

• json – whether we are sending json data

Type:

str or dict

Returns:

“Success”

upload_image_file(filename: str, rename: str | None = None, chunk_size_mb: int | None = None) → None

Parameters:

• filename – the path of the image to upload

• rename – Optional filename to rename to

• chunk_size_mb – Optional size of upload chunk (mb) (deprecated since 2.2.0)

upload_node_definition(body, json: bool = False) → str

Upload new node definition.

Parameters:

• body – node definition (yaml or json)

• json – whether we are sending json data

Type:

str or dict

Returns:

“Success”

property session: httpx.Client

Returns the used httpx client session object.

Returns:

The session object

class virl2_client.models.ResourcePoolManagement(session: httpx.Client, auto_sync=True, auto_sync_interval=1.0)

Bases: object

__init__(session: httpx.Client, auto_sync=True, auto_sync_interval=1.0)

Sync and modify resource pools.

Parameters:

• session – parent client’s httpx.Client object

• auto_sync – whether to automatically synchronize resource pools

• auto_sync_interval – how often to synchronize resource pools in seconds

create_resource_pool(label: str, **kwargs) → ResourcePool

Creates a resource pool with the given parameters.

Parameters:

• label – required pool label

• kwargs – other resource pool parameters as accepted by ResourcePool object

Returns:

resource pool object

create_resource_pools(label: str, users: list[str], **kwargs) → list[ResourcePool]

Creates a list of resource pools with the given parameters. If no template is supplied, a new template pool is created with the specified parameters, and each user is assigned a new pool with no additional limits. If a template pool is supplied, then parameters are applied to each user pool.

Parameters:

• label – pool label

• users – list of user IDs for which to create resource pools

• kwargs – other resource pool parameters as accepted by ResourcePool object

Returns:

resource pool objects, with template pool first if created

get_resource_pools_by_ids(resource_pool_ids: str | Iterable[str]) → ResourcePool | ResourcePools

Converts either one resource pool ID into a ResourcePool object or multiple IDs in an iterable into a dict of IDs and ResourcePool objects.

Parameters:

resource_pool_ids – a resource pool ID or an iterable of IDs

Returns:

a ResourcePool when one ID is passed or, when an iterable of pools is passed, a dictionary of IDs to `ResourcePool`s or `None`s when resource pool doesn’t exist

Raises:

KeyError – when one ID is passed and it doesn’t exist

sync_resource_pools() → None

sync_resource_pools_if_outdated() → None

property resource_pools: Dict[str, ResourcePool]

property session: httpx.Client

class virl2_client.models.SystemManagement(session: httpx.Client, auto_sync=True, auto_sync_interval=1.0)

Bases: object

__init__(session: httpx.Client, auto_sync=True, auto_sync_interval=1.0)

add_compute_host_local(compute_id: str, hostname: str, server_address: str, is_connector: bool, is_simulator: bool, is_connected: bool, is_synced: bool, admission_state: str, nodes: list[str] | None = None) → ComputeHost

get_external_connectors() → list[dict[str, str]]

Get the list of external connectors present on the controller

Device names are used as External Connector nodes’ configuration. Returns a list of objects with the device name and label.

get_mac_address_block() → int

Get mac address block.

Returns:

mac address block

get_web_session_timeout() → int

Get the web session timeout in seconds.

Returns:

web session timeout

set_mac_address_block(block: int) → str

Set mac address block.

Returns:

‘OK’

set_web_session_timeout(timeout: int) → str

Set the web session timeout in seconds.

Returns:

‘OK’

sync_compute_hosts() → None

sync_compute_hosts_if_outdated() → None

property compute_hosts: dict[str, ComputeHost]

property session: httpx.Client

class virl2_client.models.TokenAuth(client_library: ClientLibrary)

Bases: Auth

Initially inspired by: https://requests.readthedocs.io/en/v2.9.1/user/authentication/?highlight=AuthBase#new-forms-of-authentication and modified for httpx based on: https://www.python-httpx.org/advanced/#customizing-authentication

__init__(client_library: ClientLibrary)

auth_flow(request: Request) → Generator[Request, Response, None]

Execute the authentication flow.

To dispatch a request, yield it:

` yield request `

The client will .send() the response back into the flow generator. You can access it like so:

` response = yield request `

A return (or reaching the end of the generator) will result in the client returning the last response obtained from the server.

You can dispatch as many requests as is necessary.

logout(clear_all_sessions=False)

requires_response_body = True

property token: str | None

class virl2_client.models.UserManagement(session: httpx.Client)

Bases: object

__init__(session: httpx.Client) → None

create_user(username: str, pwd: str, fullname: str = '', description: str = '', admin: bool = False, groups: list[str] | None = None, resource_pool: str | None = None) → dict

Creates user.

Parameters:

• username – username

• pwd – desired password

• fullname – full name

• description – description

• admin – whether to create admin user

• groups – adds user to groups specified

• resource_pool – adds user to resource pool specified

Returns:

user object

delete_user(user_id: str) → None

Deletes user.

Parameters:

user_id – user uuid4

get_user(user_id: str) → dict

Gets user info.

Parameters:

user_id – user uuid4

Returns:

user object

update_user(user_id: str, fullname: str = '', description: str = '', groups: list[str] = None, admin: bool | None = None, password_dict: dict[str, str] = None, resource_pool: str | None = None) → dict

Updates user.

Parameters:

• user_id – user uuid4

• fullname – full name

• description – description

• admin – whether to create admin user

• groups – adds user to groups specified

• password_dict – dict containing old and new passwords

• resource_pool – adds user to resource pool specified

Returns:

user object

user_groups(user_id: str) → list[str]

Get the groups the user is member of.

Parameters:

user_id – user uuid4

user_id(username: str) → str

Get unique user uuid4.

Parameters:

username – user name

Returns:

user unique identifier

users() → list

Get the list of available users.

Returns:

list of user objects

property base_url: str

Submodules

virl2_client.models.authentication

class virl2_client.models.authentication.BlankAuth

Bases: Auth

auth_flow(request: Request) → Generator[Request, Response, None]

Execute the authentication flow.

To dispatch a request, yield it:

` yield request `

The client will .send() the response back into the flow generator. You can access it like so:

` response = yield request `

A return (or reaching the end of the generator) will result in the client returning the last response obtained from the server.

You can dispatch as many requests as is necessary.

class virl2_client.models.authentication.TokenAuth(client_library: ClientLibrary)

Bases: Auth

Initially inspired by: https://requests.readthedocs.io/en/v2.9.1/user/authentication/?highlight=AuthBase#new-forms-of-authentication and modified for httpx based on: https://www.python-httpx.org/advanced/#customizing-authentication

__init__(client_library: ClientLibrary)

auth_flow(request: Request) → Generator[Request, Response, None]

Execute the authentication flow.

To dispatch a request, yield it:

` yield request `

The client will .send() the response back into the flow generator. You can access it like so:

` response = yield request `

A return (or reaching the end of the generator) will result in the client returning the last response obtained from the server.

You can dispatch as many requests as is necessary.

logout(clear_all_sessions=False)

requires_response_body = True

property token: str | None

virl2_client.models.authentication.make_session(base_url: str, ssl_verify: bool = True) → Client

virl2_client.models.authentication.response_raise(response: Response) → None

Replaces the useless link to httpstatuses.com with error description.

virl2_client.models.cl_pyats

exception virl2_client.models.cl_pyats.PyatsDeviceNotFound

Bases: Exception

exception virl2_client.models.cl_pyats.PyatsNotInstalled

Bases: Exception

class virl2_client.models.cl_pyats.ClPyats(lab: Lab, hostname: str | None = None)

Bases: object

__init__(lab: Lab, hostname: str | None = None) → None

Creates a pyATS object that can be used to run commands against a device either in exec mode show version or in configuration mode interface gi0/0 \n no shut.

Parameters:

• lab – The lab which should be used with pyATS

• hostname – Force hostname/ip and port for console terminal server

Raises:

• PyatsNotInstalled – when pyATS can not be found

• PyatsDeviceNotFound – when the device can not be found

cleanup() → None

Cleans up the pyATS connections.

run_command(node_label: str, command: str) → str

Run a command on the device in exec mode.

Parameters:

• node_label – the label / title of the device

• command – the command to be run in exec mode

Returns:

The output from the device

run_config_command(node_label: str, command: str) → str

Run a command on the device in configure mode. pyATS handles the change into configure mode automatically.

Parameters:

• node_label – the label / title of the device

• command – the command to be run in exec mode

Returns:

The output from the device

sync_testbed(username: str, password: str) → None

Syncs the testbed from the server. Note that this will always fetch the latest topology data from the server.

Parameters:

• username – the username that will be inserted into the testbed data

• password – the password that will be inserted into the testbed data

virl2_client.models.groups

class virl2_client.models.groups.GroupManagement(session: httpx.Client)

Bases: object

__init__(session: httpx.Client) → None

create_group(name: str, description: str = '', members: list[str] | None = None, labs: list[dict[str, str]] | None = None) → dict

Creates a group.

Parameters:

• name – group name

• description – group description

• members – group members

• labs – group labs

Returns:

created group object

delete_group(group_id: str) → None

Deletes a group.

Parameters:

group_id – group uuid4

Returns:

None

get_group(group_id: str) → dict

Gets info for the specified group.

Parameters:

group_id – group uuid4

Returns:

group object

group_id(group_name: str) → str

Get unique user uuid4.

Parameters:

group_name – group name

Returns:

group unique identifier

group_labs(group_id: str) → list[str]

Get the list of labs that are associated with this group.

Parameters:

group_id – group uuid4

Returns:

list of labs associated with this group

group_members(group_id: str) → list[str]

Gets group members.

Parameters:

group_id – group uuid4

Returns:

list of users associated with this group

groups() → list

Get the list of available groups.

Returns:

list of group objects

update_group(group_id: str, name: str | None = None, description: str | None = None, members: list[str] | None = None, labs: list[dict[str, str]] | None = None) → dict

Updates a group.

Parameters:

• group_id – group uuid4

• name – new group name

• description – group description

• members – group members

• labs – group labs

Returns:

updated group object

property base_url: str

virl2_client.models.interface

class virl2_client.models.interface.Interface(iid: str, node: Node, label: str, slot: int | None, iface_type: str = 'physical')

Bases: object

__init__(iid: str, node: Node, label: str, slot: int | None, iface_type: str = 'physical') → None

A VIRL2 network interface, part of a node.

Parameters:

• iid – interface ID

• node – node object

• label – the label of the interface

• slot – the slot of the interface

• iface_type – the type of the interface, defaults to “physical”

as_dict() → dict[str, str]

bring_up() → None

degree()

get_link_to(other_interface: Interface) → Link | None

Returns the link between this interface and another.

Parameters:

other_interface – the other interface

Returns:

A Link

is_connected()

links()

peer_interfaces()

peer_nodes()

remove_on_server() → None

shutdown() → None

property connected: bool

Whether the interface is connected to a link.

property discovered_ipv4: str | None

property discovered_ipv6: str | None

property discovered_mac_address: str | None

property is_physical

property lab_base_url: str

property link: Link | None

Is link if connected, otherwise None.

property peer_interface: Interface | None

property peer_node: Node | None

property physical: bool

Whether the interface is physical.

property readbytes: int

property readpackets: int

property state: str | None

property writebytes: int

property writepackets: int

virl2_client.models.lab

class virl2_client.models.lab.Lab(title: str | None, lab_id: str, session: httpx.Client, username: str, password: str, auto_sync=True, auto_sync_interval=1.0, wait=True, wait_max_iterations=500, wait_time=5, hostname: str | None = None, resource_pool_manager: ResourcePoolManagement | None = None)

Bases: object

__init__(title: str | None, lab_id: str, session: httpx.Client, username: str, password: str, auto_sync=True, auto_sync_interval=1.0, wait=True, wait_max_iterations=500, wait_time=5, hostname: str | None = None, resource_pool_manager: ResourcePoolManagement | None = None) → None

A VIRL2 lab network topology. Contains nodes, links and interfaces.

Parameters:

• title – Name / title of the lab

• lab_id – A lab ID

• session – httpx Client session

• username – Username of the user to authenticate

• password – Password of the user to authenticate

• auto_sync – Should local changes sync to the server automatically

• auto_sync_interval – Interval to auto sync in seconds

• wait – Wait for convergence on backend

• wait_max_iterations – Maximum number of tries or calls for convergence

• wait_time – Time to sleep between calls for convergence on backend

• hostname – Force hostname/ip and port for pyATS console terminal server

• resource_pool_manager – a ResourcePoolManagement object shared with parent ClientLibrary

add_node_local(node_id: str, label: str, node_definition: str, image_definition: str | None, configuration: str | None, x: int, y: int, ram: int | None = None, cpus: int | None = None, cpu_limit: int | None = None, data_volume: int | None = None, boot_disk_size: int | None = None, hide_links: bool = False, tags: list | None = None, compute_id: str | None = None, resource_pool: ResourcePool | None = None) → Node

Helper function to add a node to the client library.

build_configurations() → None

Build basic configurations for all nodes in the lab which do not already have a configuration and also do support configuration building.

cleanup_pyats_connections() → None

Closes and cleans up connection that pyATS might still hold.

connect_two_nodes(node1: Node, node2: Node) → Link

Convenience method to connect two nodes within a lab.

Parameters:

• node1 – the first node object

• node2 – the second node object

Returns:

the created link

create_interface(node: Node, slot: int | None = None, wait: bool | None = None) → Interface

Create an interface in the next available slot, or, if a slot is specified, in all available slots up to and including the specified slot.

Parameters:

• node – The node on which the interface is created

• slot – (optional)

• wait – Wait for convergence (if left at default, the lab wait property takes precedence)

Returns:

The newly created interface

create_interface_local(iface_id: str, label: str, node: Node, slot: int | None, iface_type: str = 'physical') → Interface

Helper function to create an interface in the client library.

create_link(i1: Interface, i2: Interface, wait: bool | None = None) → Link

Creates a link between two interfaces

Parameters:

• i1 – the first interface object

• i2 – the second interface object

• wait – Wait for convergence (if left at default, the lab wait property takes precedence)

Returns:

the created link

create_link_local(i1: Interface, i2: Interface, link_id: str, label: str | None = None) → Link

Helper function to create a link in the client library.

create_node(label: str, node_definition: str, x: int = 0, y: int = 0, wait: bool | None = None, populate_interfaces: bool = False, **kwargs) → Node

Creates a node in the lab with the given parameters.

Parameters:

• label – Label

• node_definition – Node definition to use

• x – x coordinate

• y – y coordinate

• wait – Wait for convergence (if left at default, the lab wait property takes precedence)

• populate_interfaces – automatically create pre-defined number of interfaces on node creation

Returns:

a Node object

details() → dict[str, str | list | int]

Returns the lab details (including state) of the lab.

Returns:

The detailed lab state

download() → str

Download the lab from the server in YAML format.

Returns:

The lab in YAML format

find_nodes_by_tag(tag: str) → list[Node]

Returns the nodes identified by the given tag.

Parameters:

tag – tag of the nodes to be returned

Returns:

a list of nodes

get_interface_by_id(interface_id: str) → Interface

Returns the interface identified by the ID.

Parameters:

interface_id – ID of the interface to be returned

Returns:

An Interface object

Raises:

InterfaceNotFound – If interface not found

get_link_by_id(link_id: str) → Link

Returns the link identified by the ID.

Parameters:

link_id – ID of the interface to be returned

Returns:

A Link object

Raises:

LinkNotFound – If interface not found

static get_link_by_interfaces(iface1: Interface, iface2: Interface) → Link | None

DEPRECATED

Returns the link identified by two interface objects.

Deprecated. Use Interface.get_link_to to get link.

Parameters:

• iface1 – the first interface

• iface2 – the second interface

Returns:

the link between the interfaces

Raises:

LinkNotFound – If no such link exists

static get_link_by_nodes(node1: Node, node2: Node) → Link

DEPRECATED

Returns ONE of the links identified by two node objects.

Deprecated. Use Node.get_link_to to get one link or Node.get_links_to to get all links.

Parameters:

• node1 – the first node

• node2 – the second node

Returns:

one of links between the nodes

Raises:

LinkNotFound – If no such link exists

get_node_by_id(node_id: str) → Node

Returns the node identified by the ID.

Parameters:

node_id – ID of the node to be returned

Returns:

A Node object

Raises:

NodeNotFound – If node not found

get_node_by_label(label: str) → Node

Returns the node identified by the label.

Parameters:

label – label of the node to be returned

Returns:

A Node object

Raises:

NodeNotFound – If node not found

get_pyats_testbed(hostname: str | None = None) → str

Return lab’s pyATS YAML testbed. Example usage:

from pyats.topology import loader

lab = client_library.join_existing_lab("lab_1")
testbed_yaml = lab.get_pyats_testbed()

testbed = loader.load(io.StringIO(testbed_yaml))

# wait for lab to be ready
lab.wait_until_lab_converged()

aetest.main(testbed=testbed)

Parameters:

hostname – Force hostname/ip and port for console terminal server

Returns:

The pyATS testbed for the lab in YAML format

has_converged() → bool

import_lab(topology: dict) → None

interfaces() → list[Interface]

Returns the list of interfaces in the lab.

Returns:

A list of Interface objects

is_active() → bool

Returns whether the lab is started.

Returns:

Whether the lab is started

links() → list[Link]

Returns the list of links in the lab.

Returns:

A list of Link objects

need_to_wait(local_wait: bool | None) → bool

nodes() → list[Node]

Returns the list of nodes in the lab.

Returns:

A list of Node objects

remove() → None

Removes the lab from the server. The lab has to be stopped and wiped for this to work.

Use carefully, it removes current lab:

lab.remove()

remove_interface(iface: Interface, wait: bool | None = None) → None

Removes an interface from the lab.

Parameters:

• iface – the interface

• wait – Wait for convergence (if left at default, the lab wait property takes precedence)

remove_link(link: Link, wait: bool | None = None) → None

Removes a link from the lab.

Parameters:

• link – the link

• wait – Wait for convergence (if left at default, the lab wait property takes precedence)

remove_node(node: Node, wait: bool | None = None) → None

Removes a node from the lab.

Parameters:

• node – the node

• wait – Wait for convergence (if left at default, the lab wait property takes precedence)

remove_nodes(wait: bool | None = None) → None

Remove all nodes from the lab.

Parameters:

wait – Wait for convergence (if left at default, the lab wait property takes precedence)

start(wait: bool | None = None) → None

Start all the nodes and links in the lab.

Parameters:

wait (bool) – Wait for convergence (if left at default, the lab wait property takes precedence)

state() → str

Returns the state of the lab.

Returns:

The state as text

stop(wait: bool | None = None) → None

Stops all the nodes and links in the lab.

Parameters:

wait – Wait for convergence (if left at default, the lab wait property takes precedence)

sync(topology_only=True, with_node_configurations: bool | None = None, exclude_configurations=False) → None

Synchronize current lab applying the changes that were done in UI or in another ClientLibrary session:

lab.sync()

Parameters:

• topology_only – do not sync statistics and IP addresses

• with_node_configurations – DEPRECATED, does the opposite of what is expected: disables syncing node configuration when True

• exclude_configurations – disable syncing node configuration

sync_events() → None

sync_l3_addresses_if_outdated() → None

sync_layer3_addresses() → None

Syncs all layer 3 IP addresses from the backend server.

sync_operational() → None

sync_operational_if_outdated() → None

sync_pyats() → None

sync_states() → None

Sync all the states of the various elements with the back end server.

sync_states_if_outdated() → None

sync_statistics() → None

Retrieve the simulation statistic data from the back end server.

sync_statistics_if_outdated() → None

sync_topology_if_outdated() → None

update_lab(topology: dict, exclude_configurations: bool) → None

update_lab_groups(group_list: list[dict[str, str]]) → list[dict[str, str]]

Modifies lab / group association

Parameters:

group_list – list of objects consisting of group id and permission

Returns:

updated objects consisting of group id and permission

wait_until_lab_converged(max_iterations: int | None = None, wait_time: int | None = None) → None

Wait until lab converges.

wipe(wait: bool | None = None) → None

Wipe all the nodes and links in the lab.

Parameters:

wait – Wait for convergence (if left at default, the lab wait property takes precedence)

property description: str

Returns the description of the lab.

Returns:

The lab name

property groups: list[dict[str, str]]

Returns the groups this lab is associated with.

Returns:

associated groups

property id: str

Returns the ID of the lab (a 6 digit hex string).

Returns:

The Lab ID

property lab_base_url: str

property notes: str

Returns the notes of the lab.

Returns:

The lab name

property owner: str

Returns the owner of the lab.

Returns:

A username

property resource_pools: list[ResourcePool]

property session: httpx.Client

Returns the API client session object.

Returns:

The session object

property statistics: dict

Returns some statistics about the lab.

Returns:

A dictionary with stats of the lab

property title: str | None

Returns the title of the lab.

Returns:

The lab name

virl2_client.models.licensing

class virl2_client.models.licensing.Licensing(session: httpx.Client)

Bases: object

__init__(session: httpx.Client) → None

cancel_reservation() → bool

Cancel reservation request without completing it.

complete_reservation(authorization_code: str) → str

Complete reservation by installing authorization code from SSMS.

delete_reservation_confirmation_code() → bool

Remove the confirmation code.

delete_reservation_return_code() → bool

Remove the return code.

deregister() → int

Request deregistration from the current SSMS.

disable_reservation_mode() → None

Disable reservation mode in unregistered agent.

discard_reservation(data: str) → str

Discard a reservation authorization code for an already cancelled reservation request.

enable_reservation_mode() → None

Enable reservation mode in unregistered agent.

features() → list[dict[str, str | int]]

Get current licensing features.

get_certificate() → str | None

Get the currently installed licensing public certificate.

get_reservation_confirmation_code() → str

Get the confirmation code.

get_reservation_return_code() → str

Get the return code.

install_certificate(cert: str) → bool

Set up a licensing public certificate for internal deployment of an unregistered product instance.

register(token: str, reregister=False) → bool

Setup licensing registration.

register_renew() → bool

Request a renewal of licensing registration against current SSMS.

register_wait(token: str, reregister=False) → bool

Setup licensing registrations and wait for registration status to be COMPLETED and authorization status to be IN_COMPLIANCE.

release_reservation() → str

Return a completed reservation.

remove_certificate() → bool

Clear any licensing public certificate for internal deployment of an unregistered product instance.

renew_authorization() → bool

Renew licensing authorization with the backend.

request_reservation() → str

Initiate reservation by generating request code and message to the user.

reservation_mode(data: bool) → None

Enable or disable reservation mode in unregistered agent.

set_default_transport() → bool

Setup licensing transport configuration to default values.

set_product_license(product_license: str) → bool

Setup a product license.

set_transport(ssms: str, proxy_server: str | None = None, proxy_port: int | None = None) → bool

Setup licensing transport configuration.

status() → dict[str, Any]

Get current licensing configuration and status.

tech_support() → str

Get current licensing tech support.

update_features(features: dict[str, int] | list[dict[str, int]]) → None

Update licensing feature’s explicit count in reservation mode.

wait_for_status(what: str, *target_status: str) → None

Repeatedly check licensing registration or authorization status, until status matches one of the expected statuses or timeout is reached. :param what: “registration”, “authorization” or other status in licensing API. :param target_status: One or more expected statuses. :raises RuntimeError: When timeout is reached.

property base_url: str

max_wait = 30

wait_interval = 1.5

virl2_client.models.link

class virl2_client.models.link.Link(lab: Lab, lid: str, iface_a: Interface, iface_b: Interface, label: str | None = None)

Bases: object

__init__(lab: Lab, lid: str, iface_a: Interface, iface_b: Interface, label: str | None = None) → None

A VIRL2 network link between two nodes, connecting to two interfaces on these nodes.

Parameters:

• lab – the lab object

• lid – the link ID

• iface_a – the first interface of the link

• iface_b – the second interface of the link

• label – the link label

as_dict() → dict[str, str]

get_condition() → dict | None

Retrieves the current condition on this link. If there is no link condition specified, None is returned.

Returns:

the applied link condition

has_converged() → bool

remove_condition() → None

Removes link conditioning. If there’s no condition applied then this is a no-op for the controller.

remove_on_server() → None

set_condition(bandwidth: int, latency: int, jitter: int, loss: float) → None

Applies conditioning to this link.

Parameters:

• bandwidth (int) – desired bandwidth, 0-10000000 kbps

• latency (int) – desired latency, 0-10000 ms

• jitter (int) – desired jitter, 0-10000 ms

• loss (float) – desired loss, 0-100%

set_condition_by_name(name: str) → None

A convenience function to provide some commonly used link condition settings for various link types.

Inspired by: https://github.com/tylertreat/comcast

Name	Latency (ms)	Bandwidth	Loss (%)
gprs	500	50 kbps	2.0
edge	300	250 kbps	1.5
3g	250	750 kbps	1.5
dialup	185	40 kbps	2.0
dsl1	70	2 mbps	2.0
dsl2	40	8 mbps	0.5
wifi	10	30 mbps	0.1
wan1	80	256 kbps	0.2
wan2	80	100 mbps	0.2
satellite	1500	1 mbps	0.2

Parameters:

name – the predefined condition name as outlined in the table above

Raises:

ValueError – if the given name isn’t known

start(wait: bool | None = None) → None

stop(wait: bool | None = None) → None

wait_until_converged(max_iterations: int | None = None, wait_time: int | None = None) → None

property base_url: str

property interfaces: tuple[Interface, Interface]

property lab_base_url: str

property label: str | None

property node_a: Node

property node_b: Node

property nodes: tuple[Node, Node]

Return nodes this link connects.

property readbytes: int

property readpackets: int

property state: str | None

property writebytes: int

property writepackets: int

virl2_client.models.node

class virl2_client.models.node.Node(lab: Lab, nid: str, label: str, node_definition: str, image_definition: str | None, configuration: str | None, x: int, y: int, ram: int | None, cpus: int | None, cpu_limit: int | None, data_volume: int | None, boot_disk_size: int | None, hide_links: bool, tags: list[str], resource_pool: str | None)

Bases: object

__init__(lab: Lab, nid: str, label: str, node_definition: str, image_definition: str | None, configuration: str | None, x: int, y: int, ram: int | None, cpus: int | None, cpu_limit: int | None, data_volume: int | None, boot_disk_size: int | None, hide_links: bool, tags: list[str], resource_pool: str | None) → None

A VIRL2 Node object. Typically, a virtual machine representing a router, switch or server.

Parameters:

• lab – the Lab this node belongs to

• nid – the Node ID

• label – node label

• node_definition – The node definition of this node

• image_definition – The image definition of this node

• configuration – The initial configuration of this node

• x – X coordinate on topology canvas

• y – Y coordinate on topology canvas

• ram – memory of node in MiB (if applicable)

• cpus – Amount of CPUs in this node (if applicable)

• cpu_limit – CPU limit (default at 100%)

• data_volume – Size in GiB of 2nd HDD (if > 0)

• boot_disk_size – Size in GiB of boot disk (will expand to this size)

• hide_links – Whether node’s links should be hidden in UI visualization

• tags – List of tags

• resource_pool – A resource pool ID if the node is in a resource pool

add_tag(tag: str) → None

console_key() → str

console_logs(console_id: int, lines: int | None = None) → dict

create_interface(slot: int | None = None, wait: bool = False) → Interface

Create an interface in the specified slot or, if no slot is given, in the next available slot.

Parameters:

• slot – (optional)

• wait – Wait for the creation

Returns:

The newly created interface

Return type:

models.Interface

degree() → int

extract_configuration() → None

get_interface_by_label(label: str) → Interface

get_interface_by_slot(slot: int) → Interface

get_link_to(other_node: Node) → Link | None

Returns one link between this node and another.

Parameters:

other_node – the other node

Returns:

a link, if one exists

get_links_to(other_node: Node) → list[Link]

Returns all links between this node and another.

Parameters:

other_node – the other node

Returns:

a list of links

has_converged() → bool

interfaces() → list[Interface]

is_active() → bool

is_booted() → bool

links() → list[Link]

map_l3_addresses_to_interfaces(mapping: dict[str, dict[str, str]]) → None

next_available_interface() → Interface | None

Returns the next available physical interface on this node.

Note: On XR 9000v, the first two physical interfaces are marked as “do not use”… Only the third physical interface can be used to connect to other nodes!

Returns:

an interface or None, if all existing ones are connected

Return type:

models.Interface

peer_interfaces() → list[Interface]

peer_nodes() → list[Node]

physical_interfaces() → list[Interface]

remove_on_server() → None

remove_tag(tag: str) → None

run_pyats_command(command: str) → str

Run a pyATS command in exec mode.

Parameters:

command – the command (like “show version”)

Returns:

the output from the device

run_pyats_config_command(command: str) → str

Run a pyATS command in config mode.

Parameters:

command – the command (like “interface gi0”)

Returns:

the output from the device

start(wait=False) → None

stop(wait=False) → None

sync_layer3_addresses() → None

Acquire all L3 addresses from the controller. For this to work, the device has to be attached to the external network in bridge mode and must run DHCP to acquire an IP address.

sync_operational(response: dict[str, Any] = None)

tags() → list[str]

Returns the tags set on this node

update(node_data: dict[str, Any], exclude_configurations: bool, push_to_server: bool = False) → None

vnc_key() → str

wait_until_converged(max_iterations: int | None = None, wait_time: int | None = None) → None

wipe(wait=False) → None

property boot_disk_size: int

property compute_id

property config: str | None

property configuration: str | None

property cpu_limit: int

property cpu_usage: int | float

property cpus: int

property data_volume: int

property disk_read: int

property disk_write: int

property hide_links: bool

property image_definition: str | None

property lab_base_url: str

property label: str

property node_definition: str

property ram: int

property resource_pool: str

property state: str | None

property x: int

property y: int

virl2_client.models.node_image_definitions

class virl2_client.models.node_image_definitions.NodeImageDefinitions(session: httpx.Client)

Bases: object

__init__(session: httpx.Client) → None

VIRL2 Definition classes to specify a node VM and associated disk images.

Parameters:

session – httpx Client session

download_image_definition(definition_id: str) → str

Example:

client_library.definitions.download_image_definition("iosv-158-3")

Parameters:

definition_id – the image definition ID

Returns:

the image definition as YAML

download_image_file_list() → list[str]

download_node_definition(definition_id: str) → str

Returns the node definition for a given definition ID

Example:

client_library.definitions.download_node_definition("iosv")

Parameters:

definition_id – the node definition ID

Returns:

the node definition as YAML

image_definitions() → list[dict]

Returns all image definitions.

Returns:

list of image definitions

image_definitions_for_node_definition(definition_id: str) → list[dict]

Returns all image definitions for a given node definition

example:

client_library.definitions.image_definitions_for_node_definition("iosv")

Parameters:

definition_id – node definition id

Returns:

list of image definition objects

node_definitions() → list[dict]

Returns all node definitions.

Returns:

list of node definitions

remove_dropfolder_image(filename: str) → str

Returns:

“Success”

remove_image_definition(definition_id: str) → None

Removes the image definition with the given ID.

Example:

client_library.definitions.remove_image_definition("iosv-158-3-custom")

Parameters:

definition_id – the image definition ID to remove

remove_node_definition(definition_id: str) → None

Removes the node definition with the given ID.

Example:

client_library.definitions.remove_node_definition("iosv-custom")

Parameters:

definition_id – the definition ID to delete

upload_image_definition(body, json: bool = False) → str

Upload new image definition.

Parameters:

• body – image definition (yaml or json)

• json – whether we are sending json data

Type:

str or dict

Returns:

“Success”

upload_image_file(filename: str, rename: str | None = None, chunk_size_mb: int | None = None) → None

Parameters:

• filename – the path of the image to upload

• rename – Optional filename to rename to

• chunk_size_mb – Optional size of upload chunk (mb) (deprecated since 2.2.0)

upload_node_definition(body, json: bool = False) → str

Upload new node definition.

Parameters:

• body – node definition (yaml or json)

• json – whether we are sending json data

Type:

str or dict

Returns:

“Success”

property session: httpx.Client

Returns the used httpx client session object.

Returns:

The session object

virl2_client.models.node_image_definitions.print_progress_bar(cur: int, total: int, start_time: float, length=50) → None

virl2_client.models.resource_pools

class virl2_client.models.resource_pools.ResourcePool(resource_pools: ResourcePoolManagement, pool_id: str, label: str, description: str | None, template: str | None, licenses: int | None, ram: int | None, cpus: int | None, disk_space: int | None, external_connectors: list[str] | None, users: list[str] | None, user_pools: list[str] | None)

Bases: object

__init__(resource_pools: ResourcePoolManagement, pool_id: str, label: str, description: str | None, template: str | None, licenses: int | None, ram: int | None, cpus: int | None, disk_space: int | None, external_connectors: list[str] | None, users: list[str] | None, user_pools: list[str] | None)

get_usage() → ResourcePoolUsage

remove() → None

update(pool_data: dict[str, Any], push_to_server: bool = False)

property cpus: int

property description: str

property disk_space: int

property external_connectors: list[str] | None

property id: str

property is_template: bool

property label: str

property licenses: int

property ram: int

property template: str

property user_pools: list[str]

property users: list[str]

class virl2_client.models.resource_pools.ResourcePoolManagement(session: httpx.Client, auto_sync=True, auto_sync_interval=1.0)

Bases: object

__init__(session: httpx.Client, auto_sync=True, auto_sync_interval=1.0)

Sync and modify resource pools.

Parameters:

• session – parent client’s httpx.Client object

• auto_sync – whether to automatically synchronize resource pools

• auto_sync_interval – how often to synchronize resource pools in seconds

create_resource_pool(label: str, **kwargs) → ResourcePool

Creates a resource pool with the given parameters.

Parameters:

• label – required pool label

• kwargs – other resource pool parameters as accepted by ResourcePool object

Returns:

resource pool object

create_resource_pools(label: str, users: list[str], **kwargs) → list[ResourcePool]

Creates a list of resource pools with the given parameters. If no template is supplied, a new template pool is created with the specified parameters, and each user is assigned a new pool with no additional limits. If a template pool is supplied, then parameters are applied to each user pool.

Parameters:

• label – pool label

• users – list of user IDs for which to create resource pools

• kwargs – other resource pool parameters as accepted by ResourcePool object

Returns:

resource pool objects, with template pool first if created

get_resource_pools_by_ids(resource_pool_ids: str | Iterable[str]) → ResourcePool | ResourcePools

Converts either one resource pool ID into a ResourcePool object or multiple IDs in an iterable into a dict of IDs and ResourcePool objects.

Parameters:

resource_pool_ids – a resource pool ID or an iterable of IDs

Returns:

a ResourcePool when one ID is passed or, when an iterable of pools is passed, a dictionary of IDs to `ResourcePool`s or `None`s when resource pool doesn’t exist

Raises:

KeyError – when one ID is passed and it doesn’t exist

sync_resource_pools() → None

sync_resource_pools_if_outdated() → None

property resource_pools: Dict[str, ResourcePool]

property session: httpx.Client

class virl2_client.models.resource_pools.ResourcePoolUsage(limit: 'ResourcePoolUsageBase', usage: 'ResourcePoolUsageBase')

Bases: object

__init__(limit: ResourcePoolUsageBase, usage: ResourcePoolUsageBase) → None

limit: ResourcePoolUsageBase

usage: ResourcePoolUsageBase

class virl2_client.models.resource_pools.ResourcePoolUsageBase(licenses: 'int', cpus: 'int', ram: 'int', disk_space: 'int', external_connectors: 'list')

Bases: object

__init__(licenses: int, cpus: int, ram: int, disk_space: int, external_connectors: list) → None

cpus: int

disk_space: int

external_connectors: list

licenses: int

ram: int

virl2_client.models.system

class virl2_client.models.system.ComputeHost(system: SystemManagement, compute_id: str, hostname: str, server_address: str, is_connector: bool, is_simulator: bool, is_connected: bool, is_synced: bool, admission_state: str, nodes: list[str] | None = None)

Bases: object

__init__(system: SystemManagement, compute_id: str, hostname: str, server_address: str, is_connector: bool, is_simulator: bool, is_connected: bool, is_synced: bool, admission_state: str, nodes: list[str] | None = None)

remove() → None

update(host_data: dict[str, Any], push_to_server: bool = False)

property admission_state: str

property compute_id: str

property hostname: str

property is_connected: bool

property is_connector: bool

property is_simulator: bool

property is_synced: bool

property nodes: list[str]

property server_address: str

class virl2_client.models.system.SystemManagement(session: httpx.Client, auto_sync=True, auto_sync_interval=1.0)

Bases: object

__init__(session: httpx.Client, auto_sync=True, auto_sync_interval=1.0)

add_compute_host_local(compute_id: str, hostname: str, server_address: str, is_connector: bool, is_simulator: bool, is_connected: bool, is_synced: bool, admission_state: str, nodes: list[str] | None = None) → ComputeHost

get_external_connectors() → list[dict[str, str]]

Get the list of external connectors present on the controller

Device names are used as External Connector nodes’ configuration. Returns a list of objects with the device name and label.

get_mac_address_block() → int

Get mac address block.

Returns:

mac address block

get_web_session_timeout() → int

Get the web session timeout in seconds.

Returns:

web session timeout

set_mac_address_block(block: int) → str

Set mac address block.

Returns:

‘OK’

set_web_session_timeout(timeout: int) → str

Set the web session timeout in seconds.

Returns:

‘OK’

sync_compute_hosts() → None

sync_compute_hosts_if_outdated() → None

property compute_hosts: dict[str, ComputeHost]

property session: httpx.Client

virl2_client.models.users

class virl2_client.models.users.UserManagement(session: httpx.Client)

Bases: object

__init__(session: httpx.Client) → None

create_user(username: str, pwd: str, fullname: str = '', description: str = '', admin: bool = False, groups: list[str] | None = None, resource_pool: str | None = None) → dict

Creates user.

Parameters:

• username – username

• pwd – desired password

• fullname – full name

• description – description

• admin – whether to create admin user

• groups – adds user to groups specified

• resource_pool – adds user to resource pool specified

Returns:

user object

delete_user(user_id: str) → None

Deletes user.

Parameters:

user_id – user uuid4

get_user(user_id: str) → dict

Gets user info.

Parameters:

user_id – user uuid4

Returns:

user object

update_user(user_id: str, fullname: str = '', description: str = '', groups: list[str] = None, admin: bool | None = None, password_dict: dict[str, str] = None, resource_pool: str | None = None) → dict

Updates user.

Parameters:

• user_id – user uuid4

• fullname – full name

• description – description

• admin – whether to create admin user

• groups – adds user to groups specified

• password_dict – dict containing old and new passwords

• resource_pool – adds user to resource pool specified

Returns:

user object

user_groups(user_id: str) → list[str]

Get the groups the user is member of.

Parameters:

user_id – user uuid4

user_id(username: str) → str

Get unique user uuid4.

Parameters:

username – user name

Returns:

user unique identifier

users() → list

Get the list of available users.

Returns:

list of user objects

property base_url: str