/mandos/trunk : revision 1311

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

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

49

67

try:

50

import pydbus

51

import gi

52

dbus_python = None

68

import dbussy

69

import ravel

53

70

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

71

try:

72

import pydbus

73

import gi

74

except ImportError:

75

import dbus as dbus_python

76

62

77

63

78

# Show warnings by default

64

79

if not sys.warnoptions:

65

80

import warnings

66

81

warnings.simplefilter("default")

67

82

68

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

69

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

70

85

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

71

86

72

87

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

73

88

74

89

if sys.version_info.major == 2:

75

str = unicode

76

90

import StringIO

77

91

io.StringIO = StringIO.StringIO

78

92

79

93

locale.setlocale(locale.LC_ALL, "")

80

94

81

version = "1.8.6"

95

version = "1.8.17"

82

96

83

97

84

98

def main():

91

105

clientnames = options.client

92

106

93

107

if options.debug:

94

log.setLevel(logging.DEBUG)

108

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

95

109

96

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:

97

113

bus = pydbus_adapter.CachingBus(pydbus)

98

114

else:

99

115

bus = dbus_python_adapter.CachingBus(dbus_python)

243

259

return rfc3339_duration_to_delta(interval)

244

260

except ValueError as e:

245

261

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

246

' '.join(e.args))

262

" ".join(e.args))

247

263

return parse_pre_1_6_1_interval(interval)

248

264

249

265

379

395

380

396

381

397

def parse_pre_1_6_1_interval(interval):

382

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

383

399

and return a datetime.timedelta

384

400

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)

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)

398

414

True

399

415

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

400

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

401

418

True

402

419

403

420

"""

467

484

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

468

485

469

486

470

class dbus(object):

487

class dbus:

471

488

472

class SystemBus(object):

489

class SystemBus:

473

490

474

491

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

492

475

493

def get_managed_objects(self, busname, objectpath):

476

494

return self.call_method("GetManagedObjects", busname,

477

495

objectpath,

478

496

self.object_manager_iface)

479

497

480

498

properties_iface = "org.freedesktop.DBus.Properties"

499

481

500

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

482

501

value):

483

502

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

484

503

self.properties_iface, interface, key,

485

504

value)

486

505

506

def call_method(self, methodname, busname, objectpath,

507

interface, *args):

508

raise NotImplementedError()

487

509

488

510

class MandosBus(SystemBus):

489

511

busname_domain = "se.recompile"

521

543

pass

522

544

523

545

524

class dbus_python_adapter(object):

546

class dbus_python_adapter:

525

547

526

548

class SystemBus(dbus.MandosBus):

527

549

"""Use dbus-python"""

581

603

self.client_interface, key,

582

604

value)

583

605

584

class SilenceLogger(object):

606

class SilenceLogger:

585

607

"Simple context manager to silence a particular logger"

608

586

609

def __init__(self, loggername):

587

610

self.logger = logging.getLogger(loggername)

588

611

598

621

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

599

622

self.logger.removeFilter(self.nullfilter)

600

623

601

602

624

class CachingBus(SystemBus):

603

625

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

626

604

627

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

605

628

self.object_cache = {}

606

629

super(dbus_python_adapter.CachingBus,

607

630

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

631

608

632

def get_object(self, busname, objectpath):

609

633

try:

610

634

return self.object_cache[(busname, objectpath)]

612

636

new_object = super(

613

637

dbus_python_adapter.CachingBus,

614

638

self).get_object(busname, objectpath)

615

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

639

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

616

640

return new_object

617

641

618

642

619

class pydbus_adapter(object):

643

class pydbus_adapter:

620

644

class SystemBus(dbus.MandosBus):

621

645

def __init__(self, module=pydbus):

622

646

self.pydbus = module

665

689

666

690

class CachingBus(SystemBus):

667

691

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

692

668

693

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

669

694

self.object_cache = {}

670

695

super(pydbus_adapter.CachingBus,

671

696

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

697

672

698

def get(self, busname, objectpath):

673

699

try:

674

700

return self.object_cache[(busname, objectpath)]

675

701

except KeyError:

676

702

new_object = (super(pydbus_adapter.CachingBus, self)

677

703

.get(busname, objectpath))

678

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

679

797

return new_object

680

798

681

799

712

830

return commands

713

831

714

832

715

class command(object):

833

class command:

716

834

"""A namespace for command classes"""

717

835

718

class Base(object):

836

class Base:

719

837

"""Abstract base class for commands"""

838

720

839

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

721

840

"""Normal commands should implement run_on_one_client(),

722

841

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

726

845

for client, properties in clients.items():

727

846

self.run_on_one_client(client, properties)

728

847

729

730

848

class IsEnabled(Base):

731

849

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

732

850

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

734

852

sys.exit(0)

735

853

sys.exit(1)

736

854

737

738

855

class Approve(Base):

739

856

def run_on_one_client(self, client, properties):

740

857

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

741

858

742

743

859

class Deny(Base):

744

860

def run_on_one_client(self, client, properties):

745

861

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

746

862

747

748

863

class Remove(Base):

749

864

def run(self, clients, bus):

750

865

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

751

866

bus.call_server_method("RemoveClient", clientpath)

752

867

753

754

868

class Output(Base):

755

869

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

756

870

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

762

876

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

763

877

"LastCheckerStatus")

764

878

765

766

879

class DumpJSON(Output):

767

880

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

768

881

data = {properties["Name"]:

769

882

{key: properties[key]

770

883

for key in self.all_keywords}

771

884

for properties in clients.values()}

772

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

773

885

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

774

886

775

887

class PrintTable(Output):

776

888

def __init__(self, verbose=False):

784

896

keywords = self.all_keywords

785

897

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

786

898

787

class TableOfClients(object):

899

class TableOfClients:

788

900

tableheaders = {

789

901

"Name": "Name",

790

902

"Enabled": "Enabled",

817

929

818

930

if sys.version_info.major == 2:

819

931

__unicode__ = __str__

932

820

933

def __str__(self):

821

934

return str(self).encode(

822

935

locale.getpreferredencoding())

868

981

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

869

982

seconds=td.seconds % 60))

870

983

871

872

984

class PropertySetter(Base):

873

985

"Abstract class for Actions for setting one client property"

874

986

881

993

def propname(self):

882

994

raise NotImplementedError()

883

995

884

885

996

class Enable(PropertySetter):

886

997

propname = "Enabled"

887

998

value_to_set = True

888

999

889

890

1000

class Disable(PropertySetter):

891

1001

propname = "Enabled"

892

1002

value_to_set = False

893

1003

894

895

1004

class BumpTimeout(PropertySetter):

896

1005

propname = "LastCheckedOK"

897

1006

value_to_set = ""

898

1007

899

900

1008

class StartChecker(PropertySetter):

901

1009

propname = "CheckerRunning"

902

1010

value_to_set = True

903

1011

904

905

1012

class StopChecker(PropertySetter):

906

1013

propname = "CheckerRunning"

907

1014

value_to_set = False

908

1015

909

910

1016

class ApproveByDefault(PropertySetter):

911

1017

propname = "ApprovedByDefault"

912

1018

value_to_set = True

913

1019

914

915

1020

class DenyByDefault(PropertySetter):

916

1021

propname = "ApprovedByDefault"

917

1022

value_to_set = False

918

1023

919

920

1024

class PropertySetterValue(PropertySetter):

921

1025

"""Abstract class for PropertySetter recieving a value as

922

1026

constructor argument instead of a class attribute."""

1027

923

1028

def __init__(self, value):

924

1029

self.value_to_set = value

925

1030

932

1037

class SetChecker(PropertySetterValue):

933

1038

propname = "Checker"

934

1039

935

936

1040

class SetHost(PropertySetterValue):

937

1041

propname = "Host"

938

1042

939

940

1043

class SetSecret(PropertySetterValue):

941

1044

propname = "Secret"

942

1045

950

1053

self._vts = value.read()

951

1054

value.close()

952

1055

953

954

1056

class PropertySetterValueMilliseconds(PropertySetterValue):

955

1057

"""Abstract class for PropertySetterValue taking a value

956

1058

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

965

1067

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

966

1068

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

967

1069

968

969

1070

class SetTimeout(PropertySetterValueMilliseconds):

970

1071

propname = "Timeout"

971

1072

972

973

1073

class SetExtendedTimeout(PropertySetterValueMilliseconds):

974

1074

propname = "ExtendedTimeout"

975

1075

976

977

1076

class SetInterval(PropertySetterValueMilliseconds):

978

1077

propname = "Interval"

979

1078

980

981

1079

class SetApprovalDelay(PropertySetterValueMilliseconds):

982

1080

propname = "ApprovalDelay"

983

1081

984

985

1082

class SetApprovalDuration(PropertySetterValueMilliseconds):

986

1083

propname = "ApprovalDuration"

987

1084

1021

1118

"output"))

1022

1119

1023

1120

1024

class Unique(object):

1121

class Unique:

1025

1122

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

1026

1123

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

1027

1124

1311

1408

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1312

1409

1313

1410

def MockDBusPython_func(self, func):

1314

class mock_dbus_python(object):

1411

class mock_dbus_python:

1315

1412

"""mock dbus-python module"""

1316

class exceptions(object):

1413

class exceptions:

1317

1414

"""Pseudo-namespace"""

1318

1415

class DBusException(Exception):

1319

1416

pass

1320

class SystemBus(object):

1417

class SystemBus:

1321

1418

@staticmethod

1322

1419

def get_object(busname, objectpath):

1323

1420

DBusObject = collections.namedtuple(

1337

1434

dbus_interface=dbus_interface)

1338

1435

return DBusObject(methodname=method,

1339

1436

Set=set_property)

1340

class Boolean(object):

1437

class Boolean:

1341

1438

def __init__(self, value):

1342

1439

self.value = bool(value)

1343

1440

def __bool__(self):

1527

1624

finally:

1528

1625

dbus_logger.removeFilter(counting_handler)

1529

1626

1530

self.assertNotIsInstance(e, dbus.ConnectFailed)

1627

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

1531

1628

1532

1629

# Make sure the dbus logger was suppressed

1533

1630

self.assertEqual(0, counting_handler.count)

1555

1652

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

1556

1653

"objectpath", "interface")

1557

1654

1558

class fake_dbus_python_raises_exception_on_connect(object):

1655

class fake_dbus_python_raises_exception_on_connect:

1559

1656

"""fake dbus-python module"""

1560

class exceptions(object):

1657

class exceptions:

1561

1658

"""Pseudo-namespace"""

1562

1659

class DBusException(Exception):

1563

1660

pass

1571

1668

1572

1669

1573

1670

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1574

class mock_dbus_python(object):

1671

class mock_dbus_python:

1575

1672

"""mock dbus-python modules"""

1576

class SystemBus(object):

1673

class SystemBus:

1577

1674

@staticmethod

1578

1675

def get_object(busname, objectpath):

1579

1676

return Unique()

1625

1722

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1626

1723

1627

1724

def Stub_pydbus_func(self, func):

1628

class stub_pydbus(object):

1725

class stub_pydbus:

1629

1726

"""stub pydbus module"""

1630

class SystemBus(object):

1727

class SystemBus:

1631

1728

@staticmethod

1632

1729

def get(busname, objectpath):

1633

1730

DBusObject = collections.namedtuple(

1658

1755

self.assertIs(ret, expected_method_return)

1659

1756

1660

1757

def test_call_method_handles_exception(self):

1661

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

1662

1663

1758

def func():

1664

1759

raise gi.repository.GLib.Error()

1665

1760

1670

1765

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

1671

1766

"objectpath", "interface")

1672

1767

1673

self.assertNotIsInstance(e, dbus.ConnectFailed)

1768

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

1674

1769

1675

1770

def test_get_converts_to_correct_exception(self):

1676

1771

bus = pydbus_adapter.SystemBus(

1679

1774

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

1680

1775

"objectpath", "interface")

1681

1776

1682

class fake_pydbus_raises_exception_on_connect(object):

1777

class fake_pydbus_raises_exception_on_connect:

1683

1778

"""fake dbus-python module"""

1684

1779

@classmethod

1685

1780

def SystemBus(cls):

1689

1784

return Bus(get=get)

1690

1785

1691

1786

def test_set_property_uses_setattr(self):

1692

class Object(object):

1787

class Object:

1693

1788

pass

1694

1789

obj = Object()

1695

class pydbus_spy(object):

1696

class SystemBus(object):

1790

class pydbus_spy:

1791

class SystemBus:

1697

1792

@staticmethod

1698

1793

def get(busname, objectpath):

1699

1794

return {"interface": obj}

1706

1801

def test_get_suppresses_xml_deprecation_warning(self):

1707

1802

if sys.version_info.major >= 3:

1708

1803

return

1709

class stub_pydbus_get(object):

1710

class SystemBus(object):

1804

class stub_pydbus_get:

1805

class SystemBus:

1711

1806

@staticmethod

1712

1807

def get(busname, objectpath):

1713

1808

warnings.warn_explicit(

1721

1816

1722

1817

1723

1818

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1724

class stub_pydbus(object):

1819

class stub_pydbus:

1725

1820

"""stub pydbus module"""

1726

class SystemBus(object):

1821

class SystemBus:

1727

1822

@staticmethod

1728

1823

def get(busname, objectpath):

1729

1824

return Unique()

1771

1866

self.assertIs(obj1, obj1b)

1772

1867

1773

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

1774

2051

class Test_commands_from_options(unittest.TestCase):

1775

2052

1776

2053

def setUp(self):

2150

2427

busname = "se.recompile.Mandos"

2151

2428

client_interface = "se.recompile.Mandos.Client"

2152

2429

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

2430

self.assertTrue(self.bus.clients)

2153

2431

for clientpath in self.bus.clients:

2154

2432

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

2155

2433

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

2158

2436

busname = "se.recompile.Mandos"

2159

2437

client_interface = "se.recompile.Mandos.Client"

2160

2438

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

2439

self.assertTrue(self.bus.clients)

2161

2440

for clientpath in self.bus.clients:

2162

2441

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

2163

2442

client_interface, (False,)),

2164

2443

self.bus.calls)

2165

2444

2166

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

2167

2450

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,

2451

self.assertFalse(self.bus.clients)

2452

for clientpath in orig_clients:

2453

self.assertIn(("RemoveClient", busname,

2454

server_path, server_interface,

2171

2455

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

2172

2456

2173

2457

expected_json = {

2375

2659

else:

2376

2660

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

2377

2661

values_to_get = self.values_to_get

2662

self.assertTrue(values_to_get)

2378

2663

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

2379

2664

for clientpath in self.bus.clients:

2380

2665

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

2381

2666

Unique())

2382

2667

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

2668

self.assertTrue(self.bus.clients)

2383

2669

for clientpath in self.bus.clients:

2384

2670

value = (self.bus.clients[clientpath]

2385

2671

[self.propname])

2444

2730

class TestSetSecretCmd(TestPropertySetterCmd):

2445

2731

command = command.SetSecret

2446

2732

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]

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

2739

2451

2740

2452

2741

class TestSetTimeoutCmd(TestPropertySetterCmd):

2505

2794

2506

2795

2507

2796

2508

def should_only_run_tests():

2797

def parse_test_args():

2798

# type: () -> argparse.Namespace

2509

2799

parser = argparse.ArgumentParser(add_help=False)

2510

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

2800

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

2801

parser.add_argument("--prefix", )

2511

2802

args, unknown_args = parser.parse_known_args()

2512

run_tests = args.check

2513

if run_tests:

2514

# Remove --check argument from sys.argv

2803

if args.check:

2804

# Remove test options from sys.argv

2515

2805

sys.argv[1:] = unknown_args

2516

return run_tests

2806

return args

2517

2807

2518

2808

# Add all tests from doctest strings

2519

2809

def load_tests(loader, tests, none):

2522

2812

return tests

2523

2813

2524

2814

if __name__ == "__main__":

2815

options = parse_test_args()

2525

2816

try:

2526

if should_only_run_tests():

2527

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

2528

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

2529

2829

else:

2530

2830

main()

2531

2831

finally:

2532

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: