/mandos/trunk : revision 1082

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

556

557

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

557

558

override this run() method instead.

558

559

"""

560

self.mandos = mandos

559

561

for clientpath, properties in clients.items():

560

562

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

561

563

dbus_busname, str(clientpath))

592

594

593

595

594

596

class Remove(Base):

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)

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

601

603

602

604

603

605

class Output(Base):

611

613

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

612

614

"LastCheckerStatus")

613

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

614

622

615

623

class DumpJSON(Output):

616

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

624

def output(self, clients):

617

625

data = {client["Name"]:

618

626

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

619

627

for key in self.all_keywords}

620

for client in clients.values()}

621

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

628

for client in clients}

629

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

622

630

623

631

@staticmethod

624

632

def dbus_boolean_to_bool(value):

631

639

def __init__(self, verbose=False):

632

640

self.verbose = verbose

633

641

634

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

642

def output(self, clients):

635

643

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

636

644

"LastCheckedOK")

637

645

keywords = default_keywords

638

646

if self.verbose:

639

647

keywords = self.all_keywords

640

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

648

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

641

649

642

650

class TableOfClients(object):

643

651

tableheaders = {

724

732

seconds=td.seconds % 60))

725

733

726

734

727

class PropertySetter(Base):

735

class Property(Base):

728

736

"Abstract class for Actions for setting one client property"

729

737

730

738

def run_on_one_client(self, client, properties):

745

753

raise NotImplementedError()

746

754

747

755

748

class Enable(PropertySetter):

756

class Enable(Property):

749

757

propname = "Enabled"

750

758

value_to_set = dbus.Boolean(True)

751

759

752

760

753

class Disable(PropertySetter):

761

class Disable(Property):

754

762

propname = "Enabled"

755

763

value_to_set = dbus.Boolean(False)

756

764

757

765

758

class BumpTimeout(PropertySetter):

766

class BumpTimeout(Property):

759

767

propname = "LastCheckedOK"

760

768

value_to_set = ""

761

769

762

770

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

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"

786

793

def __init__(self, value):

787

794

self.value_to_set = value

788

795

789

796

790

class SetChecker(PropertySetterValue):

797

class SetChecker(PropertyValue):

791

798

propname = "Checker"

792

799

793

800

794

class SetHost(PropertySetterValue):

801

class SetHost(PropertyValue):

795

802

propname = "Host"

796

803

797

804

798

class SetSecret(PropertySetterValue):

805

class SetSecret(PropertyValue):

799

806

propname = "Secret"

800

807

801

808

@property

809

816

value.close()

810

817

811

818

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

819

class MillisecondsPropertyValueArgument(PropertyValue):

820

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

821

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

816

822

817

823

@property

818

824

def value_to_set(self):

824

830

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

825

831

826

832

827

class SetTimeout(PropertySetterValueMilliseconds):

833

class SetTimeout(MillisecondsPropertyValueArgument):

828

834

propname = "Timeout"

829

835

830

836

831

class SetExtendedTimeout(PropertySetterValueMilliseconds):

837

class SetExtendedTimeout(MillisecondsPropertyValueArgument):

832

838

propname = "ExtendedTimeout"

833

839

834

840

835

class SetInterval(PropertySetterValueMilliseconds):

841

class SetInterval(MillisecondsPropertyValueArgument):

836

842

propname = "Interval"

837

843

838

844

839

class SetApprovalDelay(PropertySetterValueMilliseconds):

845

class SetApprovalDelay(MillisecondsPropertyValueArgument):

840

846

propname = "ApprovalDelay"

841

847

842

848

843

class SetApprovalDuration(PropertySetterValueMilliseconds):

849

class SetApprovalDuration(MillisecondsPropertyValueArgument):

844

850

propname = "ApprovalDuration"

845

851

846

852

956

962

@staticmethod

957

963

@contextlib.contextmanager

958

964

def redirect_stderr_to_devnull():

959

old_stderr = sys.stderr

960

with contextlib.closing(open(os.devnull, "w")) as null:

961

sys.stderr = null

962

try:

963

yield

964

finally:

965

sys.stderr = old_stderr

965

null = os.open(os.path.devnull, os.O_RDWR)

966

stderrcopy = os.dup(sys.stderr.fileno())

967

os.dup2(null, sys.stderr.fileno())

968

os.close(null)

969

try:

970

yield

971

finally:

972

# restore stderr

973

os.dup2(stderrcopy, sys.stderr.fileno())

974

os.close(stderrcopy)

966

975

967

976

def check_option_syntax(self, options):

968

977

check_option_syntax(self.parser, options)

981

990

options = self.parser.parse_args()

982

991

setattr(options, action, value)

983

992

options.verbose = True

984

options.client = ["client"]

993

options.client = ["foo"]

985

994

with self.assertParseError():

986

995

self.check_option_syntax(options)

987

996

1017

1026

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

1018

1027

options = self.parser.parse_args()

1019

1028

setattr(options, action, value)

1020

options.client = ["client"]

1029

options.client = ["foo"]

1021

1030

self.check_option_syntax(options)

1022

1031

1023

1032

def test_one_client_with_all_actions_except_is_enabled(self):

1026

1035

if action == "is_enabled":

1027

1036

continue

1028

1037

setattr(options, action, value)

1029

options.client = ["client"]

1038

options.client = ["foo"]

1030

1039

self.check_option_syntax(options)

1031

1040

1032

1041

def test_two_clients_with_all_actions_except_is_enabled(self):

1035

1044

if action == "is_enabled":

1036

1045

continue

1037

1046

setattr(options, action, value)

1038

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

1047

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

1039

1048

self.check_option_syntax(options)

1040

1049

1041

1050

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1044

1053

continue

1045

1054

options = self.parser.parse_args()

1046

1055

setattr(options, action, value)

1047

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

1056

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

1048

1057

self.check_option_syntax(options)

1049

1058

1050

1059

def test_is_enabled_fails_without_client(self):

1056

1065

def test_is_enabled_fails_with_two_clients(self):

1057

1066

options = self.parser.parse_args()

1058

1067

options.is_enabled = True

1059

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

1068

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

1060

1069

with self.assertParseError():

1061

1070

self.check_option_syntax(options)

1062

1071

1089

1098

self.assertTrue(mockbus.called)

1090

1099

1091

1100

def test_logs_and_exits_on_dbus_error(self):

1092

class FailingBusStub(object):

1101

class MockBusFailing(object):

1093

1102

def get_object(self, busname, dbus_path):

1094

1103

raise dbus.exceptions.DBusException("Test")

1095

1104

1096

1105

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

1097

1106

with self.assertRaises(SystemExit) as e:

1098

bus = get_mandos_dbus_object(bus=FailingBusStub())

1107

bus = get_mandos_dbus_object(bus=MockBusFailing())

1099

1108

1100

1109

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

1101

1110

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

1105

1114

1106

1115

class Test_get_managed_objects(TestCaseWithAssertLogs):

1107

1116

def test_calls_and_returns_GetManagedObjects(self):

1108

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

1109

class ObjectManagerStub(object):

1117

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

1118

class MockObjectManager(object):

1110

1119

def GetManagedObjects(self):

1111

1120

return managed_objects

1112

retval = get_managed_objects(ObjectManagerStub())

1121

retval = get_managed_objects(MockObjectManager())

1113

1122

self.assertDictEqual(managed_objects, retval)

1114

1123

1115

1124

def test_logs_and_exits_on_dbus_error(self):

1116

1125

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

1117

1126

1118

class ObjectManagerFailingStub(object):

1127

class MockObjectManagerFailing(object):

1119

1128

def GetManagedObjects(self):

1120

1129

dbus_logger.error("Test")

1121

1130

raise dbus.exceptions.DBusException("Test")

1132

1141

try:

1133

1142

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

1134

1143

with self.assertRaises(SystemExit) as e:

1135

get_managed_objects(ObjectManagerFailingStub())

1144

get_managed_objects(MockObjectManagerFailing())

1136

1145

finally:

1137

1146

dbus_logger.removeFilter(counting_handler)

1138

1147

1155

1164

add_command_line_options(self.parser)

1156

1165

1157

1166

def test_is_enabled(self):

1158

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

1167

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

1159

1168

command.IsEnabled)

1160

1169

1161

1170

def assert_command_from_args(self, args, command_cls,

1172

1181

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

1173

1182

1174

1183

def test_is_enabled_short(self):

1175

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

1184

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

1176

1185

command.IsEnabled)

1177

1186

1178

1187

def test_approve(self):

1179

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

1188

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

1180

1189

command.Approve)

1181

1190

1182

1191

def test_approve_short(self):

1183

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

1184

command.Approve)

1192

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

1185

1193

1186

1194

def test_deny(self):

1187

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

1188

command.Deny)

1195

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

1189

1196

1190

1197

def test_deny_short(self):

1191

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

1198

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

1192

1199

1193

1200

def test_remove(self):

1194

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

1201

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

1195

1202

command.Remove)

1196

1203

1197

1204

def test_deny_before_remove(self):

1198

1205

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

1199

"client"])

1206

"foo"])

1200

1207

check_option_syntax(self.parser, options)

1201

1208

commands = commands_from_options(options)

1202

1209

self.assertEqual(2, len(commands))

1213

1220

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

1214

1221

1215

1222

def test_remove_short(self):

1216

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

1217

command.Remove)

1223

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

1218

1224

1219

1225

def test_dump_json(self):

1220

1226

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

1221

1227

command.DumpJSON)

1222

1228

1223

1229

def test_enable(self):

1224

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

1230

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

1225

1231

command.Enable)

1226

1232

1227

1233

def test_enable_short(self):

1228

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

1229

command.Enable)

1234

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

1230

1235

1231

1236

def test_disable(self):

1232

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

1237

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

1233

1238

command.Disable)

1234

1239

1235

1240

def test_disable_short(self):

1236

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

1237

command.Disable)

1241

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

1238

1242

1239

1243

def test_bump_timeout(self):

1240

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

1244

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

1241

1245

command.BumpTimeout)

1242

1246

1243

1247

def test_bump_timeout_short(self):

1244

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

1248

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

1245

1249

command.BumpTimeout)

1246

1250

1247

1251

def test_start_checker(self):

1248

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

1252

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

1249

1253

command.StartChecker)

1250

1254

1251

1255

def test_stop_checker(self):

1252

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

1256

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

1253

1257

command.StopChecker)

1254

1258

1255

1259

def test_approve_by_default(self):

1256

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

1257

"client"],

1260

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

1258

1261

command.ApproveByDefault)

1259

1262

1260

1263

def test_deny_by_default(self):

1261

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

1264

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

1262

1265

command.DenyByDefault)

1263

1266

1264

1267

def test_checker(self):

1265

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

1268

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

1266

1269

command.SetChecker,

1267

1270

value_to_set=":")

1268

1271

1269

1272

def test_checker_empty(self):

1270

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

1273

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

1271

1274

command.SetChecker,

1272

1275

value_to_set="")

1273

1276

1274

1277

def test_checker_short(self):

1275

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

1278

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

1276

1279

command.SetChecker,

1277

1280

value_to_set=":")

1278

1281

1279

1282

def test_host(self):

1280

self.assert_command_from_args(

1281

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

1282

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

1283

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

1284

"foo"], command.SetHost,

1285

value_to_set="foo.example.org")

1283

1286

1284

1287

def test_host_short(self):

1285

self.assert_command_from_args(

1286

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

1287

value_to_set="client.example.org")

1288

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

1289

"foo"], command.SetHost,

1290

value_to_set="foo.example.org")

1288

1291

1289

1292

def test_secret_devnull(self):

1290

1293

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

1291

"client"], command.SetSecret,

1294

"foo"], command.SetSecret,

1292

1295

value_to_set=b"")

1293

1296

1294

1297

def test_secret_tempfile(self):

1297

1300

f.write(value)

1298

1301

f.seek(0)

1299

1302

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

1300

"client"],

1301

command.SetSecret,

1303

"foo"], command.SetSecret,

1302

1304

value_to_set=value)

1303

1305

1304

1306

def test_secret_devnull_short(self):

1305

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

1306

"client"], command.SetSecret,

1307

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

1308

command.SetSecret,

1307

1309

value_to_set=b"")

1308

1310

1309

1311

def test_secret_tempfile_short(self):

1311

1313

value = b"secret\0xyzzy\nbar"

1312

1314

f.write(value)

1313

1315

f.seek(0)

1314

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

1316

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

1315

1317

command.SetSecret,

1316

1318

value_to_set=value)

1317

1319

1318

1320

def test_timeout(self):

1319

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

1321

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

1320

1322

command.SetTimeout,

1321

1323

value_to_set=300000)

1322

1324

1323

1325

def test_timeout_short(self):

1324

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

1326

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

1325

1327

command.SetTimeout,

1326

1328

value_to_set=300000)

1327

1329

1328

1330

def test_extended_timeout(self):

1329

1331

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

1330

"client"],

1332

"foo"],

1331

1333

command.SetExtendedTimeout,

1332

1334

value_to_set=900000)

1333

1335

1334

1336

def test_interval(self):

1335

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

1336

"client"], command.SetInterval,

1337

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

1338

command.SetInterval,

1337

1339

value_to_set=120000)

1338

1340

1339

1341

def test_interval_short(self):

1340

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

1342

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

1341

1343

command.SetInterval,

1342

1344

value_to_set=120000)

1343

1345

1344

1346

def test_approval_delay(self):

1345

1347

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

1346

"client"],

1348

"foo"],

1347

1349

command.SetApprovalDelay,

1348

1350

value_to_set=30000)

1349

1351

1350

1352

def test_approval_duration(self):

1351

1353

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

1352

"client"],

1354

"foo"],

1353

1355

command.SetApprovalDuration,

1354

1356

value_to_set=1000)

1355

1357

1438

1440

LastCheckerStatus=-2)

1439

1441

self.clients = collections.OrderedDict(

1440

1442

[

1441

(self.client.__dbus_object_path__,

1442

self.client.attributes),

1443

(self.other_client.__dbus_object_path__,

1444

self.other_client.attributes),

1443

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

1444

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

1445

])

1446

self.one_client = {self.client.__dbus_object_path__:

1447

self.client.attributes}

1446

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

1448

1447

1449

1448

@property

1450

1449

def bus(self):

1451

class MockBus(object):

1450

class Bus(object):

1452

1451

@staticmethod

1453

1452

def get_object(client_bus_name, path):

1454

1453

self.assertEqual(dbus_busname, client_bus_name)

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

1454

return {

1455

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

1456

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

1457

"/clients/foo": self.client,

1458

"/clients/barbar": self.other_client,

1459

}[path]

1460

return Bus()

1462

1461

1463

1462

1464

1463

class TestBaseCommands(TestCommand):

1495

1494

client.calls)

1496

1495

1497

1496

def test_Remove(self):

1498

class MandosSpy(object):

1497

class MockMandos(object):

1499

1498

def __init__(self):

1500

1499

self.calls = []

1501

1500

def RemoveClient(self, dbus_path):

1502

1501

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

1503

mandos = MandosSpy()

1502

mandos = MockMandos()

1504

1503

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

1505

1504

for clientpath in self.clients:

1506

1505

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

1558

1557

}

1559

1558

1560

1559

def test_DumpJSON_normal(self):

1561

with self.capture_stdout_to_buffer() as buffer:

1562

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

1563

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

1560

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

1561

json_data = json.loads(output)

1564

1562

self.assertDictEqual(self.expected_json, json_data)

1565

1563

1566

@staticmethod

1567

@contextlib.contextmanager

1568

def capture_stdout_to_buffer():

1569

capture_buffer = io.StringIO()

1570

old_stdout = sys.stdout

1571

sys.stdout = capture_buffer

1572

try:

1573

yield capture_buffer

1574

finally:

1575

sys.stdout = old_stdout

1576

1577

1564

def test_DumpJSON_one_client(self):

1578

with self.capture_stdout_to_buffer() as buffer:

1579

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

1580

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

1565

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

1566

json_data = json.loads(output)

1581

1567

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

1582

1568

self.assertDictEqual(expected_json, json_data)

1583

1569

1584

1570

def test_PrintTable_normal(self):

1585

with self.capture_stdout_to_buffer() as buffer:

1586

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

1571

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

1587

1572

expected_output = "\n".join((

1588

1573

"Name Enabled Timeout Last Successful Check",

1589

1574

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

1590

1575

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

1591

)) + "\n"

1592

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

1576

))

1577

self.assertEqual(expected_output, output)

1593

1578

1594

1579

def test_PrintTable_verbose(self):

1595

with self.capture_stdout_to_buffer() as buffer:

1596

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

1580

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

1581

self.clients.values())

1597

1582

columns = (

1598

1583

(

1599

1584

"Name ",

1681

1666

)

1682

1667

)

1683

1668

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

1684

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

1685

for rows in columns)

1686

for line in range(num_lines))

1687

+ "\n")

1688

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

1669

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

1670

for rows in columns)

1671

for line in range(num_lines))

1672

self.assertEqual(expected_output, output)

1689

1673

1690

1674

def test_PrintTable_one_client(self):

1691

with self.capture_stdout_to_buffer() as buffer:

1692

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

1675

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

1693

1676

expected_output = "\n".join((

1694

1677

"Name Enabled Timeout Last Successful Check",

1695

1678

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

1696

)) + "\n"

1697

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

1698

1699

1700

class TestPropertySetterCmd(TestCommand):

1701

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

1679

))

1680

self.assertEqual(expected_output, output)

1681

1682

1683

class TestPropertyCmd(TestCommand):

1684

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

1702

1685

def runTest(self):

1703

1686

if not hasattr(self, "command"):

1704

1687

return

1725

1708

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

1726

1709

1727

1710

1728

class TestEnableCmd(TestPropertySetterCmd):

1711

class TestEnableCmd(TestPropertyCmd):

1729

1712

command = command.Enable

1730

1713

propname = "Enabled"

1731

1714

values_to_set = [dbus.Boolean(True)]

1732

1715

1733

1716

1734

class TestDisableCmd(TestPropertySetterCmd):

1717

class TestDisableCmd(TestPropertyCmd):

1735

1718

command = command.Disable

1736

1719

propname = "Enabled"

1737

1720

values_to_set = [dbus.Boolean(False)]

1738

1721

1739

1722

1740

class TestBumpTimeoutCmd(TestPropertySetterCmd):

1723

class TestBumpTimeoutCmd(TestPropertyCmd):

1741

1724

command = command.BumpTimeout

1742

1725

propname = "LastCheckedOK"

1743

1726

values_to_set = [""]

1744

1727

1745

1728

1746

class TestStartCheckerCmd(TestPropertySetterCmd):

1729

class TestStartCheckerCmd(TestPropertyCmd):

1747

1730

command = command.StartChecker

1748

1731

propname = "CheckerRunning"

1749

1732

values_to_set = [dbus.Boolean(True)]

1750

1733

1751

1734

1752

class TestStopCheckerCmd(TestPropertySetterCmd):

1735

class TestStopCheckerCmd(TestPropertyCmd):

1753

1736

command = command.StopChecker

1754

1737

propname = "CheckerRunning"

1755

1738

values_to_set = [dbus.Boolean(False)]

1756

1739

1757

1740

1758

class TestApproveByDefaultCmd(TestPropertySetterCmd):

1741

class TestApproveByDefaultCmd(TestPropertyCmd):

1759

1742

command = command.ApproveByDefault

1760

1743

propname = "ApprovedByDefault"

1761

1744

values_to_set = [dbus.Boolean(True)]

1762

1745

1763

1746

1764

class TestDenyByDefaultCmd(TestPropertySetterCmd):

1747

class TestDenyByDefaultCmd(TestPropertyCmd):

1765

1748

command = command.DenyByDefault

1766

1749

propname = "ApprovedByDefault"

1767

1750

values_to_set = [dbus.Boolean(False)]

1768

1751

1769

1752

1770

class TestPropertySetterValueCmd(TestPropertySetterCmd):

1771

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

1753

class TestPropertyValueCmd(TestPropertyCmd):

1754

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

1772

1755

1773

1756

def runTest(self):

1774

if type(self) is TestPropertySetterValueCmd:

1757

if type(self) is TestPropertyValueCmd:

1775

1758

return

1776

return super(TestPropertySetterValueCmd, self).runTest()

1759

return super(TestPropertyValueCmd, self).runTest()

1777

1760

1778

1761

def run_command(self, value, clients):

1779

1762

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

1780

1763

1781

1764

1782

class TestSetCheckerCmd(TestPropertySetterValueCmd):

1765

class TestSetCheckerCmd(TestPropertyValueCmd):

1783

1766

command = command.SetChecker

1784

1767

propname = "Checker"

1785

1768

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

1786

1769

1787

1770

1788

class TestSetHostCmd(TestPropertySetterValueCmd):

1771

class TestSetHostCmd(TestPropertyValueCmd):

1789

1772

command = command.SetHost

1790

1773

propname = "Host"

1791

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

1792

1793

1794

class TestSetSecretCmd(TestPropertySetterValueCmd):

1774

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

1775

1776

1777

class TestSetSecretCmd(TestPropertyValueCmd):

1795

1778

command = command.SetSecret

1796

1779

propname = "Secret"

1797

1780

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

1798

1781

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

1799

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

1800

1801

1802

class TestSetTimeoutCmd(TestPropertySetterValueCmd):

1782

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

1783

1784

1785

class TestSetTimeoutCmd(TestPropertyValueCmd):

1803

1786

command = command.SetTimeout

1804

1787

propname = "Timeout"

1805

1788

values_to_set = [datetime.timedelta(),

1807

1790

datetime.timedelta(seconds=1),

1808

1791

datetime.timedelta(weeks=1),

1809

1792

datetime.timedelta(weeks=52)]

1810

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

1811

1812

1813

class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):

1793

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

1794

1795

1796

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1814

1797

command = command.SetExtendedTimeout

1815

1798

propname = "ExtendedTimeout"

1816

1799

values_to_set = [datetime.timedelta(),

1818

1801

datetime.timedelta(seconds=1),

1819

1802

datetime.timedelta(weeks=1),

1820

1803

datetime.timedelta(weeks=52)]

1821

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

1822

1823

1824

class TestSetIntervalCmd(TestPropertySetterValueCmd):

1804

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

1805

1806

1807

class TestSetIntervalCmd(TestPropertyValueCmd):

1825

1808

command = command.SetInterval

1826

1809

propname = "Interval"

1827

1810

values_to_set = [datetime.timedelta(),

1829

1812

datetime.timedelta(seconds=1),

1830

1813

datetime.timedelta(weeks=1),

1831

1814

datetime.timedelta(weeks=52)]

1832

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

1833

1834

1835

class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):

1815

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

1816

1817

1818

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1836

1819

command = command.SetApprovalDelay

1837

1820

propname = "ApprovalDelay"

1838

1821

values_to_set = [datetime.timedelta(),

1840

1823

datetime.timedelta(seconds=1),

1841

1824

datetime.timedelta(weeks=1),

1842

1825

datetime.timedelta(weeks=52)]

1843

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

1844

1845

1846

class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):

1826

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

1827

1828

1829

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1847

1830

command = command.SetApprovalDuration

1848

1831

propname = "ApprovalDuration"

1849

1832

values_to_set = [datetime.timedelta(),

1851

1834

datetime.timedelta(seconds=1),

1852

1835

datetime.timedelta(weeks=1),

1853

1836

datetime.timedelta(weeks=52)]

1854

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

1837

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

1855

1838

1856

1839

1857

1840