/mandos/trunk : revision 1089

61

62

if sys.version_info.major == 2:

63

str = unicode

64

import StringIO

65

io.StringIO = StringIO.StringIO

64

66

65

67

locale.setlocale(locale.LC_ALL, "")

66

68

81

83

82

84

def main():

83

85

parser = argparse.ArgumentParser()

84

85

86

add_command_line_options(parser)

86

87

88

options = parser.parse_args()

88

89

check_option_syntax(parser, options)

90

91

clientnames = options.client

460

461

def __enter__(self):

462

self.logger.addFilter(self.nullfilter)

463

return self

464

463

465

464

class NullFilter(logging.Filter):

466

465

def filter(self, record):

557

556

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

558

557

override this run() method instead.

559

558

"""

560

self.mandos = mandos

561

559

for clientpath, properties in clients.items():

562

560

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

563

561

dbus_busname, str(clientpath))

594

592

595

593

596

594

class Remove(Base):

597

def run_on_one_client(self, client, properties):

598

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

599

dbus_busname, server_dbus_path,

600

server_dbus_interface,

601

str(client.__dbus_object_path__))

602

self.mandos.RemoveClient(client.__dbus_object_path__)

595

def run(self, clients, bus, mandos):

596

for clientpath in clients.keys():

597

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

598

dbus_busname, server_dbus_path,

599

server_dbus_interface, clientpath)

600

mandos.RemoveClient(clientpath)

603

601

604

602

605

603

class Output(Base):

613

611

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

614

612

"LastCheckerStatus")

615

613

616

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

617

print(self.output(clients.values()))

618

619

def output(self, clients):

620

raise NotImplementedError()

621

622

614

623

615

class DumpJSON(Output):

624

def output(self, clients):

616

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

625

617

data = {client["Name"]:

626

618

{key: self.dbus_boolean_to_bool(client[key])

627

619

for key in self.all_keywords}

628

for client in clients}

629

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

620

for client in clients.values()}

621

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

630

622

631

623

@staticmethod

632

624

def dbus_boolean_to_bool(value):

639

631

def __init__(self, verbose=False):

640

632

self.verbose = verbose

641

633

642

def output(self, clients):

634

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

643

635

default_keywords = ("Name", "Enabled", "Timeout",

644

636

"LastCheckedOK")

645

637

keywords = default_keywords

646

638

if self.verbose:

647

639

keywords = self.all_keywords

648

return str(self.TableOfClients(clients, keywords))

640

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

649

641

650

642

class TableOfClients(object):

651

643

tableheaders = {

732

724

seconds=td.seconds % 60))

733

725

734

726

735

class Property(Base):

727

class PropertySetter(Base):

736

728

"Abstract class for Actions for setting one client property"

737

729

738

730

def run_on_one_client(self, client, properties):

753

745

raise NotImplementedError()

754

746

755

747

756

class Enable(Property):

748

class Enable(PropertySetter):

757

749

propname = "Enabled"

758

750

value_to_set = dbus.Boolean(True)

759

751

760

752

761

class Disable(Property):

753

class Disable(PropertySetter):

762

754

propname = "Enabled"

763

755

value_to_set = dbus.Boolean(False)

764

756

765

757

766

class BumpTimeout(Property):

758

class BumpTimeout(PropertySetter):

767

759

propname = "LastCheckedOK"

768

760

value_to_set = ""

769

761

770

762

771

class StartChecker(Property):

772

propname = "CheckerRunning"

773

value_to_set = dbus.Boolean(True)

774

775

776

class StopChecker(Property):

777

propname = "CheckerRunning"

778

value_to_set = dbus.Boolean(False)

779

780

781

class ApproveByDefault(Property):

782

propname = "ApprovedByDefault"

783

value_to_set = dbus.Boolean(True)

784

785

786

class DenyByDefault(Property):

787

propname = "ApprovedByDefault"

788

value_to_set = dbus.Boolean(False)

789

790

791

class PropertyValue(Property):

792

"Abstract class for Property recieving a value as argument"

763

class StartChecker(PropertySetter):

764

propname = "CheckerRunning"

765

value_to_set = dbus.Boolean(True)

766

767

768

class StopChecker(PropertySetter):

769

propname = "CheckerRunning"

770

value_to_set = dbus.Boolean(False)

771

772

773

class ApproveByDefault(PropertySetter):

774

propname = "ApprovedByDefault"

775

value_to_set = dbus.Boolean(True)

776

777

778

class DenyByDefault(PropertySetter):

779

propname = "ApprovedByDefault"

780

value_to_set = dbus.Boolean(False)

781

782

783

class PropertySetterValue(PropertySetter):

784

"""Abstract class for PropertySetter recieving a value as

785

constructor argument instead of a class attribute."""

793

786

def __init__(self, value):

794

787

self.value_to_set = value

795

788

796

789

797

class SetChecker(PropertyValue):

790

class SetChecker(PropertySetterValue):

798

791

propname = "Checker"

799

792

800

793

801

class SetHost(PropertyValue):

794

class SetHost(PropertySetterValue):

802

795

propname = "Host"

803

796

804

797

805

class SetSecret(PropertyValue):

798

class SetSecret(PropertySetterValue):

806

799

propname = "Secret"

807

800

808

801

@property

816

809

value.close()

817

810

818

811

819

class MillisecondsPropertyValueArgument(PropertyValue):

820

"""Abstract class for PropertyValue taking a value argument as

821

a datetime.timedelta() but should store it as milliseconds."""

812

class PropertySetterValueMilliseconds(PropertySetterValue):

813

"""Abstract class for PropertySetterValue taking a value

814

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

815

milliseconds."""

822

816

823

817

@property

824

818

def value_to_set(self):

830

824

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

831

825

832

826

833

class SetTimeout(MillisecondsPropertyValueArgument):

827

class SetTimeout(PropertySetterValueMilliseconds):

834

828

propname = "Timeout"

835

829

836

830

837

class SetExtendedTimeout(MillisecondsPropertyValueArgument):

831

class SetExtendedTimeout(PropertySetterValueMilliseconds):

838

832

propname = "ExtendedTimeout"

839

833

840

834

841

class SetInterval(MillisecondsPropertyValueArgument):

835

class SetInterval(PropertySetterValueMilliseconds):

842

836

propname = "Interval"

843

837

844

838

845

class SetApprovalDelay(MillisecondsPropertyValueArgument):

839

class SetApprovalDelay(PropertySetterValueMilliseconds):

846

840

propname = "ApprovalDelay"

847

841

848

842

849

class SetApprovalDuration(MillisecondsPropertyValueArgument):

843

class SetApprovalDuration(PropertySetterValueMilliseconds):

850

844

propname = "ApprovalDuration"

851

845

852

846

987

981

options = self.parser.parse_args()

988

982

setattr(options, action, value)

989

983

options.verbose = True

990

options.client = ["foo"]

984

options.client = ["client"]

991

985

with self.assertParseError():

992

986

self.check_option_syntax(options)

993

987

1023

1017

for action, value in self.actions.items():

1024

1018

options = self.parser.parse_args()

1025

1019

setattr(options, action, value)

1026

options.client = ["foo"]

1020

options.client = ["client"]

1027

1021

self.check_option_syntax(options)

1028

1022

1029

1023

def test_one_client_with_all_actions_except_is_enabled(self):

1032

1026

if action == "is_enabled":

1033

1027

continue

1034

1028

setattr(options, action, value)

1035

options.client = ["foo"]

1029

options.client = ["client"]

1036

1030

self.check_option_syntax(options)

1037

1031

1038

1032

def test_two_clients_with_all_actions_except_is_enabled(self):

1041

1035

if action == "is_enabled":

1042

1036

continue

1043

1037

setattr(options, action, value)

1044

options.client = ["foo", "barbar"]

1038

options.client = ["client1", "client2"]

1045

1039

self.check_option_syntax(options)

1046

1040

1047

1041

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1050

1044

continue

1051

1045

options = self.parser.parse_args()

1052

1046

setattr(options, action, value)

1053

options.client = ["foo", "barbar"]

1047

options.client = ["client1", "client2"]

1054

1048

self.check_option_syntax(options)

1055

1049

1056

1050

def test_is_enabled_fails_without_client(self):

1062

1056

def test_is_enabled_fails_with_two_clients(self):

1063

1057

options = self.parser.parse_args()

1064

1058

options.is_enabled = True

1065

options.client = ["foo", "barbar"]

1059

options.client = ["client1", "client2"]

1066

1060

with self.assertParseError():

1067

1061

self.check_option_syntax(options)

1068

1062

1095

1089

self.assertTrue(mockbus.called)

1096

1090

1097

1091

def test_logs_and_exits_on_dbus_error(self):

1098

class MockBusFailing(object):

1092

class FailingBusStub(object):

1099

1093

def get_object(self, busname, dbus_path):

1100

1094

raise dbus.exceptions.DBusException("Test")

1101

1095

1102

1096

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

1103

1097

with self.assertRaises(SystemExit) as e:

1104

bus = get_mandos_dbus_object(bus=MockBusFailing())

1098

bus = get_mandos_dbus_object(bus=FailingBusStub())

1105

1099

1106

1100

if isinstance(e.exception.code, int):

1107

1101

self.assertNotEqual(0, e.exception.code)

1111

1105

1112

1106

class Test_get_managed_objects(TestCaseWithAssertLogs):

1113

1107

def test_calls_and_returns_GetManagedObjects(self):

1114

managed_objects = {"/clients/foo": { "Name": "foo"}}

1115

class MockObjectManager(object):

1108

managed_objects = {"/clients/client": { "Name": "client"}}

1109

class ObjectManagerStub(object):

1116

1110

def GetManagedObjects(self):

1117

1111

return managed_objects

1118

retval = get_managed_objects(MockObjectManager())

1112

retval = get_managed_objects(ObjectManagerStub())

1119

1113

self.assertDictEqual(managed_objects, retval)

1120

1114

1121

1115

def test_logs_and_exits_on_dbus_error(self):

1122

1116

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

1123

1117

1124

class MockObjectManagerFailing(object):

1118

class ObjectManagerFailingStub(object):

1125

1119

def GetManagedObjects(self):

1126

1120

dbus_logger.error("Test")

1127

1121

raise dbus.exceptions.DBusException("Test")

1138

1132

try:

1139

1133

with self.assertLogs(log, logging.CRITICAL) as watcher:

1140

1134

with self.assertRaises(SystemExit) as e:

1141

get_managed_objects(MockObjectManagerFailing())

1135

get_managed_objects(ObjectManagerFailingStub())

1142

1136

finally:

1143

1137

dbus_logger.removeFilter(counting_handler)

1144

1138

1161

1155

add_command_line_options(self.parser)

1162

1156

1163

1157

def test_is_enabled(self):

1164

self.assert_command_from_args(["--is-enabled", "foo"],

1158

self.assert_command_from_args(["--is-enabled", "client"],

1165

1159

command.IsEnabled)

1166

1160

1167

1161

def assert_command_from_args(self, args, command_cls,

1178

1172

self.assertEqual(value, getattr(command, key))

1179

1173

1180

1174

def test_is_enabled_short(self):

1181

self.assert_command_from_args(["-V", "foo"],

1175

self.assert_command_from_args(["-V", "client"],

1182

1176

command.IsEnabled)

1183

1177

1184

1178

def test_approve(self):

1185

self.assert_command_from_args(["--approve", "foo"],

1179

self.assert_command_from_args(["--approve", "client"],

1186

1180

command.Approve)

1187

1181

1188

1182

def test_approve_short(self):

1189

self.assert_command_from_args(["-A", "foo"], command.Approve)

1183

self.assert_command_from_args(["-A", "client"],

1184

command.Approve)

1190

1185

1191

1186

def test_deny(self):

1192

self.assert_command_from_args(["--deny", "foo"], command.Deny)

1187

self.assert_command_from_args(["--deny", "client"],

1188

command.Deny)

1193

1189

1194

1190

def test_deny_short(self):

1195

self.assert_command_from_args(["-D", "foo"], command.Deny)

1191

self.assert_command_from_args(["-D", "client"], command.Deny)

1196

1192

1197

1193

def test_remove(self):

1198

self.assert_command_from_args(["--remove", "foo"],

1194

self.assert_command_from_args(["--remove", "client"],

1199

1195

command.Remove)

1200

1196

1201

1197

def test_deny_before_remove(self):

1202

1198

options = self.parser.parse_args(["--deny", "--remove",

1203

"foo"])

1199

"client"])

1204

1200

check_option_syntax(self.parser, options)

1205

1201

commands = commands_from_options(options)

1206

1202

self.assertEqual(2, len(commands))

1217

1213

self.assertIsInstance(commands[1], command.Remove)

1218

1214

1219

1215

def test_remove_short(self):

1220

self.assert_command_from_args(["-r", "foo"], command.Remove)

1216

self.assert_command_from_args(["-r", "client"],

1217

command.Remove)

1221

1218

1222

1219

def test_dump_json(self):

1223

1220

self.assert_command_from_args(["--dump-json"],

1224

1221

command.DumpJSON)

1225

1222

1226

1223

def test_enable(self):

1227

self.assert_command_from_args(["--enable", "foo"],

1224

self.assert_command_from_args(["--enable", "client"],

1228

1225

command.Enable)

1229

1226

1230

1227

def test_enable_short(self):

1231

self.assert_command_from_args(["-e", "foo"], command.Enable)

1228

self.assert_command_from_args(["-e", "client"],

1229

command.Enable)

1232

1230

1233

1231

def test_disable(self):

1234

self.assert_command_from_args(["--disable", "foo"],

1232

self.assert_command_from_args(["--disable", "client"],

1235

1233

command.Disable)

1236

1234

1237

1235

def test_disable_short(self):

1238

self.assert_command_from_args(["-d", "foo"], command.Disable)

1236

self.assert_command_from_args(["-d", "client"],

1237

command.Disable)

1239

1238

1240

1239

def test_bump_timeout(self):

1241

self.assert_command_from_args(["--bump-timeout", "foo"],

1240

self.assert_command_from_args(["--bump-timeout", "client"],

1242

1241

command.BumpTimeout)

1243

1242

1244

1243

def test_bump_timeout_short(self):

1245

self.assert_command_from_args(["-b", "foo"],

1244

self.assert_command_from_args(["-b", "client"],

1246

1245

command.BumpTimeout)

1247

1246

1248

1247

def test_start_checker(self):

1249

self.assert_command_from_args(["--start-checker", "foo"],

1248

self.assert_command_from_args(["--start-checker", "client"],

1250

1249

command.StartChecker)

1251

1250

1252

1251

def test_stop_checker(self):

1253

self.assert_command_from_args(["--stop-checker", "foo"],

1252

self.assert_command_from_args(["--stop-checker", "client"],

1254

1253

command.StopChecker)

1255

1254

1256

1255

def test_approve_by_default(self):

1257

self.assert_command_from_args(["--approve-by-default", "foo"],

1256

self.assert_command_from_args(["--approve-by-default",

1257

"client"],

1258

command.ApproveByDefault)

1259

1260

def test_deny_by_default(self):

1261

self.assert_command_from_args(["--deny-by-default", "foo"],

1261

self.assert_command_from_args(["--deny-by-default", "client"],

1262

command.DenyByDefault)

1263

1264

def test_checker(self):

1265

self.assert_command_from_args(["--checker", ":", "foo"],

1265

self.assert_command_from_args(["--checker", ":", "client"],

1266

command.SetChecker,

1267

value_to_set=":")

1268

1269

def test_checker_empty(self):

1270

self.assert_command_from_args(["--checker", "", "foo"],

1270

self.assert_command_from_args(["--checker", "", "client"],

1271

command.SetChecker,

1272

value_to_set="")

1273

1274

def test_checker_short(self):

1275

self.assert_command_from_args(["-c", ":", "foo"],

1275

self.assert_command_from_args(["-c", ":", "client"],

1276

command.SetChecker,

1277

value_to_set=":")

1278

1279

def test_host(self):

1280

self.assert_command_from_args(["--host", "foo.example.org",

1281

"foo"], command.SetHost,

1282

value_to_set="foo.example.org")

1280

self.assert_command_from_args(

1281

["--host", "client.example.org", "client"],

1282

command.SetHost, value_to_set="client.example.org")

1283

1284

def test_host_short(self):

1285

self.assert_command_from_args(["-H", "foo.example.org",

1286

"foo"], command.SetHost,

1287

value_to_set="foo.example.org")

1285

self.assert_command_from_args(

1286

["-H", "client.example.org", "client"], command.SetHost,

1287

value_to_set="client.example.org")

1288

1289

def test_secret_devnull(self):

1290

self.assert_command_from_args(["--secret", os.path.devnull,

1291

"foo"], command.SetSecret,

1291

"client"], command.SetSecret,

1292

value_to_set=b"")

1293

1294

def test_secret_tempfile(self):

1297

f.write(value)

1298

f.seek(0)

1299

self.assert_command_from_args(["--secret", f.name,

1300

"foo"], command.SetSecret,

1300

"client"],

1301

command.SetSecret,

1301

1302

value_to_set=value)

1302

1303

1304

def test_secret_devnull_short(self):

1304

self.assert_command_from_args(["-s", os.path.devnull, "foo"],

1305

command.SetSecret,

1305

self.assert_command_from_args(["-s", os.path.devnull,

1306

"client"], command.SetSecret,

1306

1307

value_to_set=b"")

1307

1308

1309

def test_secret_tempfile_short(self):

1310

1311

value = b"secret\0xyzzy\nbar"

1311

1312

f.write(value)

1312

1313

f.seek(0)

1313

self.assert_command_from_args(["-s", f.name, "foo"],

1314

self.assert_command_from_args(["-s", f.name, "client"],

1314

1315

command.SetSecret,

1315

1316

value_to_set=value)

1316

1317

1318

def test_timeout(self):

1318

self.assert_command_from_args(["--timeout", "PT5M", "foo"],

1319

self.assert_command_from_args(["--timeout", "PT5M", "client"],

1319

1320

command.SetTimeout,

1320

1321

value_to_set=300000)

1321

1322

1323

def test_timeout_short(self):

1323

self.assert_command_from_args(["-t", "PT5M", "foo"],

1324

self.assert_command_from_args(["-t", "PT5M", "client"],

1324

1325

command.SetTimeout,

1325

1326

value_to_set=300000)

1326

1327

1328

def test_extended_timeout(self):

1328

1329

self.assert_command_from_args(["--extended-timeout", "PT15M",

1329

"foo"],

1330

"client"],

1330

1331

command.SetExtendedTimeout,

1331

1332

value_to_set=900000)

1332

1333

1334

def test_interval(self):

1334

self.assert_command_from_args(["--interval", "PT2M", "foo"],

1335

command.SetInterval,

1335

self.assert_command_from_args(["--interval", "PT2M",

1336

"client"], command.SetInterval,

1336

1337

value_to_set=120000)

1337

1338

1339

def test_interval_short(self):

1339

self.assert_command_from_args(["-i", "PT2M", "foo"],

1340

self.assert_command_from_args(["-i", "PT2M", "client"],

1340

1341

command.SetInterval,

1341

1342

value_to_set=120000)

1342

1343

1344

def test_approval_delay(self):

1344

1345

self.assert_command_from_args(["--approval-delay", "PT30S",

1345

"foo"],

1346

"client"],

1346

1347

command.SetApprovalDelay,

1347

1348

value_to_set=30000)

1348

1349

1350

def test_approval_duration(self):

1350

1351

self.assert_command_from_args(["--approval-duration", "PT1S",

1351

"foo"],

1352

"client"],

1352

1353

command.SetApprovalDuration,

1353

1354

value_to_set=1000)

1354

1355

1437

1438

LastCheckerStatus=-2)

1438

1439

self.clients = collections.OrderedDict(

1439

1440

[

1440

("/clients/foo", self.client.attributes),

1441

("/clients/barbar", self.other_client.attributes),

1441

(self.client.__dbus_object_path__,

1442

self.client.attributes),

1443

(self.other_client.__dbus_object_path__,

1444

self.other_client.attributes),

1442

1445

])

1443

self.one_client = {"/clients/foo": self.client.attributes}

1446

self.one_client = {self.client.__dbus_object_path__:

1447

self.client.attributes}

1444

1448

1445

1449

@property

1446

1450

def bus(self):

1447

class Bus(object):

1451

class MockBus(object):

1448

1452

@staticmethod

1449

1453

def get_object(client_bus_name, path):

1450

1454

self.assertEqual(dbus_busname, client_bus_name)

1451

return {

1452

# Note: "self" here is the TestCmd instance, not

1453

# the Bus instance, since this is a static method!

1454

"/clients/foo": self.client,

1455

"/clients/barbar": self.other_client,

1456

}[path]

1457

return Bus()

1455

# Note: "self" here is the TestCmd instance, not the

1456

# MockBus instance, since this is a static method!

1457

if path == self.client.__dbus_object_path__:

1458

return self.client

1459

elif path == self.other_client.__dbus_object_path__:

1460

return self.other_client

1461

return MockBus()

1458

1462

1459

1463

1460

1464

class TestBaseCommands(TestCommand):

1491

1495

client.calls)

1492

1496

1493

1497

def test_Remove(self):

1494

class MockMandos(object):

1498

class MandosSpy(object):

1495

1499

def __init__(self):

1496

1500

self.calls = []

1497

1501

def RemoveClient(self, dbus_path):

1498

1502

self.calls.append(("RemoveClient", (dbus_path,)))

1499

mandos = MockMandos()

1503

mandos = MandosSpy()

1500

1504

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

1501

1505

for clientpath in self.clients:

1502

1506

self.assertIn(("RemoveClient", (clientpath,)),

1693

1697

self.assertEqual(expected_output, buffer.getvalue())

1694

1698

1695

1699

1696

class TestPropertyCmd(TestCommand):

1697

"""Abstract class for tests of command.Property classes"""

1700

class TestPropertySetterCmd(TestCommand):

1701

"""Abstract class for tests of command.PropertySetter classes"""

1698

1702

def runTest(self):

1699

1703

if not hasattr(self, "command"):

1700

1704

return

1721

1725

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

1722

1726

1723

1727

1724

class TestEnableCmd(TestPropertyCmd):

1728

class TestEnableCmd(TestPropertySetterCmd):

1725

1729

command = command.Enable

1726

1730

propname = "Enabled"

1727

1731

values_to_set = [dbus.Boolean(True)]

1728

1732

1729

1733

1730

class TestDisableCmd(TestPropertyCmd):

1734

class TestDisableCmd(TestPropertySetterCmd):

1731

1735

command = command.Disable

1732

1736

propname = "Enabled"

1733

1737

values_to_set = [dbus.Boolean(False)]

1734

1738

1735

1739

1736

class TestBumpTimeoutCmd(TestPropertyCmd):

1740

class TestBumpTimeoutCmd(TestPropertySetterCmd):

1737

1741

command = command.BumpTimeout

1738

1742

propname = "LastCheckedOK"

1739

1743

values_to_set = [""]

1740

1744

1741

1745

1742

class TestStartCheckerCmd(TestPropertyCmd):

1746

class TestStartCheckerCmd(TestPropertySetterCmd):

1743

1747

command = command.StartChecker

1744

1748

propname = "CheckerRunning"

1745

1749

values_to_set = [dbus.Boolean(True)]

1746

1750

1747

1751

1748

class TestStopCheckerCmd(TestPropertyCmd):

1752

class TestStopCheckerCmd(TestPropertySetterCmd):

1749

1753

command = command.StopChecker

1750

1754

propname = "CheckerRunning"

1751

1755

values_to_set = [dbus.Boolean(False)]

1752

1756

1753

1757

1754

class TestApproveByDefaultCmd(TestPropertyCmd):

1758

class TestApproveByDefaultCmd(TestPropertySetterCmd):

1755

1759

command = command.ApproveByDefault

1756

1760

propname = "ApprovedByDefault"

1757

1761

values_to_set = [dbus.Boolean(True)]

1758

1762

1759

1763

1760

class TestDenyByDefaultCmd(TestPropertyCmd):

1764

class TestDenyByDefaultCmd(TestPropertySetterCmd):

1761

1765

command = command.DenyByDefault

1762

1766

propname = "ApprovedByDefault"

1763

1767

values_to_set = [dbus.Boolean(False)]

1764

1768

1765

1769

1766

class TestPropertyValueCmd(TestPropertyCmd):

1767

"""Abstract class for tests of PropertyValueCmd classes"""

1770

class TestPropertySetterValueCmd(TestPropertySetterCmd):

1771

"""Abstract class for tests of PropertySetterValueCmd classes"""

1768

1772

1769

1773

def runTest(self):

1770

if type(self) is TestPropertyValueCmd:

1774

if type(self) is TestPropertySetterValueCmd:

1771

1775

return

1772

return super(TestPropertyValueCmd, self).runTest()

1776

return super(TestPropertySetterValueCmd, self).runTest()

1773

1777

1774

1778

def run_command(self, value, clients):

1775

1779

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

1776

1780

1777

1781

1778

class TestSetCheckerCmd(TestPropertyValueCmd):

1782

class TestSetCheckerCmd(TestPropertySetterValueCmd):

1779

1783

command = command.SetChecker

1780

1784

propname = "Checker"

1781

1785

values_to_set = ["", ":", "fping -q -- %s"]

1782

1786

1783

1787

1784

class TestSetHostCmd(TestPropertyValueCmd):

1788

class TestSetHostCmd(TestPropertySetterValueCmd):

1785

1789

command = command.SetHost

1786

1790

propname = "Host"

1787

values_to_set = ["192.0.2.3", "foo.example.org"]

1788

1789

1790

class TestSetSecretCmd(TestPropertyValueCmd):

1791

values_to_set = ["192.0.2.3", "client.example.org"]

1792

1793

1794

class TestSetSecretCmd(TestPropertySetterValueCmd):

1791

1795

command = command.SetSecret

1792

1796

propname = "Secret"

1793

1797

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

1794

1798

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

1795

values_to_get = [b"", b"secret\0xyzzy\nbar"]

1796

1797

1798

class TestSetTimeoutCmd(TestPropertyValueCmd):

1799

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

1800

1801

1802

class TestSetTimeoutCmd(TestPropertySetterValueCmd):

1799

1803

command = command.SetTimeout

1800

1804

propname = "Timeout"

1801

1805

values_to_set = [datetime.timedelta(),

1803

1807

datetime.timedelta(seconds=1),

1804

1808

datetime.timedelta(weeks=1),

1805

1809

datetime.timedelta(weeks=52)]

1806

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1807

1808

1809

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1810

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

1811

1812

1813

class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):

1810

1814

command = command.SetExtendedTimeout

1811

1815

propname = "ExtendedTimeout"

1812

1816

values_to_set = [datetime.timedelta(),

1814

1818

datetime.timedelta(seconds=1),

1815

1819

datetime.timedelta(weeks=1),

1816

1820

datetime.timedelta(weeks=52)]

1817

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1818

1819

1820

class TestSetIntervalCmd(TestPropertyValueCmd):

1821

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

1822

1823

1824

class TestSetIntervalCmd(TestPropertySetterValueCmd):

1821

1825

command = command.SetInterval

1822

1826

propname = "Interval"

1823

1827

values_to_set = [datetime.timedelta(),

1825

1829

datetime.timedelta(seconds=1),

1826

1830

datetime.timedelta(weeks=1),

1827

1831

datetime.timedelta(weeks=52)]

1828

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1829

1830

1831

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1832

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

1833

1834

1835

class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):

1832

1836

command = command.SetApprovalDelay

1833

1837

propname = "ApprovalDelay"

1834

1838

values_to_set = [datetime.timedelta(),

1836

1840

datetime.timedelta(seconds=1),

1837

1841

datetime.timedelta(weeks=1),

1838

1842

datetime.timedelta(weeks=52)]

1839

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1840

1841

1842

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1843

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

1844

1845

1846

class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):

1843

1847

command = command.SetApprovalDuration

1844

1848

propname = "ApprovalDuration"

1845

1849

values_to_set = [datetime.timedelta(),

1847

1851

datetime.timedelta(seconds=1),

1848

1852

datetime.timedelta(weeks=1),

1849

1853

datetime.timedelta(weeks=52)]

1850

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1854

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

1851

1855

1852

1856

1853

1857