/mandos/trunk : revision 1083

61

62

if sys.version_info.major == 2:

63

str = unicode

64

import StringIO

65

io.StringIO = StringIO.StringIO

66

64

67

65

locale.setlocale(locale.LC_ALL, "")

68

66

83

81

84

82

def main():

85

83

parser = argparse.ArgumentParser()

84

86

85

add_command_line_options(parser)

87

86

88

87

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

463

464

465

class NullFilter(logging.Filter):

465

466

def filter(self, record):

612

613

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

613

614

"LastCheckerStatus")

614

615

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

615

622

616

623

class DumpJSON(Output):

617

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

624

def output(self, clients):

618

625

data = {client["Name"]:

619

626

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

620

627

for key in self.all_keywords}

621

for client in clients.values()}

622

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

628

for client in clients}

629

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

623

630

624

631

@staticmethod

625

632

def dbus_boolean_to_bool(value):

632

639

def __init__(self, verbose=False):

633

640

self.verbose = verbose

634

641

635

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

642

def output(self, clients):

636

643

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

637

644

"LastCheckedOK")

638

645

keywords = default_keywords

639

646

if self.verbose:

640

647

keywords = self.all_keywords

641

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

648

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

642

649

643

650

class TableOfClients(object):

644

651

tableheaders = {

725

732

seconds=td.seconds % 60))

726

733

727

734

728

class PropertySetter(Base):

735

class Property(Base):

729

736

"Abstract class for Actions for setting one client property"

730

737

731

738

def run_on_one_client(self, client, properties):

746

753

raise NotImplementedError()

747

754

748

755

749

class Enable(PropertySetter):

756

class Enable(Property):

750

757

propname = "Enabled"

751

758

value_to_set = dbus.Boolean(True)

752

759

753

760

754

class Disable(PropertySetter):

761

class Disable(Property):

755

762

propname = "Enabled"

756

763

value_to_set = dbus.Boolean(False)

757

764

758

765

759

class BumpTimeout(PropertySetter):

766

class BumpTimeout(Property):

760

767

propname = "LastCheckedOK"

761

768

value_to_set = ""

762

769

763

770

764

class StartChecker(PropertySetter):

765

propname = "CheckerRunning"

766

value_to_set = dbus.Boolean(True)

767

768

769

class StopChecker(PropertySetter):

770

propname = "CheckerRunning"

771

value_to_set = dbus.Boolean(False)

772

773

774

class ApproveByDefault(PropertySetter):

775

propname = "ApprovedByDefault"

776

value_to_set = dbus.Boolean(True)

777

778

779

class DenyByDefault(PropertySetter):

780

propname = "ApprovedByDefault"

781

value_to_set = dbus.Boolean(False)

782

783

784

class PropertySetterValue(PropertySetter):

785

"""Abstract class for PropertySetter recieving a value as

786

constructor argument instead of a class attribute."""

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"

787

793

def __init__(self, value):

788

794

self.value_to_set = value

789

795

790

796

791

class SetChecker(PropertySetterValue):

797

class SetChecker(PropertyValue):

792

798

propname = "Checker"

793

799

794

800

795

class SetHost(PropertySetterValue):

801

class SetHost(PropertyValue):

796

802

propname = "Host"

797

803

798

804

799

class SetSecret(PropertySetterValue):

805

class SetSecret(PropertyValue):

800

806

propname = "Secret"

801

807

802

808

@property

810

816

value.close()

811

817

812

818

813

class PropertySetterValueMilliseconds(PropertySetterValue):

814

"""Abstract class for PropertySetterValue taking a value

815

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

816

milliseconds."""

819

class MillisecondsPropertyValueArgument(PropertyValue):

820

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

821

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

817

822

818

823

@property

819

824

def value_to_set(self):

825

830

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

826

831

827

832

828

class SetTimeout(PropertySetterValueMilliseconds):

833

class SetTimeout(MillisecondsPropertyValueArgument):

829

834

propname = "Timeout"

830

835

831

836

832

class SetExtendedTimeout(PropertySetterValueMilliseconds):

837

class SetExtendedTimeout(MillisecondsPropertyValueArgument):

833

838

propname = "ExtendedTimeout"

834

839

835

840

836

class SetInterval(PropertySetterValueMilliseconds):

841

class SetInterval(MillisecondsPropertyValueArgument):

837

842

propname = "Interval"

838

843

839

844

840

class SetApprovalDelay(PropertySetterValueMilliseconds):

845

class SetApprovalDelay(MillisecondsPropertyValueArgument):

841

846

propname = "ApprovalDelay"

842

847

843

848

844

class SetApprovalDuration(PropertySetterValueMilliseconds):

849

class SetApprovalDuration(MillisecondsPropertyValueArgument):

845

850

propname = "ApprovalDuration"

846

851

847

852

982

987

options = self.parser.parse_args()

983

988

setattr(options, action, value)

984

989

options.verbose = True

985

options.client = ["client"]

990

options.client = ["foo"]

986

991

with self.assertParseError():

987

992

self.check_option_syntax(options)

988

993

1018

1023

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

1019

1024

options = self.parser.parse_args()

1020

1025

setattr(options, action, value)

1021

options.client = ["client"]

1026

options.client = ["foo"]

1022

1027

self.check_option_syntax(options)

1023

1028

1024

1029

def test_one_client_with_all_actions_except_is_enabled(self):

1027

1032

if action == "is_enabled":

1028

1033

continue

1029

1034

setattr(options, action, value)

1030

options.client = ["client"]

1035

options.client = ["foo"]

1031

1036

self.check_option_syntax(options)

1032

1037

1033

1038

def test_two_clients_with_all_actions_except_is_enabled(self):

1036

1041

if action == "is_enabled":

1037

1042

continue

1038

1043

setattr(options, action, value)

1039

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

1044

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

1040

1045

self.check_option_syntax(options)

1041

1046

1042

1047

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1045

1050

continue

1046

1051

options = self.parser.parse_args()

1047

1052

setattr(options, action, value)

1048

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

1053

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

1049

1054

self.check_option_syntax(options)

1050

1055

1051

1056

def test_is_enabled_fails_without_client(self):

1057

1062

def test_is_enabled_fails_with_two_clients(self):

1058

1063

options = self.parser.parse_args()

1059

1064

options.is_enabled = True

1060

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

1065

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

1061

1066

with self.assertParseError():

1062

1067

self.check_option_syntax(options)

1063

1068

1090

1095

self.assertTrue(mockbus.called)

1091

1096

1092

1097

def test_logs_and_exits_on_dbus_error(self):

1093

class FailingBusStub(object):

1098

class MockBusFailing(object):

1094

1099

def get_object(self, busname, dbus_path):

1095

1100

raise dbus.exceptions.DBusException("Test")

1096

1101

1097

1102

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

1098

1103

with self.assertRaises(SystemExit) as e:

1099

bus = get_mandos_dbus_object(bus=FailingBusStub())

1104

bus = get_mandos_dbus_object(bus=MockBusFailing())

1100

1105

1101

1106

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

1102

1107

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

1106

1111

1107

1112

class Test_get_managed_objects(TestCaseWithAssertLogs):

1108

1113

def test_calls_and_returns_GetManagedObjects(self):

1109

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

1110

class ObjectManagerStub(object):

1114

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

1115

class MockObjectManager(object):

1111

1116

def GetManagedObjects(self):

1112

1117

return managed_objects

1113

retval = get_managed_objects(ObjectManagerStub())

1118

retval = get_managed_objects(MockObjectManager())

1114

1119

self.assertDictEqual(managed_objects, retval)

1115

1120

1116

1121

def test_logs_and_exits_on_dbus_error(self):

1117

1122

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

1118

1123

1119

class ObjectManagerFailingStub(object):

1124

class MockObjectManagerFailing(object):

1120

1125

def GetManagedObjects(self):

1121

1126

dbus_logger.error("Test")

1122

1127

raise dbus.exceptions.DBusException("Test")

1133

1138

try:

1134

1139

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

1135

1140

with self.assertRaises(SystemExit) as e:

1136

get_managed_objects(ObjectManagerFailingStub())

1141

get_managed_objects(MockObjectManagerFailing())

1137

1142

finally:

1138

1143

dbus_logger.removeFilter(counting_handler)

1139

1144

1156

1161

add_command_line_options(self.parser)

1157

1162

1158

1163

def test_is_enabled(self):

1159

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

1164

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

1160

1165

command.IsEnabled)

1161

1166

1162

1167

def assert_command_from_args(self, args, command_cls,

1173

1178

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

1174

1179

1175

1180

def test_is_enabled_short(self):

1176

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

1181

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

1177

1182

command.IsEnabled)

1178

1183

1179

1184

def test_approve(self):

1180

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

1185

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

1181

1186

command.Approve)

1182

1187

1183

1188

def test_approve_short(self):

1184

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

1185

command.Approve)

1189

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

1186

1190

1187

1191

def test_deny(self):

1188

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

1189

command.Deny)

1192

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

1190

1193

1191

1194

def test_deny_short(self):

1192

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

1195

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

1193

1196

1194

1197

def test_remove(self):

1195

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

1198

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

1196

1199

command.Remove)

1197

1200

1198

1201

def test_deny_before_remove(self):

1199

1202

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

1200

"client"])

1203

"foo"])

1201

1204

check_option_syntax(self.parser, options)

1202

1205

commands = commands_from_options(options)

1203

1206

self.assertEqual(2, len(commands))

1214

1217

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

1215

1218

1216

1219

def test_remove_short(self):

1217

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

1218

command.Remove)

1220

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

1219

1221

1220

1222

def test_dump_json(self):

1221

1223

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

1222

1224

command.DumpJSON)

1223

1225

1224

1226

def test_enable(self):

1225

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

1227

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

1226

1228

command.Enable)

1227

1229

1228

1230

def test_enable_short(self):

1229

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

1230

command.Enable)

1231

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

1231

1232

1233

def test_disable(self):

1233

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

1234

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

1234

1235

command.Disable)

1235

1236

1237

def test_disable_short(self):

1237

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

1238

command.Disable)

1238

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

1239

1240

def test_bump_timeout(self):

1241

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

1241

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

1242

command.BumpTimeout)

1243

1244

def test_bump_timeout_short(self):

1245

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

1245

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

1246

command.BumpTimeout)

1247

1248

def test_start_checker(self):

1249

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

1249

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

1250

command.StartChecker)

1251

1252

def test_stop_checker(self):

1253

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

1253

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

1254

command.StopChecker)

1255

1256

def test_approve_by_default(self):

1257

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

1258

"client"],

1257

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

1259

1258

command.ApproveByDefault)

1260

1259

1261

1260

def test_deny_by_default(self):

1262

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

1261

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

1263

1262

command.DenyByDefault)

1264

1263

1265

1264

def test_checker(self):

1266

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

1265

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

1267

1266

command.SetChecker,

1268

1267

value_to_set=":")

1269

1268

1270

1269

def test_checker_empty(self):

1271

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

1270

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

1272

1271

command.SetChecker,

1273

1272

value_to_set="")

1274

1273

1275

1274

def test_checker_short(self):

1276

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

1275

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

1277

1276

command.SetChecker,

1278

1277

value_to_set=":")

1279

1278

1280

1279

def test_host(self):

1281

self.assert_command_from_args(

1282

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

1283

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

1280

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

1281

"foo"], command.SetHost,

1282

value_to_set="foo.example.org")

1284

1283

1285

1284

def test_host_short(self):

1286

self.assert_command_from_args(

1287

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

1288

value_to_set="client.example.org")

1285

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

1286

"foo"], command.SetHost,

1287

value_to_set="foo.example.org")

1289

1288

1290

1289

def test_secret_devnull(self):

1291

1290

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

1292

"client"], command.SetSecret,

1291

"foo"], command.SetSecret,

1293

1292

value_to_set=b"")

1294

1293

1295

1294

def test_secret_tempfile(self):

1298

1297

f.write(value)

1299

1298

f.seek(0)

1300

1299

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

1301

"client"],

1302

command.SetSecret,

1300

"foo"], command.SetSecret,

1303

1301

value_to_set=value)

1304

1302

1305

1303

def test_secret_devnull_short(self):

1306

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

1307

"client"], command.SetSecret,

1304

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

1305

command.SetSecret,

1308

1306

value_to_set=b"")

1309

1307

1310

1308

def test_secret_tempfile_short(self):

1312

1310

value = b"secret\0xyzzy\nbar"

1313

1311

f.write(value)

1314

1312

f.seek(0)

1315

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

1313

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

1316

1314

command.SetSecret,

1317

1315

value_to_set=value)

1318

1316

1319

1317

def test_timeout(self):

1320

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

1318

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

1321

1319

command.SetTimeout,

1322

1320

value_to_set=300000)

1323

1321

1324

1322

def test_timeout_short(self):

1325

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

1323

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

1326

1324

command.SetTimeout,

1327

1325

value_to_set=300000)

1328

1326

1329

1327

def test_extended_timeout(self):

1330

1328

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

1331

"client"],

1329

"foo"],

1332

1330

command.SetExtendedTimeout,

1333

1331

value_to_set=900000)

1334

1332

1335

1333

def test_interval(self):

1336

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

1337

"client"], command.SetInterval,

1334

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

1335

command.SetInterval,

1338

1336

value_to_set=120000)

1339

1337

1340

1338

def test_interval_short(self):

1341

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

1339

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

1342

1340

command.SetInterval,

1343

1341

value_to_set=120000)

1344

1342

1345

1343

def test_approval_delay(self):

1346

1344

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

1347

"client"],

1345

"foo"],

1348

1346

command.SetApprovalDelay,

1349

1347

value_to_set=30000)

1350

1348

1351

1349

def test_approval_duration(self):

1352

1350

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

1353

"client"],

1351

"foo"],

1354

1352

command.SetApprovalDuration,

1355

1353

value_to_set=1000)

1356

1354

1439

1437

LastCheckerStatus=-2)

1440

1438

self.clients = collections.OrderedDict(

1441

1439

[

1442

(self.client.__dbus_object_path__,

1443

self.client.attributes),

1444

(self.other_client.__dbus_object_path__,

1445

self.other_client.attributes),

1440

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

1441

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

1446

1442

])

1447

self.one_client = {self.client.__dbus_object_path__:

1448

self.client.attributes}

1443

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

1449

1444

1450

1445

@property

1451

1446

def bus(self):

1452

class MockBus(object):

1447

class Bus(object):

1453

1448

@staticmethod

1454

1449

def get_object(client_bus_name, path):

1455

1450

self.assertEqual(dbus_busname, client_bus_name)

1456

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

1457

# MockBus instance, since this is a static method!

1458

if path == self.client.__dbus_object_path__:

1459

return self.client

1460

elif path == self.other_client.__dbus_object_path__:

1461

return self.other_client

1462

return MockBus()

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

1463

1458

1464

1459

1465

1460

class TestBaseCommands(TestCommand):

1496

1491

client.calls)

1497

1492

1498

1493

def test_Remove(self):

1499

class MandosSpy(object):

1494

class MockMandos(object):

1500

1495

def __init__(self):

1501

1496

self.calls = []

1502

1497

def RemoveClient(self, dbus_path):

1503

1498

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

1504

mandos = MandosSpy()

1499

mandos = MockMandos()

1505

1500

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

1506

1501

for clientpath in self.clients:

1507

1502

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

1559

1554

}

1560

1555

1561

1556

def test_DumpJSON_normal(self):

1562

with self.capture_stdout_to_buffer() as buffer:

1563

command.DumpJSON().run(self.clients)

1564

json_data = json.loads(buffer.getvalue())

1557

output = command.DumpJSON().output(self.clients.values())

1558

json_data = json.loads(output)

1565

1559

self.assertDictEqual(self.expected_json, json_data)

1566

1560

1567

@staticmethod

1568

@contextlib.contextmanager

1569

def capture_stdout_to_buffer():

1570

capture_buffer = io.StringIO()

1571

old_stdout = sys.stdout

1572

sys.stdout = capture_buffer

1573

try:

1574

yield capture_buffer

1575

finally:

1576

sys.stdout = old_stdout

1577

1578

1561

def test_DumpJSON_one_client(self):

1579

with self.capture_stdout_to_buffer() as buffer:

1580

command.DumpJSON().run(self.one_client)

1581

json_data = json.loads(buffer.getvalue())

1562

output = command.DumpJSON().output(self.one_client.values())

1563

json_data = json.loads(output)

1582

1564

expected_json = {"foo": self.expected_json["foo"]}

1583

1565

self.assertDictEqual(expected_json, json_data)

1584

1566

1585

1567

def test_PrintTable_normal(self):

1586

with self.capture_stdout_to_buffer() as buffer:

1587

command.PrintTable().run(self.clients)

1568

output = command.PrintTable().output(self.clients.values())

1588

1569

expected_output = "\n".join((

1589

1570

"Name Enabled Timeout Last Successful Check",

1590

1571

"foo Yes 00:05:00 2019-02-03T00:00:00 ",

1591

1572

"barbar Yes 00:05:00 2019-02-04T00:00:00 ",

1592

)) + "\n"

1593

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

1573

))

1574

self.assertEqual(expected_output, output)

1594

1575

1595

1576

def test_PrintTable_verbose(self):

1596

with self.capture_stdout_to_buffer() as buffer:

1597

command.PrintTable(verbose=True).run(self.clients)

1577

output = command.PrintTable(verbose=True).output(

1578

self.clients.values())

1598

1579

columns = (

1599

1580

(

1600

1581

"Name ",

1682

1663

)

1683

1664

)

1684

1665

num_lines = max(len(rows) for rows in columns)

1685

expected_output = ("\n".join("".join(rows[line]

1686

for rows in columns)

1687

for line in range(num_lines))

1688

+ "\n")

1689

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

1666

expected_output = "\n".join("".join(rows[line]

1667

for rows in columns)

1668

for line in range(num_lines))

1669

self.assertEqual(expected_output, output)

1690

1670

1691

1671

def test_PrintTable_one_client(self):

1692

with self.capture_stdout_to_buffer() as buffer:

1693

command.PrintTable().run(self.one_client)

1672

output = command.PrintTable().output(self.one_client.values())

1694

1673

expected_output = "\n".join((

1695

1674

"Name Enabled Timeout Last Successful Check",

1696

1675

"foo Yes 00:05:00 2019-02-03T00:00:00 ",

1697

)) + "\n"

1698

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

1699

1700

1701

class TestPropertySetterCmd(TestCommand):

1702

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

1676

))

1677

self.assertEqual(expected_output, output)

1678

1679

1680

class TestPropertyCmd(TestCommand):

1681

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

1703

1682

def runTest(self):

1704

1683

if not hasattr(self, "command"):

1705

1684

return

1726

1705

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

1727

1706

1728

1707

1729

class TestEnableCmd(TestPropertySetterCmd):

1708

class TestEnableCmd(TestPropertyCmd):

1730

1709

command = command.Enable

1731

1710

propname = "Enabled"

1732

1711

values_to_set = [dbus.Boolean(True)]

1733

1712

1734

1713

1735

class TestDisableCmd(TestPropertySetterCmd):

1714

class TestDisableCmd(TestPropertyCmd):

1736

1715

command = command.Disable

1737

1716

propname = "Enabled"

1738

1717

values_to_set = [dbus.Boolean(False)]

1739

1718

1740

1719

1741

class TestBumpTimeoutCmd(TestPropertySetterCmd):

1720

class TestBumpTimeoutCmd(TestPropertyCmd):

1742

1721

command = command.BumpTimeout

1743

1722

propname = "LastCheckedOK"

1744

1723

values_to_set = [""]

1745

1724

1746

1725

1747

class TestStartCheckerCmd(TestPropertySetterCmd):

1726

class TestStartCheckerCmd(TestPropertyCmd):

1748

1727

command = command.StartChecker

1749

1728

propname = "CheckerRunning"

1750

1729

values_to_set = [dbus.Boolean(True)]

1751

1730

1752

1731

1753

class TestStopCheckerCmd(TestPropertySetterCmd):

1732

class TestStopCheckerCmd(TestPropertyCmd):

1754

1733

command = command.StopChecker

1755

1734

propname = "CheckerRunning"

1756

1735

values_to_set = [dbus.Boolean(False)]

1757

1736

1758

1737

1759

class TestApproveByDefaultCmd(TestPropertySetterCmd):

1738

class TestApproveByDefaultCmd(TestPropertyCmd):

1760

1739

command = command.ApproveByDefault

1761

1740

propname = "ApprovedByDefault"

1762

1741

values_to_set = [dbus.Boolean(True)]

1763

1742

1764

1743

1765

class TestDenyByDefaultCmd(TestPropertySetterCmd):

1744

class TestDenyByDefaultCmd(TestPropertyCmd):

1766

1745

command = command.DenyByDefault

1767

1746

propname = "ApprovedByDefault"

1768

1747

values_to_set = [dbus.Boolean(False)]

1769

1748

1770

1749

1771

class TestPropertySetterValueCmd(TestPropertySetterCmd):

1772

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

1750

class TestPropertyValueCmd(TestPropertyCmd):

1751

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

1773

1752

1774

1753

def runTest(self):

1775

if type(self) is TestPropertySetterValueCmd:

1754

if type(self) is TestPropertyValueCmd:

1776

1755

return

1777

return super(TestPropertySetterValueCmd, self).runTest()

1756

return super(TestPropertyValueCmd, self).runTest()

1778

1757

1779

1758

def run_command(self, value, clients):

1780

1759

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

1781

1760

1782

1761

1783

class TestSetCheckerCmd(TestPropertySetterValueCmd):

1762

class TestSetCheckerCmd(TestPropertyValueCmd):

1784

1763

command = command.SetChecker

1785

1764

propname = "Checker"

1786

1765

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

1787

1766

1788

1767

1789

class TestSetHostCmd(TestPropertySetterValueCmd):

1768

class TestSetHostCmd(TestPropertyValueCmd):

1790

1769

command = command.SetHost

1791

1770

propname = "Host"

1792

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

1793

1794

1795

class TestSetSecretCmd(TestPropertySetterValueCmd):

1771

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

1772

1773

1774

class TestSetSecretCmd(TestPropertyValueCmd):

1796

1775

command = command.SetSecret

1797

1776

propname = "Secret"

1798

1777

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

1799

1778

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

1800

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

1801

1802

1803

class TestSetTimeoutCmd(TestPropertySetterValueCmd):

1779

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

1780

1781

1782

class TestSetTimeoutCmd(TestPropertyValueCmd):

1804

1783

command = command.SetTimeout

1805

1784

propname = "Timeout"

1806

1785

values_to_set = [datetime.timedelta(),

1808

1787

datetime.timedelta(seconds=1),

1809

1788

datetime.timedelta(weeks=1),

1810

1789

datetime.timedelta(weeks=52)]

1811

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

1812

1813

1814

class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):

1790

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

1791

1792

1793

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1815

1794

command = command.SetExtendedTimeout

1816

1795

propname = "ExtendedTimeout"

1817

1796

values_to_set = [datetime.timedelta(),

1819

1798

datetime.timedelta(seconds=1),

1820

1799

datetime.timedelta(weeks=1),

1821

1800

datetime.timedelta(weeks=52)]

1822

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

1823

1824

1825

class TestSetIntervalCmd(TestPropertySetterValueCmd):

1801

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

1802

1803

1804

class TestSetIntervalCmd(TestPropertyValueCmd):

1826

1805

command = command.SetInterval

1827

1806

propname = "Interval"

1828

1807

values_to_set = [datetime.timedelta(),

1830

1809

datetime.timedelta(seconds=1),

1831

1810

datetime.timedelta(weeks=1),

1832

1811

datetime.timedelta(weeks=52)]

1833

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

1834

1835

1836

class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):

1812

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

1813

1814

1815

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1837

1816

command = command.SetApprovalDelay

1838

1817

propname = "ApprovalDelay"

1839

1818

values_to_set = [datetime.timedelta(),

1841

1820

datetime.timedelta(seconds=1),

1842

1821

datetime.timedelta(weeks=1),

1843

1822

datetime.timedelta(weeks=52)]

1844

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

1845

1846

1847

class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):

1823

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

1824

1825

1826

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1848

1827

command = command.SetApprovalDuration

1849

1828

propname = "ApprovalDuration"

1850

1829

values_to_set = [datetime.timedelta(),

1852

1831

datetime.timedelta(seconds=1),

1853

1832

datetime.timedelta(weeks=1),

1854

1833

datetime.timedelta(weeks=52)]

1855

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

1834

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

1856

1835

1857

1836

1858

1837