CMLで「リンク一覧表示」「LinkDown/Up」するPythonスクリプト

import argparse
import logging

from rich import box
from rich.console import Console
from rich.table import Table
from virl2_client import ClientLibrary

URL = "https://10.0.0.1"
USER = "admin"
PASS = "pass"


def parse_args():
    parser = argparse.ArgumentParser(
        description="Display CML links or bring selected links up/down."
    )
    parser.add_argument(
        "-w",
        "--workspace",
        help="Display links only in this CML workspace (Workspace ID).",
    )
    parser.add_argument(
        "--link-id",
        nargs="+",
        metavar="LINK_ID",
        help="One or more Link IDs to operate on.",
    )
    action = parser.add_mutually_exclusive_group()
    action.add_argument(
        "--down", action="store_true", help="Bring selected links down."
    )
    action.add_argument("--up", action="store_true", help="Bring selected links up.")
    args = parser.parse_args()
    if (args.down or args.up) and not args.link_id:
        parser.error("--link-id is required with --down or --up")
    if args.link_id and not (args.down or args.up):
        parser.error("--link-id can only be used with --down or --up")
    return args


def print_links(labs, workspace=None):
    """Print links, optionally limited to one workspace."""
    title = "CML All Links" if workspace is None else f"CML Links: {workspace}"
    table = Table(title=title, box=box.MINIMAL_DOUBLE_HEAD)
    for column in (
        "Lab",
        "Link ID",
        "Node-A",
        "Node-B",
        "State",
        "Bandwidth",
        "Latency",
        "Loss",
        "Jitter",
    ):
        table.add_column(column)

    links = [
        (lab, link)
        for lab in labs
        if workspace is None or lab.id == workspace
        for link in lab.links()
    ]
    links.sort(
        key=lambda item: (
            f"{item[1].node_a.label} {item[1].interface_a.label}".casefold(),
            f"{item[1].node_b.label} {item[1].interface_b.label}".casefold(),
            item[0].title.casefold(),
            item[1].id.casefold(),
        )
    )

    synced_labs = set()

    for lab, link in links:
        if lab.id not in synced_labs:
            lab.sync_states()
            synced_labs.add(lab.id)
        is_up = str(link.state).upper() == "STARTED"
        link_state = "Up" if is_up else "-"
        condition = link.get_condition()
        values = [
            str(condition.get("bandwidth", "")),
            str(condition.get("latency", "")),
            str(condition.get("loss", "")),
            str(condition.get("jitter", "")),
        ]
        table.add_row(
            lab.title,
            link.id,
            f"{link.node_a.label} {link.interface_a.label}",
            f"{link.node_b.label} {link.interface_b.label}",
            link_state,
            *values,
            style="white" if is_up else "grey50",
        )
    Console().print(table)


def change_links(labs, link_ids, action, workspace=None):
    """Start or stop configured links in the selected workspace."""
    links_by_id = {
        link.id: link
        for lab in labs
        if workspace is None or lab.id == workspace
        for link in lab.links()
    }
    missing = [link_id for link_id in link_ids if link_id not in links_by_id]
    if missing:
        scope = f" in workspace {workspace!r}" if workspace else ""
        raise SystemExit("Link IDs not found" + scope + ": " + ", ".join(missing))

    for link_id in link_ids:
        if action == "down":
            links_by_id[link_id].stop()
        else:
            links_by_id[link_id].start()


def main():
    args = parse_args()
    logging.getLogger("virl2_client.virl2_client").addFilter(
        _SuppressSSLVerificationWarning()
    )
    cml = ClientLibrary(URL, USER, PASS, ssl_verify=False)
    cml.is_system_ready(wait=True)
    labs = cml.all_labs()

    action = "down" if args.down else "up" if args.up else None
    if action is not None:
        change_links(labs, args.link_id, action, args.workspace)
        labs = cml.all_labs()

    print_links(labs, args.workspace)


class _SuppressSSLVerificationWarning(logging.Filter):
    """Suppress only the warning emitted for the intentionally disabled SSL check."""

    def filter(self, record):
        return record.getMessage() != "SSL Verification disabled"


if __name__ == "__main__":
    main()