/mandos/trunk : revision 237.4.120

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

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.7"

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

395

382

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

396

383

and return a datetime.timedelta

397

384

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)

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)

411

398

True

412

399

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

413

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

400

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

414

401

True

415

402

416

403

"""

480

467

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

481

468

482

469

483

class dbus:

470

class dbus(object):

484

471

485

class SystemBus:

472

class SystemBus(object):

486

473

487

474

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

488

475

def get_managed_objects(self, busname, objectpath):

497

484

self.properties_iface, interface, key,

498

485

value)

499

486

500

def call_method(self, methodname, busname, objectpath,

501

interface, *args):

502

raise NotImplementedError()

503

504

487

505

488

class MandosBus(SystemBus):

506

489

busname_domain = "se.recompile"

538

521

pass

539

522

540

523

541

class dbus_python_adapter:

524

class dbus_python_adapter(object):

542

525

543

526

class SystemBus(dbus.MandosBus):

544

527

"""Use dbus-python"""

598

581

self.client_interface, key,

599

582

value)

600

583

601

class SilenceLogger:

584

class SilenceLogger(object):

602

585

"Simple context manager to silence a particular logger"

603

586

def __init__(self, loggername):

604

587

self.logger = logging.getLogger(loggername)

633

616

return new_object

634

617

635

618

636

class pydbus_adapter:

619

class pydbus_adapter(object):

637

620

class SystemBus(dbus.MandosBus):

638

621

def __init__(self, module=pydbus):

639

622

self.pydbus = module

696

679

return new_object

697

680

698

681

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

682

def commands_from_options(options):

790

683

791

684

commands = list(options.commands)

819

712

return commands

820

713

821

714

822

class command:

715

class command(object):

823

716

"""A namespace for command classes"""

824

717

825

class Base:

718

class Base(object):

826

719

"""Abstract base class for commands"""

827

720

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

828

721

"""Normal commands should implement run_on_one_client(),

876

769

{key: properties[key]

877

770

for key in self.all_keywords}

878

771

for properties in clients.values()}

879

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

772

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

880

773

881

774

882

775

class PrintTable(Output):

891

784

keywords = self.all_keywords

892

785

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

893

786

894

class TableOfClients:

787

class TableOfClients(object):

895

788

tableheaders = {

896

789

"Name": "Name",

897

790

"Enabled": "Enabled",

1128

1021

"output"))

1129

1022

1130

1023

1131

class Unique:

1024

class Unique(object):

1132

1025

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

1133

1026

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

1134

1027

1418

1311

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1419

1312

1420

1313

def MockDBusPython_func(self, func):

1421

class mock_dbus_python:

1314

class mock_dbus_python(object):

1422

1315

"""mock dbus-python module"""

1423

class exceptions:

1316

class exceptions(object):

1424

1317

"""Pseudo-namespace"""

1425

1318

class DBusException(Exception):

1426

1319

pass

1427

class SystemBus:

1320

class SystemBus(object):

1428

1321

@staticmethod

1429

1322

def get_object(busname, objectpath):

1430

1323

DBusObject = collections.namedtuple(

1444

1337

dbus_interface=dbus_interface)

1445

1338

return DBusObject(methodname=method,

1446

1339

Set=set_property)

1447

class Boolean:

1340

class Boolean(object):

1448

1341

def __init__(self, value):

1449

1342

self.value = bool(value)

1450

1343

def __bool__(self):

1634

1527

finally:

1635

1528

dbus_logger.removeFilter(counting_handler)

1636

1529

1637

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

1530

self.assertNotIsInstance(e, dbus.ConnectFailed)

1638

1531

1639

1532

# Make sure the dbus logger was suppressed

1640

1533

self.assertEqual(0, counting_handler.count)

1662

1555

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

1663

1556

"objectpath", "interface")

1664

1557

1665

class fake_dbus_python_raises_exception_on_connect:

1558

class fake_dbus_python_raises_exception_on_connect(object):

1666

1559

"""fake dbus-python module"""

1667

class exceptions:

1560

class exceptions(object):

1668

1561

"""Pseudo-namespace"""

1669

1562

class DBusException(Exception):

1670

1563

pass

1678

1571

1679

1572

1680

1573

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1681

class mock_dbus_python:

1574

class mock_dbus_python(object):

1682

1575

"""mock dbus-python modules"""

1683

class SystemBus:

1576

class SystemBus(object):

1684

1577

@staticmethod

1685

1578

def get_object(busname, objectpath):

1686

1579

return Unique()

1732

1625

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1733

1626

1734

1627

def Stub_pydbus_func(self, func):

1735

class stub_pydbus:

1628

class stub_pydbus(object):

1736

1629

"""stub pydbus module"""

1737

class SystemBus:

1630

class SystemBus(object):

1738

1631

@staticmethod

1739

1632

def get(busname, objectpath):

1740

1633

DBusObject = collections.namedtuple(

1777

1670

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

1778

1671

"objectpath", "interface")

1779

1672

1780

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

1673

self.assertNotIsInstance(e, dbus.ConnectFailed)

1781

1674

1782

1675

def test_get_converts_to_correct_exception(self):

1783

1676

bus = pydbus_adapter.SystemBus(

1786

1679

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

1787

1680

"objectpath", "interface")

1788

1681

1789

class fake_pydbus_raises_exception_on_connect:

1682

class fake_pydbus_raises_exception_on_connect(object):

1790

1683

"""fake dbus-python module"""

1791

1684

@classmethod

1792

1685

def SystemBus(cls):

1796

1689

return Bus(get=get)

1797

1690

1798

1691

def test_set_property_uses_setattr(self):

1799

class Object:

1692

class Object(object):

1800

1693

pass

1801

1694

obj = Object()

1802

class pydbus_spy:

1803

class SystemBus:

1695

class pydbus_spy(object):

1696

class SystemBus(object):

1804

1697

@staticmethod

1805

1698

def get(busname, objectpath):

1806

1699

return {"interface": obj}

1813

1706

def test_get_suppresses_xml_deprecation_warning(self):

1814

1707

if sys.version_info.major >= 3:

1815

1708

return

1816

class stub_pydbus_get:

1817

class SystemBus:

1709

class stub_pydbus_get(object):

1710

class SystemBus(object):

1818

1711

@staticmethod

1819

1712

def get(busname, objectpath):

1820

1713

warnings.warn_explicit(

1828

1721

1829

1722

1830

1723

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1831

class stub_pydbus:

1724

class stub_pydbus(object):

1832

1725

"""stub pydbus module"""

1833

class SystemBus:

1726

class SystemBus(object):

1834

1727

@staticmethod

1835

1728

def get(busname, objectpath):

1836

1729

return Unique()

1878

1771

self.assertIs(obj1, obj1b)

1879

1772

1880

1773

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

1774

class Test_commands_from_options(unittest.TestCase):

2064

1775

2065

1776

def setUp(self):

2439

2150

busname = "se.recompile.Mandos"

2440

2151

client_interface = "se.recompile.Mandos.Client"

2441

2152

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

2442

self.assertTrue(self.bus.clients)

2443

2153

for clientpath in self.bus.clients:

2444

2154

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

2445

2155

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

2448

2158

busname = "se.recompile.Mandos"

2449

2159

client_interface = "se.recompile.Mandos.Client"

2450

2160

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

2451

self.assertTrue(self.bus.clients)

2452

2161

for clientpath in self.bus.clients:

2453

2162

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

2454

2163

client_interface, (False,)),

2455

2164

self.bus.calls)

2456

2165

2457

2166

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

2167

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,

2168

for clientpath in self.bus.clients:

2169

self.assertIn(("RemoveClient", dbus_busname,

2170

dbus_server_path, dbus_server_interface,

2467

2171

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

2468

2172

2469

2173

expected_json = {

2671

2375

else:

2672

2376

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

2673

2377

values_to_get = self.values_to_get

2674

self.assertTrue(values_to_get)

2675

2378

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

2676

2379

for clientpath in self.bus.clients:

2677

2380

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

2678

2381

Unique())

2679

2382

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

2680

self.assertTrue(self.bus.clients)

2681

2383

for clientpath in self.bus.clients:

2682

2384

value = (self.bus.clients[clientpath]

2683

2385

[self.propname])

2742

2444

class TestSetSecretCmd(TestPropertySetterCmd):

2743

2445

command = command.SetSecret

2744

2446

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)

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]

2751

2450

2752

2451

2753

2452

class TestSetTimeoutCmd(TestPropertySetterCmd):

2806

2505

2807

2506

2808

2507

2809

def parse_test_args():

2810

# type: () -> argparse.Namespace

2508

def should_only_run_tests():

2811

2509

parser = argparse.ArgumentParser(add_help=False)

2812

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

2813

parser.add_argument("--prefix", )

2510

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

2814

2511

args, unknown_args = parser.parse_known_args()

2815

if args.check:

2816

# Remove test options from sys.argv

2512

run_tests = args.check

2513

if run_tests:

2514

# Remove --check argument from sys.argv

2817

2515

sys.argv[1:] = unknown_args

2818

return args

2516

return run_tests

2819

2517

2820

2518

# Add all tests from doctest strings

2821

2519

def load_tests(loader, tests, none):

2824

2522

return tests

2825

2523

2826

2524

if __name__ == "__main__":

2827

options = parse_test_args()

2828

2525

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

2526

if should_only_run_tests():

2527

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

2528

unittest.main()

2841

2529

else:

2842

2530

main()

2843

2531

finally:

2844

2532

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: