diff options
Diffstat (limited to 'examples/python/gi')
| -rwxr-xr-x | examples/python/gi/checkpoint.py | 233 | ||||
| -rwxr-xr-x | examples/python/gi/dns.py | 50 | ||||
| -rwxr-xr-x | examples/python/gi/get-devices.py | 42 | ||||
| -rwxr-xr-x | examples/python/gi/get-interface-flags.py | 23 | ||||
| -rwxr-xr-x | examples/python/gi/get-lldp-neighbors.py | 39 | ||||
| -rwxr-xr-x | examples/python/gi/gmaincontext.py | 448 | ||||
| -rwxr-xr-x | examples/python/gi/vpn-import.py | 65 | ||||
| -rwxr-xr-x | examples/python/gi/wifi-p2p.py | 117 |
8 files changed, 1017 insertions, 0 deletions
diff --git a/examples/python/gi/checkpoint.py b/examples/python/gi/checkpoint.py new file mode 100755 index 00000000..6f3194cd --- /dev/null +++ b/examples/python/gi/checkpoint.py @@ -0,0 +1,233 @@ +#!/usr/bin/env python +# SPDX-License-Identifier: GPL-2.0-or-later +# +# Copyright (C) 2017 Red Hat, Inc. +# + +import sys +import re + +import gi + +gi.require_version("NM", "1.0") +from gi.repository import GLib, NM + +############################################################################### + + +def usage(): + print("Usage: %s [COMMAND [ARG]...]" % sys.argv[0]) + print("") + print(" COMMANDS: [show]") + print( + " create TIMEOUT [--destroy-all|--delete-new-connections|--disconnect-new-devices|--allow-overlapping|DEV]..." + ) + 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) + + +def show(c, ts=None): + cr = c.get_created() + rt = c.get_rollback_timeout() + print("%s:" % c.get_path()) + print( + " created: %u%s" + % (cr, "" if ts is None else (" (%s sec ago)" % ((ts - cr) / 1000.0))) + ) + if rt == 0: + print(" timeout: infinity") + else: + print( + " timeout: %u seconds%s" + % ( + rt, + "" + if ts is None + else (" (circa %s sec left)" % ((cr + (rt * 1000) - ts) / 1000.0)), + ) + ) + print( + " devices: %s" + % (" ".join(sorted(map(lambda x: x.get_iface(), c.get_devices())))) + ) + + +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 find_checkpoint_last(nmc): + return max( + nmc.get_checkpoints(), + key=lambda c: checkpoint_path_to_num(c.get_path()), + default=None, + ) + + +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 = 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 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 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") + + timeout = int(sys.argv[2]) + devices = [] + for arg in sys.argv[3:]: + if arg == "--destroy-all": + flags |= NM.CheckpointCreateFlags.DESTROY_ALL + elif arg == "--delete-new-connections": + flags |= NM.CheckpointCreateFlags.DELETE_NEW_CONNECTIONS + elif arg == "--disconnect-new-devices": + flags |= NM.CheckpointCreateFlags.DISCONNECT_NEW_DEVICES + elif arg == "--allow-overlapping": + flags |= NM.CheckpointCreateFlags.ALLOW_OVERLAPPING + else: + d = nmc.get_device_by_iface(arg) + if d is None: + sys.exit("Unknown device %s" % arg) + devices.append(d) + + def create_cb(nmc, result, data): + try: + 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() + + nmc.checkpoint_create(devices, timeout, flags, None, create_cb, None) + + +def do_destroy(nmc): + path = validate_path_from_argv(nmc) + + def destroy_cb(nmc, result, data): + try: + if nmc.checkpoint_destroy_finish(result) == True: + print("Success") + except Exception as e: + sys.stderr.write("Failed: %s\n" % e.message) + main_loop.quit() + + nmc.checkpoint_destroy(path, None, destroy_cb, None) + + +def do_rollback(nmc): + path = validate_path_from_argv(nmc) + + def rollback_cb(nmc, result, data): + try: + res = nmc.checkpoint_rollback_finish(result) + for path in res: + d = nmc.get_device_by_path(path) + if d is None: + iface = path + else: + iface = d.get_iface() + print("%s => %s" % (iface, "OK" if res[path] == 0 else "ERROR")) + except Exception as e: + sys.stderr.write("Failed: %s\n" % e.message) + main_loop.quit() + + nmc.checkpoint_rollback(path, None, rollback_cb, None) + + +def do_adjust_rollback_timeout(nmc): + if len(sys.argv) < 3: + sys.exit("Missing checkpoint path") + if len(sys.argv) < 4: + sys.exit("Missing timeout") + try: + add_timeout = int(sys.argv[3]) + except Exception: + sys.exit("Invalid timeout") + + path = validate_path(sys.argv[2], nmc) + + def adjust_rollback_timeout_cb(nmc, result, data): + try: + nmc.checkpoint_adjust_rollback_timeout_finish(result) + print("Success") + except Exception as e: + sys.stderr.write("Failed: %s\n" % e.message) + main_loop.quit() + + nmc.checkpoint_adjust_rollback_timeout( + path, add_timeout, None, adjust_rollback_timeout_cb, None + ) + + +def do_show(nmc): + ts = NM.utils_get_timestamp_msec() + for c in nmc.get_checkpoints(): + show(c, ts) + + +if __name__ == "__main__": + nmc = NM.Client.new(None) + main_loop = GLib.MainLoop() + + if len(sys.argv) < 2 or sys.argv[1] == "show": + do_show(nmc) + sys.exit(0) + elif sys.argv[1] == "create": + do_create(nmc) + elif sys.argv[1] == "destroy": + do_destroy(nmc) + elif sys.argv[1] == "rollback": + do_rollback(nmc) + elif sys.argv[1] == "adjust-rollback-timeout": + do_adjust_rollback_timeout(nmc) + else: + usage() + + main_loop.run() diff --git a/examples/python/gi/dns.py b/examples/python/gi/dns.py new file mode 100755 index 00000000..483fe25f --- /dev/null +++ b/examples/python/gi/dns.py @@ -0,0 +1,50 @@ +#!/usr/bin/env python +# SPDX-License-Identifier: GPL-2.0-or-later +# +# Copyright (C) 2016 Red Hat, Inc. +# + +import gi + +gi.require_version("NM", "1.0") +from gi.repository import GLib, NM + +# This example shows how to monitor the DNS configuration + +main_loop = None + + +def handle_config(config): + print(" ---- new configuration ----") + for entry in config: + print(" * servers: %s" % ", ".join(map(str, entry.get_nameservers()))) + + domains = entry.get_domains() + if domains and domains[0]: + print(" domains: %s" % ", ".join(map(str, domains))) + + if entry.get_interface(): + print(" interface: %s" % entry.get_interface()) + + print(" priority: %d" % entry.get_priority()) + + if entry.get_vpn(): + print(" vpn: yes") + + print("") + + +def dns_config_changed(self, property): + handle_config(self.get_dns_configuration()) + + +main_loop = None + +if __name__ == "__main__": + c = NM.Client.new(None) + c.connect("notify::dns-configuration", dns_config_changed) + + handle_config(c.get_dns_configuration()) + + main_loop = GLib.MainLoop() + main_loop.run() diff --git a/examples/python/gi/get-devices.py b/examples/python/gi/get-devices.py new file mode 100755 index 00000000..43fd9343 --- /dev/null +++ b/examples/python/gi/get-devices.py @@ -0,0 +1,42 @@ +#!/usr/bin/env python +# SPDX-License-Identifier: GPL-2.0-or-later +# +# Copyright (C) 2014 Red Hat, Inc. +# + +# This example lists all devices, both real and placeholder ones + +import gi + +gi.require_version("NM", "1.0") +from gi.repository import NM + +if __name__ == "__main__": + client = NM.Client.new(None) + devices = client.get_all_devices() + + print("Real devices") + print("------------") + for d in devices: + if d.is_real(): + print( + "%s (%s): %s" + % ( + d.get_iface(), + d.get_type_description(), + d.get_state(), + ) + ) + + print("\nUnrealized/placeholder devices") + print("------------------------------") + for d in devices: + if not d.is_real(): + print( + "%s (%s): %s" + % ( + d.get_iface(), + d.get_type_description(), + d.get_state(), + ) + ) diff --git a/examples/python/gi/get-interface-flags.py b/examples/python/gi/get-interface-flags.py new file mode 100755 index 00000000..f7f6bf97 --- /dev/null +++ b/examples/python/gi/get-interface-flags.py @@ -0,0 +1,23 @@ +#!/usr/bin/env python +# SPDX-License-Identifier: GPL-2.0-or-later +# +# Copyright (C) 2019 Red Hat, Inc. +# + +import gi + +gi.require_version("NM", "1.0") +from gi.repository import NM + +if __name__ == "__main__": + client = NM.Client.new(None) + devices = client.get_devices() + + for d in devices: + print( + "{:<16} {:<16} {}".format( + d.get_iface(), + "(" + d.get_type_description() + ")", + NM.utils_enum_to_str(NM.DeviceInterfaceFlags, d.get_interface_flags()), + ) + ) diff --git a/examples/python/gi/get-lldp-neighbors.py b/examples/python/gi/get-lldp-neighbors.py new file mode 100755 index 00000000..2907dfa4 --- /dev/null +++ b/examples/python/gi/get-lldp-neighbors.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python +# SPDX-License-Identifier: GPL-2.0-or-later +# +# Copyright (C) 2015 Red Hat, Inc. +# + +import sys +import gi + +gi.require_version("NM", "1.0") +from gi.repository import GLib, NM + +# This example shows how to get a list of LLDP neighbors for a given interface. + +main_loop = None + +if __name__ == "__main__": + if len(sys.argv) != 2: + sys.exit("Usage: %s <interface>" % sys.argv[0]) + dev_iface = sys.argv[1] + + c = NM.Client.new(None) + dev = c.get_device_by_iface(dev_iface) + if dev is None: + sys.exit("Device '%s' not found" % dev_iface) + + neighbors = dev.get_lldp_neighbors() + for neighbor in neighbors: + ret, chassis = neighbor.get_attr_string_value("chassis-id") + ret, port = neighbor.get_attr_string_value("port-id") + print("Neighbor: %s - %s" % (chassis, port)) + for attr in neighbor.get_attr_names(): + attr_type = neighbor.get_attr_type(attr) + if attr_type.equal(GLib.VariantType.new("s")): + ret, value = neighbor.get_attr_string_value(attr) + print(" %-32s: %s" % (attr, value)) + elif attr_type.equal(GLib.VariantType.new("u")): + ret, value = neighbor.get_attr_uint_value(attr) + print(" %-32s: %u" % (attr, value)) diff --git a/examples/python/gi/gmaincontext.py b/examples/python/gi/gmaincontext.py new file mode 100755 index 00000000..90a9fa25 --- /dev/null +++ b/examples/python/gi/gmaincontext.py @@ -0,0 +1,448 @@ +#!/bin/python + +############################################################################### +# An example that creates a NMClient instance for another GMainContext +# and iterates the context while doing an async D-Bus call. +# +# D-Bus is fundamentally async. libnm's NMClient API caches D-Bus objects +# on NetworkManager's D-Bus API. As such, it is "frozen" (with the current +# content of the cache) while not iterating the GMainContext. Only by iterating +# the GMainContext any events are processed and things change. +# +# This means, NMClient heavily uses GMainContext (and GDBusConnection) +# and to operate it, you need to iterate the GMainContext. The synchronous +# API (like NM.Client.new()) is for simple programs but usually not best +# for using NMClient for real applications. +# +# To learn more about GMainContext, read https://developer.gnome.org/SearchProvider/documentation/tutorials/main-contexts.html +# When I say "mainloop" or "event loop", I mean GMainContext. GMainLoop is +# a small wrapper around GMainContext to run the context with a boolean +# flag. +# +# Usually, non trivial applications run the GMainContext (or GMainLoop) +# from the main() function and aside some setup and teardown, everything +# happens as events from the event loop. +# This example instead performs synchronous steps, and at the places where +# we need to get the result of some async operation, we iterate the GMainContext +# until we get the result. This may not be how a complex application works, +# but you might do this on a simpler application (like a script) that iterates +# the mainloop whenever it needs to wait for async operations to complete. +# +# Iterating the mainloop might dispatch any other sources that are ready. +# In this example nobody else is scheduling unrelated timers or events, but +# if that happens, your application needs to cope with that. +# E.g. while iterating the mainloop many times, still don't nest running the +# same main context (unless you really know what you do). + +############################################################################### + +import os +import sys +import time +import traceback + +import gi + +gi.require_version("NM", "1.0") +from gi.repository import NM, GLib, Gio + + +############################################################################### + + +def log(msg=None, prefix=None, suffix="\n"): + # We use nm_utils_print(), because that uses the same logging + # mechanism as if you run with "LIBNM_CLIENT_DEBUG=trace". This + # ensures that messages are in sync. + if msg is None: + NM.utils_print(0, "\n") + return + if prefix is None: + prefix = f"[{time.monotonic():.5f}] " + NM.utils_print(0, f"{prefix}{msg}{suffix}") + + +def error_is_cancelled(e): + # Whether error is due to cancellation. + if isinstance(e, GLib.GError): + if e.domain == "g-io-error-quark" and e.code == Gio.IOErrorEnum.CANCELLED: + return True + return False + + +############################################################################### + +# A Context manager for running a mainloop. Of course, this does +# not do anything magically. You can run the context/mainloop without +# this context object. +# +# This is just to show how we could iterate the GMainContext while waiting +# for an async reply. Note that many non-trivial applications that use glib +# would instead run the mainloop from the main function, only running it once, +# but for the entire duration of the program. +# +# This example and MainLoopRun instead assume that you iterate the maincontext +# for short durations at a time. In particular in this case, where there is +# a dedicated maincontext only for NMClient. +class MainLoopRun: + def __init__(self, info, ctx, timeout=None): + self._info = info + self._loop = GLib.MainLoop(ctx) + self.cancellable = Gio.Cancellable() + self._timeout = timeout + self.got_timeout = False + self.result = None + self.error = None + log(f"MainLoopRun[{self._info}]: create with timeout {self._timeout}") + + def _timeout_cb(self, _): + log(f"MainLoopRun[{self._info}]: timeout") + self.got_timeout = True + self._detach() + self.cancellable.cancel() + return False + + def _cancellable_cb(self): + log(f"MainLoopRun[{self._info}]: cancelled") + + def _detach(self): + if self._timeout_source is not None: + self._timeout_source.destroy() + self._timeout_source = None + if self._cancellable_id is not None: + self.cancellable.disconnect(self._cancellable_id) + self._cancellable_id = None + + def __enter__(self): + log(f"MainLoopRun[{self._info}]: enter") + self._timeout_source = None + if self._timeout is not None: + self._timeout_source = GLib.timeout_source_new(int(self._timeout * 1000)) + self._timeout_source.set_callback(self._timeout_cb) + self._timeout_source.attach(self._loop.get_context()) + self._cancellable_id = self.cancellable.connect(self._cancellable_cb) + self._loop.get_context().push_thread_default() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if exc_type is not None: + # Exception happened. + log(f"MainLoopRun[{self._info}]: exit with exception") + else: + log(f"MainLoopRun[{self._info}]: exit: start mainloop") + + self._loop.run() + + if self.error is not None: + log( + f"MainLoopRun[{self._info}]: exit: complete with error {self.error}" + ) + elif self.result is not None: + log( + f"MainLoopRun[{self._info}]: exit: complete with result {self.result}" + ) + else: + log(f"MainLoopRun[{self._info}]: exit: complete with success") + + self._detach() + self._loop.get_context().pop_thread_default() + return False + + def quit(self): + log(f"MainLoopRun[{self._info}]: quit mainloop") + self._detach() + self._loop.quit() + + +############################################################################### + + +def get_bus(): + # Let's get the GDBusConnection singleton by calling Gio.bus_get(). + # Since we do everything async, use Gio.bus_get() instead Gio.bus_get_sync(). + with MainLoopRun("get_bus", None, 1) as r: + + def bus_get_cb(source, result, r): + try: + c = Gio.bus_get_finish(result) + except Exception as e: + r.error = e + else: + r.result = c + r.quit() + + Gio.bus_get(Gio.BusType.SYSTEM, r.cancellable, bus_get_cb, r) + + return r.result + + +############################################################################### + + +def create_nmc(dbus_connection): + # Show how to create and initialize a NMClient asynchronously. + # + # NMClient implements GAsyncInitableIface, it thus can be initialized + # asynchronously. That has actually an advantage, because the sync + # initialization (GInitableIface) requires to create an internal GMainContext + # which has an overhead. + # + # Also, split the GObject creation and the init_async() call in two. + # That allows to pass construct-only parameters, in particular like + # the instance_flags. + + # Create a separate context for the NMClient. The NMClient is strongly + # tied to the context used at construct time. + ctx = GLib.MainContext() + ctx.push_thread_default() + + log(f"[create_nmc]: use separate context for NMClient: ctx={ctx}") + try: + # We create a client asynchronously. There is synchronous + # NM.Client(), however that requires an internal GMainContext + # and has thus an overhead. Also, it's obviously blocking. + # + # Instead, we initialize it asynchronously, which means + # we need to iterate the main context. In this case, the + # context cannot have any other sources dispatched, but + # if there would be other sources, they might be dispatched + # while iterating (so this is waiting for the result, but + # may also dispatch unrelated sources (if any), which you would need + # to handle). + # + # Also, only when using the GObject constructor directly, we can + # suppress loading the permissions and pass a D-Bus connection. + nmc = NM.Client( + instance_flags=NM.ClientInstanceFlags.NO_AUTO_FETCH_PERMISSIONS, + dbus_connection=dbus_connection, + ) + log(f"[create_nmc]: new NMClient instance: {nmc}") + finally: + # We actually don't need that the ctx is the current thread default + # later on. NMClient will automatically push it, when necessary. + ctx.pop_thread_default() + + with MainLoopRun("create_mnc", nmc.get_main_context(), 2) as r: + + def _async_init_cb(nmc, result, r): + try: + nmc.init_finish(result) + except Exception as e: + log(f"[create_nmc]: init_async() completed with error: {e}") + r.error = e + else: + log(f"[create_nmc]: init_async() completed with success") + r.quit() + + log(f"[create_nmc]: start init_async()") + nmc.init_async(GLib.PRIORITY_DEFAULT, r.cancellable, _async_init_cb, r) + + if r.error is None: + if nmc.get_nm_running(): + log( + f"[create_nmc]: completed with success (daemon version: {nmc.get_version()}, D-Bus daemon unique name: {nmc.get_dbus_name_owner()})" + ) + else: + log(f"[create_nmc]: completed with success (daemon not running)") + return nmc + if error_is_cancelled(r.error): + # Cancelled by us. This happened because we hit the timeout with + # MainLoopRun. + log(f"[create_nmc]: failed to initialize within timeout") + return None + if not nmc.get_dbus_connection(): + # The NMClient has no D-Bus connection, it usually would try + # to get one via Gio.bus_get(), but it failed. + log(f"[create_nmc]: failed to create D-Bus connection: {r.error}") + return None + + log(f"[create_nmc]: unexpected error creating NMClient ({r.error})") + # This actually should not happen. There is no other reason why + # initialization can fail. + assert False, "NMClient initialization is not supposed to fail" + return nmc + + +############################################################################### + + +def make_call(nmc): + + log("[make_call]: make some async D-Bus call") + + if not nmc: + log("[make_call]: no NMClient. Skip") + return + + with MainLoopRun("make_call", nmc.get_main_context(), 1) as r: + + # There are two reasons why async operations are preferable with + # D-Bus and libnm: + # + # - pseudo blocking messes with the ordering of events (see https://smcv.pseudorandom.co.uk/2008/11/nonblocking/). + # - blocking prevents other things from happening and combining synchronous calls is more limited. + # + # So doing async operations is mostly interesting when performing multiple operations in + # parallel, or when we still want to handle other events while waiting for the reply. + # The example here does not cover that usage well, because there is only one thing happening. + + def _dbus_call_cb(nmc, result, r): + try: + res = nmc.dbus_call_finish(result) + except Exception as e: + if error_is_cancelled(e): + log( + f"[make_call]: dbus_call() completed with cancellation after timeout" + ) + else: + log(f"[make_call]: dbus_call() completed with error: {e}") + + if False: + # I don't understand why, but if you hit this exception (e.g. by setting a low + # timeout) and pass the exception to the out context, then an additional reference + # to nmc is leaked, and destroy_nmc() will fail. Workaround + r.error = e + + r.error = str(e) + else: + log( + f"[make_call]: dbus_call() completed with success: {str(res)[:40]}..." + ) + r.quit() + + log(f"[make_call]: start GetPermissions call") + nmc.dbus_call( + NM.DBUS_PATH, + NM.DBUS_INTERFACE, + "GetPermissions", + GLib.Variant.new_tuple(), + GLib.VariantType("(a{ss})"), + 1000, + r.cancellable, + _dbus_call_cb, + r, + ) + + return r.error is None + + +############################################################################### + + +def destroy_nmc(nmc_holder): + # The way to shutdown an NMClient is just by unrefing it. + # + # At any moment, can an NMClient instance have pending async operations. + # While unrefing NMClient will cancel them right away, they are only + # reaped when we iterate the GMainContext some more. That means, if we don't + # want to leak the GMainContext and the pending operations, we must + # iterate it some more. + # + # To know how much more, there is nmc.get_context_busy_watcher(), + # We can subscribe a weak reference and keep iterating as long + # as the watcher is alive. + # + # Of course, this only applies if the application wishes to keep running + # but no longer iterating NMClient's GMainContext. Then you need to ensure + # that all pending operations in GMainContext are completed (by iterating it). + # + # In python, that is a bit tricky, because the caller of destroy_nmc() + # must give up its reference and pass it here via the @nmc_holder list. + # You must call destroy_nmc() without having any other reference on + # nmc. + # + # This is just an example. This relies that on this point we only have + # one reference to NMClient (and it's held by the nmc_holder list). + # Usually you wouldn't make assumptions about this. Instead, you just + # assume that you need to keep iterating the GMainContext as long as + # the context busy watcher is alive, regardless that at this point others + # might still hold references on the NMClient. + + # Transfer the nmc reference out of the list. + (nmc,) = nmc_holder + nmc_holder.clear() + + if not nmc: + log(f"[destroy_nmc]: nothing to destroy") + return + + log( + f"[destroy_nmc]: destroying NMClient {nmc}: pyref={sys.getrefcount(nmc)}, ref_count={nmc.ref_count}" + ) + + ctx = nmc.get_main_context() + + finished = [] + + def _weak_ref_cb(): + log(f"[destroy_nmc]: context busy watcher is gone") + finished.clear() + finished.append(True) + + # We take a weak ref on the context-busy-watcher object and give up + # our reference on nmc. This must be the last reference, which initiates + # the shutdown of the NMClient. + weak_ref = nmc.get_context_busy_watcher().weak_ref(_weak_ref_cb) + del nmc + + def _timeout_cb(unused): + if not finished: + # Somebody else holds a reference to the NMClient and keeps + # it alive. We cannot properly clean up. + log( + f"[destroy_nmc]: ERROR: timeout waiting for context busy watcher to be gone" + ) + finished.append(False) + return False + + timeout_source = GLib.timeout_source_new(1000) + timeout_source.set_callback(_timeout_cb) + timeout_source.attach(ctx) + + while not finished: + log(f"[destroy_nmc]: iterating main context") + ctx.iteration(True) + + timeout_source.destroy() + + log(f"[destroy_nmc]: done: {finished[0]}") + if not finished[0]: + weak_ref.unref() + raise Exception("Failure to destroy NMClient: something keeps it alive") + + +############################################################################### + + +def run1(): + try: + dbus_connection = get_bus() + log() + + nmc = create_nmc(dbus_connection) + log() + + make_call(nmc) + log() + + # To cleanup the NMClient, we need to give up the reference. Move + # it to a list, and destroy_nmc() will take care of it. + nmc_holder = [nmc] + del nmc + destroy_nmc(nmc_holder) + log() + log("done") + except Exception as e: + log() + log("EXCEPTION:") + log(f"{e}") + for tb in traceback.format_exception(e): + for l in tb.split("\n"): + log(f">>> {l}") + return False + return True + + +if __name__ == "__main__": + if not run1(): + sys.exit(1) diff --git a/examples/python/gi/vpn-import.py b/examples/python/gi/vpn-import.py new file mode 100755 index 00000000..0db19a25 --- /dev/null +++ b/examples/python/gi/vpn-import.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python +# SPDX-License-Identifier: GPL-2.0-or-later +# +# Copyright (C) 2014 Red Hat, Inc. +# + +# +# This example imports a VPN connection, by loading the glib based +# VPN plugin. + +import gi + +gi.require_version("NM", "1.0") +from gi.repository import GLib, NM + +import sys + +if len(sys.argv) != 2: + print("Expects one argument: the filename") + sys.exit(1) +filename = sys.argv[1] + +connection = None +for vpn_info in NM.VpnPluginInfo.list_load(): + print("TRY: plugin %s" % (vpn_info.get_filename())) + try: + vpn_plugin = vpn_info.load_editor_plugin() + except Exception as e: + print("SKIP: cannot load plugin: %s" % (e)) + continue + try: + connection = vpn_plugin.import_(filename) + except Exception as e: + print("SKIP: failure to import %s" % (e)) + continue + break + +if connection is None: + print('None of the VPN plugins was able to import "%s"' % (filename)) + sys.exit(1) + +connection.normalize() + +print( + 'connection imported from "%s" using plugin "%s" ("%s", %s)' + % (filename, vpn_info.get_filename(), connection.get_id(), connection.get_uuid()) +) + +client = NM.Client.new(None) + +main_loop = GLib.MainLoop() + + +def added_cb(client, result, data): + try: + client.add_connection_finish(result) + print("The connection profile has been successfully added to NetworkManager.") + except Exception as e: + print("ERROR: failed to add connection: %s\n" % e) + main_loop.quit() + + +client.add_connection_async(connection, True, None, added_cb, None) + +main_loop.run() diff --git a/examples/python/gi/wifi-p2p.py b/examples/python/gi/wifi-p2p.py new file mode 100755 index 00000000..f4c2278e --- /dev/null +++ b/examples/python/gi/wifi-p2p.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python +# SPDX-License-Identifier: GPL-2.0-or-later +# +# Copyright (C) 2020 Red Hat, Inc. + +# This example performs a scan of Wi-Fi P2P peers and connects to one +# of them. + +import sys +import uuid +import gi + +gi.require_version("NM", "1.0") +from gi.repository import GLib, NM + +main_loop = None +client = None + + +def create_profile(name, peer_mac): + profile = NM.SimpleConnection.new() + + s_con = NM.SettingConnection.new() + s_con.set_property(NM.SETTING_CONNECTION_ID, name) + s_con.set_property(NM.SETTING_CONNECTION_UUID, str(uuid.uuid4())) + s_con.set_property(NM.SETTING_CONNECTION_TYPE, "wifi-p2p") + s_con.set_property(NM.SETTING_CONNECTION_AUTOCONNECT, False) + + s_ip4 = NM.SettingIP4Config.new() + s_ip4.set_property(NM.SETTING_IP_CONFIG_METHOD, "auto") + + s_ip6 = NM.SettingIP6Config.new() + s_ip6.set_property(NM.SETTING_IP_CONFIG_METHOD, "auto") + + s_wifi_p2p = NM.SettingWifiP2P.new() + s_wifi_p2p.set_property(NM.SETTING_WIFI_P2P_PEER, peer_mac) + s_wifi_p2p.set_property( + NM.SETTING_WIFI_P2P_WFD_IES, + GLib.Bytes.new(b"\x00\x00\x06\x00\x90\x1c\x44\x00\xc8"), + ) + + profile.add_setting(s_con) + profile.add_setting(s_ip4) + profile.add_setting(s_ip6) + profile.add_setting(s_wifi_p2p) + + return profile + + +def activated_cb(client, result, data): + try: + client.add_and_activate_connection2_finish(result) + print(" * Connection profile activated successfully") + except Exception as e: + sys.stderr.write("Error: %s\n" % e) + main_loop.quit() + + +def scan_timeout_cb(device): + peers = device.get_peers() + if len(peers) == 0: + main_loop.quit() + sys.exit("No peer found") + + print("\n {:20} {:30} {:3} {:30}".format("MAC", "Name", "Sig", "Wfd-IEs")) + for p in peers: + if p.get_wfd_ies() is not None: + ies = p.get_wfd_ies().get_data().hex() + else: + ies = "" + print( + " {:20} {:30} {:3} {:30}".format( + p.get_hw_address(), p.get_name(), p.get_strength(), ies + ) + ) + print("") + + # Connect to first peer + profile = create_profile("P2P-connection", peers[0].get_hw_address()) + client.add_and_activate_connection2( + profile, device, "/", GLib.Variant("a{sv}", {}), None, activated_cb, None + ) + print( + " * Connecting to peer {} using profile '{}'".format( + peers[0].get_hw_address(), profile.get_id() + ) + ) + + +def start_find_cb(device, async_result, user_data): + try: + device.start_find_finish(async_result) + except Exception as e: + sys.stderr.write("Error: %s\n" % e) + main_loop.quit() + + print(" * Scanning on device {}...".format(device.get_iface())) + GLib.timeout_add(10000, scan_timeout_cb, device) + + +if __name__ == "__main__": + client = NM.Client.new(None) + device = None + + devices = client.get_devices() + for d in devices: + if d.get_device_type() == NM.DeviceType.WIFI_P2P: + device = d + break + + if device is None: + sys.exit("No Wi-Fi P2P device found") + + device.start_find(GLib.Variant("a{sv}", {}), None, start_find_cb, None) + + main_loop = GLib.MainLoop() + main_loop.run() |