diff options
author | Thomas Haller <thaller@redhat.com> | 2022-02-22 16:27:29 +0100 |
---|---|---|
committer | Thomas Haller <thaller@redhat.com> | 2022-02-23 17:07:16 +0100 |
commit | 4d70c9c4dbedb0b3904796206672b1124d66ce93 (patch) | |
tree | 769040eb02a9113476bc8809e8ed2f175ec2b722 | |
parent | 849a4eee5c06e6f5eb70a4616e0060ba66a6c5e4 (diff) | |
download | NetworkManager-4d70c9c4dbedb0b3904796206672b1124d66ce93.tar.gz |
examples: add "--last" argument to "examples/python/gi/checkpoint.py"
"examples/python/gi/checkpoint.py" is not only an example. It's also a
useful script for testing checkpoints.
Support a "--last" argument to specify the last checkpoint created.
Otherwise, when you are using this example from a test script, it
can be cumbersome to find the right checkpoint point.
Also, rename "client" and "nm_client" variables to "nmc". The purpose of
using the same variable name for the same thing is readability, but also
it works better when copy+paste snippets into the Python REPL.
-rwxr-xr-x | examples/python/gi/checkpoint.py | 117 |
1 files changed, 73 insertions, 44 deletions
diff --git a/examples/python/gi/checkpoint.py b/examples/python/gi/checkpoint.py index 0a42fa1cc0..b63b1c37a4 100755 --- a/examples/python/gi/checkpoint.py +++ b/examples/python/gi/checkpoint.py @@ -5,6 +5,7 @@ # import sys +import re import gi @@ -21,10 +22,11 @@ def usage(): print( " create TIMEOUT [--destroy-all|--delete-new-connections|--disconnect-new-devices|--allow-overlapping|DEV]..." ) - print(" destroy PATH|NUMBER") - print(" rollback PATH|NUMBER") - print(" adjust-rollback-timeout PATH|NUMBER TIMEOUT") + print(" destroy ['--last'|PATH|NUMBER[") + print(" rollback ['--last'|PATH|NUMBER]") + print(" adjust-rollback-timeout '--last'|PATH|NUMBER TIMEOUT") print("") + print(" For destroy|rollback, when omitted then '--last' is the default.") sys.exit(1) @@ -54,32 +56,65 @@ def show(c, ts=None): ) -def find_checkpoint(client, path): - for c in client.get_checkpoints(): +def checkpoint_path_to_num(path): + m = re.match(r"^/org/freedesktop/NetworkManager/Checkpoint/([1-9][0-9]*)$", path) + if m: + return int(m.group(1)) + raise Exception(f'Unexpected D-Bus path "{path}"for checkpoint') + + +def find_checkpoint(nmc, path): + for c in nmc.get_checkpoints(): if c.get_path() == path: return c return None -def validate_path(path, client): +def find_checkpoint_last(nmc): + l = [c.get_path() for c in nmc.get_checkpoints()] + if not l: + return None + l.sort(key=checkpoint_path_to_num) + return l[-1] + + +def validate_path(path, nmc): + if path == "--last": + path = find_checkpoint_last(nmc) + if path is None: + sys.exit("Has no checkpoint") + return path + try: num = int(path) - path = "/org/freedesktop/NetworkManager/Checkpoint/%u" % (num) + path = f"/org/freedesktop/NetworkManager/Checkpoint/{num}" except Exception as e: pass if not path or path[0] != "/": sys.exit('Invalid checkpoint path "%s"' % (path)) - if client is not None: - checkpoint = find_checkpoint(client, path) + if nmc is not None: + checkpoint = find_checkpoint(nmc, path) if checkpoint is None: print('WARNING: no checkpoint with path "%s" found' % (path)) return path -def do_create(client): +def validate_path_from_argv(nmc): + assert len(sys.argv) >= 2 + if len(sys.argv) == 2: + path = "--last" + elif len(sys.argv) > 3: + sys.exit("Failed: invalid extra argument") + else: + path = sys.argv[2] + + return validate_path(path, nmc) + + +def do_create(nmc): flags = NM.CheckpointCreateFlags.NONE if len(sys.argv) < 3: sys.exit("Failed: missing argument timeout") @@ -96,50 +131,44 @@ def do_create(client): elif arg == "--allow-overlapping": flags |= NM.CheckpointCreateFlags.ALLOW_OVERLAPPING else: - d = client.get_device_by_iface(arg) + d = nmc.get_device_by_iface(arg) if d is None: sys.exit("Unknown device %s" % arg) devices.append(d) - def create_cb(client, result, data): + def create_cb(nmc, result, data): try: - checkpoint = client.checkpoint_create_finish(result) + checkpoint = nmc.checkpoint_create_finish(result) print("%s" % checkpoint.get_path()) except Exception as e: sys.stderr.write("Failed: %s\n" % e.message) main_loop.quit() - client.checkpoint_create(devices, timeout, flags, None, create_cb, None) - + nmc.checkpoint_create(devices, timeout, flags, None, create_cb, None) -def do_destroy(client): - if len(sys.argv) < 3: - sys.exit("Missing checkpoint path") - path = validate_path(sys.argv[2], client) +def do_destroy(nmc): + path = validate_path_from_argv(nmc) - def destroy_cb(client, result, data): + def destroy_cb(nmc, result, data): try: - if client.checkpoint_destroy_finish(result) == True: + if nmc.checkpoint_destroy_finish(result) == True: print("Success") except Exception as e: sys.stderr.write("Failed: %s\n" % e.message) main_loop.quit() - client.checkpoint_destroy(path, None, destroy_cb, None) + nmc.checkpoint_destroy(path, None, destroy_cb, None) -def do_rollback(client): - if len(sys.argv) < 3: - sys.exit("Missing checkpoint path") - - path = validate_path(sys.argv[2], client) +def do_rollback(nmc): + path = validate_path_from_argv(nmc) - def rollback_cb(client, result, data): + def rollback_cb(nmc, result, data): try: - res = client.checkpoint_rollback_finish(result) + res = nmc.checkpoint_rollback_finish(result) for path in res: - d = client.get_device_by_path(path) + d = nmc.get_device_by_path(path) if d is None: iface = path else: @@ -149,10 +178,10 @@ def do_rollback(client): sys.stderr.write("Failed: %s\n" % e.message) main_loop.quit() - client.checkpoint_rollback(path, None, rollback_cb, None) + nmc.checkpoint_rollback(path, None, rollback_cb, None) -def do_adjust_rollback_timeout(client): +def do_adjust_rollback_timeout(nmc): if len(sys.argv) < 3: sys.exit("Missing checkpoint path") if len(sys.argv) < 4: @@ -162,42 +191,42 @@ def do_adjust_rollback_timeout(client): except Exception: sys.exit("Invalid timeout") - path = validate_path(sys.argv[2], client) + path = validate_path(sys.argv[2], nmc) - def adjust_rollback_timeout_cb(client, result, data): + def adjust_rollback_timeout_cb(nmc, result, data): try: - client.checkpoint_adjust_rollback_timeout_finish(result) + nmc.checkpoint_adjust_rollback_timeout_finish(result) print("Success") except Exception as e: sys.stderr.write("Failed: %s\n" % e.message) main_loop.quit() - client.checkpoint_adjust_rollback_timeout( + nmc.checkpoint_adjust_rollback_timeout( path, add_timeout, None, adjust_rollback_timeout_cb, None ) -def do_show(client): +def do_show(nmc): ts = NM.utils_get_timestamp_msec() - for c in client.get_checkpoints(): + for c in nmc.get_checkpoints(): show(c, ts) if __name__ == "__main__": - nm_client = NM.Client.new(None) + nmc = NM.Client.new(None) main_loop = GLib.MainLoop() if len(sys.argv) < 2 or sys.argv[1] == "show": - do_show(nm_client) + do_show(nmc) sys.exit(0) elif sys.argv[1] == "create": - do_create(nm_client) + do_create(nmc) elif sys.argv[1] == "destroy": - do_destroy(nm_client) + do_destroy(nmc) elif sys.argv[1] == "rollback": - do_rollback(nm_client) + do_rollback(nmc) elif sys.argv[1] == "adjust-rollback-timeout": - do_adjust_rollback_timeout(nm_client) + do_adjust_rollback_timeout(nmc) else: usage() |