243
259
return rfc3339_duration_to_delta(interval)
244
260
except ValueError as e:
245
261
log.warning("%s - Parsing as pre-1.6.1 interval instead",
247
263
return parse_pre_1_6_1_interval(interval)
250
266
def rfc3339_duration_to_delta(duration):
251
267
"""Parse an RFC 3339 "duration" and return a datetime.timedelta
253
>>> rfc3339_duration_to_delta("P7D")
254
datetime.timedelta(7)
255
>>> rfc3339_duration_to_delta("PT60S")
256
datetime.timedelta(0, 60)
257
>>> rfc3339_duration_to_delta("PT60M")
258
datetime.timedelta(0, 3600)
259
>>> rfc3339_duration_to_delta("P60M")
260
datetime.timedelta(1680)
261
>>> rfc3339_duration_to_delta("PT24H")
262
datetime.timedelta(1)
263
>>> rfc3339_duration_to_delta("P1W")
264
datetime.timedelta(7)
265
>>> rfc3339_duration_to_delta("PT5M30S")
266
datetime.timedelta(0, 330)
267
>>> rfc3339_duration_to_delta("P1DT3M20S")
268
datetime.timedelta(1, 200)
269
>>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)
271
>>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)
273
>>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)
276
>>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)
278
>>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)
280
>>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)
282
>>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)
284
>>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)
269
286
>>> # Can not be empty:
270
287
>>> rfc3339_duration_to_delta("")
271
288
Traceback (most recent call last):
380
397
def parse_pre_1_6_1_interval(interval):
381
"""Parse an interval string as documented by Mandos before 1.6.1,
398
r"""Parse an interval string as documented by Mandos before 1.6.1,
382
399
and return a datetime.timedelta
384
>>> parse_pre_1_6_1_interval('7d')
385
datetime.timedelta(7)
386
>>> parse_pre_1_6_1_interval('60s')
387
datetime.timedelta(0, 60)
388
>>> parse_pre_1_6_1_interval('60m')
389
datetime.timedelta(0, 3600)
390
>>> parse_pre_1_6_1_interval('24h')
391
datetime.timedelta(1)
392
>>> parse_pre_1_6_1_interval('1w')
393
datetime.timedelta(7)
394
>>> parse_pre_1_6_1_interval('5m 30s')
395
datetime.timedelta(0, 330)
396
>>> parse_pre_1_6_1_interval('')
397
datetime.timedelta(0)
401
>>> parse_pre_1_6_1_interval("7d") == datetime.timedelta(days=7)
403
>>> parse_pre_1_6_1_interval("60s") == datetime.timedelta(0, 60)
405
>>> parse_pre_1_6_1_interval("60m") == datetime.timedelta(hours=1)
407
>>> parse_pre_1_6_1_interval("24h") == datetime.timedelta(days=1)
409
>>> parse_pre_1_6_1_interval("1w") == datetime.timedelta(days=7)
411
>>> parse_pre_1_6_1_interval("5m 30s") == datetime.timedelta(0, 330)
413
>>> parse_pre_1_6_1_interval("") == datetime.timedelta(0)
398
415
>>> # Ignore unknown characters, allow any order and repetitions
399
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
400
datetime.timedelta(2, 480, 18000)
416
>>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") \
417
... == datetime.timedelta(2, 480, 18000)
665
690
class CachingBus(SystemBus):
666
691
"""A caching layer for pydbus_adapter.SystemBus"""
667
693
def __init__(self, *args, **kwargs):
668
694
self.object_cache = {}
669
695
super(pydbus_adapter.CachingBus,
670
696
self).__init__(*args, **kwargs)
671
698
def get(self, busname, objectpath):
673
700
return self.object_cache[(busname, objectpath)]
675
702
new_object = (super(pydbus_adapter.CachingBus, self)
676
703
.get(busname, objectpath))
677
self.object_cache[(busname, objectpath)] = new_object
704
self.object_cache[(busname, objectpath)] = new_object
708
class dbussy_adapter:
709
class SystemBus(dbus.SystemBus):
712
def __init__(self, dbussy, ravel):
715
self.bus = ravel.system_bus()
717
@contextlib.contextmanager
718
def convert_exception(self, exception_class=dbus.Error):
721
except self.dbussy.DBusError as e:
722
# This does what "raise from" would do
723
exc = exception_class(*e.args)
727
def call_method(self, methodname, busname, objectpath,
729
proxy_object = self.get_object(busname, objectpath)
730
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
731
interface, methodname,
732
", ".join(repr(a) for a in args))
733
iface = proxy_object.get_interface(interface)
734
method = getattr(iface, methodname)
735
with self.convert_exception(dbus.Error):
736
value = method(*args)
737
# DBussy returns values either as an empty list or as a
738
# list of one element with the return value
740
return self.type_filter(value[0])
742
def get_object(self, busname, objectpath):
743
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
745
with self.convert_exception(dbus.ConnectFailed):
746
return self.bus[busname][objectpath]
748
def type_filter(self, value):
749
"""Convert the most bothersome types to Python types"""
750
# A D-Bus Variant value is represented as the Python type
751
# Tuple[dbussy.DBUS.Signature, Any]
752
if isinstance(value, tuple):
754
and isinstance(value[0],
755
self.dbussy.DBUS.Signature)):
756
return self.type_filter(value[1])
757
elif isinstance(value, self.dbussy.DBUS.ObjectPath):
759
# Also recurse into dictionaries
760
elif isinstance(value, dict):
761
return {self.type_filter(key):
762
self.type_filter(subval)
763
for key, subval in value.items()}
766
def set_property(self, busname, objectpath, interface, key,
768
proxy_object = self.get_object(busname, objectpath)
769
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
770
objectpath, self.properties_iface, interface,
773
# DBussy wants a Byte Array to be a sequence of
774
# values, not a byte string
776
setattr(proxy_object.get_interface(interface), key, value)
778
class MandosBus(SystemBus, dbus.MandosBus):
781
class CachingBus(MandosBus):
782
"""A caching layer for dbussy_adapter.MandosBus"""
784
def __init__(self, *args, **kwargs):
785
self.object_cache = {}
786
super(dbussy_adapter.CachingBus, self).__init__(*args,
789
def get_object(self, busname, objectpath):
791
return self.object_cache[(busname, objectpath)]
794
dbussy_adapter.CachingBus,
795
self).get_object(busname, objectpath)
796
self.object_cache[(busname, objectpath)] = new_object
678
797
return new_object
1770
1866
self.assertIs(obj1, obj1b)
1869
class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):
1873
class ObjectPath(str):
1875
class DBusError(Exception):
1878
def fake_ravel_func(self, func):
1882
class DBusInterfaceProxy:
1884
def methodname(*args):
1885
return [func(*args)]
1888
def get_interface(interface):
1889
if interface == "interface":
1890
return DBusInterfaceProxy()
1891
return {"busname": {"objectpath": DBusObject()}}
1894
def call_method(self, bus, methodname, busname, objectpath,
1896
with self.assertLogs(log, logging.DEBUG):
1897
return bus.call_method(methodname, busname, objectpath,
1900
def test_call_method_returns(self):
1901
expected_method_return = Unique()
1902
method_args = (Unique(), Unique())
1904
self.assertEqual(len(method_args), len(args))
1905
for marg, arg in zip(method_args, args):
1906
self.assertIs(marg, arg)
1907
return expected_method_return
1908
fake_ravel = self.fake_ravel_func(func)
1909
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1910
ret = self.call_method(bus, "methodname", "busname",
1911
"objectpath", "interface",
1913
self.assertIs(ret, expected_method_return)
1915
def test_call_method_filters_objectpath(self):
1917
return method_return
1918
fake_ravel = self.fake_ravel_func(func)
1919
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1920
method_return = (self.dummy_dbussy.DBUS
1921
.ObjectPath("objectpath"))
1922
ret = self.call_method(bus, "methodname", "busname",
1923
"objectpath", "interface")
1924
self.assertEqual("objectpath", ret)
1925
self.assertNotIsInstance(ret,
1926
self.dummy_dbussy.DBUS.ObjectPath)
1928
def test_call_method_filters_objectpaths_in_dict(self):
1929
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1931
return method_return
1932
fake_ravel = self.fake_ravel_func(func)
1933
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1935
ObjectPath("objectpath_key_1"):
1936
ObjectPath("objectpath_value_1"),
1937
ObjectPath("objectpath_key_2"):
1938
ObjectPath("objectpath_value_2"),
1940
ret = self.call_method(bus, "methodname", "busname",
1941
"objectpath", "interface")
1942
expected_method_return = {str(key): str(value)
1944
method_return.items()}
1945
for key, value in ret.items():
1946
self.assertNotIsInstance(key, ObjectPath)
1947
self.assertNotIsInstance(value, ObjectPath)
1948
self.assertEqual(expected_method_return, ret)
1949
self.assertIsInstance(ret, dict)
1951
def test_call_method_filters_objectpaths_in_dict_in_dict(self):
1952
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1954
return method_return
1955
fake_ravel = self.fake_ravel_func(func)
1956
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1958
ObjectPath("key1"): {
1959
ObjectPath("key11"): ObjectPath("value11"),
1960
ObjectPath("key12"): ObjectPath("value12"),
1962
ObjectPath("key2"): {
1963
ObjectPath("key21"): ObjectPath("value21"),
1964
ObjectPath("key22"): ObjectPath("value22"),
1967
ret = self.call_method(bus, "methodname", "busname",
1968
"objectpath", "interface")
1969
expected_method_return = {
1970
"key1": {"key11": "value11",
1971
"key12": "value12"},
1972
"key2": {"key21": "value21",
1973
"key22": "value22"},
1975
self.assertEqual(expected_method_return, ret)
1976
for key, value in ret.items():
1977
self.assertIsInstance(value, dict)
1978
self.assertEqual(expected_method_return[key], value)
1979
self.assertNotIsInstance(key, ObjectPath)
1980
for inner_key, inner_value in value.items():
1981
self.assertIsInstance(value, dict)
1983
expected_method_return[key][inner_key],
1985
self.assertNotIsInstance(key, ObjectPath)
1987
def test_call_method_filters_objectpaths_in_dict_three_deep(self):
1988
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1990
return method_return
1991
fake_ravel = self.fake_ravel_func(func)
1992
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1994
ObjectPath("key1"): {
1995
ObjectPath("key2"): {
1996
ObjectPath("key3"): ObjectPath("value"),
2000
ret = self.call_method(bus, "methodname", "busname",
2001
"objectpath", "interface")
2002
expected_method_return = {"key1": {"key2": {"key3": "value"}}}
2003
self.assertEqual(expected_method_return, ret)
2004
self.assertIsInstance(ret, dict)
2005
self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)
2006
self.assertIsInstance(ret["key1"], dict)
2007
self.assertNotIsInstance(next(iter(ret["key1"].keys())),
2009
self.assertIsInstance(ret["key1"]["key2"], dict)
2010
self.assertNotIsInstance(
2011
next(iter(ret["key1"]["key2"].keys())),
2013
self.assertEqual("value", ret["key1"]["key2"]["key3"])
2014
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
2015
self.dummy_dbussy.DBUS.ObjectPath)
2017
def test_call_method_handles_exception(self):
2019
raise self.dummy_dbussy.DBusError()
2021
fake_ravel = self.fake_ravel_func(func)
2022
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
2024
with self.assertRaises(dbus.Error) as e:
2025
self.call_method(bus, "methodname", "busname",
2026
"objectpath", "interface")
2028
self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
2030
def test_get_object_converts_to_correct_exception(self):
2031
class fake_ravel_raises_exception_on_connect:
2036
def __getitem__(key):
2037
if key == "objectpath":
2038
raise self.dummy_dbussy.DBusError()
2039
raise Exception(key)
2040
return {"busname": Bus()}
2042
raise self.dummy_dbussy.DBusError()
2043
bus = dbussy_adapter.SystemBus(
2045
fake_ravel_raises_exception_on_connect)
2046
with self.assertRaises(dbus.ConnectFailed):
2047
self.call_method(bus, "methodname", "busname",
2048
"objectpath", "interface")
1773
2051
class Test_commands_from_options(unittest.TestCase):
1775
2053
def setUp(self):
1780
2058
self.assert_command_from_args(["--is-enabled", "client"],
1781
2059
command.IsEnabled)
1783
def assert_command_from_args(self, args, command_cls,
2061
def assert_command_from_args(self, args, command_cls, length=1,
2062
clients=None, **cmd_attrs):
1785
2063
"""Assert that parsing ARGS should result in an instance of
1786
2064
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1787
2065
options = self.parser.parse_args(args)
1788
2066
check_option_syntax(self.parser, options)
1789
2067
commands = commands_from_options(options)
1790
self.assertEqual(1, len(commands))
1791
command = commands[0]
1792
self.assertIsInstance(command, command_cls)
2068
self.assertEqual(length, len(commands))
2069
for command in commands:
2070
if isinstance(command, command_cls):
2073
self.assertIsInstance(command, command_cls)
2074
if clients is not None:
2075
self.assertEqual(clients, options.client)
1793
2076
for key, value in cmd_attrs.items():
1794
2077
self.assertEqual(value, getattr(command, key))
2079
def assert_commands_from_args(self, args, commands, clients=None):
2080
for cmd in commands:
2081
self.assert_command_from_args(args, cmd,
2082
length=len(commands),
1796
2085
def test_is_enabled_short(self):
1797
2086
self.assert_command_from_args(["-V", "client"],
1798
2087
command.IsEnabled)
2115
2436
busname = "se.recompile.Mandos"
2116
2437
client_interface = "se.recompile.Mandos.Client"
2117
2438
command.Deny().run(self.bus.clients, self.bus)
2439
self.assertTrue(self.bus.clients)
2118
2440
for clientpath in self.bus.clients:
2119
2441
self.assertIn(("Approve", busname, clientpath,
2120
2442
client_interface, (False,)),
2121
2443
self.bus.calls)
2123
2445
def test_Remove(self):
2446
busname = "se.recompile.Mandos"
2448
server_interface = "se.recompile.Mandos"
2449
orig_clients = self.bus.clients.copy()
2124
2450
command.Remove().run(self.bus.clients, self.bus)
2125
for clientpath in self.bus.clients:
2126
self.assertIn(("RemoveClient", dbus_busname,
2127
dbus_server_path, dbus_server_interface,
2451
self.assertFalse(self.bus.clients)
2452
for clientpath in orig_clients:
2453
self.assertIn(("RemoveClient", busname,
2454
server_path, server_interface,
2128
2455
(clientpath,)), self.bus.calls)
2130
2457
expected_json = {
2481
2814
if __name__ == "__main__":
2815
options = parse_test_args()
2483
if should_only_run_tests():
2484
# Call using ./tdd-python-script --check [--verbose]
2818
extra_test_prefix = options.prefix
2819
if extra_test_prefix is not None:
2820
if not (unittest.main(argv=[""], exit=False)
2821
.result.wasSuccessful()):
2823
class ExtraTestLoader(unittest.TestLoader):
2824
testMethodPrefix = extra_test_prefix
2825
# Call using ./scriptname --check [--verbose]
2826
unittest.main(argv=[""], testLoader=ExtraTestLoader())
2828
unittest.main(argv=[""])
2489
2832
logging.shutdown()
2836
# (lambda (&optional extra)
2837
# (if (not (funcall run-tests-in-test-buffer default-directory
2839
# (funcall show-test-buffer-in-test-window)
2840
# (funcall remove-test-window)
2841
# (if extra (message "Extra tests run successfully!"))))
2842
# run-tests-in-test-buffer:
2843
# (lambda (dir &optional extra)
2844
# (with-current-buffer (get-buffer-create "*Test*")
2845
# (setq buffer-read-only nil
2846
# default-directory dir)
2848
# (compilation-mode))
2849
# (let ((process-result
2850
# (let ((inhibit-read-only t))
2851
# (process-file-shell-command
2852
# (funcall get-command-line extra) nil "*Test*"))))
2853
# (and (numberp process-result)
2854
# (= process-result 0))))
2856
# (lambda (&optional extra)
2857
# (let ((quoted-script
2858
# (shell-quote-argument (funcall get-script-name))))
2860
# (concat "%s --check" (if extra " --prefix=atest" ""))
2864
# (if (fboundp 'file-local-name)
2865
# (file-local-name (buffer-file-name))
2866
# (or (file-remote-p (buffer-file-name) 'localname)
2867
# (buffer-file-name))))
2868
# remove-test-window:
2870
# (let ((test-window (get-buffer-window "*Test*")))
2871
# (if test-window (delete-window test-window))))
2872
# show-test-buffer-in-test-window:
2874
# (when (not (get-buffer-window-list "*Test*"))
2875
# (setq next-error-last-buffer (get-buffer "*Test*"))
2876
# (let* ((side (if (>= (window-width) 146) 'right 'bottom))
2877
# (display-buffer-overriding-action
2878
# `((display-buffer-in-side-window) (side . ,side)
2879
# (window-height . fit-window-to-buffer)
2880
# (window-width . fit-window-to-buffer))))
2881
# (display-buffer "*Test*"))))
2884
# (let* ((run-extra-tests (lambda () (interactive)
2885
# (funcall run-tests t)))
2886
# (inner-keymap `(keymap (116 . ,run-extra-tests))) ; t
2887
# (outer-keymap `(keymap (3 . ,inner-keymap)))) ; C-c
2888
# (setq minor-mode-overriding-map-alist
2889
# (cons `(run-tests . ,outer-keymap)
2890
# minor-mode-overriding-map-alist)))
2891
# (add-hook 'after-save-hook run-tests 90 t))