colin/NetworkManager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e860a2e09cc1ae078b51ccda60eec24d5406ec1c
NetworkManager/clients/tests/test-client.py
Thomas Haller 4b09c27d64 cli/tests: cache UUID for test replacement in stdout 
Instead of re-fetching the UUID each time with a D-Bus call,
cache/memoize it.

On my system, this improves

  $ make check-local-clients-tests-test-client

from 20.9 to 20.4 seconds (- 2.6%). That is not stellar, but noticible
enough to warrant this simple change.
2018-07-25 17:08:37 +02:00

1126 lines
44 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python
from __future__ import print_function
###############################################################################
#
# This test starts NetworkManager stub service in a user D-Bus session,
# and runs nmcli against it. The output is recorded and compared to a pre-generated
# expected output (clients/tests/test-client.check-on-disk/*.expected) which
# is also commited to git.
#
###############################################################################
#
# HOWTO: Regenerate output
#
# When adjusting the tests, or when making changes to nmcli that intentionally
# change the output, the expected output must be regenerated.
#
# $ make install
# # (step not required every time)
# # The test also compare the translated output, hence, the translation
# # file must be installed at the configured --prefix.
# # You don't need to type `make install` every time, but a suitable version
# # of translations must be installed. In practice, the tests only care about
# # Polish (pl) translations.
# # The important part is that translations work. Test
# # $ LANG=pl_PL.UTF-8 ./clients/cli/nmcli --version
# # also ensure that `locale -a` reports the Polish locale.
# $ rm -rf clients/tests/test-client.check-on-disk/*.expected
# # (step seldomly required)
# # Sometimes, if you want to be sure that the test would generate
# # exactly the same .expected files, purge the previous version first.
# # This is only necessary, when you remove test from this file.
# $ NM_TEST_REGENERATE=1 make check-local-clients-tests-test-client
# # Set NM_TEST_REGENERATE=1 to regenerate all files.
# $ git diff ... ; git add ...
# # (optional step)
# # Inspect what changed, and whether it makes sense. Then commit changes
# # to git.
#
###############################################################################
#
# Environment variables to configure test:
# (optional) The build dir. Optional, mainly used to find the nmcli binary (in case
# ENV_NM_TEST_CLIENT_NMCLI_PATH is not set.
ENV_NM_TEST_CLIENT_BUILDDIR = 'NM_TEST_CLIENT_BUILDDIR'
# (optional) Path to nmcli. By default, it looks for nmcli in build dir.
# In particular, you can test also a nmcli binary installed somewhere else.
ENV_NM_TEST_CLIENT_NMCLI_PATH = 'NM_TEST_CLIENT_NMCLI_PATH'
# (optional) The test also compares tranlsated output (l10n). This requires,
# that you first install the translation in the right place. So, by default,
# if a test for a translation fails, it will mark the test as skipped, and not
# fail the tests. Under the assumption, that the test cannot succeed currently.
# By setting NM_TEST_CLIENT_CHECK_L10N=1, you can force a failure of the test.
ENV_NM_TEST_CLIENT_CHECK_L10N = 'NM_TEST_CLIENT_CHECK_L10N'
# Regenerate the .expected files. Instead of asserting, rewrite the files
# on disk with the expected output.
ENV_NM_TEST_REGENERATE = 'NM_TEST_REGENERATE'
#
###############################################################################
import sys
try:
import gi
from gi.repository import GLib
gi.require_version('NM', '1.0')
from gi.repository import NM
except Exception as e:
GLib = None
NM = None
import os
import errno
import unittest
import socket
import itertools
import subprocess
import shlex
import re
import dbus
import time
import dbus.service
import dbus.mainloop.glib
###############################################################################
class PathConfiguration:
@staticmethod
def srcdir():
# this is the directory where the test script itself lies.
# Based on this directory, we find other parts that we expect
# in the source repository.
return os.path.dirname(os.path.abspath(__file__))
@staticmethod
def top_srcdir():
return os.path.abspath(PathConfiguration.srcdir() + "/../..")
@staticmethod
def test_networkmanager_service_path():
v = os.path.abspath(PathConfiguration.top_srcdir() + "/tools/test-networkmanager-service.py")
assert os.path.exists(v), ("Cannot find test server at \"%s\"" % (v))
return v
@staticmethod
def canonical_script_filename():
p = 'clients/tests/test-client.py'
assert (PathConfiguration.top_srcdir() + '/' + p) == os.path.abspath(__file__)
return p
###############################################################################
dbus_session_inited = False
_DEFAULT_ARG = object()
_UNSTABLE_OUTPUT = object()
###############################################################################
class Util:
@staticmethod
def python_has_version(major, minor = 0):
return sys.version_info[0] > major \
or ( sys.version_info[0] == major \
and sys.version_info[1] >= minor)
@staticmethod
def is_string(s):
if Util.python_has_version(3):
t = str
else:
t = basestring
return isinstance(s, t)
@staticmethod
def memoize_nullary(nullary_func):
result = []
def closure():
if not result:
result.append(nullary_func())
return result[0]
return closure
_find_unsafe = re.compile(r'[^\w@%+=:,./-]',
re.ASCII if sys.version_info[0] >= 3 else 0).search
@staticmethod
def quote(s):
if Util.python_has_version(3, 3):
return shlex.quote(s)
if not s:
return "''"
if Util._find_unsafe(s) is None:
return s
return "'" + s.replace("'", "'\"'\"'") + "'"
@staticmethod
def popen_wait(p, timeout = None):
# wait() has a timeout argument only since 3.3
if Util.python_has_version(3, 3):
return p.wait(timeout)
if timeout is None:
return p.wait()
start = NM.utils_get_timestamp_msec()
while True:
if p.poll() is not None:
return p.returncode
if start + (timeout * 1000) < NM.utils_get_timestamp_msec():
raise Exception("timeout expired")
time.sleep(0.05)
@staticmethod
def iter_single(itr, min_num = 1, max_num = 1):
itr = list(itr)
n = 0
v = None
for c in itr:
n += 1
if n > 1:
break
v = c
if n < min_num:
raise AssertionError("Expected at least %s elements, but %s found" % (min_num, n))
if n > max_num:
raise AssertionError("Expected at most %s elements, but %s found" % (max_num, n))
return v
@staticmethod
def file_read(filename):
try:
with open(filename, 'rb') as f:
return f.read()
except:
return None
@staticmethod
def replace_text(text, replace_arr):
if not replace_arr:
return text
text = [text]
for replace in replace_arr:
try:
v_search = replace[0]()
except TypeError:
v_search = replace[0]
assert v_search is None or Util.is_string(v_search)
if not v_search:
continue
v_replace = replace[1]
v_search = v_search.encode('utf-8')
v_replace = v_replace.encode('utf-8')
text2 = []
for t in text:
if isinstance(t, tuple):
text2.append(t)
continue
t2 = t.split(v_search)
text2.append(t2[0])
for t3 in t2[1:]:
text2.append( (v_replace,) )
text2.append(t3)
text = text2
return b''.join([(t[0] if isinstance(t, tuple) else t) for t in text])
@staticmethod
def debug_dbus_interface():
# this is for printf debugging, not used in actual code.
os.system('busctl --user --verbose call org.freedesktop.NetworkManager /org/freedesktop org.freedesktop.DBus.ObjectManager GetManagedObjects | cat')
@staticmethod
def iter_nmcli_output_modes():
for mode in [[],
['--mode', 'tabular'],
['--mode', 'multiline']]:
for fmt in [[],
['--pretty'],
['--terse']]:
for color in [[],
['--color', 'yes']]:
yield mode + fmt + color
###############################################################################
class Configuration:
def __init__(self):
self._values = {}
def get(self, name):
v = self._values.get(name, None)
if name in self._values:
return v
if name == ENV_NM_TEST_CLIENT_BUILDDIR:
v = os.environ.get(ENV_NM_TEST_CLIENT_BUILDDIR, PathConfiguration.top_srcdir())
if not os.path.isdir(v):
raise Exception("Missing builddir. Set NM_TEST_CLIENT_BUILDDIR?")
elif name == ENV_NM_TEST_CLIENT_NMCLI_PATH:
v = os.environ.get(ENV_NM_TEST_CLIENT_NMCLI_PATH, None)
if v is None:
try:
v = os.path.abspath(self.get(ENV_NM_TEST_CLIENT_BUILDDIR) + "/clients/cli/nmcli")
except:
pass
if not os.path.exists(v):
raise Exception("Missing nmcli binary. Set NM_TEST_CLIENT_NMCLI_PATH?")
elif name == ENV_NM_TEST_CLIENT_CHECK_L10N:
# if we test locales other than 'C', the output of nmcli depends on whether
# nmcli can load the translations. Unfortunately, I cannot find a way to
# make gettext use the po/*.gmo files from the build-dir.
#
# hence, such tests only work, if you also issue `make-install`
#
# Only by setting NM_TEST_CLIENT_CHECK_L10N=1, these tests are included
# as well.
v = (os.environ.get(ENV_NM_TEST_CLIENT_CHECK_L10N, '0') == '1')
elif name == ENV_NM_TEST_REGENERATE:
# in the "regenerate" mode, the tests will rewrite the files on disk against
# which we assert. That is useful, if there are intentional changes and
# we want to regenerate the expected output.
v = (os.environ.get(ENV_NM_TEST_REGENERATE, '0') == '1')
else:
raise Exception()
self._values[name] = v
return v
conf = Configuration()
###############################################################################
class NMStubServer:
@staticmethod
def _conn_get_main_object(conn):
try:
return conn.get_object('org.freedesktop.NetworkManager', '/org/freedesktop/NetworkManager')
except:
return None
def __init__(self, seed):
service_path = PathConfiguration.test_networkmanager_service_path()
self._conn = dbus.SessionBus()
env = os.environ.copy()
env['NM_TEST_NETWORKMANAGER_SERVICE_SEED'] = seed
p = subprocess.Popen([sys.executable, service_path],
stdin = subprocess.PIPE,
env = env)
start = NM.utils_get_timestamp_msec()
while True:
if p.poll() is not None:
p.stdin.close()
if p.returncode == 77:
raise unittest.SkipTest('the stub service %s exited with status 77' % (service_path))
raise Exception('the stub service %s exited unexpectedly' % (service_path))
nmobj = self._conn_get_main_object(self._conn)
if nmobj is not None:
break
if (NM.utils_get_timestamp_msec() - start) >= 2000:
p.stdin.close()
p.kill()
Util.popen_wait(p, 1000)
raise Exception("after starting stub service the D-Bus name was not claimed in time")
self._nmobj = nmobj
self._nmiface = dbus.Interface(nmobj, "org.freedesktop.NetworkManager.LibnmGlibTest")
self._p = p
def shutdown(self):
self._nmobj = None
self._nmiface = None
self._conn = None
self._p.stdin.close()
self._p.kill()
Util.popen_wait(self._p, 1000)
self._p = None
if self._conn_get_main_object(self._conn) is not None:
raise Exception("Stub service is not still here although it should shut down")
class _MethodProxy:
def __init__(self, parent, method_name):
self._parent = parent
self._method_name = method_name
def __call__(self, *args, **kwargs):
dbus_iface = kwargs.pop('dbus_iface', None)
if dbus_iface is None:
dbus_iface = self._parent._nmiface
method = dbus_iface.get_dbus_method(self._method_name)
if kwargs:
# for convenience, we allow the caller to specify arguments
# as kwargs. In this case, we construct a a{sv} array as last argument.
kwargs2 = {}
args = list(args)
args.append(kwargs2)
for k in kwargs.keys():
kwargs2[k] = kwargs[k]
return method(*args)
def __getattr__(self, member):
if not member.startswith("op_"):
raise AttributeError(member)
return self._MethodProxy(self, member[3:])
def addConnection(self, connection, do_verify_strict = True):
return self.op_AddConnection(connection, do_verify_strict)
def findConnectionUuid(self, con_id, required = True):
try:
u = Util.iter_single(self.op_FindConnections(con_id = con_id))[1]
assert u, ("Invalid uuid %s" % (u))
except Exception as e:
if not required:
return None
raise AssertionError("Unexpectedly not found connection %s: %s" % (con_id, str(e)))
return u
def setProperty(self, path, propname, value, iface_name = None):
if iface_name is None:
iface_name = ''
self.op_SetProperties([
(path, [
(iface_name, [
(propname, value),
]),
]),
])
###############################################################################
class AsyncProcess():
def __init__(self,
args,
env,
complete_cb):
self._args = args
self._env = env
self._complete_cb = complete_cb
def start(self):
if not hasattr(self, '_p'):
self._p = subprocess.Popen(self._args,
stdout = subprocess.PIPE,
stderr = subprocess.PIPE,
env = self._env)
def wait(self):
self.start()
Util.popen_wait(self._p, 2000)
(returncode, stdout, stderr) = (self._p.returncode, self._p.stdout.read(), self._p.stderr.read())
self._p.stdout.close()
self._p.stderr.close()
self._p = None
self._complete_cb(self, returncode, stdout, stderr)
###############################################################################
class NmTestBase(unittest.TestCase):
pass
class TestNmcli(NmTestBase):
@staticmethod
def _read_expected(filename):
results_expect = []
content_expect = Util.file_read(filename)
try:
base_idx = 0
size_prefix = 'size: '.encode('utf8')
while True:
if not content_expect[base_idx:base_idx + 10].startswith(size_prefix):
raise Exception("Unexpected token")
j = base_idx + len(size_prefix)
i = j
if Util.python_has_version(3, 0):
eol = ord('\n')
else:
eol = '\n'
while content_expect[i] != eol:
i += 1
i = i + 1 + int(content_expect[j:i])
results_expect.append(content_expect[base_idx:i])
if len(content_expect) == i:
break
base_idx = i
except Exception as e:
results_expect = None
return content_expect, results_expect
def call_nmcli_l(self,
args,
check_on_disk = _DEFAULT_ARG,
fatal_warnings = _DEFAULT_ARG,
expected_returncode = _DEFAULT_ARG,
expected_stdout = _DEFAULT_ARG,
expected_stderr = _DEFAULT_ARG,
replace_stdout = None,
replace_stderr = None,
sort_lines_stdout = False,
extra_env = None,
sync_barrier = False):
frame = sys._getframe(1)
for lang in [ 'C', 'pl' ]:
self._call_nmcli(args,
lang,
check_on_disk,
fatal_warnings,
expected_returncode,
expected_stdout,
expected_stderr,
replace_stdout,
replace_stderr,
sort_lines_stdout,
extra_env,
sync_barrier,
frame)
def call_nmcli(self,
args,
langs = None,
lang = None,
check_on_disk = _DEFAULT_ARG,
fatal_warnings = _DEFAULT_ARG,
expected_returncode = _DEFAULT_ARG,
expected_stdout = _DEFAULT_ARG,
expected_stderr = _DEFAULT_ARG,
replace_stdout = None,
replace_stderr = None,
sort_lines_stdout = False,
extra_env = None,
sync_barrier = None):
frame = sys._getframe(1)
if langs is not None:
assert lang is None
else:
if lang is None:
lang = 'C'
langs = [lang]
if sync_barrier is None:
sync_barrier = (len(langs) == 1)
for lang in langs:
self._call_nmcli(args,
lang,
check_on_disk,
fatal_warnings,
expected_returncode,
expected_stdout,
expected_stderr,
replace_stdout,
replace_stderr,
sort_lines_stdout,
extra_env,
sync_barrier,
frame)
def _call_nmcli(self,
args,
lang,
check_on_disk,
fatal_warnings,
expected_returncode,
expected_stdout,
expected_stderr,
replace_stdout,
replace_stderr,
sort_lines_stdout,
extra_env,
sync_barrier,
frame):
if sync_barrier:
self.async_wait()
calling_fcn = frame.f_code.co_name
calling_num = self._calling_num.get(calling_fcn, 0) + 1
self._calling_num[calling_fcn] = calling_num
test_name = '%s-%03d' % (calling_fcn, calling_num)
# we cannot use frame.f_code.co_filename directly, because it might be different depending
# on where the file lies and which is CWD. We still want to give the location of
# the file, so that the user can easier find the source (when looking at the .expected files)
self.assertTrue(os.path.abspath(frame.f_code.co_filename).endswith('/'+PathConfiguration.canonical_script_filename()))
calling_location = '%s:%d:%s()/%d' % (PathConfiguration.canonical_script_filename(), frame.f_lineno, frame.f_code.co_name, calling_num)
if lang is None or lang == 'C':
lang = 'C'
language = ''
elif lang is 'de':
lang = 'de_DE.utf8'
language = 'de'
elif lang is 'pl':
lang = 'pl_PL.UTF-8'
language = 'pl'
else:
self.fail('invalid language %s' % (lang))
env = {}
if extra_env is not None:
for k, v in extra_env.items():
env[k] = v
for k in ['LD_LIBRARY_PATH',
'DBUS_SESSION_BUS_ADDRESS']:
val = os.environ.get(k, None)
if val is not None:
env[k] = val
env['LANG'] = lang
env['LANGUAGE'] = language
env['LIBNM_USE_SESSION_BUS'] = '1'
env['LIBNM_USE_NO_UDEV'] = '1'
env['TERM'] = 'linux'
env['XDG_CONFIG_HOME'] = PathConfiguration.srcdir()
if fatal_warnings is _DEFAULT_ARG or fatal_warnings:
env['G_DEBUG'] = 'fatal-warnings'
args = [conf.get(ENV_NM_TEST_CLIENT_NMCLI_PATH)] + list(args)
if replace_stdout is not None:
replace_stdout = list(replace_stdout)
if replace_stderr is not None:
replace_stderr = list(replace_stderr)
if check_on_disk is _DEFAULT_ARG:
check_on_disk = ( expected_returncode is _DEFAULT_ARG
and (expected_stdout is _DEFAULT_ARG or expected_stdout is _UNSTABLE_OUTPUT)
and (expected_stderr is _DEFAULT_ARG or expected_stderr is _UNSTABLE_OUTPUT))
if expected_returncode is _DEFAULT_ARG:
expected_returncode = None
if expected_stdout is _DEFAULT_ARG:
expected_stdout = None
if expected_stderr is _DEFAULT_ARG:
expected_stderr = None
def complete_cb(async_job,
returncode,
stdout,
stderr):
if expected_stdout is _UNSTABLE_OUTPUT:
stdout = '<UNSTABLE OUTPUT>'.encode('utf-8')
else:
stdout = Util.replace_text(stdout, replace_stdout)
if expected_stderr is _UNSTABLE_OUTPUT:
stderr = '<UNSTABLE OUTPUT>'.encode('utf-8')
else:
stderr = Util.replace_text(stderr, replace_stderr)
if sort_lines_stdout:
stdout = b'\n'.join(sorted(stdout.split(b'\n')))
ignore_l10n_diff = ( lang != 'C'
and not conf.get(ENV_NM_TEST_CLIENT_CHECK_L10N))
if expected_stderr is not None and expected_stderr is not _UNSTABLE_OUTPUT:
if expected_stderr != stderr:
if ignore_l10n_diff:
self._skip_test_for_l10n_diff.append(test_name)
else:
self.assertEqual(expected_stderr, stderr)
if expected_stdout is not None and expected_stdout is not _UNSTABLE_OUTPUT:
if expected_stdout != stdout:
if ignore_l10n_diff:
self._skip_test_for_l10n_diff.append(test_name)
else:
self.assertEqual(expected_stdout, stdout)
if expected_returncode is not None:
self.assertEqual(expected_returncode, returncode)
if fatal_warnings is _DEFAULT_ARG:
if expected_returncode != -5:
self.assertNotEqual(returncode, -5)
elif fatal_warnings:
if expected_returncode is None:
self.assertEqual(returncode, -5)
if check_on_disk:
cmd = '$NMCLI %s' % (' '.join([Util.quote(a) for a in args[1:]])),
content = ('location: %s\n' % (calling_location)).encode('utf8') + \
('cmd: %s\n' % (cmd)).encode('utf8') + \
('lang: %s\n' % (lang)).encode('utf8') + \
('returncode: %d\n' % (returncode)).encode('utf8')
if len(stdout) > 0:
content += ('stdout: %d bytes\n>>>\n' % (len(stdout))).encode('utf8') + \
stdout + \
'\n<<<\n'.encode('utf8')
if len(stderr) > 0:
content += ('stderr: %d bytes\n>>>\n' % (len(stderr))).encode('utf8') + \
stderr + \
'\n<<<\n'.encode('utf8')
content = ('size: %s\n' % (len(content))).encode('utf8') + \
content
self._results.append({
'test_name' : test_name,
'ignore_l10n_diff' : ignore_l10n_diff,
'content' : content,
})
async_job = AsyncProcess(args = args,
env = env,
complete_cb = complete_cb)
self._async_jobs.append(async_job)
if sync_barrier:
self.async_wait()
else:
self.async_start()
def async_start(self):
# limit number parallel running jobs
for async_job in self._async_jobs[0:15]:
async_job.start()
def async_wait(self):
while self._async_jobs:
self.async_start()
self._async_jobs.pop(0).wait()
def _nm_test_pre(self):
self._calling_num = {}
self._skip_test_for_l10n_diff = []
self._async_jobs = []
self._results = []
self.srv = NMStubServer(self._testMethodName)
def _nm_test_post(self):
self.async_wait()
self.srv.shutdown()
self.srv = None
self._calling_num = None
results = self._results
self._results = None
skip_test_for_l10n_diff = self._skip_test_for_l10n_diff
self._skip_test_for_l10n_diff = None
test_name = self._testMethodName
filename = os.path.abspath(PathConfiguration.srcdir() + '/test-client.check-on-disk/' + test_name + '.expected')
regenerate = conf.get(ENV_NM_TEST_REGENERATE)
content_expect, results_expect = self._read_expected(filename)
if results_expect is None:
if not regenerate:
self.fail("Failed to parse expected file '%s'. Let the test write the file by rerunning with NM_TEST_REGENERATE=1" % (filename))
else:
for i in range(0, min(len(results_expect), len(results))):
n = results[i]
if results_expect[i] == n['content']:
continue
if regenerate:
continue
if n['ignore_l10n_diff']:
skip_test_for_l10n_diff.append(n['test_name'])
continue
print("\n\n\nThe file '%s' does not have the expected content:" % (filename))
print("ACTUAL OUTPUT:\n[[%s]]\n" % (results_expect[i]))
print("EXPECT OUTPUT:\n[[%s]]\n" % (n['content']))
print("Let the test write the file by rerunning with NM_TEST_REGENERATE=1")
print("See howto in %s for details.\n" % (PathConfiguration.canonical_script_filename()))
self.fail("Unexpected output of command, expected %s. Rerun test with NM_TEST_REGENERATE=1 to regenerate files" % (filename))
if len(results_expect) != len(results):
if not regenerate:
print("\n\n\nThe number of tests in %s does not match the expected content (%s vs %s):" % (filename, len(results_expect), len(results)))
if len(results_expect) < len(results):
print("ACTUAL OUTPUT:\n[[%s]]\n" % (results[len(results_expect)]['content']))
else:
print("EXPECT OUTPUT:\n[[%s]]\n" % (results_expect[len(results)]))
print("Let the test write the file by rerunning with NM_TEST_REGENERATE=1")
print("See howto in %s for details.\n" % (PathConfiguration.canonical_script_filename()))
self.fail("Unexpected output of command, expected %s. Rerun test with NM_TEST_REGENERATE=1 to regenerate files" % (filename))
if regenerate:
content_new = ''.join([r['content'] for r in results])
if content_new != content_expect:
try:
with open(filename, 'wb') as content_file:
content_file.write(content_new)
except Exception as e:
self.fail("Failure to write '%s': %s" % (filename, e))
if skip_test_for_l10n_diff:
# nmcli loads translations from the installation path. This failure commonly
# happens because you did not install the binary in the --prefix, before
# running the test. Hence, translations are not available or differ.
self.skipTest("Skipped asserting for localized tests %s. Set NM_TEST_CLIENT_CHECK_L10N=1 to force fail." % (','.join(skip_test_for_l10n_diff)))
def nm_test(func):
def f(self):
self._nm_test_pre()
func(self)
self._nm_test_post()
return f
def setUp(self):
if not dbus_session_inited:
self.skipTest("Own D-Bus session for testing is not initialized. Do you have dbus-run-session available?")
if NM is None:
self.skipTest("gi.NM is not available. Did you build with introspection?")
def init_001(self):
self.srv.op_AddObj('WiredDevice',
iface = 'eth0')
self.srv.op_AddObj('WiredDevice',
iface = 'eth1')
self.srv.op_AddObj('WifiDevice',
iface = 'wlan0')
self.srv.op_AddObj('WifiDevice',
iface = 'wlan1')
# add another device with an identical ifname. The D-Bus API itself
# does not enforce the ifnames are unique.
self.srv.op_AddObj('WifiDevice',
ident = 'wlan1/x',
iface = 'wlan1')
self.srv.op_AddObj('WifiAp',
device = 'wlan0')
self.srv.op_AddObj('WifiAp',
device = 'wlan0')
self.srv.op_AddObj('WifiAp',
device = 'wlan0')
self.srv.op_AddObj('WifiAp',
device = 'wlan1')
self.srv.addConnection( {
'connection': {
'type': '802-3-ethernet',
'id': 'con-1',
},
})
@nm_test
def test_001(self):
self.call_nmcli_l([])
self.call_nmcli_l(['-f', 'AP', '-mode', 'multiline', '-p', 'd', 'show', 'wlan0'])
self.call_nmcli_l(['c', 's'])
self.call_nmcli_l(['bogus', 's'])
for mode in Util.iter_nmcli_output_modes():
self.call_nmcli_l(mode + ['general', 'permissions'])
@nm_test
def test_002(self):
self.init_001()
self.call_nmcli_l(['d'])
self.call_nmcli_l(['-f', 'all', 'd'])
self.call_nmcli_l([])
self.call_nmcli_l(['-f', 'AP', '-mode', 'multiline', 'd', 'show', 'wlan0'])
self.call_nmcli_l(['-f', 'AP', '-mode', 'multiline', '-p', 'd', 'show', 'wlan0'])
self.call_nmcli_l(['-f', 'AP', '-mode', 'multiline', '-t', 'd', 'show', 'wlan0'])
self.call_nmcli_l(['-f', 'AP', '-mode', 'tabular', 'd', 'show', 'wlan0'])
self.call_nmcli_l(['-f', 'AP', '-mode', 'tabular', '-p', 'd', 'show', 'wlan0'])
self.call_nmcli_l(['-f', 'AP', '-mode', 'tabular', '-t', 'd', 'show', 'wlan0'])
self.call_nmcli_l(['-f', 'ALL', 'd', 'wifi'])
self.call_nmcli_l(['c'])
self.call_nmcli_l(['c', 's', 'con-1'])
@nm_test
def test_003(self):
self.init_001()
replace_stdout = []
replace_stdout.append((Util.memoize_nullary(lambda: self.srv.findConnectionUuid('con-xx1')), 'UUID-con-xx1-REPLACED-REPLACED-REPLA'))
self.call_nmcli(['c', 'add', 'type', 'ethernet', 'ifname', '*', 'con-name', 'con-xx1'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['c', 's'],
replace_stdout = replace_stdout)
replace_stdout.append((Util.memoize_nullary(lambda: self.srv.findConnectionUuid('ethernet')), 'UUID-ethernet-REPLACED-REPLACED-REPL'))
self.call_nmcli(['c', 'add', 'type', 'ethernet', 'ifname', '*'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['c', 's'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['-f', 'ALL', 'c', 's'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['--complete-args', '-f', 'ALL', 'c', 's', ''],
replace_stdout = replace_stdout,
sort_lines_stdout = True)
# activate the same profile on multiple devices. Our stub-implmentation
# is fine with that... although NetworkManager service would reject
# such a configuration by deactivating the profile first. But note that
# that is only an internal behavior of NetworkManager service. The D-Bus
# API perfectly allows for one profile to be active multiple times. Also
# note, that there is always a short time where one profile goes down,
# while another is activating. Hence, while real NetworkManager commonly
# does not allow that multiple profiles *stay* connected at the same
# time, there is always the possibility that a profile is activating/active
# on a device, while also activating/deactivating in parallel.
for dev in ['eth0', 'eth1']:
self.call_nmcli(['con', 'up', 'ethernet', 'ifname', dev])
self.call_nmcli_l(['con'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['-f', 'ALL', 'con'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['-f', 'ALL', 'con', 's', '-a'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['-f', 'ACTIVE-PATH,DEVICE,UUID', 'con', 's', '-act'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['-f', 'UUID,NAME', 'con', 's', '--active'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['-f', 'ALL', 'con', 's', 'ethernet'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['-f', 'GENERAL.STATE', 'con', 's', 'ethernet'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['con', 's', 'ethernet'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['-f', 'ALL', 'dev', 's', 'eth0'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['-f', 'ALL', 'dev', 'show', 'eth0'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['-f', 'ALL', '-t', 'dev', 'show', 'eth0'],
replace_stdout = replace_stdout)
self.async_wait()
self.srv.setProperty('/org/freedesktop/NetworkManager/ActiveConnection/1',
'State',
dbus.UInt32(NM.ActiveConnectionState.DEACTIVATING))
for i in [0, 1]:
if i == 1:
self.async_wait()
self.srv.op_ConnectionSetVisible(False, con_id = 'ethernet')
for mode in Util.iter_nmcli_output_modes():
self.call_nmcli_l(mode + ['-f', 'ALL', 'con'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'UUID,TYPE', 'con'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['con', 's', 'ethernet'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['c', 's', '/org/freedesktop/NetworkManager/ActiveConnection/1'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'all', 'dev', 'show', 'eth0'],
replace_stdout = replace_stdout)
@nm_test
def test_004(self):
self.init_001()
replace_stdout = []
replace_stdout.append((Util.memoize_nullary(lambda: self.srv.findConnectionUuid('con-xx1')), 'UUID-con-xx1-REPLACED-REPLACED-REPLA'))
self.call_nmcli(['c', 'add', 'type', 'wifi', 'ifname', '*', 'ssid', 'foobar', 'con-name', 'con-xx1'],
replace_stdout = replace_stdout)
self.call_nmcli(['connection', 'mod', 'con-xx1', 'ip.gateway', ''])
self.call_nmcli(['connection', 'mod', 'con-xx1', 'ipv4.gateway', '172.16.0.1'], lang = 'pl')
self.call_nmcli(['connection', 'mod', 'con-xx1', 'ipv6.gateway', '::99'])
self.call_nmcli(['connection', 'mod', 'con-xx1', '802.abc', ''])
self.call_nmcli(['connection', 'mod', 'con-xx1', '802-11-wireless.band', 'a'])
self.call_nmcli(['connection', 'mod', 'con-xx1', 'ipv4.addresses', '192.168.77.5/24', 'ipv4.routes', '2.3.4.5/32 192.168.77.1', 'ipv6.addresses', '1:2:3:4::6/64', 'ipv6.routes', '1:2:3:4:5:6::5/128'])
self.call_nmcli_l(['con', 's', 'con-xx1'],
replace_stdout = replace_stdout)
self.async_wait()
replace_stdout.append((Util.memoize_nullary(lambda: self.srv.findConnectionUuid('con-vpn-1')), 'UUID-con-vpn-1-REPLACED-REPLACED-REP'))
self.call_nmcli(['connection', 'add', 'type', 'vpn', 'con-name', 'con-vpn-1', 'ifname', '*', 'vpn-type', 'openvpn', 'vpn.data', 'key1 = val1, key2 = val2, key3=val3'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['con', 's'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['con', 's', 'con-vpn-1'],
replace_stdout = replace_stdout)
self.call_nmcli(['con', 'up', 'con-xx1'])
self.call_nmcli_l(['con', 's'],
replace_stdout = replace_stdout)
self.call_nmcli(['con', 'up', 'con-vpn-1'])
self.call_nmcli_l(['con', 's'],
replace_stdout = replace_stdout)
self.call_nmcli_l(['con', 's', 'con-vpn-1'],
replace_stdout = replace_stdout)
self.async_wait()
self.srv.setProperty('/org/freedesktop/NetworkManager/ActiveConnection/2',
'VpnState',
dbus.UInt32(NM.VpnConnectionState.ACTIVATED))
for mode in Util.iter_nmcli_output_modes():
self.call_nmcli_l(mode + ['con', 's', 'con-vpn-1'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['con', 's', 'con-vpn-1'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'ALL', 'con', 's', 'con-vpn-1'],
replace_stdout = replace_stdout)
# This only filters 'vpn' settings from the connection profile.
# Contrary to '-f GENERAL' below, it does not show the properties of
# the activated VPN connection. This is a nmcli bug.
self.call_nmcli_l(mode + ['-f', 'VPN', 'con', 's', 'con-vpn-1'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'GENERAL', 'con', 's', 'con-vpn-1'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['dev', 's'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'all', 'dev', 'status'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['dev', 'show'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'all', 'dev', 'show'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['dev', 'show', 'wlan0'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'all', 'dev', 'show', 'wlan0'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'GENERAL,GENERAL.HWADDR,WIFI-PROPERTIES', 'dev', 'show', 'wlan0'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'GENERAL,GENERAL.HWADDR,WIFI-PROPERTIES', 'dev', 'show', 'wlan0'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'DEVICE,TYPE,DBUS-PATH', 'dev'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'ALL', 'device', 'wifi', 'list' ],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'COMMON', 'device', 'wifi', 'list' ],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'NAME,SSID,SSID-HEX,BSSID,MODE,CHAN,FREQ,RATE,SIGNAL,BARS,SECURITY,WPA-FLAGS,RSN-FLAGS,DEVICE,ACTIVE,IN-USE,DBUS-PATH',
'device', 'wifi', 'list'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'ALL', 'device', 'wifi', 'list', 'bssid', 'C0:E2:BE:E8:EF:B6'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'COMMON', 'device', 'wifi', 'list', 'bssid', 'C0:E2:BE:E8:EF:B6'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'NAME,SSID,SSID-HEX,BSSID,MODE,CHAN,FREQ,RATE,SIGNAL,BARS,SECURITY,WPA-FLAGS,RSN-FLAGS,DEVICE,ACTIVE,IN-USE,DBUS-PATH',
'device', 'wifi', 'list', 'bssid', 'C0:E2:BE:E8:EF:B6'],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'ALL', 'device', 'show', 'wlan0' ],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'COMMON', 'device', 'show', 'wlan0' ],
replace_stdout = replace_stdout)
self.call_nmcli_l(mode + ['-f', 'GENERAL,CAPABILITIES,WIFI-PROPERTIES,AP,WIRED-PROPERTIES,WIMAX-PROPERTIES,NSP,IP4,DHCP4,IP6,DHCP6,BOND,TEAM,BRIDGE,VLAN,BLUETOOTH,CONNECTIONS', 'device', 'show', 'wlan0' ],
replace_stdout = replace_stdout)
###############################################################################
def main():
global dbus_session_inited
if len(sys.argv) >= 2 and sys.argv[1] == '--started-with-dbus-session':
dbus_session_inited = True
del sys.argv[1]
if not dbus_session_inited:
# we don't have yet our own dbus-session. Reexec ourself with
# a new dbus-session.
try:
try:
os.execlp('dbus-run-session', 'dbus-run-session', '--', sys.executable, __file__, '--started-with-dbus-session', *sys.argv[1:])
except OSError as e:
if e.errno != errno.ENOENT:
raise
# we have no dbus-run-session in path? Fall-through
# to skip tests gracefully
else:
raise Exception('unknown error during exec')
except Exception as e:
assert False, ("Failure to re-exec dbus-run-session: %s" % (str(e)))
if not dbus_session_inited:
# we still don't have a D-Bus session. Probably dbus-run-session is not available.
# retry with dbus-launch
if os.system('type dbus-launch 1>/dev/null') == 0:
try:
os.execlp('bash', 'bash', '-e', '-c',
'eval `dbus-launch --sh-syntax`;\n' + \
'trap "kill $DBUS_SESSION_BUS_PID" EXIT;\n' + \
'\n' + \
' '.join([Util.quote(a) for a in [sys.executable, __file__, '--started-with-dbus-session'] + sys.argv[1:]]) + ' \n' + \
'')
except Exception as e:
m = str(e)
else:
m = 'unknown error'
assert False, ('Failure to re-exec to start script with dbus-launch: %s' % (m))
r = unittest.main(exit = False)
sys.exit(not r.result.wasSuccessful())
if __name__ == '__main__':
main()
Reference in New Issue View Git Blame Copy Permalink