749
746
raise NotImplementedError()
752
class Enable(Property):
749
class Enable(PropertySetter):
753
750
propname = "Enabled"
754
751
value_to_set = dbus.Boolean(True)
757
class Disable(Property):
754
class Disable(PropertySetter):
758
755
propname = "Enabled"
759
756
value_to_set = dbus.Boolean(False)
762
class BumpTimeout(Property):
759
class BumpTimeout(PropertySetter):
763
760
propname = "LastCheckedOK"
764
761
value_to_set = ""
767
class StartChecker(Property):
768
propname = "CheckerRunning"
769
value_to_set = dbus.Boolean(True)
772
class StopChecker(Property):
773
propname = "CheckerRunning"
774
value_to_set = dbus.Boolean(False)
777
class ApproveByDefault(Property):
778
propname = "ApprovedByDefault"
779
value_to_set = dbus.Boolean(True)
782
class DenyByDefault(Property):
783
propname = "ApprovedByDefault"
784
value_to_set = dbus.Boolean(False)
787
class PropertyValue(Property):
788
"Abstract class for Property recieving a value as argument"
764
class StartChecker(PropertySetter):
765
propname = "CheckerRunning"
766
value_to_set = dbus.Boolean(True)
769
class StopChecker(PropertySetter):
770
propname = "CheckerRunning"
771
value_to_set = dbus.Boolean(False)
774
class ApproveByDefault(PropertySetter):
775
propname = "ApprovedByDefault"
776
value_to_set = dbus.Boolean(True)
779
class DenyByDefault(PropertySetter):
780
propname = "ApprovedByDefault"
781
value_to_set = dbus.Boolean(False)
784
class PropertySetterValue(PropertySetter):
785
"""Abstract class for PropertySetter recieving a value as
786
constructor argument instead of a class attribute."""
789
787
def __init__(self, value):
790
788
self.value_to_set = value
793
class SetChecker(PropertyValue):
791
class SetChecker(PropertySetterValue):
794
792
propname = "Checker"
797
class SetHost(PropertyValue):
795
class SetHost(PropertySetterValue):
798
796
propname = "Host"
801
class SetSecret(PropertyValue):
799
class SetSecret(PropertySetterValue):
802
800
propname = "Secret"
1108
1107
class Test_get_managed_objects(TestCaseWithAssertLogs):
1109
1108
def test_calls_and_returns_GetManagedObjects(self):
1110
managed_objects = {"/clients/foo": { "Name": "foo"}}
1111
class MockObjectManager(object):
1109
managed_objects = {"/clients/client": { "Name": "client"}}
1110
class ObjectManagerStub(object):
1112
1111
def GetManagedObjects(self):
1113
1112
return managed_objects
1114
retval = get_managed_objects(MockObjectManager())
1113
retval = get_managed_objects(ObjectManagerStub())
1115
1114
self.assertDictEqual(managed_objects, retval)
1117
1116
def test_logs_and_exits_on_dbus_error(self):
1118
1117
dbus_logger = logging.getLogger("dbus.proxies")
1120
class MockObjectManagerFailing(object):
1119
class ObjectManagerFailingStub(object):
1121
1120
def GetManagedObjects(self):
1122
1121
dbus_logger.error("Test")
1123
1122
raise dbus.exceptions.DBusException("Test")
1174
1173
self.assertEqual(value, getattr(command, key))
1176
1175
def test_is_enabled_short(self):
1177
self.assert_command_from_args(["-V", "foo"],
1176
self.assert_command_from_args(["-V", "client"],
1178
1177
command.IsEnabled)
1180
1179
def test_approve(self):
1181
self.assert_command_from_args(["--approve", "foo"],
1180
self.assert_command_from_args(["--approve", "client"],
1182
1181
command.Approve)
1184
1183
def test_approve_short(self):
1185
self.assert_command_from_args(["-A", "foo"], command.Approve)
1184
self.assert_command_from_args(["-A", "client"],
1187
1187
def test_deny(self):
1188
self.assert_command_from_args(["--deny", "foo"], command.Deny)
1188
self.assert_command_from_args(["--deny", "client"],
1190
1191
def test_deny_short(self):
1191
self.assert_command_from_args(["-D", "foo"], command.Deny)
1192
self.assert_command_from_args(["-D", "client"], command.Deny)
1193
1194
def test_remove(self):
1194
self.assert_command_from_args(["--remove", "foo"],
1195
self.assert_command_from_args(["--remove", "client"],
1195
1196
command.Remove)
1197
1198
def test_deny_before_remove(self):
1198
1199
options = self.parser.parse_args(["--deny", "--remove",
1200
1201
check_option_syntax(self.parser, options)
1201
1202
commands = commands_from_options(options)
1202
1203
self.assertEqual(2, len(commands))
1213
1214
self.assertIsInstance(commands[1], command.Remove)
1215
1216
def test_remove_short(self):
1216
self.assert_command_from_args(["-r", "foo"], command.Remove)
1217
self.assert_command_from_args(["-r", "client"],
1218
1220
def test_dump_json(self):
1219
1221
self.assert_command_from_args(["--dump-json"],
1220
1222
command.DumpJSON)
1222
1224
def test_enable(self):
1223
self.assert_command_from_args(["--enable", "foo"],
1225
self.assert_command_from_args(["--enable", "client"],
1224
1226
command.Enable)
1226
1228
def test_enable_short(self):
1227
self.assert_command_from_args(["-e", "foo"], command.Enable)
1229
self.assert_command_from_args(["-e", "client"],
1229
1232
def test_disable(self):
1230
self.assert_command_from_args(["--disable", "foo"],
1233
self.assert_command_from_args(["--disable", "client"],
1231
1234
command.Disable)
1233
1236
def test_disable_short(self):
1234
self.assert_command_from_args(["-d", "foo"], command.Disable)
1237
self.assert_command_from_args(["-d", "client"],
1236
1240
def test_bump_timeout(self):
1237
self.assert_command_from_args(["--bump-timeout", "foo"],
1241
self.assert_command_from_args(["--bump-timeout", "client"],
1238
1242
command.BumpTimeout)
1240
1244
def test_bump_timeout_short(self):
1241
self.assert_command_from_args(["-b", "foo"],
1245
self.assert_command_from_args(["-b", "client"],
1242
1246
command.BumpTimeout)
1244
1248
def test_start_checker(self):
1245
self.assert_command_from_args(["--start-checker", "foo"],
1249
self.assert_command_from_args(["--start-checker", "client"],
1246
1250
command.StartChecker)
1248
1252
def test_stop_checker(self):
1249
self.assert_command_from_args(["--stop-checker", "foo"],
1253
self.assert_command_from_args(["--stop-checker", "client"],
1250
1254
command.StopChecker)
1252
1256
def test_approve_by_default(self):
1253
self.assert_command_from_args(["--approve-by-default", "foo"],
1257
self.assert_command_from_args(["--approve-by-default",
1254
1259
command.ApproveByDefault)
1256
1261
def test_deny_by_default(self):
1257
self.assert_command_from_args(["--deny-by-default", "foo"],
1262
self.assert_command_from_args(["--deny-by-default", "client"],
1258
1263
command.DenyByDefault)
1260
1265
def test_checker(self):
1261
self.assert_command_from_args(["--checker", ":", "foo"],
1266
self.assert_command_from_args(["--checker", ":", "client"],
1262
1267
command.SetChecker,
1263
1268
value_to_set=":")
1265
1270
def test_checker_empty(self):
1266
self.assert_command_from_args(["--checker", "", "foo"],
1271
self.assert_command_from_args(["--checker", "", "client"],
1267
1272
command.SetChecker,
1268
1273
value_to_set="")
1270
1275
def test_checker_short(self):
1271
self.assert_command_from_args(["-c", ":", "foo"],
1276
self.assert_command_from_args(["-c", ":", "client"],
1272
1277
command.SetChecker,
1273
1278
value_to_set=":")
1275
1280
def test_host(self):
1276
self.assert_command_from_args(["--host", "foo.example.org",
1277
"foo"], command.SetHost,
1278
value_to_set="foo.example.org")
1281
self.assert_command_from_args(
1282
["--host", "client.example.org", "client"],
1283
command.SetHost, value_to_set="client.example.org")
1280
1285
def test_host_short(self):
1281
self.assert_command_from_args(["-H", "foo.example.org",
1282
"foo"], command.SetHost,
1283
value_to_set="foo.example.org")
1286
self.assert_command_from_args(
1287
["-H", "client.example.org", "client"], command.SetHost,
1288
value_to_set="client.example.org")
1285
1290
def test_secret_devnull(self):
1286
1291
self.assert_command_from_args(["--secret", os.path.devnull,
1287
"foo"], command.SetSecret,
1292
"client"], command.SetSecret,
1288
1293
value_to_set=b"")
1290
1295
def test_secret_tempfile(self):
1306
1312
value = b"secret\0xyzzy\nbar"
1309
self.assert_command_from_args(["-s", f.name, "foo"],
1315
self.assert_command_from_args(["-s", f.name, "client"],
1310
1316
command.SetSecret,
1311
1317
value_to_set=value)
1313
1319
def test_timeout(self):
1314
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1320
self.assert_command_from_args(["--timeout", "PT5M", "client"],
1315
1321
command.SetTimeout,
1316
1322
value_to_set=300000)
1318
1324
def test_timeout_short(self):
1319
self.assert_command_from_args(["-t", "PT5M", "foo"],
1325
self.assert_command_from_args(["-t", "PT5M", "client"],
1320
1326
command.SetTimeout,
1321
1327
value_to_set=300000)
1323
1329
def test_extended_timeout(self):
1324
1330
self.assert_command_from_args(["--extended-timeout", "PT15M",
1326
1332
command.SetExtendedTimeout,
1327
1333
value_to_set=900000)
1329
1335
def test_interval(self):
1330
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1331
command.SetInterval,
1336
self.assert_command_from_args(["--interval", "PT2M",
1337
"client"], command.SetInterval,
1332
1338
value_to_set=120000)
1334
1340
def test_interval_short(self):
1335
self.assert_command_from_args(["-i", "PT2M", "foo"],
1341
self.assert_command_from_args(["-i", "PT2M", "client"],
1336
1342
command.SetInterval,
1337
1343
value_to_set=120000)
1339
1345
def test_approval_delay(self):
1340
1346
self.assert_command_from_args(["--approval-delay", "PT30S",
1342
1348
command.SetApprovalDelay,
1343
1349
value_to_set=30000)
1345
1351
def test_approval_duration(self):
1346
1352
self.assert_command_from_args(["--approval-duration", "PT1S",
1348
1354
command.SetApprovalDuration,
1349
1355
value_to_set=1000)
1433
1439
LastCheckerStatus=-2)
1434
1440
self.clients = collections.OrderedDict(
1436
("/clients/foo", self.client.attributes),
1437
("/clients/barbar", self.other_client.attributes),
1442
(self.client.__dbus_object_path__,
1443
self.client.attributes),
1444
(self.other_client.__dbus_object_path__,
1445
self.other_client.attributes),
1439
self.one_client = {"/clients/foo": self.client.attributes}
1447
self.one_client = {self.client.__dbus_object_path__:
1448
self.client.attributes}
1452
class MockBus(object):
1445
1454
def get_object(client_bus_name, path):
1446
1455
self.assertEqual(dbus_busname, client_bus_name)
1448
# Note: "self" here is the TestCmd instance, not
1449
# the Bus instance, since this is a static method!
1450
"/clients/foo": self.client,
1451
"/clients/barbar": self.other_client,
1456
# Note: "self" here is the TestCmd instance, not the
1457
# MockBus instance, since this is a static method!
1458
if path == self.client.__dbus_object_path__:
1460
elif path == self.other_client.__dbus_object_path__:
1461
return self.other_client
1456
1465
class TestBaseCommands(TestCommand):
1717
1726
self.command().run(clients, self.bus)
1720
class TestEnableCmd(TestPropertyCmd):
1729
class TestEnableCmd(TestPropertySetterCmd):
1721
1730
command = command.Enable
1722
1731
propname = "Enabled"
1723
1732
values_to_set = [dbus.Boolean(True)]
1726
class TestDisableCmd(TestPropertyCmd):
1735
class TestDisableCmd(TestPropertySetterCmd):
1727
1736
command = command.Disable
1728
1737
propname = "Enabled"
1729
1738
values_to_set = [dbus.Boolean(False)]
1732
class TestBumpTimeoutCmd(TestPropertyCmd):
1741
class TestBumpTimeoutCmd(TestPropertySetterCmd):
1733
1742
command = command.BumpTimeout
1734
1743
propname = "LastCheckedOK"
1735
1744
values_to_set = [""]
1738
class TestStartCheckerCmd(TestPropertyCmd):
1747
class TestStartCheckerCmd(TestPropertySetterCmd):
1739
1748
command = command.StartChecker
1740
1749
propname = "CheckerRunning"
1741
1750
values_to_set = [dbus.Boolean(True)]
1744
class TestStopCheckerCmd(TestPropertyCmd):
1753
class TestStopCheckerCmd(TestPropertySetterCmd):
1745
1754
command = command.StopChecker
1746
1755
propname = "CheckerRunning"
1747
1756
values_to_set = [dbus.Boolean(False)]
1750
class TestApproveByDefaultCmd(TestPropertyCmd):
1759
class TestApproveByDefaultCmd(TestPropertySetterCmd):
1751
1760
command = command.ApproveByDefault
1752
1761
propname = "ApprovedByDefault"
1753
1762
values_to_set = [dbus.Boolean(True)]
1756
class TestDenyByDefaultCmd(TestPropertyCmd):
1765
class TestDenyByDefaultCmd(TestPropertySetterCmd):
1757
1766
command = command.DenyByDefault
1758
1767
propname = "ApprovedByDefault"
1759
1768
values_to_set = [dbus.Boolean(False)]
1762
class TestPropertyValueCmd(TestPropertyCmd):
1763
"""Abstract class for tests of PropertyValueCmd classes"""
1771
class TestPropertySetterValueCmd(TestPropertySetterCmd):
1772
"""Abstract class for tests of PropertySetterValueCmd classes"""
1765
1774
def runTest(self):
1766
if type(self) is TestPropertyValueCmd:
1775
if type(self) is TestPropertySetterValueCmd:
1768
return super(TestPropertyValueCmd, self).runTest()
1777
return super(TestPropertySetterValueCmd, self).runTest()
1770
1779
def run_command(self, value, clients):
1771
1780
self.command(value).run(clients, self.bus)
1774
class TestSetCheckerCmd(TestPropertyValueCmd):
1783
class TestSetCheckerCmd(TestPropertySetterValueCmd):
1775
1784
command = command.SetChecker
1776
1785
propname = "Checker"
1777
1786
values_to_set = ["", ":", "fping -q -- %s"]
1780
class TestSetHostCmd(TestPropertyValueCmd):
1789
class TestSetHostCmd(TestPropertySetterValueCmd):
1781
1790
command = command.SetHost
1782
1791
propname = "Host"
1783
values_to_set = ["192.0.2.3", "foo.example.org"]
1786
class TestSetSecretCmd(TestPropertyValueCmd):
1792
values_to_set = ["192.0.2.3", "client.example.org"]
1795
class TestSetSecretCmd(TestPropertySetterValueCmd):
1787
1796
command = command.SetSecret
1788
1797
propname = "Secret"
1789
1798
values_to_set = [io.BytesIO(b""),
1790
1799
io.BytesIO(b"secret\0xyzzy\nbar")]
1791
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1794
class TestSetTimeoutCmd(TestPropertyValueCmd):
1800
values_to_get = [f.getvalue() for f in values_to_set]
1803
class TestSetTimeoutCmd(TestPropertySetterValueCmd):
1795
1804
command = command.SetTimeout
1796
1805
propname = "Timeout"
1797
1806
values_to_set = [datetime.timedelta(),