/mandos/trunk : revision 1128

1

#!/usr/bin/python3 -bbI

2

# -*- coding: utf-8; lexical-binding: t -*-

3

#

4

# Mandos Control - Control or query the Mandos server

5

#

6

7

1

#!/usr/bin/python

2

# -*- after-save-hook: (lambda () (let ((command (if (fboundp 'file-local-name) (file-local-name (buffer-file-name)) (or (file-remote-p (buffer-file-name) 'localname) (buffer-file-name))))) (if (= (progn (if (get-buffer "*Test*") (kill-buffer "*Test*")) (process-file-shell-command (format "%s --test" (shell-quote-argument command)) nil "*Test*")) 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w))) (progn (with-current-buffer "*Test*" (compilation-mode)) (display-buffer "*Test*" '(display-buffer-in-side-window)))))); coding: utf-8 -*-

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

23

#

24

# Contact the authors at <mandos@recompile.se>.

25

#

26

27

from __future__ import (division, absolute_import, print_function,

27

28

unicode_literals)

28

29

32

33

pass

33

34

35

import sys

35

import unittest

36

import argparse

37

import logging

38

import os

39

37

import locale

40

38

import datetime

41

39

import re

40

import os

42

41

import collections

43

42

import json

43

import unittest

44

import logging

44

45

import io

45

46

import tempfile

46

47

import contextlib

47

48

if sys.version_info.major == 2:

49

__metaclass__ = type

50

str = unicode

51

input = raw_input

52

53

class gi:

54

"""Dummy gi module, for the tests"""

55

class repository:

56

class GLib:

57

class Error(Exception):

58

pass

59

dbussy = None

60

ravel = None

61

dbus_python = None

62

pydbus = None

63

64

49

try:

65

import dbussy

66

import ravel

50

import pydbus

51

import gi

52

dbus_python = None

67

53

except ImportError:

68

try:

69

import pydbus

70

import gi

71

except ImportError:

72

import dbus as dbus_python

73

54

import dbus as dbus_python

55

pydbus = None

56

class gi(object):

57

"""Dummy gi module, for the tests"""

58

class repository(object):

59

class GLib(object):

60

class Error(Exception):

61

pass

74

62

75

63

# Show warnings by default

76

64

if not sys.warnoptions:

77

65

import warnings

78

66

warnings.simplefilter("default")

79

67

80

log = logging.getLogger(os.path.basename(sys.argv[0]))

68

log = logging.getLogger(sys.argv[0])

81

69

logging.basicConfig(level="INFO", # Show info level messages

82

70

format="%(message)s") # Show basic log messages

83

71

84

72

logging.captureWarnings(True) # Show warnings via the logging system

85

73

86

74

if sys.version_info.major == 2:

75

str = unicode

87

76

import StringIO

88

77

io.StringIO = StringIO.StringIO

89

78

90

79

locale.setlocale(locale.LC_ALL, "")

91

80

92

version = "1.8.15"

81

version = "1.8.4"

93

82

94

83

95

84

def main():

102

91

clientnames = options.client

103

92

104

93

if options.debug:

105

logging.getLogger("").setLevel(logging.DEBUG)

94

log.setLevel(logging.DEBUG)

106

95

107

if dbussy is not None and ravel is not None:

108

bus = dbussy_adapter.CachingBus(dbussy, ravel)

109

elif pydbus is not None:

96

if pydbus is not None:

110

97

bus = pydbus_adapter.CachingBus(pydbus)

111

98

else:

112

99

bus = dbus_python_adapter.CachingBus(dbus_python)

256

243

return rfc3339_duration_to_delta(interval)

257

244

except ValueError as e:

258

245

log.warning("%s - Parsing as pre-1.6.1 interval instead",

259

" ".join(e.args))

246

' '.join(e.args))

260

247

return parse_pre_1_6_1_interval(interval)

261

248

262

249

263

250

def rfc3339_duration_to_delta(duration):

264

251

"""Parse an RFC 3339 "duration" and return a datetime.timedelta

265

252

266

>>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)

267

True

268

>>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)

269

True

270

>>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)

271

True

272

>>> # 60 months

273

>>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)

274

True

275

>>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)

276

True

277

>>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)

278

True

279

>>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)

280

True

281

>>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)

282

True

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)

283

269

>>> # Can not be empty:

284

270

>>> rfc3339_duration_to_delta("")

285

271

Traceback (most recent call last):

395

381

"""Parse an interval string as documented by Mandos before 1.6.1,

396

382

and return a datetime.timedelta

397

383

398

>>> parse_pre_1_6_1_interval("7d") == datetime.timedelta(days=7)

399

True

400

>>> parse_pre_1_6_1_interval("60s") == datetime.timedelta(0, 60)

401

True

402

>>> parse_pre_1_6_1_interval("60m") == datetime.timedelta(hours=1)

403

True

404

>>> parse_pre_1_6_1_interval("24h") == datetime.timedelta(days=1)

405

True

406

>>> parse_pre_1_6_1_interval("1w") == datetime.timedelta(days=7)

407

True

408

>>> parse_pre_1_6_1_interval("5m 30s") == datetime.timedelta(0, 330)

409

True

410

>>> parse_pre_1_6_1_interval("") == datetime.timedelta(0)

411

True

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)

412

398

>>> # Ignore unknown characters, allow any order and repetitions

413

>>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") == datetime.timedelta(2, 480, 18000)

414

True

399

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

400

datetime.timedelta(2, 480, 18000)

415

401

416

402

"""

417

403

480

466

parser.error("--remove can only be combined with --deny")

481

467

482

468

483

class dbus:

469

class dbus(object):

484

470

485

class SystemBus:

471

class SystemBus(object):

486

472

487

473

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

488

474

def get_managed_objects(self, busname, objectpath):

497

483

self.properties_iface, interface, key,

498

484

value)

499

485

500

def call_method(self, methodname, busname, objectpath,

501

interface, *args):

502

raise NotImplementedError()

503

504

486

505

487

class MandosBus(SystemBus):

506

488

busname_domain = "se.recompile"

538

520

pass

539

521

540

522

541

class dbus_python_adapter:

523

class dbus_python_adapter(object):

542

524

543

525

class SystemBus(dbus.MandosBus):

544

526

"""Use dbus-python"""

598

580

self.client_interface, key,

599

581

value)

600

582

601

class SilenceLogger:

583

class SilenceLogger(object):

602

584

"Simple context manager to silence a particular logger"

603

585

def __init__(self, loggername):

604

586

self.logger = logging.getLogger(loggername)

633

615

return new_object

634

616

635

617

636

class pydbus_adapter:

618

class pydbus_adapter(object):

637

619

class SystemBus(dbus.MandosBus):

638

620

def __init__(self, module=pydbus):

639

621

self.pydbus = module

696

678

return new_object

697

679

698

680

699

class dbussy_adapter:

700

class SystemBus(dbus.SystemBus):

701

"""Use DBussy"""

702

703

def __init__(self, dbussy, ravel):

704

self.dbussy = dbussy

705

self.ravel = ravel

706

self.bus = ravel.system_bus()

707

708

@contextlib.contextmanager

709

def convert_exception(self, exception_class=dbus.Error):

710

try:

711

yield

712

except self.dbussy.DBusError as e:

713

# This does what "raise from" would do

714

exc = exception_class(*e.args)

715

exc.__cause__ = e

716

raise exc

717

718

def call_method(self, methodname, busname, objectpath,

719

interface, *args):

720

proxy_object = self.get_object(busname, objectpath)

721

log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,

722

interface, methodname,

723

", ".join(repr(a) for a in args))

724

iface = proxy_object.get_interface(interface)

725

method = getattr(iface, methodname)

726

with self.convert_exception(dbus.Error):

727

value = method(*args)

728

# DBussy returns values either as an empty list or as a

729

# list of one element with the return value

730

if value:

731

return self.type_filter(value[0])

732

733

def get_object(self, busname, objectpath):

734

log.debug("D-Bus: Connect to: (busname=%r, path=%r)",

735

busname, objectpath)

736

with self.convert_exception(dbus.ConnectFailed):

737

return self.bus[busname][objectpath]

738

739

def type_filter(self, value):

740

"""Convert the most bothersome types to Python types"""

741

# A D-Bus Variant value is represented as the Python type

742

# Tuple[dbussy.DBUS.Signature, Any]

743

if isinstance(value, tuple):

744

if (len(value) == 2

745

and isinstance(value[0],

746

self.dbussy.DBUS.Signature)):

747

return self.type_filter(value[1])

748

elif isinstance(value, self.dbussy.DBUS.ObjectPath):

749

return str(value)

750

# Also recurse into dictionaries

751

elif isinstance(value, dict):

752

return {self.type_filter(key):

753

self.type_filter(subval)

754

for key, subval in value.items()}

755

return value

756

757

def set_property(self, busname, objectpath, interface, key,

758

value):

759

proxy_object = self.get_object(busname, objectpath)

760

log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,

761

objectpath, self.properties_iface, interface,

762

key, value)

763

if key == "Secret":

764

# DBussy wants a Byte Array to be a sequence of

765

# values, not a byte string

766

value = tuple(value)

767

setattr(proxy_object.get_interface(interface), key, value)

768

769

class MandosBus(SystemBus, dbus.MandosBus):

770

pass

771

772

class CachingBus(MandosBus):

773

"""A caching layer for dbussy_adapter.MandosBus"""

774

def __init__(self, *args, **kwargs):

775

self.object_cache = {}

776

super(dbussy_adapter.CachingBus, self).__init__(*args,

777

**kwargs)

778

def get_object(self, busname, objectpath):

779

try:

780

return self.object_cache[(busname, objectpath)]

781

except KeyError:

782

new_object = super(

783

dbussy_adapter.CachingBus,

784

self).get_object(busname, objectpath)

785

self.object_cache[(busname, objectpath)] = new_object

786

return new_object

787

788

789

681

def commands_from_options(options):

790

682

791

683

commands = list(options.commands)

819

711

return commands

820

712

821

713

822

class command:

714

class command(object):

823

715

"""A namespace for command classes"""

824

716

825

class Base:

717

class Base(object):

826

718

"""Abstract base class for commands"""

827

719

def run(self, clients, bus=None):

828

720

"""Normal commands should implement run_on_one_client(),

876

768

{key: properties[key]

877

769

for key in self.all_keywords}

878

770

for properties in clients.values()}

879

print(json.dumps(data, indent=4, separators=(",", ": ")))

771

print(json.dumps(data, indent=4, separators=(',', ': ')))

880

772

881

773

882

774

class PrintTable(Output):

891

783

keywords = self.all_keywords

892

784

print(self.TableOfClients(clients.values(), keywords))

893

785

894

class TableOfClients:

786

class TableOfClients(object):

895

787

tableheaders = {

896

788

"Name": "Name",

897

789

"Enabled": "Enabled",

1128

1020

"output"))

1129

1021

1130

1022

1131

class Unique:

1023

class Unique(object):

1132

1024

"""Class for objects which exist only to be unique objects, since

1133

1025

unittest.mock.sentinel only exists in Python 3.3"""

1134

1026

1418

1310

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1419

1311

1420

1312

def MockDBusPython_func(self, func):

1421

class mock_dbus_python:

1313

class mock_dbus_python(object):

1422

1314

"""mock dbus-python module"""

1423

class exceptions:

1315

class exceptions(object):

1424

1316

"""Pseudo-namespace"""

1425

1317

class DBusException(Exception):

1426

1318

pass

1427

class SystemBus:

1319

class SystemBus(object):

1428

1320

@staticmethod

1429

1321

def get_object(busname, objectpath):

1430

1322

DBusObject = collections.namedtuple(

1444

1336

dbus_interface=dbus_interface)

1445

1337

return DBusObject(methodname=method,

1446

1338

Set=set_property)

1447

class Boolean:

1339

class Boolean(object):

1448

1340

def __init__(self, value):

1449

1341

self.value = bool(value)

1450

1342

def __bool__(self):

1634

1526

finally:

1635

1527

dbus_logger.removeFilter(counting_handler)

1636

1528

1637

self.assertNotIsInstance(e.exception, dbus.ConnectFailed)

1529

self.assertNotIsInstance(e, dbus.ConnectFailed)

1638

1530

1639

1531

# Make sure the dbus logger was suppressed

1640

1532

self.assertEqual(0, counting_handler.count)

1662

1554

self.call_method(bus, "methodname", "busname",

1663

1555

"objectpath", "interface")

1664

1556

1665

class fake_dbus_python_raises_exception_on_connect:

1557

class fake_dbus_python_raises_exception_on_connect(object):

1666

1558

"""fake dbus-python module"""

1667

class exceptions:

1559

class exceptions(object):

1668

1560

"""Pseudo-namespace"""

1669

1561

class DBusException(Exception):

1670

1562

pass

1678

1570

1679

1571

1680

1572

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1681

class mock_dbus_python:

1573

class mock_dbus_python(object):

1682

1574

"""mock dbus-python modules"""

1683

class SystemBus:

1575

class SystemBus(object):

1684

1576

@staticmethod

1685

1577

def get_object(busname, objectpath):

1686

1578

return Unique()

1732

1624

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1733

1625

1734

1626

def Stub_pydbus_func(self, func):

1735

class stub_pydbus:

1627

class stub_pydbus(object):

1736

1628

"""stub pydbus module"""

1737

class SystemBus:

1629

class SystemBus(object):

1738

1630

@staticmethod

1739

1631

def get(busname, objectpath):

1740

1632

DBusObject = collections.namedtuple(

1777

1669

self.call_method(bus, "methodname", "busname",

1778

1670

"objectpath", "interface")

1779

1671

1780

self.assertNotIsInstance(e.exception, dbus.ConnectFailed)

1672

self.assertNotIsInstance(e, dbus.ConnectFailed)

1781

1673

1782

1674

def test_get_converts_to_correct_exception(self):

1783

1675

bus = pydbus_adapter.SystemBus(

1786

1678

self.call_method(bus, "methodname", "busname",

1787

1679

"objectpath", "interface")

1788

1680

1789

class fake_pydbus_raises_exception_on_connect:

1681

class fake_pydbus_raises_exception_on_connect(object):

1790

1682

"""fake dbus-python module"""

1791

1683

@classmethod

1792

1684

def SystemBus(cls):

1796

1688

return Bus(get=get)

1797

1689

1798

1690

def test_set_property_uses_setattr(self):

1799

class Object:

1691

class Object(object):

1800

1692

pass

1801

1693

obj = Object()

1802

class pydbus_spy:

1803

class SystemBus:

1694

class pydbus_spy(object):

1695

class SystemBus(object):

1804

1696

@staticmethod

1805

1697

def get(busname, objectpath):

1806

1698

return {"interface": obj}

1813

1705

def test_get_suppresses_xml_deprecation_warning(self):

1814

1706

if sys.version_info.major >= 3:

1815

1707

return

1816

class stub_pydbus_get:

1817

class SystemBus:

1708

class stub_pydbus_get(object):

1709

class SystemBus(object):

1818

1710

@staticmethod

1819

1711

def get(busname, objectpath):

1820

1712

warnings.warn_explicit(

1828

1720

1829

1721

1830

1722

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1831

class stub_pydbus:

1723

class stub_pydbus(object):

1832

1724

"""stub pydbus module"""

1833

class SystemBus:

1725

class SystemBus(object):

1834

1726

@staticmethod

1835

1727

def get(busname, objectpath):

1836

1728

return Unique()

1878

1770

self.assertIs(obj1, obj1b)

1879

1771

1880

1772

1881

class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):

1882

1883

class dummy_dbussy:

1884

class DBUS:

1885

class ObjectPath(str):

1886

pass

1887

class DBusError(Exception):

1888

pass

1889

1890

def fake_ravel_func(self, func):

1891

class fake_ravel:

1892

@staticmethod

1893

def system_bus():

1894

class DBusInterfaceProxy:

1895

@staticmethod

1896

def methodname(*args):

1897

return [func(*args)]

1898

class DBusObject:

1899

@staticmethod

1900

def get_interface(interface):

1901

if interface == "interface":

1902

return DBusInterfaceProxy()

1903

return {"busname": {"objectpath": DBusObject()}}

1904

return fake_ravel

1905

1906

def call_method(self, bus, methodname, busname, objectpath,

1907

interface, *args):

1908

with self.assertLogs(log, logging.DEBUG):

1909

return bus.call_method(methodname, busname, objectpath,

1910

interface, *args)

1911

1912

def test_call_method_returns(self):

1913

expected_method_return = Unique()

1914

method_args = (Unique(), Unique())

1915

def func(*args):

1916

self.assertEqual(len(method_args), len(args))

1917

for marg, arg in zip(method_args, args):

1918

self.assertIs(marg, arg)

1919

return expected_method_return

1920

fake_ravel = self.fake_ravel_func(func)

1921

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1922

ret = self.call_method(bus, "methodname", "busname",

1923

"objectpath", "interface",

1924

*method_args)

1925

self.assertIs(ret, expected_method_return)

1926

1927

def test_call_method_filters_objectpath(self):

1928

def func():

1929

return method_return

1930

fake_ravel = self.fake_ravel_func(func)

1931

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1932

method_return = (self.dummy_dbussy.DBUS

1933

.ObjectPath("objectpath"))

1934

ret = self.call_method(bus, "methodname", "busname",

1935

"objectpath", "interface")

1936

self.assertEqual("objectpath", ret)

1937

self.assertNotIsInstance(ret,

1938

self.dummy_dbussy.DBUS.ObjectPath)

1939

1940

def test_call_method_filters_objectpaths_in_dict(self):

1941

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1942

def func():

1943

return method_return

1944

fake_ravel = self.fake_ravel_func(func)

1945

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1946

method_return = {

1947

ObjectPath("objectpath_key_1"):

1948

ObjectPath("objectpath_value_1"),

1949

ObjectPath("objectpath_key_2"):

1950

ObjectPath("objectpath_value_2"),

1951

}

1952

ret = self.call_method(bus, "methodname", "busname",

1953

"objectpath", "interface")

1954

expected_method_return = {str(key): str(value)

1955

for key, value in

1956

method_return.items()}

1957

for key, value in ret.items():

1958

self.assertNotIsInstance(key, ObjectPath)

1959

self.assertNotIsInstance(value, ObjectPath)

1960

self.assertEqual(expected_method_return, ret)

1961

self.assertIsInstance(ret, dict)

1962

1963

def test_call_method_filters_objectpaths_in_dict_in_dict(self):

1964

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1965

def func():

1966

return method_return

1967

fake_ravel = self.fake_ravel_func(func)

1968

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1969

method_return = {

1970

ObjectPath("key1"): {

1971

ObjectPath("key11"): ObjectPath("value11"),

1972

ObjectPath("key12"): ObjectPath("value12"),

1973

},

1974

ObjectPath("key2"): {

1975

ObjectPath("key21"): ObjectPath("value21"),

1976

ObjectPath("key22"): ObjectPath("value22"),

1977

},

1978

}

1979

ret = self.call_method(bus, "methodname", "busname",

1980

"objectpath", "interface")

1981

expected_method_return = {

1982

"key1": {"key11": "value11",

1983

"key12": "value12"},

1984

"key2": {"key21": "value21",

1985

"key22": "value22"},

1986

}

1987

self.assertEqual(expected_method_return, ret)

1988

for key, value in ret.items():

1989

self.assertIsInstance(value, dict)

1990

self.assertEqual(expected_method_return[key], value)

1991

self.assertNotIsInstance(key, ObjectPath)

1992

for inner_key, inner_value in value.items():

1993

self.assertIsInstance(value, dict)

1994

self.assertEqual(

1995

expected_method_return[key][inner_key],

1996

inner_value)

1997

self.assertNotIsInstance(key, ObjectPath)

1998

1999

def test_call_method_filters_objectpaths_in_dict_three_deep(self):

2000

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

2001

def func():

2002

return method_return

2003

fake_ravel = self.fake_ravel_func(func)

2004

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

2005

method_return = {

2006

ObjectPath("key1"): {

2007

ObjectPath("key2"): {

2008

ObjectPath("key3"): ObjectPath("value"),

2009

},

2010

},

2011

}

2012

ret = self.call_method(bus, "methodname", "busname",

2013

"objectpath", "interface")

2014

expected_method_return = {"key1": {"key2": {"key3": "value"}}}

2015

self.assertEqual(expected_method_return, ret)

2016

self.assertIsInstance(ret, dict)

2017

self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)

2018

self.assertIsInstance(ret["key1"], dict)

2019

self.assertNotIsInstance(next(iter(ret["key1"].keys())),

2020

ObjectPath)

2021

self.assertIsInstance(ret["key1"]["key2"], dict)

2022

self.assertNotIsInstance(

2023

next(iter(ret["key1"]["key2"].keys())),

2024

ObjectPath)

2025

self.assertEqual("value", ret["key1"]["key2"]["key3"])

2026

self.assertNotIsInstance(ret["key1"]["key2"]["key3"],

2027

self.dummy_dbussy.DBUS.ObjectPath)

2028

2029

def test_call_method_handles_exception(self):

2030

def func():

2031

raise self.dummy_dbussy.DBusError()

2032

2033

fake_ravel = self.fake_ravel_func(func)

2034

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

2035

2036

with self.assertRaises(dbus.Error) as e:

2037

self.call_method(bus, "methodname", "busname",

2038

"objectpath", "interface")

2039

2040

self.assertNotIsInstance(e.exception, dbus.ConnectFailed)

2041

2042

def test_get_object_converts_to_correct_exception(self):

2043

class fake_ravel_raises_exception_on_connect:

2044

@staticmethod

2045

def system_bus():

2046

class Bus:

2047

@staticmethod

2048

def __getitem__(key):

2049

if key == "objectpath":

2050

raise self.dummy_dbussy.DBusError()

2051

raise Exception(key)

2052

return {"busname": Bus()}

2053

def func():

2054

raise self.dummy_dbussy.DBusError()

2055

bus = dbussy_adapter.SystemBus(

2056

self.dummy_dbussy,

2057

fake_ravel_raises_exception_on_connect)

2058

with self.assertRaises(dbus.ConnectFailed):

2059

self.call_method(bus, "methodname", "busname",

2060

"objectpath", "interface")

2061

2062

2063

1773

class Test_commands_from_options(unittest.TestCase):

2064

1774

2065

1775

def setUp(self):

2291

2001

2292

2002

2293

2003

def test_manual_page_example_1(self):

2294

self.assert_command_from_args("",

2004

self.assert_command_from_args("--verbose".split(),

2295

2005

command.PrintTable,

2296

2006

clients=[],

2297

verbose=False)

2007

verbose=True)

2298

2008

2299

2009

def test_manual_page_example_2(self):

2300

2010

self.assert_command_from_args(

2439

2149

busname = "se.recompile.Mandos"

2440

2150

client_interface = "se.recompile.Mandos.Client"

2441

2151

command.Approve().run(self.bus.clients, self.bus)

2442

self.assertTrue(self.bus.clients)

2443

2152

for clientpath in self.bus.clients:

2444

2153

self.assertIn(("Approve", busname, clientpath,

2445

2154

client_interface, (True,)), self.bus.calls)

2448

2157

busname = "se.recompile.Mandos"

2449

2158

client_interface = "se.recompile.Mandos.Client"

2450

2159

command.Deny().run(self.bus.clients, self.bus)

2451

self.assertTrue(self.bus.clients)

2452

2160

for clientpath in self.bus.clients:

2453

2161

self.assertIn(("Approve", busname, clientpath,

2454

2162

client_interface, (False,)),

2455

2163

self.bus.calls)

2456

2164

2457

2165

def test_Remove(self):

2458

busname = "se.recompile.Mandos"

2459

server_path = "/"

2460

server_interface = "se.recompile.Mandos"

2461

orig_clients = self.bus.clients.copy()

2462

2166

command.Remove().run(self.bus.clients, self.bus)

2463

self.assertFalse(self.bus.clients)

2464

for clientpath in orig_clients:

2465

self.assertIn(("RemoveClient", busname,

2466

server_path, server_interface,

2167

for clientpath in self.bus.clients:

2168

self.assertIn(("RemoveClient", dbus_busname,

2169

dbus_server_path, dbus_server_interface,

2467

2170

(clientpath,)), self.bus.calls)

2468

2171

2469

2172

expected_json = {

2671

2374

else:

2672

2375

cmd_args = [() for x in range(len(self.values_to_get))]

2673

2376

values_to_get = self.values_to_get

2674

self.assertTrue(values_to_get)

2675

2377

for value_to_get, cmd_arg in zip(values_to_get, cmd_args):

2676

2378

for clientpath in self.bus.clients:

2677

2379

self.bus.clients[clientpath][self.propname] = (

2678

2380

Unique())

2679

2381

self.command(*cmd_arg).run(self.bus.clients, self.bus)

2680

self.assertTrue(self.bus.clients)

2681

2382

for clientpath in self.bus.clients:

2682

2383

value = (self.bus.clients[clientpath]

2683

2384

[self.propname])

2742

2443

class TestSetSecretCmd(TestPropertySetterCmd):

2743

2444

command = command.SetSecret

2744

2445

propname = "Secret"

2745

def __init__(self, *args, **kwargs):

2746

self.values_to_set = [io.BytesIO(b""),

2747

io.BytesIO(b"secret\0xyzzy\nbar")]

2748

self.values_to_get = [f.getvalue() for f in

2749

self.values_to_set]

2750

super(TestSetSecretCmd, self).__init__(*args, **kwargs)

2446

values_to_set = [io.BytesIO(b""),

2447

io.BytesIO(b"secret\0xyzzy\nbar")]

2448

values_to_get = [f.getvalue() for f in values_to_set]

2751

2449

2752

2450

2753

2451

class TestSetTimeoutCmd(TestPropertySetterCmd):

2806

2504

2807

2505

2808

2506

2809

def parse_test_args():

2810

# type: () -> argparse.Namespace

2507

def should_only_run_tests():

2811

2508

parser = argparse.ArgumentParser(add_help=False)

2812

parser.add_argument("--check", action="store_true")

2813

parser.add_argument("--prefix", )

2509

parser.add_argument("--check", action='store_true')

2814

2510

args, unknown_args = parser.parse_known_args()

2815

if args.check:

2816

# Remove test options from sys.argv

2511

run_tests = args.check

2512

if run_tests:

2513

# Remove --check argument from sys.argv

2817

2514

sys.argv[1:] = unknown_args

2818

return args

2515

return run_tests

2819

2516

2820

2517

# Add all tests from doctest strings

2821

2518

def load_tests(loader, tests, none):

2824

2521

return tests

2825

2522

2826

2523

if __name__ == "__main__":

2827

options = parse_test_args()

2828

2524

try:

2829

if options.check:

2830

extra_test_prefix = options.prefix

2831

if extra_test_prefix is not None:

2832

if not (unittest.main(argv=[""], exit=False)

2833

.result.wasSuccessful()):

2834

sys.exit(1)

2835

class ExtraTestLoader(unittest.TestLoader):

2836

testMethodPrefix = extra_test_prefix

2837

# Call using ./scriptname --check [--verbose]

2838

unittest.main(argv=[""], testLoader=ExtraTestLoader())

2839

else:

2840

unittest.main(argv=[""])

2525

if should_only_run_tests():

2526

# Call using ./tdd-python-script --check [--verbose]

2527

unittest.main()

2841

2528

else:

2842

2529

main()

2843

2530

finally:

2844

2531

logging.shutdown()

2845

2846

# Local Variables:

2847

# run-tests:

2848

# (lambda (&optional extra)

2849

# (if (not (funcall run-tests-in-test-buffer default-directory

2850

# extra))

2851

# (funcall show-test-buffer-in-test-window)

2852

# (funcall remove-test-window)

2853

# (if extra (message "Extra tests run successfully!"))))

2854

# run-tests-in-test-buffer:

2855

# (lambda (dir &optional extra)

2856

# (with-current-buffer (get-buffer-create "*Test*")

2857

# (setq buffer-read-only nil

2858

# default-directory dir)

2859

# (erase-buffer)

2860

# (compilation-mode))

2861

# (let ((process-result

2862

# (let ((inhibit-read-only t))

2863

# (process-file-shell-command

2864

# (funcall get-command-line extra) nil "*Test*"))))

2865

# (and (numberp process-result)

2866

# (= process-result 0))))

2867

# get-command-line:

2868

# (lambda (&optional extra)

2869

# (let ((quoted-script

2870

# (shell-quote-argument (funcall get-script-name))))

2871

# (format

2872

# (concat "%s --check" (if extra " --prefix=atest" ""))

2873

# quoted-script)))

2874

# get-script-name:

2875

# (lambda ()

2876

# (if (fboundp 'file-local-name)

2877

# (file-local-name (buffer-file-name))

2878

# (or (file-remote-p (buffer-file-name) 'localname)

2879

# (buffer-file-name))))

2880

# remove-test-window:

2881

# (lambda ()

2882

# (let ((test-window (get-buffer-window "*Test*")))

2883

# (if test-window (delete-window test-window))))

2884

# show-test-buffer-in-test-window:

2885

# (lambda ()

2886

# (when (not (get-buffer-window-list "*Test*"))

2887

# (setq next-error-last-buffer (get-buffer "*Test*"))

2888

# (let* ((side (if (>= (window-width) 146) 'right 'bottom))

2889

# (display-buffer-overriding-action

2890

# `((display-buffer-in-side-window) (side . ,side)

2891

# (window-height . fit-window-to-buffer)

2892

# (window-width . fit-window-to-buffer))))

2893

# (display-buffer "*Test*"))))

2894

# eval:

2895

# (progn

2896

# (let* ((run-extra-tests (lambda () (interactive)

2897

# (funcall run-tests t)))

2898

# (inner-keymap `(keymap (116 . ,run-extra-tests))) ; t

2899

# (outer-keymap `(keymap (3 . ,inner-keymap)))) ; C-c

2900

# (setq minor-mode-overriding-map-alist

2901

# (cons `(run-tests . ,outer-keymap)

2902

# minor-mode-overriding-map-alist)))

2903

# (add-hook 'after-save-hook run-tests 90 t))

2904

# End: