/mandos/release : revision 237.7.703

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

49

if sys.version_info.major == 2:

49

50

__metaclass__ = type

50

str = unicode

51

input = raw_input

52

53

54

class gi:

55

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

56

class repository:

57

class GLib:

58

class Error(Exception):

59

pass

60

61

62

dbussy = None

63

ravel = None

64

dbus_python = None

65

pydbus = None

66

51

67

52

try:

68

import dbussy

69

import ravel

53

import pydbus

54

import gi

55

dbus_python = None

70

56

except ImportError:

71

try:

72

import pydbus

73

import gi

74

except ImportError:

75

import dbus as dbus_python

76

57

import dbus as dbus_python

58

pydbus = None

59

class gi:

60

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

61

class repository:

62

class GLib:

63

class Error(Exception):

64

pass

77

65

78

66

# Show warnings by default

79

67

if not sys.warnoptions:

80

68

import warnings

81

69

warnings.simplefilter("default")

82

70

83

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

84

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

71

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

72

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

85

73

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

86

74

87

75

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

88

76

89

77

if sys.version_info.major == 2:

78

str = unicode

90

79

import StringIO

91

80

io.StringIO = StringIO.StringIO

92

81

93

82

locale.setlocale(locale.LC_ALL, "")

94

83

95

version = "1.8.16"

84

version = "1.8.7"

96

85

97

86

98

87

def main():

105

94

clientnames = options.client

106

95

107

96

if options.debug:

108

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

97

log.setLevel(logging.DEBUG)

109

98

110

if dbussy is not None and ravel is not None:

111

bus = dbussy_adapter.CachingBus(dbussy, ravel)

112

elif pydbus is not None:

99

if pydbus is not None:

113

100

bus = pydbus_adapter.CachingBus(pydbus)

114

101

else:

115

102

bus = dbus_python_adapter.CachingBus(dbus_python)

259

246

return rfc3339_duration_to_delta(interval)

260

247

except ValueError as e:

261

248

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

262

" ".join(e.args))

249

' '.join(e.args))

263

250

return parse_pre_1_6_1_interval(interval)

264

251

265

252

395

382

396

383

397

384

def parse_pre_1_6_1_interval(interval):

398

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

385

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

399

386

and return a datetime.timedelta

400

387

401

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

402

True

403

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

404

True

405

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

406

True

407

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

408

True

409

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

410

True

411

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

412

True

413

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

388

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

389

True

390

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

391

True

392

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

393

True

394

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

395

True

396

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

397

True

398

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

399

True

400

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

414

401

True

415

402

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

416

>>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") \

417

... == datetime.timedelta(2, 480, 18000)

403

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

418

404

True

419

405

420

406

"""

489

475

class SystemBus:

490

476

491

477

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

492

493

478

def get_managed_objects(self, busname, objectpath):

494

479

return self.call_method("GetManagedObjects", busname,

495

480

objectpath,

496

481

self.object_manager_iface)

497

482

498

483

properties_iface = "org.freedesktop.DBus.Properties"

499

500

484

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

501

485

value):

502

486

self.call_method("Set", busname, objectpath,

503

487

self.properties_iface, interface, key,

504

488

value)

505

489

506

def call_method(self, methodname, busname, objectpath,

507

interface, *args):

508

raise NotImplementedError()

509

490

510

491

class MandosBus(SystemBus):

511

492

busname_domain = "se.recompile"

605

586

606

587

class SilenceLogger:

607

588

"Simple context manager to silence a particular logger"

608

609

589

def __init__(self, loggername):

610

590

self.logger = logging.getLogger(loggername)

611

591

621

601

def __exit__(self, exc_type, exc_val, exc_tb):

622

602

self.logger.removeFilter(self.nullfilter)

623

603

604

624

605

class CachingBus(SystemBus):

625

606

"""A caching layer for dbus_python_adapter.SystemBus"""

626

627

607

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

628

608

self.object_cache = {}

629

609

super(dbus_python_adapter.CachingBus,

630

610

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

631

632

611

def get_object(self, busname, objectpath):

633

612

try:

634

613

return self.object_cache[(busname, objectpath)]

636

615

new_object = super(

637

616

dbus_python_adapter.CachingBus,

638

617

self).get_object(busname, objectpath)

639

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

618

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

640

619

return new_object

641

620

642

621

689

668

690

669

class CachingBus(SystemBus):

691

670

"""A caching layer for pydbus_adapter.SystemBus"""

692

693

671

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

694

672

self.object_cache = {}

695

673

super(pydbus_adapter.CachingBus,

696

674

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

697

698

675

def get(self, busname, objectpath):

699

676

try:

700

677

return self.object_cache[(busname, objectpath)]

701

678

except KeyError:

702

679

new_object = (super(pydbus_adapter.CachingBus, self)

703

680

.get(busname, objectpath))

704

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

705

return new_object

706

707

708

class dbussy_adapter:

709

class SystemBus(dbus.SystemBus):

710

"""Use DBussy"""

711

712

def __init__(self, dbussy, ravel):

713

self.dbussy = dbussy

714

self.ravel = ravel

715

self.bus = ravel.system_bus()

716

717

@contextlib.contextmanager

718

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

719

try:

720

yield

721

except self.dbussy.DBusError as e:

722

# This does what "raise from" would do

723

exc = exception_class(*e.args)

724

exc.__cause__ = e

725

raise exc

726

727

def call_method(self, methodname, busname, objectpath,

728

interface, *args):

729

proxy_object = self.get_object(busname, objectpath)

730

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

731

interface, methodname,

732

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

733

iface = proxy_object.get_interface(interface)

734

method = getattr(iface, methodname)

735

with self.convert_exception(dbus.Error):

736

value = method(*args)

737

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

738

# list of one element with the return value

739

if value:

740

return self.type_filter(value[0])

741

742

def get_object(self, busname, objectpath):

743

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

744

busname, objectpath)

745

with self.convert_exception(dbus.ConnectFailed):

746

return self.bus[busname][objectpath]

747

748

def type_filter(self, value):

749

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

750

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

751

# Tuple[dbussy.DBUS.Signature, Any]

752

if isinstance(value, tuple):

753

if (len(value) == 2

754

and isinstance(value[0],

755

self.dbussy.DBUS.Signature)):

756

return self.type_filter(value[1])

757

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

758

return str(value)

759

# Also recurse into dictionaries

760

elif isinstance(value, dict):

761

return {self.type_filter(key):

762

self.type_filter(subval)

763

for key, subval in value.items()}

764

return value

765

766

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

767

value):

768

proxy_object = self.get_object(busname, objectpath)

769

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

770

objectpath, self.properties_iface, interface,

771

key, value)

772

if key == "Secret":

773

# DBussy wants a Byte Array to be a sequence of

774

# values, not a byte string

775

value = tuple(value)

776

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

777

778

class MandosBus(SystemBus, dbus.MandosBus):

779

pass

780

781

class CachingBus(MandosBus):

782

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

783

784

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

785

self.object_cache = {}

786

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

787

**kwargs)

788

789

def get_object(self, busname, objectpath):

790

try:

791

return self.object_cache[(busname, objectpath)]

792

except KeyError:

793

new_object = super(

794

dbussy_adapter.CachingBus,

795

self).get_object(busname, objectpath)

796

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

681

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

797

682

return new_object

798

683

799

684

835

720

836

721

class Base:

837

722

"""Abstract base class for commands"""

838

839

723

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

840

724

"""Normal commands should implement run_on_one_client(),

841

725

but commands which want to operate on all clients at the same time can

845

729

for client, properties in clients.items():

846

730

self.run_on_one_client(client, properties)

847

731

732

848

733

class IsEnabled(Base):

849

734

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

850

735

properties = next(iter(clients.values()))

852

737

sys.exit(0)

853

738

sys.exit(1)

854

739

740

855

741

class Approve(Base):

856

742

def run_on_one_client(self, client, properties):

857

743

self.bus.call_client_method(client, "Approve", True)

858

744

745

859

746

class Deny(Base):

860

747

def run_on_one_client(self, client, properties):

861

748

self.bus.call_client_method(client, "Approve", False)

862

749

750

863

751

class Remove(Base):

864

752

def run(self, clients, bus):

865

753

for clientpath in frozenset(clients.keys()):

866

754

bus.call_server_method("RemoveClient", clientpath)

867

755

756

868

757

class Output(Base):

869

758

"""Abstract class for commands outputting client details"""

870

759

all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",

876

765

"Checker", "ExtendedTimeout", "Expires",

877

766

"LastCheckerStatus")

878

767

768

879

769

class DumpJSON(Output):

880

770

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

881

771

data = {properties["Name"]:

882

772

{key: properties[key]

883

773

for key in self.all_keywords}

884

774

for properties in clients.values()}

885

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

775

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

776

886

777

887

778

class PrintTable(Output):

888

779

def __init__(self, verbose=False):

929

820

930

821

if sys.version_info.major == 2:

931

822

__unicode__ = __str__

932

933

823

def __str__(self):

934

824

return str(self).encode(

935

825

locale.getpreferredencoding())

981

871

minutes=(td.seconds % 3600) // 60,

982

872

seconds=td.seconds % 60))

983

873

874

984

875

class PropertySetter(Base):

985

876

"Abstract class for Actions for setting one client property"

986

877

993

884

def propname(self):

994

885

raise NotImplementedError()

995

886

887

996

888

class Enable(PropertySetter):

997

889

propname = "Enabled"

998

890

value_to_set = True

999

891

892

1000

893

class Disable(PropertySetter):

1001

894

propname = "Enabled"

1002

895

value_to_set = False

1003

896

897

1004

898

class BumpTimeout(PropertySetter):

1005

899

propname = "LastCheckedOK"

1006

900

value_to_set = ""

1007

901

902

1008

903

class StartChecker(PropertySetter):

1009

904

propname = "CheckerRunning"

1010

905

value_to_set = True

1011

906

907

1012

908

class StopChecker(PropertySetter):

1013

909

propname = "CheckerRunning"

1014

910

value_to_set = False

1015

911

912

1016

913

class ApproveByDefault(PropertySetter):

1017

914

propname = "ApprovedByDefault"

1018

915

value_to_set = True

1019

916

917

1020

918

class DenyByDefault(PropertySetter):

1021

919

propname = "ApprovedByDefault"

1022

920

value_to_set = False

1023

921

922

1024

923

class PropertySetterValue(PropertySetter):

1025

924

"""Abstract class for PropertySetter recieving a value as

1026

925

constructor argument instead of a class attribute."""

1027

1028

926

def __init__(self, value):

1029

927

self.value_to_set = value

1030

928

1037

935

class SetChecker(PropertySetterValue):

1038

936

propname = "Checker"

1039

937

938

1040

939

class SetHost(PropertySetterValue):

1041

940

propname = "Host"

1042

941

942

1043

943

class SetSecret(PropertySetterValue):

1044

944

propname = "Secret"

1045

945

1053

953

self._vts = value.read()

1054

954

value.close()

1055

955

956

1056

957

class PropertySetterValueMilliseconds(PropertySetterValue):

1057

958

"""Abstract class for PropertySetterValue taking a value

1058

959

argument as a datetime.timedelta() but should store it as

1067

968

"When setting, convert value from a datetime.timedelta"

1068

969

self._vts = int(round(value.total_seconds() * 1000))

1069

970

971

1070

972

class SetTimeout(PropertySetterValueMilliseconds):

1071

973

propname = "Timeout"

1072

974

975

1073

976

class SetExtendedTimeout(PropertySetterValueMilliseconds):

1074

977

propname = "ExtendedTimeout"

1075

978

979

1076

980

class SetInterval(PropertySetterValueMilliseconds):

1077

981

propname = "Interval"

1078

982

983

1079

984

class SetApprovalDelay(PropertySetterValueMilliseconds):

1080

985

propname = "ApprovalDelay"

1081

986

987

1082

988

class SetApprovalDuration(PropertySetterValueMilliseconds):

1083

989

propname = "ApprovalDuration"

1084

990

1624

1530

finally:

1625

1531

dbus_logger.removeFilter(counting_handler)

1626

1532

1627

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

1533

self.assertNotIsInstance(e, dbus.ConnectFailed)

1628

1534

1629

1535

# Make sure the dbus logger was suppressed

1630

1536

self.assertEqual(0, counting_handler.count)

1755

1661

self.assertIs(ret, expected_method_return)

1756

1662

1757

1663

def test_call_method_handles_exception(self):

1664

dbus_logger = logging.getLogger("dbus.proxies")

1665

1758

1666

def func():

1759

1667

raise gi.repository.GLib.Error()

1760

1668

1765

1673

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

1766

1674

"objectpath", "interface")

1767

1675

1768

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

1676

self.assertNotIsInstance(e, dbus.ConnectFailed)

1769

1677

1770

1678

def test_get_converts_to_correct_exception(self):

1771

1679

bus = pydbus_adapter.SystemBus(

1866

1774

self.assertIs(obj1, obj1b)

1867

1775

1868

1776

1869

class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):

1870

1871

class dummy_dbussy:

1872

class DBUS:

1873

class ObjectPath(str):

1874

pass

1875

class DBusError(Exception):

1876

pass

1877

1878

def fake_ravel_func(self, func):

1879

class fake_ravel:

1880

@staticmethod

1881

def system_bus():

1882

class DBusInterfaceProxy:

1883

@staticmethod

1884

def methodname(*args):

1885

return [func(*args)]

1886

class DBusObject:

1887

@staticmethod

1888

def get_interface(interface):

1889

if interface == "interface":

1890

return DBusInterfaceProxy()

1891

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

1892

return fake_ravel

1893

1894

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

1895

interface, *args):

1896

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

1897

return bus.call_method(methodname, busname, objectpath,

1898

interface, *args)

1899

1900

def test_call_method_returns(self):

1901

expected_method_return = Unique()

1902

method_args = (Unique(), Unique())

1903

def func(*args):

1904

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

1905

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

1906

self.assertIs(marg, arg)

1907

return expected_method_return

1908

fake_ravel = self.fake_ravel_func(func)

1909

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

1910

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

1911

"objectpath", "interface",

1912

*method_args)

1913

self.assertIs(ret, expected_method_return)

1914

1915

def test_call_method_filters_objectpath(self):

1916

def func():

1917

return method_return

1918

fake_ravel = self.fake_ravel_func(func)

1919

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

1920

method_return = (self.dummy_dbussy.DBUS

1921

.ObjectPath("objectpath"))

1922

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

1923

"objectpath", "interface")

1924

self.assertEqual("objectpath", ret)

1925

self.assertNotIsInstance(ret,

1926

self.dummy_dbussy.DBUS.ObjectPath)

1927

1928

def test_call_method_filters_objectpaths_in_dict(self):

1929

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1930

def func():

1931

return method_return

1932

fake_ravel = self.fake_ravel_func(func)

1933

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

1934

method_return = {

1935

ObjectPath("objectpath_key_1"):

1936

ObjectPath("objectpath_value_1"),

1937

ObjectPath("objectpath_key_2"):

1938

ObjectPath("objectpath_value_2"),

1939

}

1940

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

1941

"objectpath", "interface")

1942

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

1943

for key, value in

1944

method_return.items()}

1945

for key, value in ret.items():

1946

self.assertNotIsInstance(key, ObjectPath)

1947

self.assertNotIsInstance(value, ObjectPath)

1948

self.assertEqual(expected_method_return, ret)

1949

self.assertIsInstance(ret, dict)

1950

1951

def test_call_method_filters_objectpaths_in_dict_in_dict(self):

1952

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1953

def func():

1954

return method_return

1955

fake_ravel = self.fake_ravel_func(func)

1956

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

1957

method_return = {

1958

ObjectPath("key1"): {

1959

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

1960

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

1961

},

1962

ObjectPath("key2"): {

1963

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

1964

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

1965

},

1966

}

1967

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

1968

"objectpath", "interface")

1969

expected_method_return = {

1970

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

1971

"key12": "value12"},

1972

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

1973

"key22": "value22"},

1974

}

1975

self.assertEqual(expected_method_return, ret)

1976

for key, value in ret.items():

1977

self.assertIsInstance(value, dict)

1978

self.assertEqual(expected_method_return[key], value)

1979

self.assertNotIsInstance(key, ObjectPath)

1980

for inner_key, inner_value in value.items():

1981

self.assertIsInstance(value, dict)

1982

self.assertEqual(

1983

expected_method_return[key][inner_key],

1984

inner_value)

1985

self.assertNotIsInstance(key, ObjectPath)

1986

1987

def test_call_method_filters_objectpaths_in_dict_three_deep(self):

1988

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1989

def func():

1990

return method_return

1991

fake_ravel = self.fake_ravel_func(func)

1992

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

1993

method_return = {

1994

ObjectPath("key1"): {

1995

ObjectPath("key2"): {

1996

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

1997

},

1998

},

1999

}

2000

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

2001

"objectpath", "interface")

2002

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

2003

self.assertEqual(expected_method_return, ret)

2004

self.assertIsInstance(ret, dict)

2005

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

2006

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

2007

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

2008

ObjectPath)

2009

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

2010

self.assertNotIsInstance(

2011

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

2012

ObjectPath)

2013

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

2014

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

2015

self.dummy_dbussy.DBUS.ObjectPath)

2016

2017

def test_call_method_handles_exception(self):

2018

def func():

2019

raise self.dummy_dbussy.DBusError()

2020

2021

fake_ravel = self.fake_ravel_func(func)

2022

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

2023

2024

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

2025

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

2026

"objectpath", "interface")

2027

2028

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

2029

2030

def test_get_object_converts_to_correct_exception(self):

2031

class fake_ravel_raises_exception_on_connect:

2032

@staticmethod

2033

def system_bus():

2034

class Bus:

2035

@staticmethod

2036

def __getitem__(key):

2037

if key == "objectpath":

2038

raise self.dummy_dbussy.DBusError()

2039

raise Exception(key)

2040

return {"busname": Bus()}

2041

def func():

2042

raise self.dummy_dbussy.DBusError()

2043

bus = dbussy_adapter.SystemBus(

2044

self.dummy_dbussy,

2045

fake_ravel_raises_exception_on_connect)

2046

with self.assertRaises(dbus.ConnectFailed):

2047

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

2048

"objectpath", "interface")

2049

2050

2051

1777

class Test_commands_from_options(unittest.TestCase):

2052

1778

2053

1779

def setUp(self):

2427

2153

busname = "se.recompile.Mandos"

2428

2154

client_interface = "se.recompile.Mandos.Client"

2429

2155

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

2430

self.assertTrue(self.bus.clients)

2431

2156

for clientpath in self.bus.clients:

2432

2157

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

2433

2158

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

2436

2161

busname = "se.recompile.Mandos"

2437

2162

client_interface = "se.recompile.Mandos.Client"

2438

2163

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

2439

self.assertTrue(self.bus.clients)

2440

2164

for clientpath in self.bus.clients:

2441

2165

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

2442

2166

client_interface, (False,)),

2443

2167

self.bus.calls)

2444

2168

2445

2169

def test_Remove(self):

2446

busname = "se.recompile.Mandos"

2447

server_path = "/"

2448

server_interface = "se.recompile.Mandos"

2449

orig_clients = self.bus.clients.copy()

2450

2170

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

2451

self.assertFalse(self.bus.clients)

2452

for clientpath in orig_clients:

2453

self.assertIn(("RemoveClient", busname,

2454

server_path, server_interface,

2171

for clientpath in self.bus.clients:

2172

self.assertIn(("RemoveClient", dbus_busname,

2173

dbus_server_path, dbus_server_interface,

2455

2174

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

2456

2175

2457

2176

expected_json = {

2659

2378

else:

2660

2379

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

2661

2380

values_to_get = self.values_to_get

2662

self.assertTrue(values_to_get)

2663

2381

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

2664

2382

for clientpath in self.bus.clients:

2665

2383

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

2666

2384

Unique())

2667

2385

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

2668

self.assertTrue(self.bus.clients)

2669

2386

for clientpath in self.bus.clients:

2670

2387

value = (self.bus.clients[clientpath]

2671

2388

[self.propname])

2730

2447

class TestSetSecretCmd(TestPropertySetterCmd):

2731

2448

command = command.SetSecret

2732

2449

propname = "Secret"

2733

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

2734

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

2735

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

2736

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

2737

self.values_to_set]

2738

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

2450

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

2451

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

2452

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

2739

2453

2740

2454

2741

2455

class TestSetTimeoutCmd(TestPropertySetterCmd):

2794

2508

2795

2509

2796

2510

2797

def parse_test_args():

2798

# type: () -> argparse.Namespace

2511

def should_only_run_tests():

2799

2512

parser = argparse.ArgumentParser(add_help=False)

2800

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

2801

parser.add_argument("--prefix", )

2513

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

2802

2514

args, unknown_args = parser.parse_known_args()

2803

if args.check:

2804

# Remove test options from sys.argv

2515

run_tests = args.check

2516

if run_tests:

2517

# Remove --check argument from sys.argv

2805

2518

sys.argv[1:] = unknown_args

2806

return args

2519

return run_tests

2807

2520

2808

2521

# Add all tests from doctest strings

2809

2522

def load_tests(loader, tests, none):

2812

2525

return tests

2813

2526

2814

2527

if __name__ == "__main__":

2815

options = parse_test_args()

2816

2528

try:

2817

if options.check:

2818

extra_test_prefix = options.prefix

2819

if extra_test_prefix is not None:

2820

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

2821

.result.wasSuccessful()):

2822

sys.exit(1)

2823

class ExtraTestLoader(unittest.TestLoader):

2824

testMethodPrefix = extra_test_prefix

2825

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

2826

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

2827

else:

2828

unittest.main(argv=[""])

2529

if should_only_run_tests():

2530

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

2531

unittest.main()

2829

2532

else:

2830

2533

main()

2831

2534

finally:

2832

2535

logging.shutdown()

2833

2834

# Local Variables:

2835

# run-tests:

2836

# (lambda (&optional extra)

2837

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

2838

# extra))

2839

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

2840

# (funcall remove-test-window)

2841

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

2842

# run-tests-in-test-buffer:

2843

# (lambda (dir &optional extra)

2844

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

2845

# (setq buffer-read-only nil

2846

# default-directory dir)

2847

# (erase-buffer)

2848

# (compilation-mode))

2849

# (let ((process-result

2850

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

2851

# (process-file-shell-command

2852

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

2853

# (and (numberp process-result)

2854

# (= process-result 0))))

2855

# get-command-line:

2856

# (lambda (&optional extra)

2857

# (let ((quoted-script

2858

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

2859

# (format

2860

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

2861

# quoted-script)))

2862

# get-script-name:

2863

# (lambda ()

2864

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

2865

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

2866

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

2867

# (buffer-file-name))))

2868

# remove-test-window:

2869

# (lambda ()

2870

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

2871

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

2872

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

2873

# (lambda ()

2874

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

2875

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

2876

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

2877

# (display-buffer-overriding-action

2878

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

2879

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

2880

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

2881

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

2882

# eval:

2883

# (progn

2884

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

2885

# (funcall run-tests t)))

2886

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

2887

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

2888

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

2889

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

2890

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

2891

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

2892

# End: