/mandos/release : revision 237.7.695

1

#!/usr/bin/python3 -bbI

2

# -*- coding: utf-8; lexical-binding: t -*-

3

#

4

# Mandos Control - Control or query the Mandos server

5

#

6

7

1

#!/usr/bin/python

2

# -*- after-save-hook: (lambda () (let ((command (if (fboundp 'file-local-name) (file-local-name (buffer-file-name)) (or (file-remote-p (buffer-file-name) 'localname) (buffer-file-name))))) (if (= (progn (if (get-buffer "*Test*") (kill-buffer "*Test*")) (process-file-shell-command (format "%s --check" (shell-quote-argument command)) nil "*Test*")) 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w))) (progn (with-current-buffer "*Test*" (compilation-mode)) (display-buffer "*Test*" '(display-buffer-in-side-window)))))); coding: utf-8 -*-

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

23

#

24

# Contact the authors at <mandos@recompile.se>.

25

#

26

27

from __future__ import (division, absolute_import, print_function,

27

28

unicode_literals)

28

29

32

33

pass

33

34

35

import sys

35

import unittest

36

import argparse

37

import logging

38

import os

39

37

import locale

40

38

import datetime

41

39

import re

40

import os

42

41

import collections

43

42

import json

43

import unittest

44

import logging

44

45

import io

45

46

import tempfile

46

47

import contextlib

47

48

if sys.version_info.major == 2:

49

__metaclass__ = type

50

str = unicode

51

input = raw_input

52

53

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

67

49

try:

68

import dbussy

69

import ravel

50

import pydbus

51

import gi

52

dbus_python = None

70

53

except ImportError:

71

try:

72

import pydbus

73

import gi

74

except ImportError:

75

import dbus as dbus_python

76

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

77

62

78

63

# Show warnings by default

79

64

if not sys.warnoptions:

80

65

import warnings

81

66

warnings.simplefilter("default")

82

67

83

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

84

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

68

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

69

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

85

70

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

86

71

87

72

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

88

73

89

74

if sys.version_info.major == 2:

75

str = unicode

90

76

import StringIO

91

77

io.StringIO = StringIO.StringIO

92

78

93

79

locale.setlocale(locale.LC_ALL, "")

94

80

95

version = "1.8.16"

81

version = "1.8.6"

96

82

97

83

98

84

def main():

105

91

clientnames = options.client

106

92

107

93

if options.debug:

108

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

94

log.setLevel(logging.DEBUG)

109

95

110

if dbussy is not None and ravel is not None:

111

bus = dbussy_adapter.CachingBus(dbussy, ravel)

112

elif pydbus is not None:

96

if pydbus is not None:

113

97

bus = pydbus_adapter.CachingBus(pydbus)

114

98

else:

115

99

bus = dbus_python_adapter.CachingBus(dbus_python)

259

243

return rfc3339_duration_to_delta(interval)

260

244

except ValueError as e:

261

245

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

262

" ".join(e.args))

246

' '.join(e.args))

263

247

return parse_pre_1_6_1_interval(interval)

264

248

265

249

395

379

396

380

397

381

def parse_pre_1_6_1_interval(interval):

398

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

382

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

399

383

and return a datetime.timedelta

400

384

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)

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)

414

398

True

415

399

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

416

>>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") \

417

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

400

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

418

401

True

419

402

420

403

"""

484

467

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

485

468

486

469

487

class dbus:

470

class dbus(object):

488

471

489

class SystemBus:

472

class SystemBus(object):

490

473

491

474

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

492

493

475

def get_managed_objects(self, busname, objectpath):

494

476

return self.call_method("GetManagedObjects", busname,

495

477

objectpath,

496

478

self.object_manager_iface)

497

479

498

480

properties_iface = "org.freedesktop.DBus.Properties"

499

500

481

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

501

482

value):

502

483

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

503

484

self.properties_iface, interface, key,

504

485

value)

505

486

506

def call_method(self, methodname, busname, objectpath,

507

interface, *args):

508

raise NotImplementedError()

509

487

510

488

class MandosBus(SystemBus):

511

489

busname_domain = "se.recompile"

543

521

pass

544

522

545

523

546

class dbus_python_adapter:

524

class dbus_python_adapter(object):

547

525

548

526

class SystemBus(dbus.MandosBus):

549

527

"""Use dbus-python"""

603

581

self.client_interface, key,

604

582

value)

605

583

606

class SilenceLogger:

584

class SilenceLogger(object):

607

585

"Simple context manager to silence a particular logger"

608

609

586

def __init__(self, loggername):

610

587

self.logger = logging.getLogger(loggername)

611

588

621

598

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

622

599

self.logger.removeFilter(self.nullfilter)

623

600

601

624

602

class CachingBus(SystemBus):

625

603

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

626

627

604

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

628

605

self.object_cache = {}

629

606

super(dbus_python_adapter.CachingBus,

630

607

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

631

632

608

def get_object(self, busname, objectpath):

633

609

try:

634

610

return self.object_cache[(busname, objectpath)]

636

612

new_object = super(

637

613

dbus_python_adapter.CachingBus,

638

614

self).get_object(busname, objectpath)

639

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

615

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

640

616

return new_object

641

617

642

618

643

class pydbus_adapter:

619

class pydbus_adapter(object):

644

620

class SystemBus(dbus.MandosBus):

645

621

def __init__(self, module=pydbus):

646

622

self.pydbus = module

689

665

690

666

class CachingBus(SystemBus):

691

667

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

692

693

668

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

694

669

self.object_cache = {}

695

670

super(pydbus_adapter.CachingBus,

696

671

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

697

698

672

def get(self, busname, objectpath):

699

673

try:

700

674

return self.object_cache[(busname, objectpath)]

701

675

except KeyError:

702

676

new_object = (super(pydbus_adapter.CachingBus, self)

703

677

.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

678

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

797

679

return new_object

798

680

799

681

830

712

return commands

831

713

832

714

833

class command:

715

class command(object):

834

716

"""A namespace for command classes"""

835

717

836

class Base:

718

class Base(object):

837

719

"""Abstract base class for commands"""

838

839

720

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

840

721

"""Normal commands should implement run_on_one_client(),

841

722

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

845

726

for client, properties in clients.items():

846

727

self.run_on_one_client(client, properties)

847

728

729

848

730

class IsEnabled(Base):

849

731

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

850

732

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

852

734

sys.exit(0)

853

735

sys.exit(1)

854

736

737

855

738

class Approve(Base):

856

739

def run_on_one_client(self, client, properties):

857

740

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

858

741

742

859

743

class Deny(Base):

860

744

def run_on_one_client(self, client, properties):

861

745

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

862

746

747

863

748

class Remove(Base):

864

749

def run(self, clients, bus):

865

750

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

866

751

bus.call_server_method("RemoveClient", clientpath)

867

752

753

868

754

class Output(Base):

869

755

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

870

756

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

876

762

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

877

763

"LastCheckerStatus")

878

764

765

879

766

class DumpJSON(Output):

880

767

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

881

768

data = {properties["Name"]:

882

769

{key: properties[key]

883

770

for key in self.all_keywords}

884

771

for properties in clients.values()}

885

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

772

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

773

886

774

887

775

class PrintTable(Output):

888

776

def __init__(self, verbose=False):

896

784

keywords = self.all_keywords

897

785

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

898

786

899

class TableOfClients:

787

class TableOfClients(object):

900

788

tableheaders = {

901

789

"Name": "Name",

902

790

"Enabled": "Enabled",

929

817

930

818

if sys.version_info.major == 2:

931

819

__unicode__ = __str__

932

933

820

def __str__(self):

934

821

return str(self).encode(

935

822

locale.getpreferredencoding())

981

868

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

982

869

seconds=td.seconds % 60))

983

870

871

984

872

class PropertySetter(Base):

985

873

"Abstract class for Actions for setting one client property"

986

874

993

881

def propname(self):

994

882

raise NotImplementedError()

995

883

884

996

885

class Enable(PropertySetter):

997

886

propname = "Enabled"

998

887

value_to_set = True

999

888

889

1000

890

class Disable(PropertySetter):

1001

891

propname = "Enabled"

1002

892

value_to_set = False

1003

893

894

1004

895

class BumpTimeout(PropertySetter):

1005

896

propname = "LastCheckedOK"

1006

897

value_to_set = ""

1007

898

899

1008

900

class StartChecker(PropertySetter):

1009

901

propname = "CheckerRunning"

1010

902

value_to_set = True

1011

903

904

1012

905

class StopChecker(PropertySetter):

1013

906

propname = "CheckerRunning"

1014

907

value_to_set = False

1015

908

909

1016

910

class ApproveByDefault(PropertySetter):

1017

911

propname = "ApprovedByDefault"

1018

912

value_to_set = True

1019

913

914

1020

915

class DenyByDefault(PropertySetter):

1021

916

propname = "ApprovedByDefault"

1022

917

value_to_set = False

1023

918

919

1024

920

class PropertySetterValue(PropertySetter):

1025

921

"""Abstract class for PropertySetter recieving a value as

1026

922

constructor argument instead of a class attribute."""

1027

1028

923

def __init__(self, value):

1029

924

self.value_to_set = value

1030

925

1037

932

class SetChecker(PropertySetterValue):

1038

933

propname = "Checker"

1039

934

935

1040

936

class SetHost(PropertySetterValue):

1041

937

propname = "Host"

1042

938

939

1043

940

class SetSecret(PropertySetterValue):

1044

941

propname = "Secret"

1045

942

1053

950

self._vts = value.read()

1054

951

value.close()

1055

952

953

1056

954

class PropertySetterValueMilliseconds(PropertySetterValue):

1057

955

"""Abstract class for PropertySetterValue taking a value

1058

956

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

1067

965

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

1068

966

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

1069

967

968

1070

969

class SetTimeout(PropertySetterValueMilliseconds):

1071

970

propname = "Timeout"

1072

971

972

1073

973

class SetExtendedTimeout(PropertySetterValueMilliseconds):

1074

974

propname = "ExtendedTimeout"

1075

975

976

1076

977

class SetInterval(PropertySetterValueMilliseconds):

1077

978

propname = "Interval"

1078

979

980

1079

981

class SetApprovalDelay(PropertySetterValueMilliseconds):

1080

982

propname = "ApprovalDelay"

1081

983

984

1082

985

class SetApprovalDuration(PropertySetterValueMilliseconds):

1083

986

propname = "ApprovalDuration"

1084

987

1118

1021

"output"))

1119

1022

1120

1023

1121

class Unique:

1024

class Unique(object):

1122

1025

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

1123

1026

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

1124

1027

1408

1311

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1409

1312

1410

1313

def MockDBusPython_func(self, func):

1411

class mock_dbus_python:

1314

class mock_dbus_python(object):

1412

1315

"""mock dbus-python module"""

1413

class exceptions:

1316

class exceptions(object):

1414

1317

"""Pseudo-namespace"""

1415

1318

class DBusException(Exception):

1416

1319

pass

1417

class SystemBus:

1320

class SystemBus(object):

1418

1321

@staticmethod

1419

1322

def get_object(busname, objectpath):

1420

1323

DBusObject = collections.namedtuple(

1434

1337

dbus_interface=dbus_interface)

1435

1338

return DBusObject(methodname=method,

1436

1339

Set=set_property)

1437

class Boolean:

1340

class Boolean(object):

1438

1341

def __init__(self, value):

1439

1342

self.value = bool(value)

1440

1343

def __bool__(self):

1624

1527

finally:

1625

1528

dbus_logger.removeFilter(counting_handler)

1626

1529

1627

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

1530

self.assertNotIsInstance(e, dbus.ConnectFailed)

1628

1531

1629

1532

# Make sure the dbus logger was suppressed

1630

1533

self.assertEqual(0, counting_handler.count)

1652

1555

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

1653

1556

"objectpath", "interface")

1654

1557

1655

class fake_dbus_python_raises_exception_on_connect:

1558

class fake_dbus_python_raises_exception_on_connect(object):

1656

1559

"""fake dbus-python module"""

1657

class exceptions:

1560

class exceptions(object):

1658

1561

"""Pseudo-namespace"""

1659

1562

class DBusException(Exception):

1660

1563

pass

1668

1571

1669

1572

1670

1573

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1671

class mock_dbus_python:

1574

class mock_dbus_python(object):

1672

1575

"""mock dbus-python modules"""

1673

class SystemBus:

1576

class SystemBus(object):

1674

1577

@staticmethod

1675

1578

def get_object(busname, objectpath):

1676

1579

return Unique()

1722

1625

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1723

1626

1724

1627

def Stub_pydbus_func(self, func):

1725

class stub_pydbus:

1628

class stub_pydbus(object):

1726

1629

"""stub pydbus module"""

1727

class SystemBus:

1630

class SystemBus(object):

1728

1631

@staticmethod

1729

1632

def get(busname, objectpath):

1730

1633

DBusObject = collections.namedtuple(

1755

1658

self.assertIs(ret, expected_method_return)

1756

1659

1757

1660

def test_call_method_handles_exception(self):

1661

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

1662

1758

1663

def func():

1759

1664

raise gi.repository.GLib.Error()

1760

1665

1765

1670

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

1766

1671

"objectpath", "interface")

1767

1672

1768

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

1673

self.assertNotIsInstance(e, dbus.ConnectFailed)

1769

1674

1770

1675

def test_get_converts_to_correct_exception(self):

1771

1676

bus = pydbus_adapter.SystemBus(

1774

1679

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

1775

1680

"objectpath", "interface")

1776

1681

1777

class fake_pydbus_raises_exception_on_connect:

1682

class fake_pydbus_raises_exception_on_connect(object):

1778

1683

"""fake dbus-python module"""

1779

1684

@classmethod

1780

1685

def SystemBus(cls):

1784

1689

return Bus(get=get)

1785

1690

1786

1691

def test_set_property_uses_setattr(self):

1787

class Object:

1692

class Object(object):

1788

1693

pass

1789

1694

obj = Object()

1790

class pydbus_spy:

1791

class SystemBus:

1695

class pydbus_spy(object):

1696

class SystemBus(object):

1792

1697

@staticmethod

1793

1698

def get(busname, objectpath):

1794

1699

return {"interface": obj}

1801

1706

def test_get_suppresses_xml_deprecation_warning(self):

1802

1707

if sys.version_info.major >= 3:

1803

1708

return

1804

class stub_pydbus_get:

1805

class SystemBus:

1709

class stub_pydbus_get(object):

1710

class SystemBus(object):

1806

1711

@staticmethod

1807

1712

def get(busname, objectpath):

1808

1713

warnings.warn_explicit(

1816

1721

1817

1722

1818

1723

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1819

class stub_pydbus:

1724

class stub_pydbus(object):

1820

1725

"""stub pydbus module"""

1821

class SystemBus:

1726

class SystemBus(object):

1822

1727

@staticmethod

1823

1728

def get(busname, objectpath):

1824

1729

return Unique()

1866

1771

self.assertIs(obj1, obj1b)

1867

1772

1868

1773

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

1774

class Test_commands_from_options(unittest.TestCase):

2052

1775

2053

1776

def setUp(self):

2427

2150

busname = "se.recompile.Mandos"

2428

2151

client_interface = "se.recompile.Mandos.Client"

2429

2152

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

2430

self.assertTrue(self.bus.clients)

2431

2153

for clientpath in self.bus.clients:

2432

2154

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

2433

2155

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

2436

2158

busname = "se.recompile.Mandos"

2437

2159

client_interface = "se.recompile.Mandos.Client"

2438

2160

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

2439

self.assertTrue(self.bus.clients)

2440

2161

for clientpath in self.bus.clients:

2441

2162

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

2442

2163

client_interface, (False,)),

2443

2164

self.bus.calls)

2444

2165

2445

2166

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

2167

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,

2168

for clientpath in self.bus.clients:

2169

self.assertIn(("RemoveClient", dbus_busname,

2170

dbus_server_path, dbus_server_interface,

2455

2171

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

2456

2172

2457

2173

expected_json = {

2659

2375

else:

2660

2376

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

2661

2377

values_to_get = self.values_to_get

2662

self.assertTrue(values_to_get)

2663

2378

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

2664

2379

for clientpath in self.bus.clients:

2665

2380

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

2666

2381

Unique())

2667

2382

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

2668

self.assertTrue(self.bus.clients)

2669

2383

for clientpath in self.bus.clients:

2670

2384

value = (self.bus.clients[clientpath]

2671

2385

[self.propname])

2730

2444

class TestSetSecretCmd(TestPropertySetterCmd):

2731

2445

command = command.SetSecret

2732

2446

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)

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]

2739

2450

2740

2451

2741

2452

class TestSetTimeoutCmd(TestPropertySetterCmd):

2794

2505

2795

2506

2796

2507

2797

def parse_test_args():

2798

# type: () -> argparse.Namespace

2508

def should_only_run_tests():

2799

2509

parser = argparse.ArgumentParser(add_help=False)

2800

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

2801

parser.add_argument("--prefix", )

2510

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

2802

2511

args, unknown_args = parser.parse_known_args()

2803

if args.check:

2804

# Remove test options from sys.argv

2512

run_tests = args.check

2513

if run_tests:

2514

# Remove --check argument from sys.argv

2805

2515

sys.argv[1:] = unknown_args

2806

return args

2516

return run_tests

2807

2517

2808

2518

# Add all tests from doctest strings

2809

2519

def load_tests(loader, tests, none):

2812

2522

return tests

2813

2523

2814

2524

if __name__ == "__main__":

2815

options = parse_test_args()

2816

2525

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

2526

if should_only_run_tests():

2527

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

2528

unittest.main()

2829

2529

else:

2830

2530

main()

2831

2531

finally:

2832

2532

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: