Creating Metadata Plugin (#558)

Author: amisiorek-akamai

linodecli/plugins/metadata.py

@@ -0,0 +1,219 @@
+"""
+This plugin allows users to access the metadata service while in a Linode.
+
+Usage:
+
+   linode-cli metadata [ENDPOINT]
+"""
+
+import argparse
+import sys
+
+from linode_metadata import MetadataClient
+from linode_metadata.objects.error import ApiError
+from linode_metadata.objects.instance import ResponseBase
+from requests import ConnectTimeout
+from rich import print as rprint
+from rich.table import Table
+
+PLUGIN_BASE = "linode-cli metadata"
+
+
+def process_sub_columns(subcolumn: ResponseBase, table: Table, values_row):
+    """
+    Helper method to process embedded ResponseBase objects
+    """
+    for key, value in vars(subcolumn).items():
+        if isinstance(value, ResponseBase):
+            process_sub_columns(value, table, values_row)
+        else:
+            table.add_column(key)
+            values_row.append(str(value))
+
+
+def print_instance_table(data):
+    """
+    Prints the table that contains information about the current instance
+    """
+    attributes = vars(data)
+    values_row = []
+
+    table = Table()
+
+    for key, value in attributes.items():
+        if isinstance(value, ResponseBase):
+            process_sub_columns(value, table, values_row)
+        else:
+            table.add_column(key)
+            values_row.append(str(value))
+
+    table.add_row(*values_row)
+    rprint(table)
+
+
+def print_ssh_keys_table(data):
+    """
+    Prints the table that contains information about the SSH keys for the current instance
+    """
+    table = Table(show_lines=True)
+
+    table.add_column("ssh keys")
+
+    if data.users.root is not None:
+        for key in data.users.root:
+            table.add_row(key)
+
+    rprint(table)
+
+
+def print_networking_tables(data):
+    """
+    Prints the table that contains information about the network of the current instance
+    """
+    interfaces = Table(title="Interfaces", show_lines=True)
+
+    interfaces.add_column("label")
+    interfaces.add_column("purpose")
+    interfaces.add_column("ipam addresses")
+
+    for interface in data.interfaces:
+        attributes = vars(interface)
+        interface_row = []
+        for _, value in attributes.items():
+            interface_row.append(str(value))
+        interfaces.add_row(*interface_row)
+
+    ipv4 = Table(title="IPv4")
+    ipv4.add_column("ip address")
+    ipv4.add_column("type")
+    attributes = vars(data.ipv4)
+    for key, value in attributes.items():
+        for address in value:
+            ipv4.add_row(*[address, key])
+
+    ipv6 = Table(title="IPv6")
+    ipv6_data = data.ipv6
+    ipv6.add_column("slaac")
+    ipv6.add_column("link local")
+    ipv6.add_column("ranges")
+    ipv6.add_column("shared ranges")
+    ipv6.add_row(
+        *[
+            ipv6_data.slaac,
+            ipv6_data.link_local,
+            str(ipv6_data.ranges),
+            str(ipv6_data.shared_ranges),
+        ]
+    )
+
+    rprint(interfaces)
+    rprint(ipv4)
+    rprint(ipv6)
+
+
+def get_instance(client: MetadataClient):
+    """
+    Get information about your instance, including plan resources
+    """
+    data = client.get_instance()
+    print_instance_table(data)
+
+
+def get_user_data(client: MetadataClient):
+    """
+    Get your user data
+    """
+    data = client.get_user_data()
+    rprint(data)
+
+
+def get_network(client: MetadataClient):
+    """
+    Get information about your instance’s IP addresses
+    """
+    data = client.get_network()
+    print_networking_tables(data)
+
+
+def get_ssh_keys(client: MetadataClient):
+    """
+    Get information about public SSH Keys configured on your instance
+    """
+    data = client.get_ssh_keys()
+    print_ssh_keys_table(data)
+
+
+COMMAND_MAP = {
+    "instance": get_instance,
+    "user-data": get_user_data,
+    "networking": get_network,
+    "sshkeys": get_ssh_keys,
+}
+
+
+def print_help(parser: argparse.ArgumentParser):
+    """
+    Print out the help info to the standard output
+    """
+    parser.print_help()
+
+    # additional help
+    print()
+    print("Available endpoints: ")
+
+    command_help_map = [
+        [name, func.__doc__.strip()]
+        for name, func in sorted(COMMAND_MAP.items())
+    ]
+
+    tab = Table(show_header=False)
+    for row in command_help_map:
+        tab.add_row(*row)
+    rprint(tab)
+
+
+def get_metadata_parser():
+    """
+    Builds argparser for Metadata plug-in
+    """
+    parser = argparse.ArgumentParser(PLUGIN_BASE, add_help=False)
+
+    parser.add_argument(
+        "endpoint",
+        metavar="ENDPOINT",
+        nargs="?",
+        type=str,
+        help="The API endpoint to be called from the Metadata service.",
+    )
+
+    return parser
+
+
+def call(args, _):
+    """
+    The entrypoint for this plugin
+    """
+
+    parser = get_metadata_parser()
+    parsed, args = parser.parse_known_args(args)
+
+    if not parsed.endpoint in COMMAND_MAP or len(args) != 0:
+        print_help(parser)
+        sys.exit(0)
+
+    # make a client, but only if we weren't printing help and endpoint is valid
+    if "--help" not in args:
+        try:
+            client = MetadataClient()
+        except ConnectTimeout as exc:
+            raise ConnectionError(
+                "Can't access Metadata service. Please verify that you are inside a Linode."
+            ) from exc
+    else:
+        print_help(parser)
+        sys.exit(0)
+
+    try:
+        COMMAND_MAP[parsed.endpoint](client)
+    except ApiError as e:
+        sys.exit(f"Error: {e}")

requirements.txt

@@ -4,3 +4,4 @@ PyYAML
 packaging
 rich
 urllib3<3
+linode-metadata

tests/unit/test_plugin_metadata.py

@@ -0,0 +1,128 @@
+import importlib
+
+import pytest
+from linode_metadata.objects.instance import InstanceResponse
+from linode_metadata.objects.networking import NetworkResponse
+from linode_metadata.objects.ssh_keys import SSHKeysResponse
+from pytest import CaptureFixture
+
+from linodecli.plugins.metadata import (
+    print_instance_table,
+    print_networking_tables,
+    print_ssh_keys_table,
+)
+
+plugin = importlib.import_module("linodecli.plugins.metadata")
+
+INSTANCE = InstanceResponse(
+    json_data={
+        "id": 1,
+        "host_uuid": "test_uuid",
+        "label": "test-label",
+        "region": "us-southeast",
+        "tags": "test-tag",
+        "type": "g6-standard-1",
+        "specs": {"vcpus": 2, "disk": 3, "memory": 4, "transfer": 5, "gpus": 6},
+        "backups": {"enabled": False, "status": ["test1", "test2"]},
+    }
+)
+
+NETWORKING = NetworkResponse(
+    json_data={
+        "interfaces": [
+            {
+                "label": "interface-label-1",
+                "purpose": "purpose-1",
+                "ipam_address": ["address1", "address2"],
+            },
+            {
+                "label": "interface-label-2",
+                "purpose": "purpose-2",
+                "ipam_address": ["address3", "address4"],
+            },
+        ],
+        "ipv4": {
+            "public": ["public-1", "public-2"],
+            "private": ["private-1", "private-2"],
+            "shared": ["shared-1", "shared-2"],
+        },
+        "ipv6": {
+            "slaac": "slaac-1",
+            "link_local": "link-local-1",
+            "ranges": ["range-1", "range-2"],
+            "shared_ranges": ["shared-range-1", "shared-range-2"],
+        },
+    }
+)
+
+SSH_KEYS = SSHKeysResponse(
+    json_data={"users": {"root": ["ssh-key-1", "ssh-key-2"]}}
+)
+
+SSH_KEYS_EMPTY = SSHKeysResponse(json_data={"users": {"root": None}})
+
+
+def test_print_help(capsys: CaptureFixture):
+    with pytest.raises(SystemExit) as err:
+        plugin.call(["--help"], None)
+
+    captured_text = capsys.readouterr().out
+
+    assert err.value.code == 0
+    assert "Available endpoints: " in captured_text
+    assert "Get information about public SSH Keys" in captured_text
+
+
+def test_faulty_endpoint(capsys: CaptureFixture):
+    with pytest.raises(SystemExit) as err:
+        plugin.call(["blah"], None)
+
+    captured_text = capsys.readouterr().out
+
+    assert err.value.code == 0
+    assert "Available endpoints: " in captured_text
+    assert "Get information about public SSH Keys" in captured_text
+
+
+def test_instance_table(capsys: CaptureFixture):
+    # Note: Test is brief since table is very large with all values included and captured text abbreviates a lot of values
+    print_instance_table(INSTANCE)
+    captured_text = capsys.readouterr()
+
+    assert "id" in captured_text.out
+    assert "1" in captured_text.out
+
+    assert "3" in captured_text.out
+    assert "2" in captured_text.out
+
+
+def test_networking_table(capsys: CaptureFixture):
+    print_networking_tables(NETWORKING)
+    captured_text = capsys.readouterr()
+
+    assert "purpose" in captured_text.out
+    assert "purpose-1" in captured_text.out
+
+    assert "ip address" in captured_text.out
+    assert "private-1" in captured_text.out
+    assert "type" in captured_text.out
+    assert "shared" in captured_text.out
+
+    assert "slaac" in captured_text.out
+    assert "slaac-1" in captured_text.out
+
+
+def test_ssh_key_table(capsys: CaptureFixture):
+    print_ssh_keys_table(SSH_KEYS)
+    captured_text = capsys.readouterr()
+
+    assert "ssh keys" in captured_text.out
+    assert "ssh-key-1" in captured_text.out
+    assert "ssh-key-2" in captured_text.out
+
+
+def test_empty_ssh_key_table(capsys: CaptureFixture):
+    print_ssh_keys_table(SSH_KEYS_EMPTY)
+    captured_text = capsys.readouterr()
+
+    assert "ssh keys" in captured_text.out