/mandos/trunk : revision 1035

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

47

49

48

import dbus

50

49

294

293

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

295

294

"Expires", "LastCheckerStatus")

296

295

def run(self, mandos, clients):

297

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

298

def output(self, clients):

299

raise NotImplementedError()

296

print(self.output(clients))

300

297

301

298

class PropertyCmd(Command):

302

299

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

303

300

def run_on_one_client(self, client, properties):

304

301

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

305

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

306

client.__dbus_object_path__,

307

dbus.PROPERTIES_IFACE, client_interface,

308

self.propname, self.value_to_set

309

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

310

else bool(self.value_to_set))

311

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

302

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

312

303

dbus_interface=dbus.PROPERTIES_IFACE)

313

@property

314

def propname(self):

315

raise NotImplementedError()

316

304

317

305

class ValueArgumentMixIn(object):

318

306

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

341

329

keywords = default_keywords

342

330

if self.verbose:

343

331

keywords = self.all_keywords

344

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

332

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

345

333

346

334

class TableOfClients(object):

347

335

tableheaders = {

438

426

sys.exit(0)

439

427

sys.exit(1)

440

428

def is_enabled(self, client, properties):

441

return properties["Enabled"]

429

return bool(properties["Enabled"])

442

430

443

431

class RemoveCmd(Command):

444

432

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

433

self.mandos.RemoveClient(client.__dbus_object_path__)

449

434

450

435

class ApproveCmd(Command):

451

436

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

437

client.Approve(dbus.Boolean(True),

455

438

dbus_interface=client_interface)

456

439

457

440

class DenyCmd(Command):

458

441

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

442

client.Approve(dbus.Boolean(False),

462

443

dbus_interface=client_interface)

463

444

464

445

class EnableCmd(PropertyCmd):

465

propname = "Enabled"

446

property = "Enabled"

466

447

value_to_set = dbus.Boolean(True)

467

448

468

449

class DisableCmd(PropertyCmd):

469

propname = "Enabled"

450

property = "Enabled"

470

451

value_to_set = dbus.Boolean(False)

471

452

472

453

class BumpTimeoutCmd(PropertyCmd):

473

propname = "LastCheckedOK"

454

property = "LastCheckedOK"

474

455

value_to_set = ""

475

456

476

457

class StartCheckerCmd(PropertyCmd):

477

propname = "CheckerRunning"

458

property = "CheckerRunning"

478

459

value_to_set = dbus.Boolean(True)

479

460

480

461

class StopCheckerCmd(PropertyCmd):

481

propname = "CheckerRunning"

462

property = "CheckerRunning"

482

463

value_to_set = dbus.Boolean(False)

483

464

484

465

class ApproveByDefaultCmd(PropertyCmd):

485

propname = "ApprovedByDefault"

466

property = "ApprovedByDefault"

486

467

value_to_set = dbus.Boolean(True)

487

468

488

469

class DenyByDefaultCmd(PropertyCmd):

489

propname = "ApprovedByDefault"

470

property = "ApprovedByDefault"

490

471

value_to_set = dbus.Boolean(False)

491

472

492

473

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

493

propname = "Checker"

474

property = "Checker"

494

475

495

476

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

496

propname = "Host"

477

property = "Host"

497

478

498

479

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

499

propname = "Secret"

500

480

@property

501

481

def value_to_set(self):

502

482

return self._vts

505

485

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

506

486

self._vts = value.read()

507

487

value.close()

488

property = "Secret"

508

489

509

490

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

510

propname = "Timeout"

491

property = "Timeout"

511

492

512

493

class SetExtendedTimeoutCmd(PropertyCmd,

513

494

MillisecondsValueArgumentMixIn):

514

propname = "ExtendedTimeout"

495

property = "ExtendedTimeout"

515

496

516

497

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

517

propname = "Interval"

498

property = "Interval"

518

499

519

500

class SetApprovalDelayCmd(PropertyCmd,

520

501

MillisecondsValueArgumentMixIn):

521

propname = "ApprovalDelay"

502

property = "ApprovalDelay"

522

503

523

504

class SetApprovalDurationCmd(PropertyCmd,

524

505

MillisecondsValueArgumentMixIn):

525

propname = "ApprovalDuration"

506

property = "ApprovalDuration"

526

507

527

508

def add_command_line_options(parser):

528

509

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

584

565

help="Approve any current client request")

585

566

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

586

567

help="Deny any current client request")

587

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

588

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

589

568

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

590

569

help="Run self-test")

591

570

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

616

595

if options.is_enabled:

617

596

commands.append(IsEnabledCmd())

618

597

598

if options.remove:

599

commands.append(RemoveCmd())

600

619

601

if options.checker is not None:

620

602

commands.append(SetCheckerCmd(options.checker))

621

603

654

636

if options.deny:

655

637

commands.append(DenyCmd())

656

638

657

if options.remove:

658

commands.append(RemoveCmd())

659

660

639

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

661

640

# optionally verbosely

662

641

if not commands:

665

644

return commands

666

645

667

646

668

def check_option_syntax(parser, options):

669

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

670

argparse"""

647

def main():

648

parser = argparse.ArgumentParser()

649

650

add_command_line_options(parser)

651

652

options = parser.parse_args()

671

653

672

654

def has_actions(options):

673

655

return any((options.enable,

700

682

parser.error("--all requires an action.")

701

683

if options.is_enabled and len(options.client) > 1:

702

684

parser.error("--is-enabled requires exactly one client")

703

if options.remove:

704

options.remove = False

705

if has_actions(options) and not options.deny:

706

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

707

options.remove = True

708

709

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

685

719

686

clientnames = options.client

720

687

721

if options.debug:

722

log.setLevel(logging.DEBUG)

723

724

688

try:

725

689

bus = dbus.SystemBus()

726

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

727

server_path)

728

690

mandos_dbus_objc = bus.get_object(busname, server_path)

729

691

except dbus.exceptions.DBusException:

730

692

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

743

705

dbus_filter = NullFilter()

744

706

try:

745

707

dbus_logger.addFilter(dbus_filter)

746

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

747

server_path, dbus.OBJECT_MANAGER_IFACE)

748

708

mandos_clients = {path: ifs_and_props[client_interface]

749

709

for path, ifs_and_props in

750

710

mandos_serv_object_manager

762

722

clients = {}

763

723

764

724

if not clientnames:

765

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

766

busname, str(path)) and False) or

767

bus.get_object(busname, path): properties

725

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

768

726

for path, properties in mandos_clients.items()}

769

727

else:

770

728

for name in clientnames:

771

729

for path, client in mandos_clients.items():

772

730

if client["Name"] == name:

773

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

774

busname, str(path))

775

731

client_objc = bus.get_object(busname, path)

776

732

clients[client_objc] = client

777

733

break

841

797

self.attributes = attributes

842

798

self.attributes["Name"] = name

843

799

self.calls = []

844

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

845

testcase.assertEqual(interface, client_interface)

846

testcase.assertEqual(dbus_interface,

847

dbus.PROPERTIES_IFACE)

848

self.attributes[propname] = value

849

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

850

testcase.assertEqual(interface, client_interface)

851

testcase.assertEqual(dbus_interface,

852

dbus.PROPERTIES_IFACE)

853

return self.attributes[propname]

800

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

801

testcase.assertEqual(interface, client_interface)

802

testcase.assertEqual(dbus_interface,

803

dbus.PROPERTIES_IFACE)

804

self.attributes[property] = value

805

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

806

testcase.assertEqual(interface, client_interface)

807

testcase.assertEqual(dbus_interface,

808

dbus.PROPERTIES_IFACE)

809

return self.attributes[property]

854

810

def Approve(self, approve, dbus_interface):

855

811

testcase.assertEqual(dbus_interface, client_interface)

856

812

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

912

868

913

869

class TestPrintTableCmd(TestCmd):

914

870

def test_normal(self):

915

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

871

output = PrintTableCmd().output(self.clients)

916

872

expected_output = """

917

873

Name Enabled Timeout Last Successful Check

918

874

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

920

876

"""[1:-1]

921

877

self.assertEqual(output, expected_output)

922

878

def test_verbose(self):

923

output = PrintTableCmd(verbose=True).output(

924

self.clients.values())

879

output = PrintTableCmd(verbose=True).output(self.clients)

925

880

expected_output = """

926

881

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

927

882

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

929

884

"""[1:-1]

930

885

self.assertEqual(output, expected_output)

931

886

def test_one_client(self):

932

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

887

output = PrintTableCmd().output(self.one_client)

933

888

expected_output = """

934

889

Name Enabled Timeout Last Successful Check

935

890

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

1079

1034

for value_to_set, value_to_get in zip(self.values_to_set,

1080

1035

values_to_get):

1081

1036

for client in self.clients:

1082

old_value = client.attributes[self.propname]

1037

old_value = client.attributes[self.property]

1083

1038

self.assertNotIsInstance(old_value, Unique)

1084

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

1039

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

1085

1040

self.run_command(value_to_set, self.clients)

1086

1041

for client in self.clients:

1087

value = client.attributes[self.propname]

1042

value = client.attributes[self.property]

1088

1043

self.assertNotIsInstance(value, Unique)

1089

1044

self.assertEqual(value, value_to_get)

1090

1045

def run_command(self, value, clients):

1092

1047

1093

1048

class TestBumpTimeoutCmd(TestPropertyCmd):

1094

1049

command = BumpTimeoutCmd

1095

propname = "LastCheckedOK"

1050

property = "LastCheckedOK"

1096

1051

values_to_set = [""]

1097

1052

1098

1053

class TestStartCheckerCmd(TestPropertyCmd):

1099

1054

command = StartCheckerCmd

1100

propname = "CheckerRunning"

1055

property = "CheckerRunning"

1101

1056

values_to_set = [dbus.Boolean(True)]

1102

1057

1103

1058

class TestStopCheckerCmd(TestPropertyCmd):

1104

1059

command = StopCheckerCmd

1105

propname = "CheckerRunning"

1060

property = "CheckerRunning"

1106

1061

values_to_set = [dbus.Boolean(False)]

1107

1062

1108

1063

class TestApproveByDefaultCmd(TestPropertyCmd):

1109

1064

command = ApproveByDefaultCmd

1110

propname = "ApprovedByDefault"

1065

property = "ApprovedByDefault"

1111

1066

values_to_set = [dbus.Boolean(True)]

1112

1067

1113

1068

class TestDenyByDefaultCmd(TestPropertyCmd):

1114

1069

command = DenyByDefaultCmd

1115

propname = "ApprovedByDefault"

1070

property = "ApprovedByDefault"

1116

1071

values_to_set = [dbus.Boolean(False)]

1117

1072

1118

1073

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1127

1082

1128

1083

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1129

1084

command = SetCheckerCmd

1130

propname = "Checker"

1085

property = "Checker"

1131

1086

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

1132

1087

1133

1088

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1134

1089

command = SetHostCmd

1135

propname = "Host"

1090

property = "Host"

1136

1091

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

1137

1092

1138

1093

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1139

1094

command = SetSecretCmd

1140

propname = "Secret"

1141

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

1095

property = "Secret"

1096

values_to_set = [open("/dev/null", "rb"),

1142

1097

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

1143

1098

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

1144

1099

1145

1100

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1146

1101

command = SetTimeoutCmd

1147

propname = "Timeout"

1102

property = "Timeout"

1148

1103

values_to_set = [datetime.timedelta(),

1149

1104

datetime.timedelta(minutes=5),

1150

1105

datetime.timedelta(seconds=1),

1154

1109

1155

1110

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1156

1111

command = SetExtendedTimeoutCmd

1157

propname = "ExtendedTimeout"

1112

property = "ExtendedTimeout"

1158

1113

values_to_set = [datetime.timedelta(),

1159

1114

datetime.timedelta(minutes=5),

1160

1115

datetime.timedelta(seconds=1),

1164

1119

1165

1120

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1166

1121

command = SetIntervalCmd

1167

propname = "Interval"

1122

property = "Interval"

1168

1123

values_to_set = [datetime.timedelta(),

1169

1124

datetime.timedelta(minutes=5),

1170

1125

datetime.timedelta(seconds=1),

1174

1129

1175

1130

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1176

1131

command = SetApprovalDelayCmd

1177

propname = "ApprovalDelay"

1132

property = "ApprovalDelay"

1178

1133

values_to_set = [datetime.timedelta(),

1179

1134

datetime.timedelta(minutes=5),

1180

1135

datetime.timedelta(seconds=1),

1184

1139

1185

1140

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1186

1141

command = SetApprovalDurationCmd

1187

propname = "ApprovalDuration"

1142

property = "ApprovalDuration"

1188

1143

values_to_set = [datetime.timedelta(),

1189

1144

datetime.timedelta(minutes=5),

1190

1145

datetime.timedelta(seconds=1),

1200

1155

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

1201

1156

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

1202

1157

options = self.parser.parse_args(args)

1203

check_option_syntax(self.parser, options)

1204

1158

commands = commands_from_options(options)

1205

1159

self.assertEqual(len(commands), 1)

1206

1160

command = commands[0]

1215

1169

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

1216

1170

verbose=True)

1217

1171

1218

def test_print_table_verbose_short(self):

1219

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

1220

verbose=True)

1221

1222

1172

def test_enable(self):

1223

1173

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

1224

1174

1225

def test_enable_short(self):

1226

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

1227

1228

1175

def test_disable(self):

1229

1176

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

1230

1177

DisableCmd)

1231

1178

1232

def test_disable_short(self):

1233

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

1234

1235

1179

def test_bump_timeout(self):

1236

1180

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

1237

1181

BumpTimeoutCmd)

1238

1182

1239

def test_bump_timeout_short(self):

1240

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

1241

1242

1183

def test_start_checker(self):

1243

1184

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

1244

1185

StartCheckerCmd)

1251

1192

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

1252

1193

RemoveCmd)

1253

1194

1254

def test_remove_short(self):

1255

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

1256

1257

1195

def test_checker(self):

1258

1196

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

1259

1197

SetCheckerCmd, value_to_set=":")

1262

1200

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

1263

1201

SetCheckerCmd, value_to_set="")

1264

1202

1265

def test_checker_short(self):

1266

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

1267

SetCheckerCmd, value_to_set=":")

1268

1269

1203

def test_timeout(self):

1270

1204

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

1271

1205

SetTimeoutCmd,

1272

1206

value_to_set=300000)

1273

1207

1274

def test_timeout_short(self):

1275

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

1276

SetTimeoutCmd,

1277

value_to_set=300000)

1278

1279

1208

def test_extended_timeout(self):

1280

1209

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

1281

1210

"foo"],

1287

1216

SetIntervalCmd,

1288

1217

value_to_set=120000)

1289

1218

1290

def test_interval_short(self):

1291

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

1292

SetIntervalCmd,

1293

value_to_set=120000)

1294

1295

1219

def test_approve_by_default(self):

1296

1220

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

1297

1221

ApproveByDefaultCmd)

1315

1239

"foo"], SetHostCmd,

1316

1240

value_to_set="foo.example.org")

1317

1241

1318

def test_host_short(self):

1319

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

1320

"foo"], SetHostCmd,

1321

value_to_set="foo.example.org")

1322

1323

1242

def test_secret_devnull(self):

1324

1243

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

1325

1244

"foo"], SetSecretCmd,

1334

1253

"foo"], SetSecretCmd,

1335

1254

value_to_set=value)

1336

1255

1337

def test_secret_devnull_short(self):

1338

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

1339

SetSecretCmd, value_to_set=b"")

1340

1341

def test_secret_tempfile_short(self):

1342

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

1343

value = b"secret\0xyzzy\nbar"

1344

f.write(value)

1345

f.seek(0)

1346

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

1347

SetSecretCmd,

1348

value_to_set=value)

1349

1350

1256

def test_approve(self):

1351

1257

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

1352

1258

ApproveCmd)

1353

1259

1354

def test_approve_short(self):

1355

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

1356

1357

1260

def test_deny(self):

1358

1261

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

1359

1262

1360

def test_deny_short(self):

1361

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

1362

1363

1263

def test_dump_json(self):

1364

1264

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

1365

1265

1367

1267

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

1368

1268

IsEnabledCmd)

1369

1269

1370

def test_is_enabled_short(self):

1371

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

1372

1373

def test_deny_before_remove(self):

1374

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

1375

check_option_syntax(self.parser, options)

1376

commands = commands_from_options(options)

1377

self.assertEqual(len(commands), 2)

1378

self.assertIsInstance(commands[0], DenyCmd)

1379

self.assertIsInstance(commands[1], RemoveCmd)

1380

1381

def test_deny_before_remove_reversed(self):

1382

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

1383

check_option_syntax(self.parser, options)

1384

commands = commands_from_options(options)

1385

self.assertEqual(len(commands), 2)

1386

self.assertIsInstance(commands[0], DenyCmd)

1387

self.assertIsInstance(commands[1], RemoveCmd)

1388

1389

1390

class Test_check_option_syntax(unittest.TestCase):

1391

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

1392

# check_option_syntax()

1393

actions = {

1394

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

1395

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

1396

# never used

1397

"enable": True,

1398

"disable": True,

1399

"bump_timeout": True,

1400

"start_checker": True,

1401

"stop_checker": True,

1402

"is_enabled": True,

1403

"remove": True,

1404

"checker": "x",

1405

"timeout": datetime.timedelta(),

1406

"extended_timeout": datetime.timedelta(),

1407

"interval": datetime.timedelta(),

1408

"approved_by_default": True,

1409

"approval_delay": datetime.timedelta(),

1410

"approval_duration": datetime.timedelta(),

1411

"host": "x",

1412

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

1413

"approve": True,

1414

"deny": True,

1415

}

1416

1417

def setUp(self):

1418

self.parser = argparse.ArgumentParser()

1419

add_command_line_options(self.parser)

1420

1421

@contextlib.contextmanager

1422

def assertParseError(self):

1423

with self.assertRaises(SystemExit) as e:

1424

with self.temporarily_suppress_stderr():

1425

yield

1426

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

1427

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

1428

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

1429

1430

@staticmethod

1431

@contextlib.contextmanager

1432

def temporarily_suppress_stderr():

1433

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

1434

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

1435

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

1436

os.close(null)

1437

try:

1438

yield

1439

finally:

1440

# restore stderr

1441

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

1442

os.close(stderrcopy)

1443

1444

def check_option_syntax(self, options):

1445

check_option_syntax(self.parser, options)

1446

1447

def test_actions_requires_client_or_all(self):

1448

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

1449

options = self.parser.parse_args()

1450

setattr(options, action, value)

1451

with self.assertParseError():

1452

self.check_option_syntax(options)

1453

1454

def test_actions_conflicts_with_verbose(self):

1455

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

1456

options = self.parser.parse_args()

1457

setattr(options, action, value)

1458

options.verbose = True

1459

with self.assertParseError():

1460

self.check_option_syntax(options)

1461

1462

def test_dump_json_conflicts_with_verbose(self):

1463

options = self.parser.parse_args()

1464

options.dump_json = True

1465

options.verbose = True

1466

with self.assertParseError():

1467

self.check_option_syntax(options)

1468

1469

def test_dump_json_conflicts_with_action(self):

1470

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

1471

options = self.parser.parse_args()

1472

setattr(options, action, value)

1473

options.dump_json = True

1474

with self.assertParseError():

1475

self.check_option_syntax(options)

1476

1477

def test_all_can_not_be_alone(self):

1478

options = self.parser.parse_args()

1479

options.all = True

1480

with self.assertParseError():

1481

self.check_option_syntax(options)

1482

1483

def test_all_is_ok_with_any_action(self):

1484

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

1485

options = self.parser.parse_args()

1486

setattr(options, action, value)

1487

options.all = True

1488

self.check_option_syntax(options)

1489

1490

def test_is_enabled_fails_without_client(self):

1491

options = self.parser.parse_args()

1492

options.is_enabled = True

1493

with self.assertParseError():

1494

self.check_option_syntax(options)

1495

1496

def test_is_enabled_works_with_one_client(self):

1497

options = self.parser.parse_args()

1498

options.is_enabled = True

1499

options.client = ["foo"]

1500

self.check_option_syntax(options)

1501

1502

def test_is_enabled_fails_with_two_clients(self):

1503

options = self.parser.parse_args()

1504

options.is_enabled = True

1505

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

1506

with self.assertParseError():

1507

self.check_option_syntax(options)

1508

1509

def test_remove_can_only_be_combined_with_action_deny(self):

1510

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

1511

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

1512

continue

1513

options = self.parser.parse_args()

1514

setattr(options, action, value)

1515

options.all = True

1516

options.remove = True

1517

with self.assertParseError():

1518

self.check_option_syntax(options)

1519

1520

1270

1521

1271

1522

1272

def should_only_run_tests():