/mandos/release : revision 409

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

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

8

#

9

# This file is part of Mandos.

10

#

23

#

24

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

25

#

26

27

26

from __future__ import (division, absolute_import, print_function,

28

27

unicode_literals)

29

28

33

32

pass

34

33

35

34

import sys

35

import unittest

36

import argparse

37

import logging

38

import os

37

39

import locale

38

40

import datetime

39

41

import re

40

import os

41

42

import collections

42

43

import json

43

import unittest

44

import logging

45

44

import io

46

45

import tempfile

47

46

import contextlib

48

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

49

64

try:

50

import pydbus

51

import gi

52

dbus_python = None

65

import dbussy

66

import ravel

53

67

except ImportError:

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

68

try:

69

import pydbus

70

import gi

71

except ImportError:

72

import dbus as dbus_python

73

62

74

63

75

# Show warnings by default

64

76

if not sys.warnoptions:

65

77

import warnings

66

78

warnings.simplefilter("default")

67

79

68

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

80

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

69

81

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

70

82

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

71

83

72

84

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

73

85

74

86

if sys.version_info.major == 2:

75

str = unicode

76

87

import StringIO

77

88

io.StringIO = StringIO.StringIO

78

89

79

90

locale.setlocale(locale.LC_ALL, "")

80

91

81

version = "1.8.4"

92

version = "1.8.16"

82

93

83

94

84

95

def main():

91

102

clientnames = options.client

92

103

93

104

if options.debug:

94

log.setLevel(logging.DEBUG)

105

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

95

106

96

if pydbus is not None:

107

if dbussy is not None and ravel is not None:

108

bus = dbussy_adapter.CachingBus(dbussy, ravel)

109

elif pydbus is not None:

97

110

bus = pydbus_adapter.CachingBus(pydbus)

98

111

else:

99

112

bus = dbus_python_adapter.CachingBus(dbus_python)

243

256

return rfc3339_duration_to_delta(interval)

244

257

except ValueError as e:

245

258

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

246

' '.join(e.args))

259

" ".join(e.args))

247

260

return parse_pre_1_6_1_interval(interval)

248

261

249

262

250

263

def rfc3339_duration_to_delta(duration):

251

264

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

252

265

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)

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

269

283

>>> # Can not be empty:

270

284

>>> rfc3339_duration_to_delta("")

271

285

Traceback (most recent call last):

381

395

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

382

396

and return a datetime.timedelta

383

397

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)

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

398

412

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

399

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

400

datetime.timedelta(2, 480, 18000)

413

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

414

True

401

415

402

416

"""

403

417

466

480

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

467

481

468

482

469

class dbus(object):

483

class dbus:

470

484

471

class SystemBus(object):

485

class SystemBus:

472

486

473

487

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

474

488

def get_managed_objects(self, busname, objectpath):

483

497

self.properties_iface, interface, key,

484

498

value)

485

499

500

def call_method(self, methodname, busname, objectpath,

501

interface, *args):

502

raise NotImplementedError()

503

486

504

487

505

class MandosBus(SystemBus):

488

506

busname_domain = "se.recompile"

520

538

pass

521

539

522

540

523

class dbus_python_adapter(object):

541

class dbus_python_adapter:

524

542

525

543

class SystemBus(dbus.MandosBus):

526

544

"""Use dbus-python"""

580

598

self.client_interface, key,

581

599

value)

582

600

583

class SilenceLogger(object):

601

class SilenceLogger:

584

602

"Simple context manager to silence a particular logger"

585

603

def __init__(self, loggername):

586

604

self.logger = logging.getLogger(loggername)

615

633

return new_object

616

634

617

635

618

class pydbus_adapter(object):

636

class pydbus_adapter:

619

637

class SystemBus(dbus.MandosBus):

620

638

def __init__(self, module=pydbus):

621

639

self.pydbus = module

678

696

return new_object

679

697

680

698

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

681

789

def commands_from_options(options):

682

790

683

791

commands = list(options.commands)

711

819

return commands

712

820

713

821

714

class command(object):

822

class command:

715

823

"""A namespace for command classes"""

716

824

717

class Base(object):

825

class Base:

718

826

"""Abstract base class for commands"""

719

827

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

720

828

"""Normal commands should implement run_on_one_client(),

768

876

{key: properties[key]

769

877

for key in self.all_keywords}

770

878

for properties in clients.values()}

771

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

879

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

772

880

773

881

774

882

class PrintTable(Output):

783

891

keywords = self.all_keywords

784

892

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

785

893

786

class TableOfClients(object):

894

class TableOfClients:

787

895

tableheaders = {

788

896

"Name": "Name",

789

897

"Enabled": "Enabled",

1020

1128

"output"))

1021

1129

1022

1130

1023

class Unique(object):

1131

class Unique:

1024

1132

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

1025

1133

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

1026

1134

1310

1418

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1311

1419

1312

1420

def MockDBusPython_func(self, func):

1313

class mock_dbus_python(object):

1421

class mock_dbus_python:

1314

1422

"""mock dbus-python module"""

1315

class exceptions(object):

1423

class exceptions:

1316

1424

"""Pseudo-namespace"""

1317

1425

class DBusException(Exception):

1318

1426

pass

1319

class SystemBus(object):

1427

class SystemBus:

1320

1428

@staticmethod

1321

1429

def get_object(busname, objectpath):

1322

1430

DBusObject = collections.namedtuple(

1336

1444

dbus_interface=dbus_interface)

1337

1445

return DBusObject(methodname=method,

1338

1446

Set=set_property)

1339

class Boolean(object):

1447

class Boolean:

1340

1448

def __init__(self, value):

1341

1449

self.value = bool(value)

1342

1450

def __bool__(self):

1526

1634

finally:

1527

1635

dbus_logger.removeFilter(counting_handler)

1528

1636

1529

self.assertNotIsInstance(e, dbus.ConnectFailed)

1637

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

1530

1638

1531

1639

# Make sure the dbus logger was suppressed

1532

1640

self.assertEqual(0, counting_handler.count)

1554

1662

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

1555

1663

"objectpath", "interface")

1556

1664

1557

class fake_dbus_python_raises_exception_on_connect(object):

1665

class fake_dbus_python_raises_exception_on_connect:

1558

1666

"""fake dbus-python module"""

1559

class exceptions(object):

1667

class exceptions:

1560

1668

"""Pseudo-namespace"""

1561

1669

class DBusException(Exception):

1562

1670

pass

1570

1678

1571

1679

1572

1680

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1573

class mock_dbus_python(object):

1681

class mock_dbus_python:

1574

1682

"""mock dbus-python modules"""

1575

class SystemBus(object):

1683

class SystemBus:

1576

1684

@staticmethod

1577

1685

def get_object(busname, objectpath):

1578

1686

return Unique()

1624

1732

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1625

1733

1626

1734

def Stub_pydbus_func(self, func):

1627

class stub_pydbus(object):

1735

class stub_pydbus:

1628

1736

"""stub pydbus module"""

1629

class SystemBus(object):

1737

class SystemBus:

1630

1738

@staticmethod

1631

1739

def get(busname, objectpath):

1632

1740

DBusObject = collections.namedtuple(

1669

1777

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

1670

1778

"objectpath", "interface")

1671

1779

1672

self.assertNotIsInstance(e, dbus.ConnectFailed)

1780

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

1673

1781

1674

1782

def test_get_converts_to_correct_exception(self):

1675

1783

bus = pydbus_adapter.SystemBus(

1678

1786

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

1679

1787

"objectpath", "interface")

1680

1788

1681

class fake_pydbus_raises_exception_on_connect(object):

1789

class fake_pydbus_raises_exception_on_connect:

1682

1790

"""fake dbus-python module"""

1683

1791

@classmethod

1684

1792

def SystemBus(cls):

1688

1796

return Bus(get=get)

1689

1797

1690

1798

def test_set_property_uses_setattr(self):

1691

class Object(object):

1799

class Object:

1692

1800

pass

1693

1801

obj = Object()

1694

class pydbus_spy(object):

1695

class SystemBus(object):

1802

class pydbus_spy:

1803

class SystemBus:

1696

1804

@staticmethod

1697

1805

def get(busname, objectpath):

1698

1806

return {"interface": obj}

1705

1813

def test_get_suppresses_xml_deprecation_warning(self):

1706

1814

if sys.version_info.major >= 3:

1707

1815

return

1708

class stub_pydbus_get(object):

1709

class SystemBus(object):

1816

class stub_pydbus_get:

1817

class SystemBus:

1710

1818

@staticmethod

1711

1819

def get(busname, objectpath):

1712

1820

warnings.warn_explicit(

1720

1828

1721

1829

1722

1830

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1723

class stub_pydbus(object):

1831

class stub_pydbus:

1724

1832

"""stub pydbus module"""

1725

class SystemBus(object):

1833

class SystemBus:

1726

1834

@staticmethod

1727

1835

def get(busname, objectpath):

1728

1836

return Unique()

1770

1878

self.assertIs(obj1, obj1b)

1771

1879

1772

1880

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

1773

2063

class Test_commands_from_options(unittest.TestCase):

1774

2064

1775

2065

def setUp(self):

2001

2291

2002

2292

2003

2293

def test_manual_page_example_1(self):

2004

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

2294

self.assert_command_from_args("",

2005

2295

command.PrintTable,

2006

2296

clients=[],

2007

verbose=True)

2297

verbose=False)

2008

2298

2009

2299

def test_manual_page_example_2(self):

2010

2300

self.assert_command_from_args(

2149

2439

busname = "se.recompile.Mandos"

2150

2440

client_interface = "se.recompile.Mandos.Client"

2151

2441

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

2442

self.assertTrue(self.bus.clients)

2152

2443

for clientpath in self.bus.clients:

2153

2444

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

2154

2445

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

2157

2448

busname = "se.recompile.Mandos"

2158

2449

client_interface = "se.recompile.Mandos.Client"

2159

2450

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

2451

self.assertTrue(self.bus.clients)

2160

2452

for clientpath in self.bus.clients:

2161

2453

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

2162

2454

client_interface, (False,)),

2163

2455

self.bus.calls)

2164

2456

2165

2457

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()

2166

2462

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

2167

for clientpath in self.bus.clients:

2168

self.assertIn(("RemoveClient", dbus_busname,

2169

dbus_server_path, dbus_server_interface,

2463

self.assertFalse(self.bus.clients)

2464

for clientpath in orig_clients:

2465

self.assertIn(("RemoveClient", busname,

2466

server_path, server_interface,

2170

2467

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

2171

2468

2172

2469

expected_json = {

2374

2671

else:

2375

2672

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

2376

2673

values_to_get = self.values_to_get

2674

self.assertTrue(values_to_get)

2377

2675

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

2378

2676

for clientpath in self.bus.clients:

2379

2677

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

2380

2678

Unique())

2381

2679

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

2680

self.assertTrue(self.bus.clients)

2382

2681

for clientpath in self.bus.clients:

2383

2682

value = (self.bus.clients[clientpath]

2384

2683

[self.propname])

2443

2742

class TestSetSecretCmd(TestPropertySetterCmd):

2444

2743

command = command.SetSecret

2445

2744

propname = "Secret"

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]

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)

2449

2751

2450

2752

2451

2753

class TestSetTimeoutCmd(TestPropertySetterCmd):

2504

2806

2505

2807

2506

2808

2507

def should_only_run_tests():

2809

def parse_test_args():

2810

# type: () -> argparse.Namespace

2508

2811

parser = argparse.ArgumentParser(add_help=False)

2509

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

2812

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

2813

parser.add_argument("--prefix", )

2510

2814

args, unknown_args = parser.parse_known_args()

2511

run_tests = args.check

2512

if run_tests:

2513

# Remove --check argument from sys.argv

2815

if args.check:

2816

# Remove test options from sys.argv

2514

2817

sys.argv[1:] = unknown_args

2515

return run_tests

2818

return args

2516

2819

2517

2820

# Add all tests from doctest strings

2518

2821

def load_tests(loader, tests, none):

2521

2824

return tests

2522

2825

2523

2826

if __name__ == "__main__":

2827

options = parse_test_args()

2524

2828

try:

2525

if should_only_run_tests():

2526

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

2527

unittest.main()

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=[""])

2528

2841

else:

2529

2842

main()

2530

2843

finally:

2531

2844

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: