/mandos/trunk : revision 1305

1

#!/usr/bin/python3 -bbI

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

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

49

48

if sys.version_info.major == 2:

50

49

__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

51

66

52

67

try:

53

import pydbus

54

import gi

55

dbus_python = None

68

import dbussy

69

import ravel

56

70

except ImportError:

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

71

try:

72

import pydbus

73

import gi

74

except ImportError:

75

import dbus as dbus_python

76

65

77

66

78

# Show warnings by default

67

79

if not sys.warnoptions:

68

80

import warnings

69

81

warnings.simplefilter("default")

70

82

71

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

72

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

83

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

84

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

73

85

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

74

86

75

87

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

76

88

77

89

if sys.version_info.major == 2:

78

str = unicode

79

90

import StringIO

80

91

io.StringIO = StringIO.StringIO

81

92

82

93

locale.setlocale(locale.LC_ALL, "")

83

94

84

version = "1.8.9"

95

version = "1.8.17"

85

96

86

97

87

98

def main():

94

105

clientnames = options.client

95

106

96

107

if options.debug:

97

log.setLevel(logging.DEBUG)

108

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

98

109

99

if pydbus is not None:

110

if dbussy is not None and ravel is not None:

111

bus = dbussy_adapter.CachingBus(dbussy, ravel)

112

elif pydbus is not None:

100

113

bus = pydbus_adapter.CachingBus(pydbus)

101

114

else:

102

115

bus = dbus_python_adapter.CachingBus(dbus_python)

246

259

return rfc3339_duration_to_delta(interval)

247

260

except ValueError as e:

248

261

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

249

' '.join(e.args))

262

" ".join(e.args))

250

263

return parse_pre_1_6_1_interval(interval)

251

264

252

265

382

395

383

396

384

397

def parse_pre_1_6_1_interval(interval):

385

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

398

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

386

399

and return a datetime.timedelta

387

400

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)

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)

401

414

True

402

415

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

403

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

416

>>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") \

417

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

404

418

True

405

419

406

420

"""

475

489

class SystemBus:

476

490

477

491

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

492

478

493

def get_managed_objects(self, busname, objectpath):

479

494

return self.call_method("GetManagedObjects", busname,

480

495

objectpath,

481

496

self.object_manager_iface)

482

497

483

498

properties_iface = "org.freedesktop.DBus.Properties"

499

484

500

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

485

501

value):

486

502

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

487

503

self.properties_iface, interface, key,

488

504

value)

489

505

506

def call_method(self, methodname, busname, objectpath,

507

interface, *args):

508

raise NotImplementedError()

490

509

491

510

class MandosBus(SystemBus):

492

511

busname_domain = "se.recompile"

586

605

587

606

class SilenceLogger:

588

607

"Simple context manager to silence a particular logger"

608

589

609

def __init__(self, loggername):

590

610

self.logger = logging.getLogger(loggername)

591

611

601

621

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

602

622

self.logger.removeFilter(self.nullfilter)

603

623

604

605

624

class CachingBus(SystemBus):

606

625

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

626

607

627

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

608

628

self.object_cache = {}

609

629

super(dbus_python_adapter.CachingBus,

610

630

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

631

611

632

def get_object(self, busname, objectpath):

612

633

try:

613

634

return self.object_cache[(busname, objectpath)]

615

636

new_object = super(

616

637

dbus_python_adapter.CachingBus,

617

638

self).get_object(busname, objectpath)

618

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

639

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

619

640

return new_object

620

641

621

642

668

689

669

690

class CachingBus(SystemBus):

670

691

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

692

671

693

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

672

694

self.object_cache = {}

673

695

super(pydbus_adapter.CachingBus,

674

696

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

697

675

698

def get(self, busname, objectpath):

676

699

try:

677

700

return self.object_cache[(busname, objectpath)]

678

701

except KeyError:

679

702

new_object = (super(pydbus_adapter.CachingBus, self)

680

703

.get(busname, objectpath))

681

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

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

682

797

return new_object

683

798

684

799

720

835

721

836

class Base:

722

837

"""Abstract base class for commands"""

838

723

839

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

724

840

"""Normal commands should implement run_on_one_client(),

725

841

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

729

845

for client, properties in clients.items():

730

846

self.run_on_one_client(client, properties)

731

847

732

733

848

class IsEnabled(Base):

734

849

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

735

850

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

737

852

sys.exit(0)

738

853

sys.exit(1)

739

854

740

741

855

class Approve(Base):

742

856

def run_on_one_client(self, client, properties):

743

857

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

744

858

745

746

859

class Deny(Base):

747

860

def run_on_one_client(self, client, properties):

748

861

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

749

862

750

751

863

class Remove(Base):

752

864

def run(self, clients, bus):

753

865

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

754

866

bus.call_server_method("RemoveClient", clientpath)

755

867

756

757

868

class Output(Base):

758

869

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

759

870

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

765

876

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

766

877

"LastCheckerStatus")

767

878

768

769

879

class DumpJSON(Output):

770

880

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

771

881

data = {properties["Name"]:

772

882

{key: properties[key]

773

883

for key in self.all_keywords}

774

884

for properties in clients.values()}

775

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

776

885

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

777

886

778

887

class PrintTable(Output):

779

888

def __init__(self, verbose=False):

820

929

821

930

if sys.version_info.major == 2:

822

931

__unicode__ = __str__

932

823

933

def __str__(self):

824

934

return str(self).encode(

825

935

locale.getpreferredencoding())

871

981

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

872

982

seconds=td.seconds % 60))

873

983

874

875

984

class PropertySetter(Base):

876

985

"Abstract class for Actions for setting one client property"

877

986

884

993

def propname(self):

885

994

raise NotImplementedError()

886

995

887

888

996

class Enable(PropertySetter):

889

997

propname = "Enabled"

890

998

value_to_set = True

891

999

892

893

1000

class Disable(PropertySetter):

894

1001

propname = "Enabled"

895

1002

value_to_set = False

896

1003

897

898

1004

class BumpTimeout(PropertySetter):

899

1005

propname = "LastCheckedOK"

900

1006

value_to_set = ""

901

1007

902

903

1008

class StartChecker(PropertySetter):

904

1009

propname = "CheckerRunning"

905

1010

value_to_set = True

906

1011

907

908

1012

class StopChecker(PropertySetter):

909

1013

propname = "CheckerRunning"

910

1014

value_to_set = False

911

1015

912

913

1016

class ApproveByDefault(PropertySetter):

914

1017

propname = "ApprovedByDefault"

915

1018

value_to_set = True

916

1019

917

918

1020

class DenyByDefault(PropertySetter):

919

1021

propname = "ApprovedByDefault"

920

1022

value_to_set = False

921

1023

922

923

1024

class PropertySetterValue(PropertySetter):

924

1025

"""Abstract class for PropertySetter recieving a value as

925

1026

constructor argument instead of a class attribute."""

1027

926

1028

def __init__(self, value):

927

1029

self.value_to_set = value

928

1030

935

1037

class SetChecker(PropertySetterValue):

936

1038

propname = "Checker"

937

1039

938

939

1040

class SetHost(PropertySetterValue):

940

1041

propname = "Host"

941

1042

942

943

1043

class SetSecret(PropertySetterValue):

944

1044

propname = "Secret"

945

1045

953

1053

self._vts = value.read()

954

1054

value.close()

955

1055

956

957

1056

class PropertySetterValueMilliseconds(PropertySetterValue):

958

1057

"""Abstract class for PropertySetterValue taking a value

959

1058

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

968

1067

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

969

1068

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

970

1069

971

972

1070

class SetTimeout(PropertySetterValueMilliseconds):

973

1071

propname = "Timeout"

974

1072

975

976

1073

class SetExtendedTimeout(PropertySetterValueMilliseconds):

977

1074

propname = "ExtendedTimeout"

978

1075

979

980

1076

class SetInterval(PropertySetterValueMilliseconds):

981

1077

propname = "Interval"

982

1078

983

984

1079

class SetApprovalDelay(PropertySetterValueMilliseconds):

985

1080

propname = "ApprovalDelay"

986

1081

987

988

1082

class SetApprovalDuration(PropertySetterValueMilliseconds):

989

1083

propname = "ApprovalDuration"

990

1084

1530

1624

finally:

1531

1625

dbus_logger.removeFilter(counting_handler)

1532

1626

1533

self.assertNotIsInstance(e, dbus.ConnectFailed)

1627

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

1534

1628

1535

1629

# Make sure the dbus logger was suppressed

1536

1630

self.assertEqual(0, counting_handler.count)

1661

1755

self.assertIs(ret, expected_method_return)

1662

1756

1663

1757

def test_call_method_handles_exception(self):

1664

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

1665

1666

1758

def func():

1667

1759

raise gi.repository.GLib.Error()

1668

1760

1673

1765

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

1674

1766

"objectpath", "interface")

1675

1767

1676

self.assertNotIsInstance(e, dbus.ConnectFailed)

1768

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

1677

1769

1678

1770

def test_get_converts_to_correct_exception(self):

1679

1771

bus = pydbus_adapter.SystemBus(

1774

1866

self.assertIs(obj1, obj1b)

1775

1867

1776

1868

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

1777

2051

class Test_commands_from_options(unittest.TestCase):

1778

2052

1779

2053

def setUp(self):

2153

2427

busname = "se.recompile.Mandos"

2154

2428

client_interface = "se.recompile.Mandos.Client"

2155

2429

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

2430

self.assertTrue(self.bus.clients)

2156

2431

for clientpath in self.bus.clients:

2157

2432

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

2158

2433

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

2161

2436

busname = "se.recompile.Mandos"

2162

2437

client_interface = "se.recompile.Mandos.Client"

2163

2438

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

2439

self.assertTrue(self.bus.clients)

2164

2440

for clientpath in self.bus.clients:

2165

2441

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

2166

2442

client_interface, (False,)),

2167

2443

self.bus.calls)

2168

2444

2169

2445

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

2170

2450

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

2171

for clientpath in self.bus.clients:

2172

self.assertIn(("RemoveClient", dbus_busname,

2173

dbus_server_path, dbus_server_interface,

2451

self.assertFalse(self.bus.clients)

2452

for clientpath in orig_clients:

2453

self.assertIn(("RemoveClient", busname,

2454

server_path, server_interface,

2174

2455

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

2175

2456

2176

2457

expected_json = {

2378

2659

else:

2379

2660

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

2380

2661

values_to_get = self.values_to_get

2662

self.assertTrue(values_to_get)

2381

2663

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

2382

2664

for clientpath in self.bus.clients:

2383

2665

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

2384

2666

Unique())

2385

2667

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

2668

self.assertTrue(self.bus.clients)

2386

2669

for clientpath in self.bus.clients:

2387

2670

value = (self.bus.clients[clientpath]

2388

2671

[self.propname])

2447

2730

class TestSetSecretCmd(TestPropertySetterCmd):

2448

2731

command = command.SetSecret

2449

2732

propname = "Secret"

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]

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)

2453

2739

2454

2740

2455

2741

class TestSetTimeoutCmd(TestPropertySetterCmd):

2508

2794

2509

2795

2510

2796

2511

def should_only_run_tests():

2797

def parse_test_args():

2798

# type: () -> argparse.Namespace

2512

2799

parser = argparse.ArgumentParser(add_help=False)

2513

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

2800

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

2801

parser.add_argument("--prefix", )

2514

2802

args, unknown_args = parser.parse_known_args()

2515

run_tests = args.check

2516

if run_tests:

2517

# Remove --check argument from sys.argv

2803

if args.check:

2804

# Remove test options from sys.argv

2518

2805

sys.argv[1:] = unknown_args

2519

return run_tests

2806

return args

2520

2807

2521

2808

# Add all tests from doctest strings

2522

2809

def load_tests(loader, tests, none):

2525

2812

return tests

2526

2813

2527

2814

if __name__ == "__main__":

2815

options = parse_test_args()

2528

2816

try:

2529

if should_only_run_tests():

2530

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

2531

unittest.main()

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

2532

2829

else:

2533

2830

main()

2534

2831

finally:

2535

2832

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: