/mandos/release : revision 237.7.631

To get this branch, use:

bzr branch
http://bzr.recompile.se/loggerhead/mandos/release

« back to all changes in this revision

Viewing changes to mandos-ctl

Committer: Teddy Hogeborn
Date: 2019-03-17 18:17:02 UTC
mto: This revision was merged to the branch mainline in revision 382.
Revision ID: teddy@recompile.se-20190317181702-tf6uvnb0gnyw9z8v

mandos-ctl: Refactor tests

* mandos-ctl (Test_check_option_syntax.redirect_stderr_to_devnull):
Override file objects instead of raw file descriptors.

files modified:
TODO

mandos-ctl

Show diffs side-by-side

added added

removed removed

mandos-ctl

locale.setlocale(locale.LC_ALL, "")

domain = "se.recompile"

busname = domain + ".Mandos"

server_path = "/"

server_interface = domain + ".Mandos"

client_interface = domain + ".Mandos.Client"

dbus_busname_domain = "se.recompile"

dbus_busname = dbus_busname_domain + ".Mandos"

server_dbus_path = "/"

server_dbus_interface = dbus_busname_domain + ".Mandos"

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

del dbus_busname_domain

version = "1.8.3"

dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"

def milliseconds_to_string(ms):

td = datetime.timedelta(0, 0, 0, ms)

return ("{days}{hours:02}:{minutes:02}:{seconds:02}"

.format(days="{}T".format(td.days) if td.days else "",

hours=td.seconds // 3600,

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

seconds=td.seconds % 60))

def main():

parser = argparse.ArgumentParser()

add_command_line_options(parser)

options = parser.parse_args()

check_option_syntax(parser, options)

clientnames = options.client

if options.debug:

log.setLevel(logging.DEBUG)

bus = dbus.SystemBus()

mandos_dbus_object = get_mandos_dbus_object(bus)

100

mandos_serv = dbus.Interface(

101

mandos_dbus_object, dbus_interface=server_dbus_interface)

102

mandos_serv_object_manager = dbus.Interface(

103

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

104

105

managed_objects = get_managed_objects(mandos_serv_object_manager)

106

107

all_clients = {}

108

for path, ifs_and_props in managed_objects.items():

109

try:

110

all_clients[path] = ifs_and_props[client_dbus_interface]

111

except KeyError:

112

pass

113

114

# Compile dict of (clientpath: properties) to process

115

if not clientnames:

116

clients = all_clients

117

else:

118

clients = {}

119

for name in clientnames:

120

for objpath, properties in all_clients.items():

121

if properties["Name"] == name:

122

clients[objpath] = properties

123

break

124

else:

125

log.critical("Client not found on server: %r", name)

126

sys.exit(1)

127

128

commands = commands_from_options(options)

129

130

for command in commands:

131

command.run(clients, bus, mandos_serv)

132

133

134

def add_command_line_options(parser):

135

parser.add_argument("--version", action="version",

136

version="%(prog)s {}".format(version),

137

help="show version number and exit")

138

parser.add_argument("-a", "--all", action="store_true",

139

help="Select all clients")

140

parser.add_argument("-v", "--verbose", action="store_true",

141

help="Print all fields")

142

parser.add_argument("-j", "--dump-json", action="store_true",

143

help="Dump client data in JSON format")

144

enable_disable = parser.add_mutually_exclusive_group()

145

enable_disable.add_argument("-e", "--enable", action="store_true",

146

help="Enable client")

147

enable_disable.add_argument("-d", "--disable",

148

action="store_true",

149

help="disable client")

150

parser.add_argument("-b", "--bump-timeout", action="store_true",

151

help="Bump timeout for client")

152

start_stop_checker = parser.add_mutually_exclusive_group()

153

start_stop_checker.add_argument("--start-checker",

154

action="store_true",

155

help="Start checker for client")

156

start_stop_checker.add_argument("--stop-checker",

157

action="store_true",

158

help="Stop checker for client")

159

parser.add_argument("-V", "--is-enabled", action="store_true",

160

help="Check if client is enabled")

161

parser.add_argument("-r", "--remove", action="store_true",

162

help="Remove client")

163

parser.add_argument("-c", "--checker",

164

help="Set checker command for client")

165

parser.add_argument("-t", "--timeout", type=string_to_delta,

166

help="Set timeout for client")

167

parser.add_argument("--extended-timeout", type=string_to_delta,

168

help="Set extended timeout for client")

169

parser.add_argument("-i", "--interval", type=string_to_delta,

170

help="Set checker interval for client")

171

approve_deny_default = parser.add_mutually_exclusive_group()

172

approve_deny_default.add_argument(

173

"--approve-by-default", action="store_true",

174

default=None, dest="approved_by_default",

175

help="Set client to be approved by default")

176

approve_deny_default.add_argument(

177

"--deny-by-default", action="store_false",

178

dest="approved_by_default",

179

help="Set client to be denied by default")

180

parser.add_argument("--approval-delay", type=string_to_delta,

181

help="Set delay before client approve/deny")

182

parser.add_argument("--approval-duration", type=string_to_delta,

183

help="Set duration of one client approval")

184

parser.add_argument("-H", "--host", help="Set host for client")

185

parser.add_argument("-s", "--secret",

186

type=argparse.FileType(mode="rb"),

187

help="Set password blob (file) for client")

188

approve_deny = parser.add_mutually_exclusive_group()

189

approve_deny.add_argument(

190

"-A", "--approve", action="store_true",

191

help="Approve any current client request")

192

approve_deny.add_argument("-D", "--deny", action="store_true",

193

help="Deny any current client request")

194

parser.add_argument("--debug", action="store_true",

195

help="Debug mode (show D-Bus commands)")

196

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

197

help="Run self-test")

198

parser.add_argument("client", nargs="*", help="Client name")

199

200

201

def string_to_delta(interval):

202

"""Parse a string and return a datetime.timedelta"""

203

204

try:

205

return rfc3339_duration_to_delta(interval)

206

except ValueError as e:

207

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

208

' '.join(e.args))

209

return parse_pre_1_6_1_interval(interval)

210

211

212

def rfc3339_duration_to_delta(duration):

110

232

>>> rfc3339_duration_to_delta("")

111

233

Traceback (most recent call last):

112

234

...

113

ValueError: Invalid RFC 3339 duration: u''

235

ValueError: Invalid RFC 3339 duration: ""

114

236

>>> # Must start with "P":

115

237

>>> rfc3339_duration_to_delta("1D")

116

238

Traceback (most recent call last):

117

239

...

118

ValueError: Invalid RFC 3339 duration: u'1D'

240

ValueError: Invalid RFC 3339 duration: "1D"

119

241

>>> # Must use correct order

120

242

>>> rfc3339_duration_to_delta("PT1S2M")

121

243

Traceback (most recent call last):

122

244

...

123

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

245

ValueError: Invalid RFC 3339 duration: "PT1S2M"

124

246

>>> # Time needs time marker

125

247

>>> rfc3339_duration_to_delta("P1H2S")

126

248

Traceback (most recent call last):

127

249

...

128

ValueError: Invalid RFC 3339 duration: u'P1H2S'

250

ValueError: Invalid RFC 3339 duration: "P1H2S"

129

251

>>> # Weeks can not be combined with anything else

130

252

>>> rfc3339_duration_to_delta("P1D2W")

131

253

Traceback (most recent call last):

132

254

...

133

ValueError: Invalid RFC 3339 duration: u'P1D2W'

255

ValueError: Invalid RFC 3339 duration: "P1D2W"

134

256

>>> rfc3339_duration_to_delta("P2W2H")

135

257

Traceback (most recent call last):

136

258

...

137

ValueError: Invalid RFC 3339 duration: u'P2W2H'

259

ValueError: Invalid RFC 3339 duration: "P2W2H"

138

260

"""

139

261

140

262

# Parsing an RFC 3339 duration with regular expressions is not

211

333

break

212

334

else:

213

335

# No currently valid tokens were found

214

raise ValueError("Invalid RFC 3339 duration: {!r}"

336

raise ValueError("Invalid RFC 3339 duration: \"{}\""

215

337

.format(duration))

216

338

# End token found

217

339

return value

218

340

219

341

220

def string_to_delta(interval):

221

"""Parse a string and return a datetime.timedelta"""

222

223

try:

224

return rfc3339_duration_to_delta(interval)

225

except ValueError as e:

226

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

227

' '.join(e.args))

228

return parse_pre_1_6_1_interval(interval)

229

230

231

342

def parse_pre_1_6_1_interval(interval):

232

343

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

233

344

and return a datetime.timedelta

271

382

return value

272

383

273

384

274

## Classes for commands.

275

276

# Abstract classes first

277

class Command(object):

278

"""Abstract class for commands"""

279

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

280

"""Normal commands should implement run_on_one_client(), but

281

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

282

can override this run() method instead."""

283

self.mandos = mandos

284

for client, properties in clients.items():

285

self.run_on_one_client(client, properties)

286

287

class PrintCmd(Command):

288

"""Abstract class for commands printing client details"""

289

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

290

"Created", "Interval", "Host", "KeyID",

291

"Fingerprint", "CheckerRunning", "LastEnabled",

292

"ApprovalPending", "ApprovedByDefault",

293

"LastApprovalRequest", "ApprovalDelay",

294

"ApprovalDuration", "Checker", "ExtendedTimeout",

295

"Expires", "LastCheckerStatus")

296

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

297

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

298

299

class PropertyCmd(Command):

300

"""Abstract class for Actions for setting one client property"""

301

def run_on_one_client(self, client, properties):

302

"""Set the Client's D-Bus property"""

303

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

304

client.__dbus_object_path__,

305

dbus.PROPERTIES_IFACE, client_interface,

306

self.property, self.value_to_set

307

if not isinstance(self.value_to_set, dbus.Boolean)

308

else bool(self.value_to_set))

309

client.Set(client_interface, self.property, self.value_to_set,

310

dbus_interface=dbus.PROPERTIES_IFACE)

311

312

class ValueArgumentMixIn(object):

313

"""Mixin class for commands taking a value as argument"""

314

def __init__(self, value):

315

self.value_to_set = value

316

317

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

318

"""Mixin class for commands taking a value argument as

319

milliseconds."""

320

@property

321

def value_to_set(self):

322

return self._vts

323

@value_to_set.setter

324

def value_to_set(self, value):

325

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

326

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

327

328

# Actual (non-abstract) command classes

329

330

class PrintTableCmd(PrintCmd):

331

def __init__(self, verbose=False):

332

self.verbose = verbose

333

334

def output(self, mandos_objmgr, clients):

335

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

336

keywords = default_keywords

337

if self.verbose:

338

keywords = self.all_keywords

339

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

340

341

class TableOfClients(object):

342

tableheaders = {

343

"Name": "Name",

344

"Enabled": "Enabled",

345

"Timeout": "Timeout",

346

"LastCheckedOK": "Last Successful Check",

347

"LastApprovalRequest": "Last Approval Request",

348

"Created": "Created",

349

"Interval": "Interval",

350

"Host": "Host",

351

"Fingerprint": "Fingerprint",

352

"KeyID": "Key ID",

353

"CheckerRunning": "Check Is Running",

354

"LastEnabled": "Last Enabled",

355

"ApprovalPending": "Approval Is Pending",

356

"ApprovedByDefault": "Approved By Default",

357

"ApprovalDelay": "Approval Delay",

358

"ApprovalDuration": "Approval Duration",

359

"Checker": "Checker",

360

"ExtendedTimeout": "Extended Timeout",

361

"Expires": "Expires",

362

"LastCheckerStatus": "Last Checker Status",

363

}

364

365

def __init__(self, clients, keywords, tableheaders=None):

366

self.clients = clients

367

self.keywords = keywords

368

if tableheaders is not None:

369

self.tableheaders = tableheaders

370

371

def __str__(self):

372

return "\n".join(self.rows())

373

374

if sys.version_info.major == 2:

375

__unicode__ = __str__

376

def __str__(self):

377

return str(self).encode(locale.getpreferredencoding())

378

379

def rows(self):

380

format_string = self.row_formatting_string()

381

rows = [self.header_line(format_string)]

382

rows.extend(self.client_line(client, format_string)

383

for client in self.clients)

384

return rows

385

386

def row_formatting_string(self):

387

"Format string used to format table rows"

388

return " ".join("{{{key}:{width}}}".format(

389

width=max(len(self.tableheaders[key]),

390

*(len(self.string_from_client(client, key))

391

for client in self.clients)),

392

key=key)

393

for key in self.keywords)

394

395

def string_from_client(self, client, key):

396

return self.valuetostring(client[key], key)

397

398

@staticmethod

399

def valuetostring(value, keyword):

400

if isinstance(value, dbus.Boolean):

401

return "Yes" if value else "No"

402

if keyword in ("Timeout", "Interval", "ApprovalDelay",

403

"ApprovalDuration", "ExtendedTimeout"):

404

return milliseconds_to_string(value)

405

return str(value)

406

407

def header_line(self, format_string):

408

return format_string.format(**self.tableheaders)

409

410

def client_line(self, client, format_string):

411

return format_string.format(

412

**{key: self.string_from_client(client, key)

413

for key in self.keywords})

414

415

416

417

class DumpJSONCmd(PrintCmd):

418

def output(self, mandos_objmgr, clients):

419

data = {client["Name"]:

420

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

421

for key in self.all_keywords}

422

for client in clients.values()}

423

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

424

@staticmethod

425

def dbus_boolean_to_bool(value):

426

if isinstance(value, dbus.Boolean):

427

value = bool(value)

428

return value

429

430

class IsEnabledCmd(Command):

431

def run_on_one_client(self, client, properties):

432

if self.is_enabled(client, properties):

433

sys.exit(0)

434

sys.exit(1)

435

def is_enabled(self, client, properties):

436

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

437

client.__dbus_object_path__,

438

dbus.PROPERTIES_IFACE, client_interface,

439

"Enabled")

440

return bool(client.Get(client_interface, "Enabled",

441

dbus_interface=dbus.PROPERTIES_IFACE))

442

443

class RemoveCmd(Command):

444

def run_on_one_client(self, client, properties):

445

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

446

server_path, server_interface,

447

str(client.__dbus_object_path__))

448

self.mandos.RemoveClient(client.__dbus_object_path__)

449

450

class ApproveCmd(Command):

451

def run_on_one_client(self, client, properties):

452

log.debug("D-Bus: %s:%s.Approve(True)",

453

client.__dbus_object_path__, client_interface)

454

client.Approve(dbus.Boolean(True),

455

dbus_interface=client_interface)

456

457

class DenyCmd(Command):

458

def run_on_one_client(self, client, properties):

459

log.debug("D-Bus: %s:%s.Approve(False)",

460

client.__dbus_object_path__, client_interface)

461

client.Approve(dbus.Boolean(False),

462

dbus_interface=client_interface)

463

464

class EnableCmd(PropertyCmd):

465

property = "Enabled"

466

value_to_set = dbus.Boolean(True)

467

468

class DisableCmd(PropertyCmd):

469

property = "Enabled"

470

value_to_set = dbus.Boolean(False)

471

472

class BumpTimeoutCmd(PropertyCmd):

473

property = "LastCheckedOK"

474

value_to_set = ""

475

476

class StartCheckerCmd(PropertyCmd):

477

property = "CheckerRunning"

478

value_to_set = dbus.Boolean(True)

479

480

class StopCheckerCmd(PropertyCmd):

481

property = "CheckerRunning"

482

value_to_set = dbus.Boolean(False)

483

484

class ApproveByDefaultCmd(PropertyCmd):

485

property = "ApprovedByDefault"

486

value_to_set = dbus.Boolean(True)

487

488

class DenyByDefaultCmd(PropertyCmd):

489

property = "ApprovedByDefault"

490

value_to_set = dbus.Boolean(False)

491

492

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

493

property = "Checker"

494

495

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

496

property = "Host"

497

498

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

499

@property

500

def value_to_set(self):

501

return self._vts

502

@value_to_set.setter

503

def value_to_set(self, value):

504

"""When setting, read data from supplied file object"""

505

self._vts = value.read()

506

value.close()

507

property = "Secret"

508

509

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

510

property = "Timeout"

511

512

class SetExtendedTimeoutCmd(PropertyCmd,

513

MillisecondsValueArgumentMixIn):

514

property = "ExtendedTimeout"

515

516

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

517

property = "Interval"

518

519

class SetApprovalDelayCmd(PropertyCmd,

520

MillisecondsValueArgumentMixIn):

521

property = "ApprovalDelay"

522

523

class SetApprovalDurationCmd(PropertyCmd,

524

MillisecondsValueArgumentMixIn):

525

property = "ApprovalDuration"

526

527

def add_command_line_options(parser):

528

parser.add_argument("--version", action="version",

529

version="%(prog)s {}".format(version),

530

help="show version number and exit")

531

parser.add_argument("-a", "--all", action="store_true",

532

help="Select all clients")

533

parser.add_argument("-v", "--verbose", action="store_true",

534

help="Print all fields")

535

parser.add_argument("-j", "--dump-json", action="store_true",

536

help="Dump client data in JSON format")

537

enable_disable = parser.add_mutually_exclusive_group()

538

enable_disable.add_argument("-e", "--enable", action="store_true",

539

help="Enable client")

540

enable_disable.add_argument("-d", "--disable",

541

action="store_true",

542

help="disable client")

543

parser.add_argument("-b", "--bump-timeout", action="store_true",

544

help="Bump timeout for client")

545

start_stop_checker = parser.add_mutually_exclusive_group()

546

start_stop_checker.add_argument("--start-checker",

547

action="store_true",

548

help="Start checker for client")

549

start_stop_checker.add_argument("--stop-checker",

550

action="store_true",

551

help="Stop checker for client")

552

parser.add_argument("-V", "--is-enabled", action="store_true",

553

help="Check if client is enabled")

554

parser.add_argument("-r", "--remove", action="store_true",

555

help="Remove client")

556

parser.add_argument("-c", "--checker",

557

help="Set checker command for client")

558

parser.add_argument("-t", "--timeout", type=string_to_delta,

559

help="Set timeout for client")

560

parser.add_argument("--extended-timeout", type=string_to_delta,

561

help="Set extended timeout for client")

562

parser.add_argument("-i", "--interval", type=string_to_delta,

563

help="Set checker interval for client")

564

approve_deny_default = parser.add_mutually_exclusive_group()

565

approve_deny_default.add_argument(

566

"--approve-by-default", action="store_true",

567

default=None, dest="approved_by_default",

568

help="Set client to be approved by default")

569

approve_deny_default.add_argument(

570

"--deny-by-default", action="store_false",

571

dest="approved_by_default",

572

help="Set client to be denied by default")

573

parser.add_argument("--approval-delay", type=string_to_delta,

574

help="Set delay before client approve/deny")

575

parser.add_argument("--approval-duration", type=string_to_delta,

576

help="Set duration of one client approval")

577

parser.add_argument("-H", "--host", help="Set host for client")

578

parser.add_argument("-s", "--secret",

579

type=argparse.FileType(mode="rb"),

580

help="Set password blob (file) for client")

581

approve_deny = parser.add_mutually_exclusive_group()

582

approve_deny.add_argument(

583

"-A", "--approve", action="store_true",

584

help="Approve any current client request")

585

approve_deny.add_argument("-D", "--deny", action="store_true",

586

help="Deny any current client request")

587

parser.add_argument("--debug", action="store_true",

588

help="Debug mode (show D-Bus commands)")

589

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

590

help="Run self-test")

591

parser.add_argument("client", nargs="*", help="Client name")

592

593

594

def commands_from_options(options):

595

596

commands = []

597

598

if options.dump_json:

599

commands.append(DumpJSONCmd())

600

601

if options.enable:

602

commands.append(EnableCmd())

603

604

if options.disable:

605

commands.append(DisableCmd())

606

607

if options.bump_timeout:

608

commands.append(BumpTimeoutCmd())

609

610

if options.start_checker:

611

commands.append(StartCheckerCmd())

612

613

if options.stop_checker:

614

commands.append(StopCheckerCmd())

615

616

if options.is_enabled:

617

commands.append(IsEnabledCmd())

618

619

if options.checker is not None:

620

commands.append(SetCheckerCmd(options.checker))

621

622

if options.timeout is not None:

623

commands.append(SetTimeoutCmd(options.timeout))

624

625

if options.extended_timeout:

626

commands.append(

627

SetExtendedTimeoutCmd(options.extended_timeout))

628

629

if options.interval is not None:

630

commands.append(SetIntervalCmd(options.interval))

631

632

if options.approved_by_default is not None:

633

if options.approved_by_default:

634

commands.append(ApproveByDefaultCmd())

635

else:

636

commands.append(DenyByDefaultCmd())

637

638

if options.approval_delay is not None:

639

commands.append(SetApprovalDelayCmd(options.approval_delay))

640

641

if options.approval_duration is not None:

642

commands.append(

643

SetApprovalDurationCmd(options.approval_duration))

644

645

if options.host is not None:

646

commands.append(SetHostCmd(options.host))

647

648

if options.secret is not None:

649

commands.append(SetSecretCmd(options.secret))

650

651

if options.approve:

652

commands.append(ApproveCmd())

653

654

if options.deny:

655

commands.append(DenyCmd())

656

657

if options.remove:

658

commands.append(RemoveCmd())

659

660

# If no command option has been given, show table of clients,

661

# optionally verbosely

662

if not commands:

663

commands.append(PrintTableCmd(verbose=options.verbose))

664

665

return commands

666

667

668

385

def check_option_syntax(parser, options):

669

386

"""Apply additional restrictions on options, not expressible in

670

387

argparse"""

707

424

options.remove = True

708

425

709

426

710

def main():

711

parser = argparse.ArgumentParser()

712

713

add_command_line_options(parser)

714

715

options = parser.parse_args()

716

717

check_option_syntax(parser, options)

718

719

clientnames = options.client

720

721

if options.debug:

722

log.setLevel(logging.DEBUG)

723

427

def get_mandos_dbus_object(bus):

428

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

429

dbus_busname, server_dbus_path)

430

with if_dbus_exception_log_with_exception_and_exit(

431

"Could not connect to Mandos server: %s"):

432

mandos_dbus_object = bus.get_object(dbus_busname,

433

server_dbus_path)

434

return mandos_dbus_object

435

436

437

@contextlib.contextmanager

438

def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):

724

439

try:

725

bus = dbus.SystemBus()

726

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

727

server_path)

728

mandos_dbus_objc = bus.get_object(busname, server_path)

729

except dbus.exceptions.DBusException:

730

log.critical("Could not connect to Mandos server")

440

yield

441

except dbus.exceptions.DBusException as e:

442

log.critical(*(args + (e,)), **kwargs)

731

443

sys.exit(1)

732

444

733

mandos_serv = dbus.Interface(mandos_dbus_objc,

734

dbus_interface=server_interface)

735

mandos_serv_object_manager = dbus.Interface(

736

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

737

738

# Filter out log message from dbus module

739

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

445

446

def get_managed_objects(object_manager):

447

log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,

448

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

449

with if_dbus_exception_log_with_exception_and_exit(

450

"Failed to access Mandos server through D-Bus:\n%s"):

451

with SilenceLogger("dbus.proxies"):

452

managed_objects = object_manager.GetManagedObjects()

453

return managed_objects

454

455

456

class SilenceLogger(object):

457

"Simple context manager to silence a particular logger"

458

def __init__(self, loggername):

459

self.logger = logging.getLogger(loggername)

460

461

def __enter__(self):

462

self.logger.addFilter(self.nullfilter)

463

return self

464

740

465

class NullFilter(logging.Filter):

741

466

def filter(self, record):

742

467

return False

743

dbus_filter = NullFilter()

744

try:

745

dbus_logger.addFilter(dbus_filter)

746

log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", busname,

747

server_path, dbus.OBJECT_MANAGER_IFACE)

748

mandos_clients = {path: ifs_and_props[client_interface]

749

for path, ifs_and_props in

750

mandos_serv_object_manager

751

.GetManagedObjects().items()

752

if client_interface in ifs_and_props}

753

except dbus.exceptions.DBusException as e:

754

log.critical("Failed to access Mandos server through D-Bus:"

755

"\n%s", e)

756

sys.exit(1)

757

finally:

758

# restore dbus logger

759

dbus_logger.removeFilter(dbus_filter)

760

761

# Compile dict of (clients: properties) to process

762

clients = {}

763

764

if not clientnames:

765

clients = {bus.get_object(busname, path): properties

766

for path, properties in mandos_clients.items()}

767

else:

768

for name in clientnames:

769

for path, client in mandos_clients.items():

770

if client["Name"] == name:

771

client_objc = bus.get_object(busname, path)

772

clients[client_objc] = client

773

break

774

else:

775

log.critical("Client not found on server: %r", name)

776

sys.exit(1)

777

778

# Run all commands on clients

779

commands = commands_from_options(options)

780

for command in commands:

781

command.run(mandos_serv, mandos_serv_object_manager, clients)

468

469

nullfilter = NullFilter()

470

471

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

472

self.logger.removeFilter(self.nullfilter)

473

474

475

def commands_from_options(options):

476

477

commands = []

478

479

if options.is_enabled:

480

commands.append(command.IsEnabled())

481

482

if options.approve:

483

commands.append(command.Approve())

484

485

if options.deny:

486

commands.append(command.Deny())

487

488

if options.remove:

489

commands.append(command.Remove())

490

491

if options.dump_json:

492

commands.append(command.DumpJSON())

493

494

if options.enable:

495

commands.append(command.Enable())

496

497

if options.disable:

498

commands.append(command.Disable())

499

500

if options.bump_timeout:

501

commands.append(command.BumpTimeout())

502

503

if options.start_checker:

504

commands.append(command.StartChecker())

505

506

if options.stop_checker:

507

commands.append(command.StopChecker())

508

509

if options.approved_by_default is not None:

510

if options.approved_by_default:

511

commands.append(command.ApproveByDefault())

512

else:

513

commands.append(command.DenyByDefault())

514

515

if options.checker is not None:

516

commands.append(command.SetChecker(options.checker))

517

518

if options.host is not None:

519

commands.append(command.SetHost(options.host))

520

521

if options.secret is not None:

522

commands.append(command.SetSecret(options.secret))

523

524

if options.timeout is not None:

525

commands.append(command.SetTimeout(options.timeout))

526

527

if options.extended_timeout:

528

commands.append(

529

command.SetExtendedTimeout(options.extended_timeout))

530

531

if options.interval is not None:

532

commands.append(command.SetInterval(options.interval))

533

534

if options.approval_delay is not None:

535

commands.append(

536

command.SetApprovalDelay(options.approval_delay))

537

538

if options.approval_duration is not None:

539

commands.append(

540

command.SetApprovalDuration(options.approval_duration))

541

542

# If no command option has been given, show table of clients,

543

# optionally verbosely

544

if not commands:

545

commands.append(command.PrintTable(verbose=options.verbose))

546

547

return commands

548

549

550

class command(object):

551

"""A namespace for command classes"""

552

553

class Base(object):

554

"""Abstract base class for commands"""

555

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

556

"""Normal commands should implement run_on_one_client(),

557

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

558

override this run() method instead.

559

"""

560

self.mandos = mandos

561

for clientpath, properties in clients.items():

562

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

563

dbus_busname, str(clientpath))

564

client = bus.get_object(dbus_busname, clientpath)

565

self.run_on_one_client(client, properties)

566

567

568

class IsEnabled(Base):

569

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

570

client, properties = next(iter(clients.items()))

571

if self.is_enabled(client, properties):

572

sys.exit(0)

573

sys.exit(1)

574

def is_enabled(self, client, properties):

575

return properties["Enabled"]

576

577

578

class Approve(Base):

579

def run_on_one_client(self, client, properties):

580

log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,

581

client.__dbus_object_path__,

582

client_dbus_interface)

583

client.Approve(dbus.Boolean(True),

584

dbus_interface=client_dbus_interface)

585

586

587

class Deny(Base):

588

def run_on_one_client(self, client, properties):

589

log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,

590

client.__dbus_object_path__,

591

client_dbus_interface)

592

client.Approve(dbus.Boolean(False),

593

dbus_interface=client_dbus_interface)

594

595

596

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

603

604

605

class Output(Base):

606

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

607

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

608

"Created", "Interval", "Host", "KeyID",

609

"Fingerprint", "CheckerRunning",

610

"LastEnabled", "ApprovalPending",

611

"ApprovedByDefault", "LastApprovalRequest",

612

"ApprovalDelay", "ApprovalDuration",

613

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

614

"LastCheckerStatus")

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

622

623

class DumpJSON(Output):

624

def output(self, clients):

625

data = {client["Name"]:

626

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

627

for key in self.all_keywords}

628

for client in clients}

629

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

630

631

@staticmethod

632

def dbus_boolean_to_bool(value):

633

if isinstance(value, dbus.Boolean):

634

value = bool(value)

635

return value

636

637

638

class PrintTable(Output):

639

def __init__(self, verbose=False):

640

self.verbose = verbose

641

642

def output(self, clients):

643

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

644

"LastCheckedOK")

645

keywords = default_keywords

646

if self.verbose:

647

keywords = self.all_keywords

648

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

649

650

class TableOfClients(object):

651

tableheaders = {

652

"Name": "Name",

653

"Enabled": "Enabled",

654

"Timeout": "Timeout",

655

"LastCheckedOK": "Last Successful Check",

656

"LastApprovalRequest": "Last Approval Request",

657

"Created": "Created",

658

"Interval": "Interval",

659

"Host": "Host",

660

"Fingerprint": "Fingerprint",

661

"KeyID": "Key ID",

662

"CheckerRunning": "Check Is Running",

663

"LastEnabled": "Last Enabled",

664

"ApprovalPending": "Approval Is Pending",

665

"ApprovedByDefault": "Approved By Default",

666

"ApprovalDelay": "Approval Delay",

667

"ApprovalDuration": "Approval Duration",

668

"Checker": "Checker",

669

"ExtendedTimeout": "Extended Timeout",

670

"Expires": "Expires",

671

"LastCheckerStatus": "Last Checker Status",

672

}

673

674

def __init__(self, clients, keywords):

675

self.clients = clients

676

self.keywords = keywords

677

678

def __str__(self):

679

return "\n".join(self.rows())

680

681

if sys.version_info.major == 2:

682

__unicode__ = __str__

683

def __str__(self):

684

return str(self).encode(

685

locale.getpreferredencoding())

686

687

def rows(self):

688

format_string = self.row_formatting_string()

689

rows = [self.header_line(format_string)]

690

rows.extend(self.client_line(client, format_string)

691

for client in self.clients)

692

return rows

693

694

def row_formatting_string(self):

695

"Format string used to format table rows"

696

return " ".join("{{{key}:{width}}}".format(

697

width=max(len(self.tableheaders[key]),

698

*(len(self.string_from_client(client,

699

key))

700

for client in self.clients)),

701

key=key)

702

for key in self.keywords)

703

704

def string_from_client(self, client, key):

705

return self.valuetostring(client[key], key)

706

707

@classmethod

708

def valuetostring(cls, value, keyword):

709

if isinstance(value, dbus.Boolean):

710

return "Yes" if value else "No"

711

if keyword in ("Timeout", "Interval", "ApprovalDelay",

712

"ApprovalDuration", "ExtendedTimeout"):

713

return cls.milliseconds_to_string(value)

714

return str(value)

715

716

def header_line(self, format_string):

717

return format_string.format(**self.tableheaders)

718

719

def client_line(self, client, format_string):

720

return format_string.format(

721

**{key: self.string_from_client(client, key)

722

for key in self.keywords})

723

724

@staticmethod

725

def milliseconds_to_string(ms):

726

td = datetime.timedelta(0, 0, 0, ms)

727

return ("{days}{hours:02}:{minutes:02}:{seconds:02}"

728

.format(days="{}T".format(td.days)

729

if td.days else "",

730

hours=td.seconds // 3600,

731

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

732

seconds=td.seconds % 60))

733

734

735

class Property(Base):

736

"Abstract class for Actions for setting one client property"

737

738

def run_on_one_client(self, client, properties):

739

"""Set the Client's D-Bus property"""

740

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

741

client.__dbus_object_path__,

742

dbus.PROPERTIES_IFACE, client_dbus_interface,

743

self.propname, self.value_to_set

744

if not isinstance(self.value_to_set,

745

dbus.Boolean)

746

else bool(self.value_to_set))

747

client.Set(client_dbus_interface, self.propname,

748

self.value_to_set,

749

dbus_interface=dbus.PROPERTIES_IFACE)

750

751

@property

752

def propname(self):

753

raise NotImplementedError()

754

755

756

class Enable(Property):

757

propname = "Enabled"

758

value_to_set = dbus.Boolean(True)

759

760

761

class Disable(Property):

762

propname = "Enabled"

763

value_to_set = dbus.Boolean(False)

764

765

766

class BumpTimeout(Property):

767

propname = "LastCheckedOK"

768

value_to_set = ""

769

770

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"

793

def __init__(self, value):

794

self.value_to_set = value

795

796

797

class SetChecker(PropertyValue):

798

propname = "Checker"

799

800

801

class SetHost(PropertyValue):

802

propname = "Host"

803

804

805

class SetSecret(PropertyValue):

806

propname = "Secret"

807

808

@property

809

def value_to_set(self):

810

return self._vts

811

812

@value_to_set.setter

813

def value_to_set(self, value):

814

"""When setting, read data from supplied file object"""

815

self._vts = value.read()

816

value.close()

817

818

819

class MillisecondsPropertyValueArgument(PropertyValue):

820

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

821

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

822

823

@property

824

def value_to_set(self):

825

return self._vts

826

827

@value_to_set.setter

828

def value_to_set(self, value):

829

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

830

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

831

832

833

class SetTimeout(MillisecondsPropertyValueArgument):

834

propname = "Timeout"

835

836

837

class SetExtendedTimeout(MillisecondsPropertyValueArgument):

838

propname = "ExtendedTimeout"

839

840

841

class SetInterval(MillisecondsPropertyValueArgument):

842

propname = "Interval"

843

844

845

class SetApprovalDelay(MillisecondsPropertyValueArgument):

846

propname = "ApprovalDelay"

847

848

849

class SetApprovalDuration(MillisecondsPropertyValueArgument):

850

propname = "ApprovalDuration"

851

782

852

783

853

784

class Test_milliseconds_to_string(unittest.TestCase):

785

def test_all(self):

786

self.assertEqual(milliseconds_to_string(93785000),

787

"1T02:03:05")

788

def test_no_days(self):

789

self.assertEqual(milliseconds_to_string(7385000), "02:03:05")

790

def test_all_zero(self):

791

self.assertEqual(milliseconds_to_string(0), "00:00:00")

792

def test_no_fractional_seconds(self):

793

self.assertEqual(milliseconds_to_string(400), "00:00:00")

794

self.assertEqual(milliseconds_to_string(900), "00:00:00")

795

self.assertEqual(milliseconds_to_string(1900), "00:00:01")

796

797

class Test_string_to_delta(unittest.TestCase):

798

def test_handles_basic_rfc3339(self):

799

self.assertEqual(string_to_delta("PT0S"),

800

datetime.timedelta())

801

self.assertEqual(string_to_delta("P0D"),

802

datetime.timedelta())

803

self.assertEqual(string_to_delta("PT1S"),

804

datetime.timedelta(0, 1))

805

self.assertEqual(string_to_delta("PT2H"),

806

datetime.timedelta(0, 7200))

854

class TestCaseWithAssertLogs(unittest.TestCase):

855

"""unittest.TestCase.assertLogs only exists in Python 3.4"""

856

857

if not hasattr(unittest.TestCase, "assertLogs"):

858

@contextlib.contextmanager

859

def assertLogs(self, logger, level=logging.INFO):

860

capturing_handler = self.CapturingLevelHandler(level)

861

old_level = logger.level

862

old_propagate = logger.propagate

863

logger.addHandler(capturing_handler)

864

logger.setLevel(level)

865

logger.propagate = False

866

try:

867

yield capturing_handler.watcher

868

finally:

869

logger.propagate = old_propagate

870

logger.removeHandler(capturing_handler)

871

logger.setLevel(old_level)

872

self.assertGreater(len(capturing_handler.watcher.records),

873

874

875

class CapturingLevelHandler(logging.Handler):

876

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

877

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

878

self.watcher = self.LoggingWatcher([], [])

879

def emit(self, record):

880

self.watcher.records.append(record)

881

self.watcher.output.append(self.format(record))

882

883

LoggingWatcher = collections.namedtuple("LoggingWatcher",

884

("records",

885

"output"))

886

887

888

class Test_string_to_delta(TestCaseWithAssertLogs):

889

# Just test basic RFC 3339 functionality here, the doc string for

890

# rfc3339_duration_to_delta() already has more comprehensive

891

# tests, which is run by doctest.

892

893

def test_rfc3339_zero_seconds(self):

894

self.assertEqual(datetime.timedelta(),

895

string_to_delta("PT0S"))

896

897

def test_rfc3339_zero_days(self):

898

self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))

899

900

def test_rfc3339_one_second(self):

901

self.assertEqual(datetime.timedelta(0, 1),

902

string_to_delta("PT1S"))

903

904

def test_rfc3339_two_hours(self):

905

self.assertEqual(datetime.timedelta(0, 7200),

906

string_to_delta("PT2H"))

907

807

908

def test_falls_back_to_pre_1_6_1_with_warning(self):

808

# assertLogs only exists in Python 3.4

809

if hasattr(self, "assertLogs"):

810

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

811

value = string_to_delta("2h")

812

else:

813

class WarningFilter(logging.Filter):

814

"""Don't show, but record the presence of, warnings"""

815

def filter(self, record):

816

is_warning = record.levelno >= logging.WARNING

817

self.found = is_warning or getattr(self, "found",

818

False)

819

return not is_warning

820

warning_filter = WarningFilter()

821

log.addFilter(warning_filter)

909

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

910

value = string_to_delta("2h")

911

self.assertEqual(datetime.timedelta(0, 7200), value)

912

913

914

class Test_check_option_syntax(unittest.TestCase):

915

def setUp(self):

916

self.parser = argparse.ArgumentParser()

917

add_command_line_options(self.parser)

918

919

def test_actions_requires_client_or_all(self):

920

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

921

options = self.parser.parse_args()

922

setattr(options, action, value)

923

with self.assertParseError():

924

self.check_option_syntax(options)

925

926

# This mostly corresponds to the definition from has_actions() in

927

# check_option_syntax()

928

actions = {

929

# The actual values set here are not that important, but we do

930

# at least stick to the correct types, even though they are

931

# never used

932

"enable": True,

933

"disable": True,

934

"bump_timeout": True,

935

"start_checker": True,

936

"stop_checker": True,

937

"is_enabled": True,

938

"remove": True,

939

"checker": "x",

940

"timeout": datetime.timedelta(),

941

"extended_timeout": datetime.timedelta(),

942

"interval": datetime.timedelta(),

943

"approved_by_default": True,

944

"approval_delay": datetime.timedelta(),

945

"approval_duration": datetime.timedelta(),

946

"host": "x",

947

"secret": io.BytesIO(b"x"),

948

"approve": True,

949

"deny": True,

950

}

951

952

@contextlib.contextmanager

953

def assertParseError(self):

954

with self.assertRaises(SystemExit) as e:

955

with self.redirect_stderr_to_devnull():

956

yield

957

# Exit code from argparse is guaranteed to be "2". Reference:

958

# https://docs.python.org/3/library

959

# /argparse.html#exiting-methods

960

self.assertEqual(2, e.exception.code)

961

962

@staticmethod

963

@contextlib.contextmanager

964

def redirect_stderr_to_devnull():

965

old_stderr = sys.stderr

966

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

967

sys.stderr = null

822

968

try:

823

value = string_to_delta("2h")

969

yield

824

970

finally:

825

log.removeFilter(warning_filter)

826

self.assertTrue(getattr(warning_filter, "found", False))

827

self.assertEqual(value, datetime.timedelta(0, 7200))

828

829

830

class TestCmd(unittest.TestCase):

971

sys.stderr = old_stderr

972

973

def check_option_syntax(self, options):

974

check_option_syntax(self.parser, options)

975

976

def test_actions_all_conflicts_with_verbose(self):

977

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

978

options = self.parser.parse_args()

979

setattr(options, action, value)

980

options.all = True

981

options.verbose = True

982

with self.assertParseError():

983

self.check_option_syntax(options)

984

985

def test_actions_with_client_conflicts_with_verbose(self):

986

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

987

options = self.parser.parse_args()

988

setattr(options, action, value)

989

options.verbose = True

990

options.client = ["foo"]

991

with self.assertParseError():

992

self.check_option_syntax(options)

993

994

def test_dump_json_conflicts_with_verbose(self):

995

options = self.parser.parse_args()

996

options.dump_json = True

997

options.verbose = True

998

with self.assertParseError():

999

self.check_option_syntax(options)

1000

1001

def test_dump_json_conflicts_with_action(self):

1002

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

1003

options = self.parser.parse_args()

1004

setattr(options, action, value)

1005

options.dump_json = True

1006

with self.assertParseError():

1007

self.check_option_syntax(options)

1008

1009

def test_all_can_not_be_alone(self):

1010

options = self.parser.parse_args()

1011

options.all = True

1012

with self.assertParseError():

1013

self.check_option_syntax(options)

1014

1015

def test_all_is_ok_with_any_action(self):

1016

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

1017

options = self.parser.parse_args()

1018

setattr(options, action, value)

1019

options.all = True

1020

self.check_option_syntax(options)

1021

1022

def test_any_action_is_ok_with_one_client(self):

1023

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

1024

options = self.parser.parse_args()

1025

setattr(options, action, value)

1026

options.client = ["foo"]

1027

self.check_option_syntax(options)

1028

1029

def test_one_client_with_all_actions_except_is_enabled(self):

1030

options = self.parser.parse_args()

1031

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

1032

if action == "is_enabled":

1033

continue

1034

setattr(options, action, value)

1035

options.client = ["foo"]

1036

self.check_option_syntax(options)

1037

1038

def test_two_clients_with_all_actions_except_is_enabled(self):

1039

options = self.parser.parse_args()

1040

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

1041

if action == "is_enabled":

1042

continue

1043

setattr(options, action, value)

1044

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

1045

self.check_option_syntax(options)

1046

1047

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1048

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

1049

if action == "is_enabled":

1050

continue

1051

options = self.parser.parse_args()

1052

setattr(options, action, value)

1053

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

1054

self.check_option_syntax(options)

1055

1056

def test_is_enabled_fails_without_client(self):

1057

options = self.parser.parse_args()

1058

options.is_enabled = True

1059

with self.assertParseError():

1060

self.check_option_syntax(options)

1061

1062

def test_is_enabled_fails_with_two_clients(self):

1063

options = self.parser.parse_args()

1064

options.is_enabled = True

1065

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

1066

with self.assertParseError():

1067

self.check_option_syntax(options)

1068

1069

def test_remove_can_only_be_combined_with_action_deny(self):

1070

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

1071

if action in {"remove", "deny"}:

1072

continue

1073

options = self.parser.parse_args()

1074

setattr(options, action, value)

1075

options.all = True

1076

options.remove = True

1077

with self.assertParseError():

1078

self.check_option_syntax(options)

1079

1080

1081

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1082

def test_calls_and_returns_get_object_on_bus(self):

1083

class MockBus(object):

1084

called = False

1085

def get_object(mockbus_self, busname, dbus_path):

1086

# Note that "self" is still the testcase instance,

1087

# this MockBus instance is in "mockbus_self".

1088

self.assertEqual(dbus_busname, busname)

1089

self.assertEqual(server_dbus_path, dbus_path)

1090

mockbus_self.called = True

1091

return mockbus_self

1092

1093

mockbus = get_mandos_dbus_object(bus=MockBus())

1094

self.assertIsInstance(mockbus, MockBus)

1095

self.assertTrue(mockbus.called)

1096

1097

def test_logs_and_exits_on_dbus_error(self):

1098

class MockBusFailing(object):

1099

def get_object(self, busname, dbus_path):

1100

raise dbus.exceptions.DBusException("Test")

1101

1102

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

1103

with self.assertRaises(SystemExit) as e:

1104

bus = get_mandos_dbus_object(bus=MockBusFailing())

1105

1106

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

1107

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

1108

else:

1109

self.assertIsNotNone(e.exception.code)

1110

1111

1112

class Test_get_managed_objects(TestCaseWithAssertLogs):

1113

def test_calls_and_returns_GetManagedObjects(self):

1114

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

1115

class MockObjectManager(object):

1116

def GetManagedObjects(self):

1117

return managed_objects

1118

retval = get_managed_objects(MockObjectManager())

1119

self.assertDictEqual(managed_objects, retval)

1120

1121

def test_logs_and_exits_on_dbus_error(self):

1122

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

1123

1124

class MockObjectManagerFailing(object):

1125

def GetManagedObjects(self):

1126

dbus_logger.error("Test")

1127

raise dbus.exceptions.DBusException("Test")

1128

1129

class CountingHandler(logging.Handler):

1130

count = 0

1131

def emit(self, record):

1132

self.count += 1

1133

1134

counting_handler = CountingHandler()

1135

1136

dbus_logger.addHandler(counting_handler)

1137

1138

try:

1139

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

1140

with self.assertRaises(SystemExit) as e:

1141

get_managed_objects(MockObjectManagerFailing())

1142

finally:

1143

dbus_logger.removeFilter(counting_handler)

1144

1145

# Make sure the dbus logger was suppressed

1146

self.assertEqual(0, counting_handler.count)

1147

1148

# Test that the dbus_logger still works

1149

with self.assertLogs(dbus_logger, logging.ERROR):

1150

dbus_logger.error("Test")

1151

1152

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

1153

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

1154

else:

1155

self.assertIsNotNone(e.exception.code)

1156

1157

1158

class Test_commands_from_options(unittest.TestCase):

1159

def setUp(self):

1160

self.parser = argparse.ArgumentParser()

1161

add_command_line_options(self.parser)

1162

1163

def test_is_enabled(self):

1164

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

1165

command.IsEnabled)

1166

1167

def assert_command_from_args(self, args, command_cls,

1168

**cmd_attrs):

1169

"""Assert that parsing ARGS should result in an instance of

1170

COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""

1171

options = self.parser.parse_args(args)

1172

check_option_syntax(self.parser, options)

1173

commands = commands_from_options(options)

1174

self.assertEqual(1, len(commands))

1175

command = commands[0]

1176

self.assertIsInstance(command, command_cls)

1177

for key, value in cmd_attrs.items():

1178

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

1179

1180

def test_is_enabled_short(self):

1181

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

1182

command.IsEnabled)

1183

1184

def test_approve(self):

1185

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

1186

command.Approve)

1187

1188

def test_approve_short(self):

1189

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

1190

1191

def test_deny(self):

1192

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

1193

1194

def test_deny_short(self):

1195

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

1196

1197

def test_remove(self):

1198

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

1199

command.Remove)

1200

1201

def test_deny_before_remove(self):

1202

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

1203

"foo"])

1204

check_option_syntax(self.parser, options)

1205

commands = commands_from_options(options)

1206

self.assertEqual(2, len(commands))

1207

self.assertIsInstance(commands[0], command.Deny)

1208

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

1209

1210

def test_deny_before_remove_reversed(self):

1211

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

1212

"--all"])

1213

check_option_syntax(self.parser, options)

1214

commands = commands_from_options(options)

1215

self.assertEqual(2, len(commands))

1216

self.assertIsInstance(commands[0], command.Deny)

1217

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

1218

1219

def test_remove_short(self):

1220

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

1221

1222

def test_dump_json(self):

1223

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

1224

command.DumpJSON)

1225

1226

def test_enable(self):

1227

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

1228

command.Enable)

1229

1230

def test_enable_short(self):

1231

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

1232

1233

def test_disable(self):

1234

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

1235

command.Disable)

1236

1237

def test_disable_short(self):

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

1242

command.BumpTimeout)

1243

1244

def test_bump_timeout_short(self):

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

1250

command.StartChecker)

1251

1252

def test_stop_checker(self):

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

1258

command.ApproveByDefault)

1259

1260

def test_deny_by_default(self):

1261

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

1262

command.DenyByDefault)

1263

1264

def test_checker(self):

1265

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

1266

command.SetChecker,

1267

value_to_set=":")

1268

1269

def test_checker_empty(self):

1270

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

1271

command.SetChecker,

1272

value_to_set="")

1273

1274

def test_checker_short(self):

1275

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

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

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

1288

1289

def test_secret_devnull(self):

1290

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

1291

"foo"], command.SetSecret,

1292

value_to_set=b"")

1293

1294

def test_secret_tempfile(self):

1295

with tempfile.NamedTemporaryFile(mode="r+b") as f:

1296

value = b"secret\0xyzzy\nbar"

1297

f.write(value)

1298

f.seek(0)

1299

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

1300

"foo"], command.SetSecret,

1301

value_to_set=value)

1302

1303

def test_secret_devnull_short(self):

1304

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

1305

command.SetSecret,

1306

value_to_set=b"")

1307

1308

def test_secret_tempfile_short(self):

1309

with tempfile.NamedTemporaryFile(mode="r+b") as f:

1310

value = b"secret\0xyzzy\nbar"

1311

f.write(value)

1312

f.seek(0)

1313

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

1314

command.SetSecret,

1315

value_to_set=value)

1316

1317

def test_timeout(self):

1318

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

1319

command.SetTimeout,

1320

value_to_set=300000)

1321

1322

def test_timeout_short(self):

1323

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

1324

command.SetTimeout,

1325

value_to_set=300000)

1326

1327

def test_extended_timeout(self):

1328

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

1329

"foo"],

1330

command.SetExtendedTimeout,

1331

value_to_set=900000)

1332

1333

def test_interval(self):

1334

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

1335

command.SetInterval,

1336

value_to_set=120000)

1337

1338

def test_interval_short(self):

1339

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

1340

command.SetInterval,

1341

value_to_set=120000)

1342

1343

def test_approval_delay(self):

1344

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

1345

"foo"],

1346

command.SetApprovalDelay,

1347

value_to_set=30000)

1348

1349

def test_approval_duration(self):

1350

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

1351

"foo"],

1352

command.SetApprovalDuration,

1353

value_to_set=1000)

1354

1355

def test_print_table(self):

1356

self.assert_command_from_args([], command.PrintTable,

1357

verbose=False)

1358

1359

def test_print_table_verbose(self):

1360

self.assert_command_from_args(["--verbose"],

1361

command.PrintTable,

1362

verbose=True)

1363

1364

def test_print_table_verbose_short(self):

1365

self.assert_command_from_args(["-v"], command.PrintTable,

1366

verbose=True)

1367

1368

1369

class TestCommand(unittest.TestCase):

831

1370

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

1371

832

1372

def setUp(self):

833

1373

testcase = self

834

1374

class MockClient(object):

835

1375

def __init__(self, name, **attributes):

836

self.__dbus_object_path__ = "objpath_{}".format(name)

1376

self.__dbus_object_path__ = "/clients/{}".format(name)

837

1377

self.attributes = attributes

838

1378

self.attributes["Name"] = name

839

1379

self.calls = []

840

def Set(self, interface, property, value, dbus_interface):

841

testcase.assertEqual(interface, client_interface)

842

testcase.assertEqual(dbus_interface,

843

dbus.PROPERTIES_IFACE)

844

self.attributes[property] = value

845

def Get(self, interface, property, dbus_interface):

846

testcase.assertEqual(interface, client_interface)

847

testcase.assertEqual(dbus_interface,

848

dbus.PROPERTIES_IFACE)

849

return self.attributes[property]

1380

def Set(self, interface, propname, value, dbus_interface):

1381

testcase.assertEqual(client_dbus_interface, interface)

1382

testcase.assertEqual(dbus.PROPERTIES_IFACE,

1383

dbus_interface)

1384

self.attributes[propname] = value

850

1385

def Approve(self, approve, dbus_interface):

851

testcase.assertEqual(dbus_interface, client_interface)

1386

testcase.assertEqual(client_dbus_interface,

1387

dbus_interface)

852

1388

self.calls.append(("Approve", (approve,

853

1389

dbus_interface)))

854

1390

self.client = MockClient(

894

1430

ApprovedByDefault=dbus.Boolean(False),

895

1431

LastApprovalRequest="2019-01-03T00:00:00",

896

1432

ApprovalDelay=30000,

897

ApprovalDuration=1000,

1433

ApprovalDuration=93785000,

898

1434

Checker=":",

899

1435

ExtendedTimeout=900000,

900

1436

Expires="2019-02-05T00:00:00",

901

1437

LastCheckerStatus=-2)

902

1438

self.clients = collections.OrderedDict(

903

1439

[

904

(self.client, self.client.attributes),

905

(self.other_client, self.other_client.attributes),

1440

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

1441

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

906

1442

])

907

self.one_client = {self.client: self.client.attributes}

908

909

class TestPrintTableCmd(TestCmd):

910

def test_normal(self):

911

output = PrintTableCmd().output(None, self.clients.values())

912

expected_output = """

913

Name Enabled Timeout Last Successful Check

914

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

915

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

916

"""[1:-1]

917

self.assertEqual(output, expected_output)

918

def test_verbose(self):

919

output = PrintTableCmd(verbose=True).output(

920

None, self.clients.values())

921

expected_output = """

922

Name Enabled Timeout Last Successful Check Created Interval Host Key ID Fingerprint Check Is Running Last Enabled Approval Is Pending Approved By Default Last Approval Request Approval Delay Approval Duration Checker Extended Timeout Expires Last Checker Status

923

foo Yes 00:05:00 2019-02-03T00:00:00 2019-01-02T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03T00:00:00 No Yes 00:00:00 00:00:01 fping -q -- %(host)s 00:15:00 2019-02-04T00:00:00 0

924

barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04T00:00:00 No No 2019-01-03T00:00:00 00:00:30 00:00:01 : 00:15:00 2019-02-05T00:00:00 -2

925

"""[1:-1]

926

self.assertEqual(output, expected_output)

927

def test_one_client(self):

928

output = PrintTableCmd().output(None, self.one_client.values())

929

expected_output = """

930

Name Enabled Timeout Last Successful Check

931

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

932

"""[1:-1]

933

self.assertEqual(output, expected_output)

934

935

class TestDumpJSONCmd(TestCmd):

936

def setUp(self):

937

self.expected_json = {

938

"foo": {

939

"Name": "foo",

940

"KeyID": ("92ed150794387c03ce684574b1139a65"

941

"94a34f895daaaf09fd8ea90a27cddb12"),

942

"Host": "foo.example.org",

943

"Enabled": True,

944

"Timeout": 300000,

945

"LastCheckedOK": "2019-02-03T00:00:00",

946

"Created": "2019-01-02T00:00:00",

947

"Interval": 120000,

948

"Fingerprint": ("778827225BA7DE539C5A"

949

"7CFA59CFF7CDBD9A5920"),

950

"CheckerRunning": False,

951

"LastEnabled": "2019-01-03T00:00:00",

952

"ApprovalPending": False,

953

"ApprovedByDefault": True,

954

"LastApprovalRequest": "",

955

"ApprovalDelay": 0,

956

"ApprovalDuration": 1000,

957

"Checker": "fping -q -- %(host)s",

958

"ExtendedTimeout": 900000,

959

"Expires": "2019-02-04T00:00:00",

960

"LastCheckerStatus": 0,

961

962

"barbar": {

963

"Name": "barbar",

964

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

965

"6ab612cff5ad227247e46c2b020f441c"),

966

"Host": "192.0.2.3",

967

"Enabled": True,

968

"Timeout": 300000,

969

"LastCheckedOK": "2019-02-04T00:00:00",

970

"Created": "2019-01-03T00:00:00",

971

"Interval": 120000,

972

"Fingerprint": ("3E393AEAEFB84C7E89E2"

973

"F547B3A107558FCA3A27"),

974

"CheckerRunning": True,

975

"LastEnabled": "2019-01-04T00:00:00",

976

"ApprovalPending": False,

977

"ApprovedByDefault": False,

978

"LastApprovalRequest": "2019-01-03T00:00:00",

979

"ApprovalDelay": 30000,

980

"ApprovalDuration": 1000,

981

"Checker": ":",

982

"ExtendedTimeout": 900000,

983

"Expires": "2019-02-05T00:00:00",

984

"LastCheckerStatus": -2,

985

986

}

987

return super(TestDumpJSONCmd, self).setUp()

988

def test_normal(self):

989

json_data = json.loads(DumpJSONCmd().output(None,

990

self.clients))

991

self.assertDictEqual(json_data, self.expected_json)

992

def test_one_client(self):

993

clients = self.one_client

994

json_data = json.loads(DumpJSONCmd().output(None, clients))

995

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

996

self.assertDictEqual(json_data, expected_json)

997

998

class TestIsEnabledCmd(TestCmd):

999

def test_is_enabled(self):

1000

self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)

1001

for client, properties in self.clients.items()))

1002

def test_is_enabled_run_exits_successfully(self):

1443

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

1444

1445

@property

1446

def bus(self):

1447

class Bus(object):

1448

@staticmethod

1449

def get_object(client_bus_name, path):

1450

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

1458

1459

1460

class TestBaseCommands(TestCommand):

1461

1462

def test_IsEnabled_exits_successfully(self):

1003

1463

with self.assertRaises(SystemExit) as e:

1004

IsEnabledCmd().run(None, None, self.one_client)

1464

command.IsEnabled().run(self.one_client)

1005

1465

if e.exception.code is not None:

1006

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

1466

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

1007

1467

else:

1008

1468

self.assertIsNone(e.exception.code)

1009

def test_is_enabled_run_exits_with_failure(self):

1469

1470

def test_IsEnabled_exits_with_failure(self):

1010

1471

self.client.attributes["Enabled"] = dbus.Boolean(False)

1011

1472

with self.assertRaises(SystemExit) as e:

1012

IsEnabledCmd().run(None, None, self.one_client)

1473

command.IsEnabled().run(self.one_client)

1013

1474

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

1014

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

1475

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

1015

1476

else:

1016

1477

self.assertIsNotNone(e.exception.code)

1017

1478

1018

class TestRemoveCmd(TestCmd):

1019

def test_remove(self):

1479

def test_Approve(self):

1480

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

1481

for clientpath in self.clients:

1482

client = self.bus.get_object(dbus_busname, clientpath)

1483

self.assertIn(("Approve", (True, client_dbus_interface)),

1484

client.calls)

1485

1486

def test_Deny(self):

1487

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

1488

for clientpath in self.clients:

1489

client = self.bus.get_object(dbus_busname, clientpath)

1490

self.assertIn(("Approve", (False, client_dbus_interface)),

1491

client.calls)

1492

1493

def test_Remove(self):

1020

1494

class MockMandos(object):

1021

1495

def __init__(self):

1022

1496

self.calls = []

1023

1497

def RemoveClient(self, dbus_path):

1024

1498

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

1025

1499

mandos = MockMandos()

1026

super(TestRemoveCmd, self).setUp()

1027

RemoveCmd().run(mandos, None, self.clients)

1028

self.assertEqual(len(mandos.calls), 2)

1029

for client in self.clients:

1030

self.assertIn(("RemoveClient",

1031

(client.__dbus_object_path__,)),

1500

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

1501

for clientpath in self.clients:

1502

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

1032

1503

mandos.calls)

1033

1504

1034

class TestApproveCmd(TestCmd):

1035

def test_approve(self):

1036

ApproveCmd().run(None, None, self.clients)

1037

for client in self.clients:

1038

self.assertIn(("Approve", (True, client_interface)),

1039

client.calls)

1040

1041

class TestDenyCmd(TestCmd):

1042

def test_deny(self):

1043

DenyCmd().run(None, None, self.clients)

1044

for client in self.clients:

1045

self.assertIn(("Approve", (False, client_interface)),

1046

client.calls)

1047

1048

class TestEnableCmd(TestCmd):

1049

def test_enable(self):

1050

for client in self.clients:

1051

client.attributes["Enabled"] = False

1052

1053

EnableCmd().run(None, None, self.clients)

1054

1055

for client in self.clients:

1056

self.assertTrue(client.attributes["Enabled"])

1057

1058

class TestDisableCmd(TestCmd):

1059

def test_disable(self):

1060

DisableCmd().run(None, None, self.clients)

1061

1062

for client in self.clients:

1063

self.assertFalse(client.attributes["Enabled"])

1064

1065

class Unique(object):

1066

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

1067

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

1068

1069

class TestPropertyCmd(TestCmd):

1070

"""Abstract class for tests of PropertyCmd classes"""

1505

expected_json = {

1506

"foo": {

1507

"Name": "foo",

1508

"KeyID": ("92ed150794387c03ce684574b1139a65"

1509

"94a34f895daaaf09fd8ea90a27cddb12"),

1510

"Host": "foo.example.org",

1511

"Enabled": True,

1512

"Timeout": 300000,

1513

"LastCheckedOK": "2019-02-03T00:00:00",

1514

"Created": "2019-01-02T00:00:00",

1515

"Interval": 120000,

1516

"Fingerprint": ("778827225BA7DE539C5A"

1517

"7CFA59CFF7CDBD9A5920"),

1518

"CheckerRunning": False,

1519

"LastEnabled": "2019-01-03T00:00:00",

1520

"ApprovalPending": False,

1521

"ApprovedByDefault": True,

1522

"LastApprovalRequest": "",

1523

"ApprovalDelay": 0,

1524

"ApprovalDuration": 1000,

1525

"Checker": "fping -q -- %(host)s",

1526

"ExtendedTimeout": 900000,

1527

"Expires": "2019-02-04T00:00:00",

1528

"LastCheckerStatus": 0,

1529

1530

"barbar": {

1531

"Name": "barbar",

1532

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1533

"6ab612cff5ad227247e46c2b020f441c"),

1534

"Host": "192.0.2.3",

1535

"Enabled": True,

1536

"Timeout": 300000,

1537

"LastCheckedOK": "2019-02-04T00:00:00",

1538

"Created": "2019-01-03T00:00:00",

1539

"Interval": 120000,

1540

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1541

"F547B3A107558FCA3A27"),

1542

"CheckerRunning": True,

1543

"LastEnabled": "2019-01-04T00:00:00",

1544

"ApprovalPending": False,

1545

"ApprovedByDefault": False,

1546

"LastApprovalRequest": "2019-01-03T00:00:00",

1547

"ApprovalDelay": 30000,

1548

"ApprovalDuration": 93785000,

1549

"Checker": ":",

1550

"ExtendedTimeout": 900000,

1551

"Expires": "2019-02-05T00:00:00",

1552

"LastCheckerStatus": -2,

1553

1554

}

1555

1556

def test_DumpJSON_normal(self):

1557

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

1558

json_data = json.loads(output)

1559

self.assertDictEqual(self.expected_json, json_data)

1560

1561

def test_DumpJSON_one_client(self):

1562

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

1563

json_data = json.loads(output)

1564

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

1565

self.assertDictEqual(expected_json, json_data)

1566

1567

def test_PrintTable_normal(self):

1568

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

1569

expected_output = "\n".join((

1570

"Name Enabled Timeout Last Successful Check",

1571

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

1572

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

1573

))

1574

self.assertEqual(expected_output, output)

1575

1576

def test_PrintTable_verbose(self):

1577

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

1578

self.clients.values())

1579

columns = (

1580

(

1581

"Name ",

1582

"foo ",

1583

"barbar ",

1584

),(

1585

"Enabled ",

1586

"Yes ",

1587

"Yes ",

1588

),(

1589

"Timeout ",

1590

"00:05:00 ",

1591

"00:05:00 ",

1592

),(

1593

"Last Successful Check ",

1594

"2019-02-03T00:00:00 ",

1595

"2019-02-04T00:00:00 ",

1596

),(

1597

"Created ",

1598

"2019-01-02T00:00:00 ",

1599

"2019-01-03T00:00:00 ",

1600

),(

1601

"Interval ",

1602

"00:02:00 ",

1603

"00:02:00 ",

1604

),(

1605

"Host ",

1606

"foo.example.org ",

1607

"192.0.2.3 ",

1608

),(

1609

("Key ID "

1610

" "),

1611

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1612

"ea90a27cddb12 "),

1613

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1614

"46c2b020f441c "),

1615

),(

1616

"Fingerprint ",

1617

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1618

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1619

),(

1620

"Check Is Running ",

1621

"No ",

1622

"Yes ",

1623

),(

1624

"Last Enabled ",

1625

"2019-01-03T00:00:00 ",

1626

"2019-01-04T00:00:00 ",

1627

),(

1628

"Approval Is Pending ",

1629

"No ",

1630

"No ",

1631

),(

1632

"Approved By Default ",

1633

"Yes ",

1634

"No ",

1635

),(

1636

"Last Approval Request ",

1637

" ",

1638

"2019-01-03T00:00:00 ",

1639

),(

1640

"Approval Delay ",

1641

"00:00:00 ",

1642

"00:00:30 ",

1643

),(

1644

"Approval Duration ",

1645

"00:00:01 ",

1646

"1T02:03:05 ",

1647

),(

1648

"Checker ",

1649

"fping -q -- %(host)s ",

1650

": ",

1651

),(

1652

"Extended Timeout ",

1653

"00:15:00 ",

1654

"00:15:00 ",

1655

),(

1656

"Expires ",

1657

"2019-02-04T00:00:00 ",

1658

"2019-02-05T00:00:00 ",

1659

),(

1660

"Last Checker Status",

1661

"0 ",

1662

"-2 ",

1663

)

1664

)

1665

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

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)

1670

1671

def test_PrintTable_one_client(self):

1672

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

1673

expected_output = "\n".join((

1674

"Name Enabled Timeout Last Successful Check",

1675

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

1676

))

1677

self.assertEqual(expected_output, output)

1678

1679

1680

class TestPropertyCmd(TestCommand):

1681

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

1071

1682

def runTest(self):

1072

1683

if not hasattr(self, "command"):

1073

1684

return

1075

1686

self.values_to_set)

1076

1687

for value_to_set, value_to_get in zip(self.values_to_set,

1077

1688

values_to_get):

1078

for client in self.clients:

1079

old_value = client.attributes[self.property]

1080

self.assertNotIsInstance(old_value, Unique)

1081

client.attributes[self.property] = Unique()

1689

for clientpath in self.clients:

1690

client = self.bus.get_object(dbus_busname, clientpath)

1691

old_value = client.attributes[self.propname]

1692

client.attributes[self.propname] = self.Unique()

1082

1693

self.run_command(value_to_set, self.clients)

1083

for client in self.clients:

1084

value = client.attributes[self.property]

1085

self.assertNotIsInstance(value, Unique)

1086

self.assertEqual(value, value_to_get)

1694

for clientpath in self.clients:

1695

client = self.bus.get_object(dbus_busname, clientpath)

1696

value = client.attributes[self.propname]

1697

self.assertNotIsInstance(value, self.Unique)

1698

self.assertEqual(value_to_get, value)

1699

1700

class Unique(object):

1701

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

1702

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

1703

1087

1704

def run_command(self, value, clients):

1088

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

1705

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

1706

1707

1708

class TestEnableCmd(TestPropertyCmd):

1709

command = command.Enable

1710

propname = "Enabled"

1711

values_to_set = [dbus.Boolean(True)]

1712

1713

1714

class TestDisableCmd(TestPropertyCmd):

1715

command = command.Disable

1716

propname = "Enabled"

1717

values_to_set = [dbus.Boolean(False)]

1718

1089

1719

1090

1720

class TestBumpTimeoutCmd(TestPropertyCmd):

1091

command = BumpTimeoutCmd

1092

property = "LastCheckedOK"

1721

command = command.BumpTimeout

1722

propname = "LastCheckedOK"

1093

1723

values_to_set = [""]

1094

1724

1725

1095

1726

class TestStartCheckerCmd(TestPropertyCmd):

1096

command = StartCheckerCmd

1097

property = "CheckerRunning"

1727

command = command.StartChecker

1728

propname = "CheckerRunning"

1098

1729

values_to_set = [dbus.Boolean(True)]

1099

1730

1731

1100

1732

class TestStopCheckerCmd(TestPropertyCmd):

1101

command = StopCheckerCmd

1102

property = "CheckerRunning"

1733

command = command.StopChecker

1734

propname = "CheckerRunning"

1103

1735

values_to_set = [dbus.Boolean(False)]

1104

1736

1737

1105

1738

class TestApproveByDefaultCmd(TestPropertyCmd):

1106

command = ApproveByDefaultCmd

1107

property = "ApprovedByDefault"

1739

command = command.ApproveByDefault

1740

propname = "ApprovedByDefault"

1108

1741

values_to_set = [dbus.Boolean(True)]

1109

1742

1743

1110

1744

class TestDenyByDefaultCmd(TestPropertyCmd):

1111

command = DenyByDefaultCmd

1112

property = "ApprovedByDefault"

1745

command = command.DenyByDefault

1746

propname = "ApprovedByDefault"

1113

1747

values_to_set = [dbus.Boolean(False)]

1114

1748

1115

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1116

"""Abstract class for tests of PropertyCmd classes using the

1117

ValueArgumentMixIn"""

1749

1750

class TestPropertyValueCmd(TestPropertyCmd):

1751

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

1752

1118

1753

def runTest(self):

1119

if type(self) is TestValueArgumentPropertyCmd:

1754

if type(self) is TestPropertyValueCmd:

1120

1755

return

1121

return super(TestValueArgumentPropertyCmd, self).runTest()

1756

return super(TestPropertyValueCmd, self).runTest()

1757

1122

1758

def run_command(self, value, clients):

1123

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

1124

1125

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1126

command = SetCheckerCmd

1127

property = "Checker"

1759

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

1760

1761

1762

class TestSetCheckerCmd(TestPropertyValueCmd):

1763

command = command.SetChecker

1764

propname = "Checker"

1128

1765

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

1129

1766

1130

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1131

command = SetHostCmd

1132

property = "Host"

1767

1768

class TestSetHostCmd(TestPropertyValueCmd):

1769

command = command.SetHost

1770

propname = "Host"

1133

1771

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

1134

1772

1135

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1136

command = SetSecretCmd

1137

property = "Secret"

1773

1774

class TestSetSecretCmd(TestPropertyValueCmd):

1775

command = command.SetSecret

1776

propname = "Secret"

1138

1777

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

1139

1778

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

1140

1779

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

1141

1780

1142

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1143

command = SetTimeoutCmd

1144

property = "Timeout"

1145

values_to_set = [datetime.timedelta(),

1146

datetime.timedelta(minutes=5),

1147

datetime.timedelta(seconds=1),

1148

datetime.timedelta(weeks=1),

1149

datetime.timedelta(weeks=52)]

1150

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

1151

1152

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1153

command = SetExtendedTimeoutCmd

1154

property = "ExtendedTimeout"

1155

values_to_set = [datetime.timedelta(),

1156

datetime.timedelta(minutes=5),

1157

datetime.timedelta(seconds=1),

1158

datetime.timedelta(weeks=1),

1159

datetime.timedelta(weeks=52)]

1160

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

1161

1162

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1163

command = SetIntervalCmd

1164

property = "Interval"

1165

values_to_set = [datetime.timedelta(),

1166

datetime.timedelta(minutes=5),

1167

datetime.timedelta(seconds=1),

1168

datetime.timedelta(weeks=1),

1169

datetime.timedelta(weeks=52)]

1170

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

1171

1172

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1173

command = SetApprovalDelayCmd

1174

property = "ApprovalDelay"

1175

values_to_set = [datetime.timedelta(),

1176

datetime.timedelta(minutes=5),

1177

datetime.timedelta(seconds=1),

1178

datetime.timedelta(weeks=1),

1179

datetime.timedelta(weeks=52)]

1180

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

1181

1182

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1183

command = SetApprovalDurationCmd

1184

property = "ApprovalDuration"

1185

values_to_set = [datetime.timedelta(),

1186

datetime.timedelta(minutes=5),

1187

datetime.timedelta(seconds=1),

1188

datetime.timedelta(weeks=1),

1189

datetime.timedelta(weeks=52)]

1190

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

1191

1192

class Test_command_from_options(unittest.TestCase):

1193

def setUp(self):

1194

self.parser = argparse.ArgumentParser()

1195

add_command_line_options(self.parser)

1196

def assert_command_from_args(self, args, command_cls, **cmd_attrs):

1197

"""Assert that parsing ARGS should result in an instance of

1198

COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""

1199

options = self.parser.parse_args(args)

1200

check_option_syntax(self.parser, options)

1201

commands = commands_from_options(options)

1202

self.assertEqual(len(commands), 1)

1203

command = commands[0]

1204

self.assertIsInstance(command, command_cls)

1205

for key, value in cmd_attrs.items():

1206

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

1207

def test_print_table(self):

1208

self.assert_command_from_args([], PrintTableCmd,

1209

verbose=False)

1210

1211

def test_print_table_verbose(self):

1212

self.assert_command_from_args(["--verbose"], PrintTableCmd,

1213

verbose=True)

1214

1215

def test_print_table_verbose_short(self):

1216

self.assert_command_from_args(["-v"], PrintTableCmd,

1217

verbose=True)

1218

1219

def test_enable(self):

1220

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

1221

1222

def test_enable_short(self):

1223

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

1224

1225

def test_disable(self):

1226

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

1227

DisableCmd)

1228

1229

def test_disable_short(self):

1230

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

1231

1232

def test_bump_timeout(self):

1233

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

1234

BumpTimeoutCmd)

1235

1236

def test_bump_timeout_short(self):

1237

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

1238

1239

def test_start_checker(self):

1240

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

1241

StartCheckerCmd)

1242

1243

def test_stop_checker(self):

1244

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

1245

StopCheckerCmd)

1246

1247

def test_remove(self):

1248

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

1249

RemoveCmd)

1250

1251

def test_remove_short(self):

1252

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

1253

1254

def test_checker(self):

1255

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

1256

SetCheckerCmd, value_to_set=":")

1257

1258

def test_checker_empty(self):

1259

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

1260

SetCheckerCmd, value_to_set="")

1261

1262

def test_checker_short(self):

1263

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

1264

SetCheckerCmd, value_to_set=":")

1265

1266

def test_timeout(self):

1267

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

1268

SetTimeoutCmd,

1269

value_to_set=300000)

1270

1271

def test_timeout_short(self):

1272

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

1273

SetTimeoutCmd,

1274

value_to_set=300000)

1275

1276

def test_extended_timeout(self):

1277

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

1278

"foo"],

1279

SetExtendedTimeoutCmd,

1280

value_to_set=900000)

1281

1282

def test_interval(self):

1283

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

1284

SetIntervalCmd,

1285

value_to_set=120000)

1286

1287

def test_interval_short(self):

1288

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

1289

SetIntervalCmd,

1290

value_to_set=120000)

1291

1292

def test_approve_by_default(self):

1293

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

1294

ApproveByDefaultCmd)

1295

1296

def test_deny_by_default(self):

1297

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

1298

DenyByDefaultCmd)

1299

1300

def test_approval_delay(self):

1301

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

1302

"foo"], SetApprovalDelayCmd,

1303

value_to_set=30000)

1304

1305

def test_approval_duration(self):

1306

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

1307

"foo"], SetApprovalDurationCmd,

1308

value_to_set=1000)

1309

1310

def test_host(self):

1311

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

1312

"foo"], SetHostCmd,

1313

value_to_set="foo.example.org")

1314

1315

def test_host_short(self):

1316

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

1317

"foo"], SetHostCmd,

1318

value_to_set="foo.example.org")

1319

1320

def test_secret_devnull(self):

1321

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

1322

"foo"], SetSecretCmd,

1323

value_to_set=b"")

1324

1325

def test_secret_tempfile(self):

1326

with tempfile.NamedTemporaryFile(mode="r+b") as f:

1327

value = b"secret\0xyzzy\nbar"

1328

f.write(value)

1329

f.seek(0)

1330

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

1331

"foo"], SetSecretCmd,

1332

value_to_set=value)

1333

1334

def test_secret_devnull_short(self):

1335

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

1336

SetSecretCmd, value_to_set=b"")

1337

1338

def test_secret_tempfile_short(self):

1339

with tempfile.NamedTemporaryFile(mode="r+b") as f:

1340

value = b"secret\0xyzzy\nbar"

1341

f.write(value)

1342

f.seek(0)

1343

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

1344

SetSecretCmd,

1345

value_to_set=value)

1346

1347

def test_approve(self):

1348

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

1349

ApproveCmd)

1350

1351

def test_approve_short(self):

1352

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

1353

1354

def test_deny(self):

1355

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

1356

1357

def test_deny_short(self):

1358

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

1359

1360

def test_dump_json(self):

1361

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

1362

1363

def test_is_enabled(self):

1364

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

1365

IsEnabledCmd)

1366

1367

def test_is_enabled_short(self):

1368

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

1369

1370

def test_deny_before_remove(self):

1371

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

1372

check_option_syntax(self.parser, options)

1373

commands = commands_from_options(options)

1374

self.assertEqual(len(commands), 2)

1375

self.assertIsInstance(commands[0], DenyCmd)

1376

self.assertIsInstance(commands[1], RemoveCmd)

1377

1378

def test_deny_before_remove_reversed(self):

1379

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

1380

check_option_syntax(self.parser, options)

1381

commands = commands_from_options(options)

1382

self.assertEqual(len(commands), 2)

1383

self.assertIsInstance(commands[0], DenyCmd)

1384

self.assertIsInstance(commands[1], RemoveCmd)

1385

1386

1387

class Test_check_option_syntax(unittest.TestCase):

1388

# This mostly corresponds to the definition from has_actions() in

1389

# check_option_syntax()

1390

actions = {

1391

# The actual values set here are not that important, but we do

1392

# at least stick to the correct types, even though they are

1393

# never used

1394

"enable": True,

1395

"disable": True,

1396

"bump_timeout": True,

1397

"start_checker": True,

1398

"stop_checker": True,

1399

"is_enabled": True,

1400

"remove": True,

1401

"checker": "x",

1402

"timeout": datetime.timedelta(),

1403

"extended_timeout": datetime.timedelta(),

1404

"interval": datetime.timedelta(),

1405

"approved_by_default": True,

1406

"approval_delay": datetime.timedelta(),

1407

"approval_duration": datetime.timedelta(),

1408

"host": "x",

1409

"secret": io.BytesIO(b"x"),

1410

"approve": True,

1411

"deny": True,

1412

}

1413

1414

def setUp(self):

1415

self.parser = argparse.ArgumentParser()

1416

add_command_line_options(self.parser)

1417

1418

@contextlib.contextmanager

1419

def assertParseError(self):

1420

with self.assertRaises(SystemExit) as e:

1421

with self.temporarily_suppress_stderr():

1422

yield

1423

# Exit code from argparse is guaranteed to be "2". Reference:

1424

# https://docs.python.org/3/library/argparse.html#exiting-methods

1425

self.assertEqual(e.exception.code, 2)

1426

1427

@staticmethod

1428

@contextlib.contextmanager

1429

def temporarily_suppress_stderr():

1430

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

1431

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

1432

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

1433

os.close(null)

1434

try:

1435

yield

1436

finally:

1437

# restore stderr

1438

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

1439

os.close(stderrcopy)

1440

1441

def check_option_syntax(self, options):

1442

check_option_syntax(self.parser, options)

1443

1444

def test_actions_requires_client_or_all(self):

1445

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

1446

options = self.parser.parse_args()

1447

setattr(options, action, value)

1448

with self.assertParseError():

1449

self.check_option_syntax(options)

1450

1451

def test_actions_conflicts_with_verbose(self):

1452

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

1453

options = self.parser.parse_args()

1454

setattr(options, action, value)

1455

options.verbose = True

1456

with self.assertParseError():

1457

self.check_option_syntax(options)

1458

1459

def test_dump_json_conflicts_with_verbose(self):

1460

options = self.parser.parse_args()

1461

options.dump_json = True

1462

options.verbose = True

1463

with self.assertParseError():

1464

self.check_option_syntax(options)

1465

1466

def test_dump_json_conflicts_with_action(self):

1467

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

1468

options = self.parser.parse_args()

1469

setattr(options, action, value)

1470

options.dump_json = True

1471

with self.assertParseError():

1472

self.check_option_syntax(options)

1473

1474

def test_all_can_not_be_alone(self):

1475

options = self.parser.parse_args()

1476

options.all = True

1477

with self.assertParseError():

1478

self.check_option_syntax(options)

1479

1480

def test_all_is_ok_with_any_action(self):

1481

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

1482

options = self.parser.parse_args()

1483

setattr(options, action, value)

1484

options.all = True

1485

self.check_option_syntax(options)

1486

1487

def test_is_enabled_fails_without_client(self):

1488

options = self.parser.parse_args()

1489

options.is_enabled = True

1490

with self.assertParseError():

1491

self.check_option_syntax(options)

1492

1493

def test_is_enabled_works_with_one_client(self):

1494

options = self.parser.parse_args()

1495

options.is_enabled = True

1496

options.client = ["foo"]

1497

self.check_option_syntax(options)

1498

1499

def test_is_enabled_fails_with_two_clients(self):

1500

options = self.parser.parse_args()

1501

options.is_enabled = True

1502

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

1503

with self.assertParseError():

1504

self.check_option_syntax(options)

1505

1506

def test_remove_can_only_be_combined_with_action_deny(self):

1507

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

1508

if action in {"remove", "deny"}:

1509

continue

1510

options = self.parser.parse_args()

1511

setattr(options, action, value)

1512

options.all = True

1513

options.remove = True

1514

with self.assertParseError():

1515

self.check_option_syntax(options)

1781

1782

class TestSetTimeoutCmd(TestPropertyValueCmd):

1783

command = command.SetTimeout

1784

propname = "Timeout"

1785

values_to_set = [datetime.timedelta(),

1786

datetime.timedelta(minutes=5),

1787

datetime.timedelta(seconds=1),

1788

datetime.timedelta(weeks=1),

1789

datetime.timedelta(weeks=52)]

1790

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

1791

1792

1793

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1794

command = command.SetExtendedTimeout

1795

propname = "ExtendedTimeout"

1796

values_to_set = [datetime.timedelta(),

1797

datetime.timedelta(minutes=5),

1798

datetime.timedelta(seconds=1),

1799

datetime.timedelta(weeks=1),

1800

datetime.timedelta(weeks=52)]

1801

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

1802

1803

1804

class TestSetIntervalCmd(TestPropertyValueCmd):

1805

command = command.SetInterval

1806

propname = "Interval"

1807

values_to_set = [datetime.timedelta(),

1808

datetime.timedelta(minutes=5),

1809

datetime.timedelta(seconds=1),

1810

datetime.timedelta(weeks=1),

1811

datetime.timedelta(weeks=52)]

1812

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

1813

1814

1815

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1816

command = command.SetApprovalDelay

1817

propname = "ApprovalDelay"

1818

values_to_set = [datetime.timedelta(),

1819

datetime.timedelta(minutes=5),

1820

datetime.timedelta(seconds=1),

1821

datetime.timedelta(weeks=1),

1822

datetime.timedelta(weeks=52)]

1823

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

1824

1825

1826

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1827

command = command.SetApprovalDuration

1828

propname = "ApprovalDuration"

1829

values_to_set = [datetime.timedelta(),

1830

datetime.timedelta(minutes=5),

1831

datetime.timedelta(seconds=1),

1832

datetime.timedelta(weeks=1),

1833

datetime.timedelta(weeks=52)]

1834

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

1516

1835

1517

1836

1518

1837

1533

1852

return tests

1534

1853

1535

1854

if __name__ == "__main__":

1536

if should_only_run_tests():

1537

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

1538

unittest.main()

1539

else:

1540

main()

1855

try:

1856

if should_only_run_tests():

1857

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

1858

unittest.main()

1859

else:

1860

main()

1861

finally:

1862

logging.shutdown()

Older »