colin/NetworkManager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

examples: extend "ovs-external-ids.py" to support nmcli device modify

Browse Source 
The python example is (also) used to test the feature. That is because
currently nmcli does not yet have support for ovs.external-ids and this
API is only accessible via D-Bus (or a tool like this example).
This commit is contained in:
Thomas Haller
2020-11-11 12:23:46 +01:00
parent afd1d58af5
commit c45041f1fc
1 changed files with 306 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

374
examples/python/gi/ovs-external-ids.py
View File

@@ -12,25 +12,68 @@ import sys
import os
import re
import pprint
import subprocess
import gi
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
###############################################################################
MODE_GET = "get"
MODE_SET = "set"
MODE_APPLY = "apply"
def memoize0(f):
result = []
def helper():
if len(result) == 0:
result.append(f())
return result[0]
return helper
def memoize(f):
memo = {}
def helper(x):
if x not in memo:
memo[x] = f(x)
return memo[x]
return helper
def pr(v):
pprint.pprint(v, indent=4, depth=5, width=60)
HAS_LIBNM_DEBUG = os.getenv("LIBNM_CLIENT_DEBUG") is not None
@memoize0
def is_libnm_debug():
return os.getenv("LIBNM_CLIENT_DEBUG") is not None
@memoize0
def can_sudo():
try:
return (
subprocess.run(
["sudo", "-n", "true"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
).returncode
== 0
)
except:
return False
def _print(msg=""):
if HAS_LIBNM_DEBUG:
if is_libnm_debug():
# we want to use the same logging mechanism as libnm's debug
# logging with "LIBNM_CLIENT_DEBUG=trace,stdout".
NM.utils_print(0, msg + "\n")
@@ -63,12 +106,106 @@ def mainloop_run(timeout_msec=0, mainloop=None):
return not timeout_reached
###############################################################################
def connection_update2(remote_connection, connection):
mainloop = GLib.MainLoop()
result_error = []
def cb(c, result):
try:
c.update2_finish(result)
except Exception as e:
result_error.append(e)
mainloop.quit()
remote_connection.update2(
connection.to_dbus(NM.ConnectionSerializationFlags.ALL),
NM.SettingsUpdate2Flags.NO_REAPPLY,
None,
None,
cb,
)
mainloop_run(mainloop=mainloop)
if result_error:
raise result_error[0]
def device_get_applied_connection(device):
mainloop = GLib.MainLoop()
rr = []
def cb(c, result):
try:
con, version_id = c.get_applied_connection_finish(result)
except Exception as e:
rr.append(e)
else:
rr.append(con)
rr.append(version_id)
mainloop.quit()
device.get_applied_connection_async(0, None, cb)
mainloop_run(mainloop=mainloop)
if len(rr) == 1:
raise rr[0]
return rr[0], rr[1]
def device_reapply(device, connection, version_id):
mainloop = GLib.MainLoop()
result_error = []
def cb(d, result):
try:
d.reapply_finish(result)
except Exception as e:
result_error.append(e)
mainloop.quit()
device.reapply_async(connection, version_id, 0, None, cb)
mainloop_run(mainloop=mainloop)
if len(result_error) == 1:
raise result_error[0]
def ovs_print_external_ids(prefix):
if not can_sudo():
_print(prefix + ": not running as root and cannot call ovs-vsctl")
return
cmds = [["ovs-vsctl", "show"]]
for typ in ["Bridge", "Port", "Interface"]:
cmds += [["ovs-vsctl", "--columns=name,external-ids", "list", typ]]
out = ""
for cmd in cmds:
p = subprocess.run(cmd, stdout=subprocess.PIPE, check=True,)
out += p.stdout.decode("utf-8") + "\n"
out = "\n".join([prefix + s for s in out.split("\n")])
_print(out)
###############################################################################
def usage():
_print("%s g[et] PROFILE [ GETTER ]" % (sys.argv[0]))
_print("%s s[et] [--test] PROFILE SETTER" % (sys.argv[0]))
_print("%s g[et] PROFILE [ GETTER ]" % (sys.argv[0]))
_print("%s s[et] PROFILE SETTER [--test]" % (sys.argv[0]))
_print("%s a[pply] DEVICE SETTER [--test]" % (sys.argv[0]))
_print(
" PROFILE := [id | uuid | type] STRING | [ ~id | ~type ] REGEX_STRING | STRING"
)
_print(" DEVICE := [iface] STRING")
_print(" GETTER := ( KEY | ~REGEX_KEY ) [... GETTER]")
_print(" SETTER := ( + | - | -KEY | [+]KEY VALUE ) [... SETTER]")
@@ -88,7 +225,7 @@ def parse_args(argv):
had_dash_dash = False
args = {
"mode": MODE_GET,
"profile_arg": None,
"select_arg": None,
"ids_arg": [],
"do_test": False,
}
@@ -101,6 +238,8 @@ def parse_args(argv):
args["mode"] = MODE_SET
elif a in ["g", "get"]:
args["mode"] = MODE_GET
elif a in ["a", "apply"]:
args["mode"] = MODE_APPLY
else:
die_usage("unexpected mode argument '%s'" % (a))
i += 1
@@ -111,17 +250,22 @@ def parse_args(argv):
i += 1
continue
if args["profile_arg"] is None:
if a in ["id", "~id", "uuid", "type", "~type"]:
if args["select_arg"] is None:
if args["mode"] == MODE_APPLY:
possible_selects = ["iface"]
else:
possible_selects = ["id", "~id", "uuid", "type", "~type"]
if a in possible_selects:
if i + 1 >= len(argv):
die_usage("'%s' requires an argument'" % (a))
args["profile_arg"] = (a, argv[i + 1])
args["select_arg"] = (a, argv[i + 1])
i += 2
continue
if a == "*":
a = None
args["profile_arg"] = ("*", a)
args["select_arg"] = ("*", a)
i += 1
continue
@@ -156,6 +300,12 @@ def parse_args(argv):
return args
def device_to_str(device, show_type=False):
if show_type:
return "%s (%s)" % (device.get_iface(), device.get_type_desc())
return "%s" % (device.get_iface(),)
def connection_to_str(connection, show_type=False):
if show_type:
return "%s (%s, %s)" % (
@@ -166,15 +316,38 @@ def connection_to_str(connection, show_type=False):
return "%s (%s)" % (connection.get_id(), connection.get_uuid())
def connections_filter(connections, profile_arg):
def devices_filter(devices, select_arg):
devices = list(sorted(devices, key=device_to_str))
if not select_arg:
return devices
# we preserve the order of the selected devices. And
# if devices are selected multiple times, we return
# them multiple times.
l = []
f = select_arg
for d in devices:
if f[0] == "iface":
if f[1] == d.get_iface():
l.append(d)
else:
assert f[0] == "*"
if f[1] is None:
l.append(d)
else:
if f[1] in [d.get_iface()]:
l.append(d)
return l
def connections_filter(connections, select_arg):
connections = list(sorted(connections, key=connection_to_str))
if not profile_arg:
if not select_arg:
return connections
# we preserve the order of the selected connections. And
# if connections are selected multiple times, we return
# them multiple times.
l = []
f = profile_arg
f = select_arg
for c in connections:
if f[0] == "id":
if f[1] == c.get_id():
@@ -218,6 +391,7 @@ def ids_select(ids, mode, ids_arg):
if d not in requested:
requested.append(d)
else:
assert mode in [MODE_SET, MODE_APPLY]
d2 = d[0]
assert d2[0] in ["-", "+"]
d3 = d2[1:]
@@ -241,7 +415,8 @@ def connection_print(connection, mode, ids_arg, dbus_path, prefix=""):
_print(
"%s%s [%s]" % (prefix, connection_to_str(connection, show_type=True), num_str)
)
_print("%s %s" % (prefix, dbus_path))
if dbus_path:
_print("%s %s" % (prefix, dbus_path))
if sett is not None:
dd = sett.get_property(NM.SETTING_OVS_EXTERNAL_IDS_DATA)
else:
@@ -255,25 +430,7 @@ def connection_print(connection, mode, ids_arg, dbus_path, prefix=""):
_print('%s "%s" = <unset>' % (prefix, k))
def do_get(connections, ids_arg):
first_line = True
for c in connections:
if first_line:
first_line = False
else:
_print()
connection_print(c, MODE_GET, ids_arg, dbus_path=c.get_path())
def do_set(nmc, connection, ids_arg, do_test):
remote_connection = connection
connection = NM.SimpleConnection.new_clone(remote_connection)
connection_print(
connection, MODE_SET, [], remote_connection.get_path(), prefix="BEFORE: "
)
_print()
def sett_update(connection, ids_arg):
sett = connection.get_setting(NM.SettingOvsExternalIDs)
@@ -334,34 +491,39 @@ def do_set(nmc, connection, ids_arg, do_test):
_print(' SET: "%s" = "%s" (unchanged)' % (key, val))
sett.set_data(key, val)
def do_get(connections, ids_arg):
first_line = True
for c in connections:
if first_line:
first_line = False
else:
_print()
connection_print(c, MODE_GET, ids_arg, dbus_path=c.get_path())
def do_set(nmc, connection, ids_arg, do_test):
remote_connection = connection
connection = NM.SimpleConnection.new_clone(remote_connection)
connection_print(
connection, MODE_SET, [], remote_connection.get_path(), prefix="BEFORE: "
)
_print()
sett_update(connection, ids_arg)
if do_test:
_print()
_print("Only show. Run without --test to set")
return
mainloop = GLib.MainLoop()
result_error = []
def callback(c, result):
try:
c.update2_finish(result)
except Exception as e:
result_error.append(e)
mainloop.quit()
remote_connection.update2(
connection.to_dbus(NM.ConnectionSerializationFlags.ALL),
NM.SettingsUpdate2Flags.NO_REAPPLY,
None,
None,
callback,
)
mainloop_run(mainloop=mainloop)
if result_error:
try:
connection_update2(remote_connection, connection)
except Exception as e:
_print()
_print("FAILURE to commit connection: %s" % (result_error[0]))
_print("FAILURE to commit connection: %s" % (e))
return
# NMClient received the completion of Update2() call. It also received
@@ -397,6 +559,68 @@ def do_set(nmc, connection, ids_arg, do_test):
_print("WARNING: resulting connection is not as expected")
def do_apply(nmc, device, ids_arg, do_test):
try:
connection_orig, version_id = device_get_applied_connection(device)
except Exception as e:
_print(
'failure to get applied connection for %s: %s"' % (device_to_str(device), e)
)
die("The device does not seem active? Nothing to reapply")
_print(
"REAPPLY device %s (%s) with connection %s (version-id = %s)"
% (
device_to_str(device),
NM.Object.get_path(device),
connection_to_str(connection_orig),
version_id,
)
)
_print()
ovs_print_external_ids("BEFORE-OVS-VSCTL: ")
_print()
connection = NM.SimpleConnection.new_clone(connection_orig)
connection_print(connection, MODE_APPLY, [], device.get_path(), prefix="BEFORE: ")
_print()
sett_update(connection, ids_arg)
if do_test:
_print()
_print("Only show. Run without --test to set")
return
_print()
_print("reapply...")
try:
device_reapply(device, connection, version_id)
except Exception as e:
_print()
_print("FAILURE to commit connection: %s" % (e))
return
try:
connection_after, version_id = device_get_applied_connection(device)
except Exception as e:
_print(
'failure to get applied connection after reapply for device %s: %s"'
% (device_to_str(device), e)
)
die("FAILURE to get applied connection after reapply")
_print()
connection_print(connection, MODE_APPLY, [], device.get_path(), prefix="AFTER: ")
_print()
ovs_print_external_ids("AFTER-OVS-VSCTL: ")
###############################################################################
if __name__ == "__main__":
@@ -405,21 +629,35 @@ if __name__ == "__main__":
nmc = NM.Client.new(None)
connections = connections_filter(nmc.get_connections(), args["profile_arg"])
if args["mode"] == MODE_APPLY:
if args["mode"] == MODE_SET:
if len(connections) != 1:
devices = devices_filter(nmc.get_devices(), args["select_arg"])
if len(devices) != 1:
_print(
"To set the external-ids of a connection, exactly one connection must be selected via id|uuid. Instead, %s connection matched ([%s])"
% (
len(connections),
", ".join([connection_to_str(c) for c in connections]),
)
"To apply the external-ids of a device, exactly one connection must be selected. Instead, %s devices matched ([%s])"
% (len(devices), ", ".join([device_to_str(c) for c in devices]),)
)
die_usage("Select unique connection to set")
do_set(nmc, connections[0], args["ids_arg"], do_test=args["do_test"])
die_usage("Select unique device to apply")
do_apply(nmc, devices[0], args["ids_arg"], do_test=args["do_test"])
else:
if len(connections) < 1:
_print("No connection selected for printing the external ids")
die_usage("Select connection to get")
do_get(connections, args["ids_arg"])
connections = connections_filter(nmc.get_connections(), args["select_arg"])
if args["mode"] == MODE_SET:
if len(connections) != 1:
_print(
"To set the external-ids of a connection, exactly one connection must be selected via id|uuid. Instead, %s connection matched ([%s])"
% (
len(connections),
", ".join([connection_to_str(c) for c in connections]),
)
)
die_usage("Select unique connection to set")
do_set(nmc, connections[0], args["ids_arg"], do_test=args["do_test"])
else:
if len(connections) < 1:
_print("No connection selected for printing the external ids")
die_usage("Select connection to get")
do_get(connections, args["ids_arg"])