/mandos/trunk : revision 1261

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 --check" (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.6"

92

version = "1.8.15"

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

382

395

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

383

396

and return a datetime.timedelta

384

397

385

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

386

True

387

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

388

True

389

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

390

True

391

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

392

True

393

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

394

True

395

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

396

True

397

>>> parse_pre_1_6_1_interval('') == 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)

398

411

True

399

412

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

400

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

413

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

401

414

True

402

415

403

416

"""

467

480

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

468

481

469

482

470

class dbus(object):

483

class dbus:

471

484

472

class SystemBus(object):

485

class SystemBus:

473

486

474

487

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

475

488

def get_managed_objects(self, busname, objectpath):

484

497

self.properties_iface, interface, key,

485

498

value)

486

499

500

def call_method(self, methodname, busname, objectpath,

501

interface, *args):

502

raise NotImplementedError()

503

487

504

488

505

class MandosBus(SystemBus):

489

506

busname_domain = "se.recompile"

521

538

pass

522

539

523

540

524

class dbus_python_adapter(object):

541

class dbus_python_adapter:

525

542

526

543

class SystemBus(dbus.MandosBus):

527

544

"""Use dbus-python"""

581

598

self.client_interface, key,

582

599

value)

583

600

584

class SilenceLogger(object):

601

class SilenceLogger:

585

602

"Simple context manager to silence a particular logger"

586

603

def __init__(self, loggername):

587

604

self.logger = logging.getLogger(loggername)

616

633

return new_object

617

634

618

635

619

class pydbus_adapter(object):

636

class pydbus_adapter:

620

637

class SystemBus(dbus.MandosBus):

621

638

def __init__(self, module=pydbus):

622

639

self.pydbus = module

679

696

return new_object

680

697

681

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

682

789

def commands_from_options(options):

683

790

684

791

commands = list(options.commands)

712

819

return commands

713

820

714

821

715

class command(object):

822

class command:

716

823

"""A namespace for command classes"""

717

824

718

class Base(object):

825

class Base:

719

826

"""Abstract base class for commands"""

720

827

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

721

828

"""Normal commands should implement run_on_one_client(),

769

876

{key: properties[key]

770

877

for key in self.all_keywords}

771

878

for properties in clients.values()}

772

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

879

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

773

880

774

881

775

882

class PrintTable(Output):

784

891

keywords = self.all_keywords

785

892

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

786

893

787

class TableOfClients(object):

894

class TableOfClients:

788

895

tableheaders = {

789

896

"Name": "Name",

790

897

"Enabled": "Enabled",

1021

1128

"output"))

1022

1129

1023

1130

1024

class Unique(object):

1131

class Unique:

1025

1132

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

1026

1133

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

1027

1134

1311

1418

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1312

1419

1313

1420

def MockDBusPython_func(self, func):

1314

class mock_dbus_python(object):

1421

class mock_dbus_python:

1315

1422

"""mock dbus-python module"""

1316

class exceptions(object):

1423

class exceptions:

1317

1424

"""Pseudo-namespace"""

1318

1425

class DBusException(Exception):

1319

1426

pass

1320

class SystemBus(object):

1427

class SystemBus:

1321

1428

@staticmethod

1322

1429

def get_object(busname, objectpath):

1323

1430

DBusObject = collections.namedtuple(

1337

1444

dbus_interface=dbus_interface)

1338

1445

return DBusObject(methodname=method,

1339

1446

Set=set_property)

1340

class Boolean(object):

1447

class Boolean:

1341

1448

def __init__(self, value):

1342

1449

self.value = bool(value)

1343

1450

def __bool__(self):

1527

1634

finally:

1528

1635

dbus_logger.removeFilter(counting_handler)

1529

1636

1530

self.assertNotIsInstance(e, dbus.ConnectFailed)

1637

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

1531

1638

1532

1639

# Make sure the dbus logger was suppressed

1533

1640

self.assertEqual(0, counting_handler.count)

1555

1662

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

1556

1663

"objectpath", "interface")

1557

1664

1558

class fake_dbus_python_raises_exception_on_connect(object):

1665

class fake_dbus_python_raises_exception_on_connect:

1559

1666

"""fake dbus-python module"""

1560

class exceptions(object):

1667

class exceptions:

1561

1668

"""Pseudo-namespace"""

1562

1669

class DBusException(Exception):

1563

1670

pass

1571

1678

1572

1679

1573

1680

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1574

class mock_dbus_python(object):

1681

class mock_dbus_python:

1575

1682

"""mock dbus-python modules"""

1576

class SystemBus(object):

1683

class SystemBus:

1577

1684

@staticmethod

1578

1685

def get_object(busname, objectpath):

1579

1686

return Unique()

1625

1732

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1626

1733

1627

1734

def Stub_pydbus_func(self, func):

1628

class stub_pydbus(object):

1735

class stub_pydbus:

1629

1736

"""stub pydbus module"""

1630

class SystemBus(object):

1737

class SystemBus:

1631

1738

@staticmethod

1632

1739

def get(busname, objectpath):

1633

1740

DBusObject = collections.namedtuple(

1670

1777

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

1671

1778

"objectpath", "interface")

1672

1779

1673

self.assertNotIsInstance(e, dbus.ConnectFailed)

1780

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

1674

1781

1675

1782

def test_get_converts_to_correct_exception(self):

1676

1783

bus = pydbus_adapter.SystemBus(

1679

1786

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

1680

1787

"objectpath", "interface")

1681

1788

1682

class fake_pydbus_raises_exception_on_connect(object):

1789

class fake_pydbus_raises_exception_on_connect:

1683

1790

"""fake dbus-python module"""

1684

1791

@classmethod

1685

1792

def SystemBus(cls):

1689

1796

return Bus(get=get)

1690

1797

1691

1798

def test_set_property_uses_setattr(self):

1692

class Object(object):

1799

class Object:

1693

1800

pass

1694

1801

obj = Object()

1695

class pydbus_spy(object):

1696

class SystemBus(object):

1802

class pydbus_spy:

1803

class SystemBus:

1697

1804

@staticmethod

1698

1805

def get(busname, objectpath):

1699

1806

return {"interface": obj}

1706

1813

def test_get_suppresses_xml_deprecation_warning(self):

1707

1814

if sys.version_info.major >= 3:

1708

1815

return

1709

class stub_pydbus_get(object):

1710

class SystemBus(object):

1816

class stub_pydbus_get:

1817

class SystemBus:

1711

1818

@staticmethod

1712

1819

def get(busname, objectpath):

1713

1820

warnings.warn_explicit(

1721

1828

1722

1829

1723

1830

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1724

class stub_pydbus(object):

1831

class stub_pydbus:

1725

1832

"""stub pydbus module"""

1726

class SystemBus(object):

1833

class SystemBus:

1727

1834

@staticmethod

1728

1835

def get(busname, objectpath):

1729

1836

return Unique()

1771

1878

self.assertIs(obj1, obj1b)

1772

1879

1773

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

1774

2063

class Test_commands_from_options(unittest.TestCase):

1775

2064

1776

2065

def setUp(self):

2150

2439

busname = "se.recompile.Mandos"

2151

2440

client_interface = "se.recompile.Mandos.Client"

2152

2441

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

2442

self.assertTrue(self.bus.clients)

2153

2443

for clientpath in self.bus.clients:

2154

2444

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

2155

2445

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

2158

2448

busname = "se.recompile.Mandos"

2159

2449

client_interface = "se.recompile.Mandos.Client"

2160

2450

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

2451

self.assertTrue(self.bus.clients)

2161

2452

for clientpath in self.bus.clients:

2162

2453

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

2163

2454

client_interface, (False,)),

2164

2455

self.bus.calls)

2165

2456

2166

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

2167

2462

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

2168

for clientpath in self.bus.clients:

2169

self.assertIn(("RemoveClient", dbus_busname,

2170

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,

2171

2467

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

2172

2468

2173

2469

expected_json = {

2375

2671

else:

2376

2672

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

2377

2673

values_to_get = self.values_to_get

2674

self.assertTrue(values_to_get)

2378

2675

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

2379

2676

for clientpath in self.bus.clients:

2380

2677

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

2381

2678

Unique())

2382

2679

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

2680

self.assertTrue(self.bus.clients)

2383

2681

for clientpath in self.bus.clients:

2384

2682

value = (self.bus.clients[clientpath]

2385

2683

[self.propname])

2444

2742

class TestSetSecretCmd(TestPropertySetterCmd):

2445

2743

command = command.SetSecret

2446

2744

propname = "Secret"

2447

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

2448

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

2449

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)

2450

2751

2451

2752

2452

2753

class TestSetTimeoutCmd(TestPropertySetterCmd):

2505

2806

2506

2807

2507

2808

2508

def should_only_run_tests():

2809

def parse_test_args():

2810

# type: () -> argparse.Namespace

2509

2811

parser = argparse.ArgumentParser(add_help=False)

2510

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

2812

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

2813

parser.add_argument("--prefix", )

2511

2814

args, unknown_args = parser.parse_known_args()

2512

run_tests = args.check

2513

if run_tests:

2514

# Remove --check argument from sys.argv

2815

if args.check:

2816

# Remove test options from sys.argv

2515

2817

sys.argv[1:] = unknown_args

2516

return run_tests

2818

return args

2517

2819

2518

2820

# Add all tests from doctest strings

2519

2821

def load_tests(loader, tests, none):

2522

2824

return tests

2523

2825

2524

2826

if __name__ == "__main__":

2827

options = parse_test_args()

2525

2828

try:

2526

if should_only_run_tests():

2527

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

2528

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

2529

2841

else:

2530

2842

main()

2531

2843

finally:

2532

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: