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