To get this branch, use:
bzr branch
http://bzr.recompile.se/loggerhead/mandos/trunk
« back to all changes in this revision
Viewing changes to mandos-ctl
- browse files at revision 1086
- compare with another revision
- download diff
- download tarball
- view history from revision 1086
- Committer: Teddy Hogeborn
- Date: 2019-03-18 22:29:25 UTC
- Revision ID: teddy@recompile.se-20190318222925-jvhek84dgcfgj6g3
mandos-ctl: Refactor tests
* mandos-ctl: Where the clients names "foo" and "barbar" do not refer
to the actual mock clients in the TestCommand class,
change all occurrences of these names to "client1" and
"client2" (or just "client" when only one is used) .
Also change all test doubles to use correct terminology;
some things called mocks are actually stubs or spies,
and rename all true mocks to have "mock" in their names.
Also eliminate duplicate values in tests; derive values
from previously defined values whenever possible.
* mandos-ctl: Where the clients names "foo" and "barbar" do not refer
to the actual mock clients in the TestCommand class,
change all occurrences of these names to "client1" and
"client2" (or just "client" when only one is used) .
Also change all test doubles to use correct terminology;
some things called mocks are actually stubs or spies,
and rename all true mocks to have "mock" in their names.
Also eliminate duplicate values in tests; derive values
from previously defined values whenever possible.
- files modified:
- TODO
added
removed
mandos-ctl
61
61
62
62
if sys.version_info.major == 2:
63
63
str = unicode
64
import StringIO
65
io.StringIO = StringIO.StringIO
64
66
65
67
locale.setlocale(locale.LC_ALL, "")
66
68
125
127
log.critical("Client not found on server: %r", name)
126
128
sys.exit(1)
127
129
128
# Run all commands on clients
129
130
commands = commands_from_options(options)
131
130
132
for command in commands:
131
133
command.run(clients, bus, mandos_serv)
132
134
232
234
>>> rfc3339_duration_to_delta("")
233
235
Traceback (most recent call last):
234
236
...
235
ValueError: Invalid RFC 3339 duration: u''
237
ValueError: Invalid RFC 3339 duration: ""
236
238
>>> # Must start with "P":
237
239
>>> rfc3339_duration_to_delta("1D")
238
240
Traceback (most recent call last):
239
241
...
240
ValueError: Invalid RFC 3339 duration: u'1D'
242
ValueError: Invalid RFC 3339 duration: "1D"
241
243
>>> # Must use correct order
242
244
>>> rfc3339_duration_to_delta("PT1S2M")
243
245
Traceback (most recent call last):
244
246
...
245
ValueError: Invalid RFC 3339 duration: u'PT1S2M'
247
ValueError: Invalid RFC 3339 duration: "PT1S2M"
246
248
>>> # Time needs time marker
247
249
>>> rfc3339_duration_to_delta("P1H2S")
248
250
Traceback (most recent call last):
249
251
...
250
ValueError: Invalid RFC 3339 duration: u'P1H2S'
252
ValueError: Invalid RFC 3339 duration: "P1H2S"
251
253
>>> # Weeks can not be combined with anything else
252
254
>>> rfc3339_duration_to_delta("P1D2W")
253
255
Traceback (most recent call last):
254
256
...
255
ValueError: Invalid RFC 3339 duration: u'P1D2W'
257
ValueError: Invalid RFC 3339 duration: "P1D2W"
256
258
>>> rfc3339_duration_to_delta("P2W2H")
257
259
Traceback (most recent call last):
258
260
...
259
ValueError: Invalid RFC 3339 duration: u'P2W2H'
261
ValueError: Invalid RFC 3339 duration: "P2W2H"
260
262
"""
261
263
262
264
# Parsing an RFC 3339 duration with regular expressions is not
333
335
break
334
336
else:
335
337
# No currently valid tokens were found
336
raise ValueError("Invalid RFC 3339 duration: {!r}"
338
raise ValueError("Invalid RFC 3339 duration: \"{}\""
337
339
.format(duration))
338
340
# End token found
339
341
return value
477
479
commands = []
478
480
479
481
if options.is_enabled:
480
commands.append(IsEnabledCmd())
482
commands.append(command.IsEnabled())
481
483
482
484
if options.approve:
483
commands.append(ApproveCmd())
485
commands.append(command.Approve())
484
486
485
487
if options.deny:
486
commands.append(DenyCmd())
488
commands.append(command.Deny())
487
489
488
490
if options.remove:
489
commands.append(RemoveCmd())
491
commands.append(command.Remove())
490
492
491
493
if options.dump_json:
492
commands.append(DumpJSONCmd())
494
commands.append(command.DumpJSON())
493
495
494
496
if options.enable:
495
commands.append(EnableCmd())
497
commands.append(command.Enable())
496
498
497
499
if options.disable:
498
commands.append(DisableCmd())
500
commands.append(command.Disable())
499
501
500
502
if options.bump_timeout:
501
commands.append(BumpTimeoutCmd())
503
commands.append(command.BumpTimeout())
502
504
503
505
if options.start_checker:
504
commands.append(StartCheckerCmd())
506
commands.append(command.StartChecker())
505
507
506
508
if options.stop_checker:
507
commands.append(StopCheckerCmd())
509
commands.append(command.StopChecker())
508
510
509
511
if options.approved_by_default is not None:
510
512
if options.approved_by_default:
511
commands.append(ApproveByDefaultCmd())
513
commands.append(command.ApproveByDefault())
512
514
else:
513
commands.append(DenyByDefaultCmd())
515
commands.append(command.DenyByDefault())
514
516
515
517
if options.checker is not None:
516
commands.append(SetCheckerCmd(options.checker))
518
commands.append(command.SetChecker(options.checker))
517
519
518
520
if options.host is not None:
519
commands.append(SetHostCmd(options.host))
521
commands.append(command.SetHost(options.host))
520
522
521
523
if options.secret is not None:
522
commands.append(SetSecretCmd(options.secret))
524
commands.append(command.SetSecret(options.secret))
523
525
524
526
if options.timeout is not None:
525
commands.append(SetTimeoutCmd(options.timeout))
527
commands.append(command.SetTimeout(options.timeout))
526
528
527
529
if options.extended_timeout:
528
530
commands.append(
529
SetExtendedTimeoutCmd(options.extended_timeout))
531
command.SetExtendedTimeout(options.extended_timeout))
530
532
531
533
if options.interval is not None:
532
commands.append(SetIntervalCmd(options.interval))
534
commands.append(command.SetInterval(options.interval))
533
535
534
536
if options.approval_delay is not None:
535
commands.append(SetApprovalDelayCmd(options.approval_delay))
537
commands.append(
538
command.SetApprovalDelay(options.approval_delay))
536
539
537
540
if options.approval_duration is not None:
538
541
commands.append(
539
SetApprovalDurationCmd(options.approval_duration))
542
command.SetApprovalDuration(options.approval_duration))
540
543
541
544
# If no command option has been given, show table of clients,
542
545
# optionally verbosely
543
546
if not commands:
544
commands.append(PrintTableCmd(verbose=options.verbose))
547
commands.append(command.PrintTable(verbose=options.verbose))
545
548
546
549
return commands
547
550
548
551
549
class Command(object):
550
"""Abstract class for commands"""
551
def run(self, clients, bus=None, mandos=None):
552
"""Normal commands should implement run_on_one_client(), but
553
commands which want to operate on all clients at the same time
554
can override this run() method instead."""
555
self.mandos = mandos
556
for clientpath, properties in clients.items():
557
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
558
dbus_busname, str(clientpath))
559
client = bus.get_object(dbus_busname, clientpath)
560
self.run_on_one_client(client, properties)
561
562
563
class IsEnabledCmd(Command):
564
def run(self, clients, bus=None, mandos=None):
565
client, properties = next(iter(clients.items()))
566
if self.is_enabled(client, properties):
567
sys.exit(0)
568
sys.exit(1)
569
def is_enabled(self, client, properties):
570
return properties["Enabled"]
571
572
573
class ApproveCmd(Command):
574
def run_on_one_client(self, client, properties):
575
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
576
client.__dbus_object_path__, client_dbus_interface)
577
client.Approve(dbus.Boolean(True),
578
dbus_interface=client_dbus_interface)
579
580
581
class DenyCmd(Command):
582
def run_on_one_client(self, client, properties):
583
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
584
client.__dbus_object_path__, client_dbus_interface)
585
client.Approve(dbus.Boolean(False),
586
dbus_interface=client_dbus_interface)
587
588
589
class RemoveCmd(Command):
590
def run_on_one_client(self, client, properties):
591
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", dbus_busname,
592
server_dbus_path, server_dbus_interface,
593
str(client.__dbus_object_path__))
594
self.mandos.RemoveClient(client.__dbus_object_path__)
595
596
597
class OutputCmd(Command):
598
"""Abstract class for commands outputting client details"""
599
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
600
"Created", "Interval", "Host", "KeyID",
601
"Fingerprint", "CheckerRunning", "LastEnabled",
602
"ApprovalPending", "ApprovedByDefault",
603
"LastApprovalRequest", "ApprovalDelay",
604
"ApprovalDuration", "Checker", "ExtendedTimeout",
605
"Expires", "LastCheckerStatus")
606
607
def run(self, clients, bus=None, mandos=None):
608
print(self.output(clients.values()))
609
610
def output(self, clients):
611
raise NotImplementedError()
612
613
614
class DumpJSONCmd(OutputCmd):
615
def output(self, clients):
616
data = {client["Name"]:
617
{key: self.dbus_boolean_to_bool(client[key])
618
for key in self.all_keywords}
619
for client in clients}
620
return json.dumps(data, indent=4, separators=(',', ': '))
621
622
@staticmethod
623
def dbus_boolean_to_bool(value):
624
if isinstance(value, dbus.Boolean):
625
value = bool(value)
626
return value
627
628
629
class PrintTableCmd(OutputCmd):
630
def __init__(self, verbose=False):
631
self.verbose = verbose
632
633
def output(self, clients):
634
default_keywords = ("Name", "Enabled", "Timeout",
635
"LastCheckedOK")
636
keywords = default_keywords
637
if self.verbose:
638
keywords = self.all_keywords
639
return str(self.TableOfClients(clients, keywords))
640
641
class TableOfClients(object):
642
tableheaders = {
643
"Name": "Name",
644
"Enabled": "Enabled",
645
"Timeout": "Timeout",
646
"LastCheckedOK": "Last Successful Check",
647
"LastApprovalRequest": "Last Approval Request",
648
"Created": "Created",
649
"Interval": "Interval",
650
"Host": "Host",
651
"Fingerprint": "Fingerprint",
652
"KeyID": "Key ID",
653
"CheckerRunning": "Check Is Running",
654
"LastEnabled": "Last Enabled",
655
"ApprovalPending": "Approval Is Pending",
656
"ApprovedByDefault": "Approved By Default",
657
"ApprovalDelay": "Approval Delay",
658
"ApprovalDuration": "Approval Duration",
659
"Checker": "Checker",
660
"ExtendedTimeout": "Extended Timeout",
661
"Expires": "Expires",
662
"LastCheckerStatus": "Last Checker Status",
663
}
664
665
def __init__(self, clients, keywords):
666
self.clients = clients
667
self.keywords = keywords
668
669
def __str__(self):
670
return "\n".join(self.rows())
671
672
if sys.version_info.major == 2:
673
__unicode__ = __str__
552
class command(object):
553
"""A namespace for command classes"""
554
555
class Base(object):
556
"""Abstract base class for commands"""
557
def run(self, clients, bus=None, mandos=None):
558
"""Normal commands should implement run_on_one_client(),
559
but commands which want to operate on all clients at the same time can
560
override this run() method instead.
561
"""
562
self.mandos = mandos
563
for clientpath, properties in clients.items():
564
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
565
dbus_busname, str(clientpath))
566
client = bus.get_object(dbus_busname, clientpath)
567
self.run_on_one_client(client, properties)
568
569
570
class IsEnabled(Base):
571
def run(self, clients, bus=None, mandos=None):
572
client, properties = next(iter(clients.items()))
573
if self.is_enabled(client, properties):
574
sys.exit(0)
575
sys.exit(1)
576
def is_enabled(self, client, properties):
577
return properties["Enabled"]
578
579
580
class Approve(Base):
581
def run_on_one_client(self, client, properties):
582
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
583
client.__dbus_object_path__,
584
client_dbus_interface)
585
client.Approve(dbus.Boolean(True),
586
dbus_interface=client_dbus_interface)
587
588
589
class Deny(Base):
590
def run_on_one_client(self, client, properties):
591
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
592
client.__dbus_object_path__,
593
client_dbus_interface)
594
client.Approve(dbus.Boolean(False),
595
dbus_interface=client_dbus_interface)
596
597
598
class Remove(Base):
599
def run_on_one_client(self, client, properties):
600
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)",
601
dbus_busname, server_dbus_path,
602
server_dbus_interface,
603
str(client.__dbus_object_path__))
604
self.mandos.RemoveClient(client.__dbus_object_path__)
605
606
607
class Output(Base):
608
"""Abstract class for commands outputting client details"""
609
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
610
"Created", "Interval", "Host", "KeyID",
611
"Fingerprint", "CheckerRunning",
612
"LastEnabled", "ApprovalPending",
613
"ApprovedByDefault", "LastApprovalRequest",
614
"ApprovalDelay", "ApprovalDuration",
615
"Checker", "ExtendedTimeout", "Expires",
616
"LastCheckerStatus")
617
618
619
class DumpJSON(Output):
620
def run(self, clients, bus=None, mandos=None):
621
data = {client["Name"]:
622
{key: self.dbus_boolean_to_bool(client[key])
623
for key in self.all_keywords}
624
for client in clients.values()}
625
print(json.dumps(data, indent=4, separators=(',', ': ')))
626
627
@staticmethod
628
def dbus_boolean_to_bool(value):
629
if isinstance(value, dbus.Boolean):
630
value = bool(value)
631
return value
632
633
634
class PrintTable(Output):
635
def __init__(self, verbose=False):
636
self.verbose = verbose
637
638
def run(self, clients, bus=None, mandos=None):
639
default_keywords = ("Name", "Enabled", "Timeout",
640
"LastCheckedOK")
641
keywords = default_keywords
642
if self.verbose:
643
keywords = self.all_keywords
644
print(self.TableOfClients(clients.values(), keywords))
645
646
class TableOfClients(object):
647
tableheaders = {
648
"Name": "Name",
649
"Enabled": "Enabled",
650
"Timeout": "Timeout",
651
"LastCheckedOK": "Last Successful Check",
652
"LastApprovalRequest": "Last Approval Request",
653
"Created": "Created",
654
"Interval": "Interval",
655
"Host": "Host",
656
"Fingerprint": "Fingerprint",
657
"KeyID": "Key ID",
658
"CheckerRunning": "Check Is Running",
659
"LastEnabled": "Last Enabled",
660
"ApprovalPending": "Approval Is Pending",
661
"ApprovedByDefault": "Approved By Default",
662
"ApprovalDelay": "Approval Delay",
663
"ApprovalDuration": "Approval Duration",
664
"Checker": "Checker",
665
"ExtendedTimeout": "Extended Timeout",
666
"Expires": "Expires",
667
"LastCheckerStatus": "Last Checker Status",
668
}
669
670
def __init__(self, clients, keywords):
671
self.clients = clients
672
self.keywords = keywords
673
674
674
def __str__(self):
675
return str(self).encode(locale.getpreferredencoding())
676
677
def rows(self):
678
format_string = self.row_formatting_string()
679
rows = [self.header_line(format_string)]
680
rows.extend(self.client_line(client, format_string)
681
for client in self.clients)
682
return rows
683
684
def row_formatting_string(self):
685
"Format string used to format table rows"
686
return " ".join("{{{key}:{width}}}".format(
687
width=max(len(self.tableheaders[key]),
688
*(len(self.string_from_client(client, key))
689
for client in self.clients)),
690
key=key)
691
for key in self.keywords)
692
693
def string_from_client(self, client, key):
694
return self.valuetostring(client[key], key)
695
696
@classmethod
697
def valuetostring(cls, value, keyword):
698
if isinstance(value, dbus.Boolean):
699
return "Yes" if value else "No"
700
if keyword in ("Timeout", "Interval", "ApprovalDelay",
701
"ApprovalDuration", "ExtendedTimeout"):
702
return cls.milliseconds_to_string(value)
703
return str(value)
704
705
def header_line(self, format_string):
706
return format_string.format(**self.tableheaders)
707
708
def client_line(self, client, format_string):
709
return format_string.format(
710
**{key: self.string_from_client(client, key)
711
for key in self.keywords})
712
713
@staticmethod
714
def milliseconds_to_string(ms):
715
td = datetime.timedelta(0, 0, 0, ms)
716
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
717
.format(days="{}T".format(td.days)
718
if td.days else "",
719
hours=td.seconds // 3600,
720
minutes=(td.seconds % 3600) // 60,
721
seconds=td.seconds % 60))
722
723
724
class PropertyCmd(Command):
725
"""Abstract class for Actions for setting one client property"""
726
727
def run_on_one_client(self, client, properties):
728
"""Set the Client's D-Bus property"""
729
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
730
client.__dbus_object_path__,
731
dbus.PROPERTIES_IFACE, client_dbus_interface,
732
self.propname, self.value_to_set
733
if not isinstance(self.value_to_set, dbus.Boolean)
734
else bool(self.value_to_set))
735
client.Set(client_dbus_interface, self.propname,
736
self.value_to_set,
737
dbus_interface=dbus.PROPERTIES_IFACE)
738
739
@property
740
def propname(self):
741
raise NotImplementedError()
742
743
744
class EnableCmd(PropertyCmd):
745
propname = "Enabled"
746
value_to_set = dbus.Boolean(True)
747
748
749
class DisableCmd(PropertyCmd):
750
propname = "Enabled"
751
value_to_set = dbus.Boolean(False)
752
753
754
class BumpTimeoutCmd(PropertyCmd):
755
propname = "LastCheckedOK"
756
value_to_set = ""
757
758
759
class StartCheckerCmd(PropertyCmd):
760
propname = "CheckerRunning"
761
value_to_set = dbus.Boolean(True)
762
763
764
class StopCheckerCmd(PropertyCmd):
765
propname = "CheckerRunning"
766
value_to_set = dbus.Boolean(False)
767
768
769
class ApproveByDefaultCmd(PropertyCmd):
770
propname = "ApprovedByDefault"
771
value_to_set = dbus.Boolean(True)
772
773
774
class DenyByDefaultCmd(PropertyCmd):
775
propname = "ApprovedByDefault"
776
value_to_set = dbus.Boolean(False)
777
778
779
class PropertyValueCmd(PropertyCmd):
780
"""Abstract class for PropertyCmd recieving a value as argument"""
781
def __init__(self, value):
782
self.value_to_set = value
783
784
785
class SetCheckerCmd(PropertyValueCmd):
786
propname = "Checker"
787
788
789
class SetHostCmd(PropertyValueCmd):
790
propname = "Host"
791
792
793
class SetSecretCmd(PropertyValueCmd):
794
propname = "Secret"
795
796
@property
797
def value_to_set(self):
798
return self._vts
799
800
@value_to_set.setter
801
def value_to_set(self, value):
802
"""When setting, read data from supplied file object"""
803
self._vts = value.read()
804
value.close()
805
806
807
class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):
808
"""Abstract class for PropertyValueCmd taking a value argument as
675
return "\n".join(self.rows())
676
677
if sys.version_info.major == 2:
678
__unicode__ = __str__
679
def __str__(self):
680
return str(self).encode(
681
locale.getpreferredencoding())
682
683
def rows(self):
684
format_string = self.row_formatting_string()
685
rows = [self.header_line(format_string)]
686
rows.extend(self.client_line(client, format_string)
687
for client in self.clients)
688
return rows
689
690
def row_formatting_string(self):
691
"Format string used to format table rows"
692
return " ".join("{{{key}:{width}}}".format(
693
width=max(len(self.tableheaders[key]),
694
*(len(self.string_from_client(client,
695
key))
696
for client in self.clients)),
697
key=key)
698
for key in self.keywords)
699
700
def string_from_client(self, client, key):
701
return self.valuetostring(client[key], key)
702
703
@classmethod
704
def valuetostring(cls, value, keyword):
705
if isinstance(value, dbus.Boolean):
706
return "Yes" if value else "No"
707
if keyword in ("Timeout", "Interval", "ApprovalDelay",
708
"ApprovalDuration", "ExtendedTimeout"):
709
return cls.milliseconds_to_string(value)
710
return str(value)
711
712
def header_line(self, format_string):
713
return format_string.format(**self.tableheaders)
714
715
def client_line(self, client, format_string):
716
return format_string.format(
717
**{key: self.string_from_client(client, key)
718
for key in self.keywords})
719
720
@staticmethod
721
def milliseconds_to_string(ms):
722
td = datetime.timedelta(0, 0, 0, ms)
723
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
724
.format(days="{}T".format(td.days)
725
if td.days else "",
726
hours=td.seconds // 3600,
727
minutes=(td.seconds % 3600) // 60,
728
seconds=td.seconds % 60))
729
730
731
class Property(Base):
732
"Abstract class for Actions for setting one client property"
733
734
def run_on_one_client(self, client, properties):
735
"""Set the Client's D-Bus property"""
736
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
737
client.__dbus_object_path__,
738
dbus.PROPERTIES_IFACE, client_dbus_interface,
739
self.propname, self.value_to_set
740
if not isinstance(self.value_to_set,
741
dbus.Boolean)
742
else bool(self.value_to_set))
743
client.Set(client_dbus_interface, self.propname,
744
self.value_to_set,
745
dbus_interface=dbus.PROPERTIES_IFACE)
746
747
@property
748
def propname(self):
749
raise NotImplementedError()
750
751
752
class Enable(Property):
753
propname = "Enabled"
754
value_to_set = dbus.Boolean(True)
755
756
757
class Disable(Property):
758
propname = "Enabled"
759
value_to_set = dbus.Boolean(False)
760
761
762
class BumpTimeout(Property):
763
propname = "LastCheckedOK"
764
value_to_set = ""
765
766
767
class StartChecker(Property):
768
propname = "CheckerRunning"
769
value_to_set = dbus.Boolean(True)
770
771
772
class StopChecker(Property):
773
propname = "CheckerRunning"
774
value_to_set = dbus.Boolean(False)
775
776
777
class ApproveByDefault(Property):
778
propname = "ApprovedByDefault"
779
value_to_set = dbus.Boolean(True)
780
781
782
class DenyByDefault(Property):
783
propname = "ApprovedByDefault"
784
value_to_set = dbus.Boolean(False)
785
786
787
class PropertyValue(Property):
788
"Abstract class for Property recieving a value as argument"
789
def __init__(self, value):
790
self.value_to_set = value
791
792
793
class SetChecker(PropertyValue):
794
propname = "Checker"
795
796
797
class SetHost(PropertyValue):
798
propname = "Host"
799
800
801
class SetSecret(PropertyValue):
802
propname = "Secret"
803
804
@property
805
def value_to_set(self):
806
return self._vts
807
808
@value_to_set.setter
809
def value_to_set(self, value):
810
"""When setting, read data from supplied file object"""
811
self._vts = value.read()
812
value.close()
813
814
815
class MillisecondsPropertyValueArgument(PropertyValue):
816
"""Abstract class for PropertyValue taking a value argument as
809
817
a datetime.timedelta() but should store it as milliseconds."""
810
818
811
@property
812
def value_to_set(self):
813
return self._vts
814
815
@value_to_set.setter
816
def value_to_set(self, value):
817
"""When setting, convert value from a datetime.timedelta"""
818
self._vts = int(round(value.total_seconds() * 1000))
819
820
821
class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
822
propname = "Timeout"
823
824
825
class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
826
propname = "ExtendedTimeout"
827
828
829
class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):
830
propname = "Interval"
831
832
833
class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):
834
propname = "ApprovalDelay"
835
836
837
class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):
838
propname = "ApprovalDuration"
819
@property
820
def value_to_set(self):
821
return self._vts
822
823
@value_to_set.setter
824
def value_to_set(self, value):
825
"When setting, convert value from a datetime.timedelta"
826
self._vts = int(round(value.total_seconds() * 1000))
827
828
829
class SetTimeout(MillisecondsPropertyValueArgument):
830
propname = "Timeout"
831
832
833
class SetExtendedTimeout(MillisecondsPropertyValueArgument):
834
propname = "ExtendedTimeout"
835
836
837
class SetInterval(MillisecondsPropertyValueArgument):
838
propname = "Interval"
839
840
841
class SetApprovalDelay(MillisecondsPropertyValueArgument):
842
propname = "ApprovalDelay"
843
844
845
class SetApprovalDuration(MillisecondsPropertyValueArgument):
846
propname = "ApprovalDuration"
839
847
840
848
841
849
872
880
("records",
873
881
"output"))
874
882
883
875
884
class Test_string_to_delta(TestCaseWithAssertLogs):
876
def test_handles_basic_rfc3339(self):
877
self.assertEqual(string_to_delta("PT0S"),
878
datetime.timedelta())
879
self.assertEqual(string_to_delta("P0D"),
880
datetime.timedelta())
881
self.assertEqual(string_to_delta("PT1S"),
882
datetime.timedelta(0, 1))
883
self.assertEqual(string_to_delta("PT2H"),
884
datetime.timedelta(0, 7200))
885
# Just test basic RFC 3339 functionality here, the doc string for
886
# rfc3339_duration_to_delta() already has more comprehensive
887
# tests, which is run by doctest.
888
889
def test_rfc3339_zero_seconds(self):
890
self.assertEqual(datetime.timedelta(),
891
string_to_delta("PT0S"))
892
893
def test_rfc3339_zero_days(self):
894
self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))
895
896
def test_rfc3339_one_second(self):
897
self.assertEqual(datetime.timedelta(0, 1),
898
string_to_delta("PT1S"))
899
900
def test_rfc3339_two_hours(self):
901
self.assertEqual(datetime.timedelta(0, 7200),
902
string_to_delta("PT2H"))
885
903
886
904
def test_falls_back_to_pre_1_6_1_with_warning(self):
887
905
with self.assertLogs(log, logging.WARNING):
888
906
value = string_to_delta("2h")
889
self.assertEqual(value, datetime.timedelta(0, 7200))
907
self.assertEqual(datetime.timedelta(0, 7200), value)
890
908
891
909
892
910
class Test_check_option_syntax(unittest.TestCase):
930
948
@contextlib.contextmanager
931
949
def assertParseError(self):
932
950
with self.assertRaises(SystemExit) as e:
933
with self.temporarily_suppress_stderr():
951
with self.redirect_stderr_to_devnull():
934
952
yield
935
953
# Exit code from argparse is guaranteed to be "2". Reference:
936
954
# https://docs.python.org/3/library
937
955
# /argparse.html#exiting-methods
938
self.assertEqual(e.exception.code, 2)
956
self.assertEqual(2, e.exception.code)
939
957
940
958
@staticmethod
941
959
@contextlib.contextmanager
942
def temporarily_suppress_stderr():
943
null = os.open(os.path.devnull, os.O_RDWR)
944
stderrcopy = os.dup(sys.stderr.fileno())
945
os.dup2(null, sys.stderr.fileno())
946
os.close(null)
947
try:
948
yield
949
finally:
950
# restore stderr
951
os.dup2(stderrcopy, sys.stderr.fileno())
952
os.close(stderrcopy)
960
def redirect_stderr_to_devnull():
961
old_stderr = sys.stderr
962
with contextlib.closing(open(os.devnull, "w")) as null:
963
sys.stderr = null
964
try:
965
yield
966
finally:
967
sys.stderr = old_stderr
953
968
954
969
def check_option_syntax(self, options):
955
970
check_option_syntax(self.parser, options)
968
983
options = self.parser.parse_args()
969
984
setattr(options, action, value)
970
985
options.verbose = True
971
options.client = ["foo"]
986
options.client = ["client"]
972
987
with self.assertParseError():
973
988
self.check_option_syntax(options)
974
989
1004
1019
for action, value in self.actions.items():
1005
1020
options = self.parser.parse_args()
1006
1021
setattr(options, action, value)
1007
options.client = ["foo"]
1022
options.client = ["client"]
1008
1023
self.check_option_syntax(options)
1009
1024
1010
def test_actions_except_is_enabled_are_ok_with_two_clients(self):
1025
def test_one_client_with_all_actions_except_is_enabled(self):
1026
options = self.parser.parse_args()
1027
for action, value in self.actions.items():
1028
if action == "is_enabled":
1029
continue
1030
setattr(options, action, value)
1031
options.client = ["client"]
1032
self.check_option_syntax(options)
1033
1034
def test_two_clients_with_all_actions_except_is_enabled(self):
1035
options = self.parser.parse_args()
1036
for action, value in self.actions.items():
1037
if action == "is_enabled":
1038
continue
1039
setattr(options, action, value)
1040
options.client = ["client1", "client2"]
1041
self.check_option_syntax(options)
1042
1043
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1011
1044
for action, value in self.actions.items():
1012
1045
if action == "is_enabled":
1013
1046
continue
1014
1047
options = self.parser.parse_args()
1015
1048
setattr(options, action, value)
1016
options.client = ["foo", "barbar"]
1049
options.client = ["client1", "client2"]
1017
1050
self.check_option_syntax(options)
1018
1051
1019
1052
def test_is_enabled_fails_without_client(self):
1025
1058
def test_is_enabled_fails_with_two_clients(self):
1026
1059
options = self.parser.parse_args()
1027
1060
options.is_enabled = True
1028
options.client = ["foo", "barbar"]
1061
options.client = ["client1", "client2"]
1029
1062
with self.assertParseError():
1030
1063
self.check_option_syntax(options)
1031
1064
1048
1081
def get_object(mockbus_self, busname, dbus_path):
1049
1082
# Note that "self" is still the testcase instance,
1050
1083
# this MockBus instance is in "mockbus_self".
1051
self.assertEqual(busname, dbus_busname)
1052
self.assertEqual(dbus_path, server_dbus_path)
1084
self.assertEqual(dbus_busname, busname)
1085
self.assertEqual(server_dbus_path, dbus_path)
1053
1086
mockbus_self.called = True
1054
1087
return mockbus_self
1055
1088
1058
1091
self.assertTrue(mockbus.called)
1059
1092
1060
1093
def test_logs_and_exits_on_dbus_error(self):
1061
class MockBusFailing(object):
1094
class FailingBusStub(object):
1062
1095
def get_object(self, busname, dbus_path):
1063
1096
raise dbus.exceptions.DBusException("Test")
1064
1097
1065
1098
with self.assertLogs(log, logging.CRITICAL):
1066
1099
with self.assertRaises(SystemExit) as e:
1067
bus = get_mandos_dbus_object(bus=MockBusFailing())
1100
bus = get_mandos_dbus_object(bus=FailingBusStub())
1068
1101
1069
1102
if isinstance(e.exception.code, int):
1070
self.assertNotEqual(e.exception.code, 0)
1103
self.assertNotEqual(0, e.exception.code)
1071
1104
else:
1072
1105
self.assertIsNotNone(e.exception.code)
1073
1106
1074
1107
1075
1108
class Test_get_managed_objects(TestCaseWithAssertLogs):
1076
1109
def test_calls_and_returns_GetManagedObjects(self):
1077
managed_objects = {"/clients/foo": { "Name": "foo"}}
1078
class MockObjectManager(object):
1110
managed_objects = {"/clients/client": { "Name": "client"}}
1111
class ObjectManagerStub(object):
1079
1112
def GetManagedObjects(self):
1080
1113
return managed_objects
1081
retval = get_managed_objects(MockObjectManager())
1114
retval = get_managed_objects(ObjectManagerStub())
1082
1115
self.assertDictEqual(managed_objects, retval)
1083
1116
1084
1117
def test_logs_and_exits_on_dbus_error(self):
1085
1118
dbus_logger = logging.getLogger("dbus.proxies")
1086
1119
1087
class MockObjectManagerFailing(object):
1120
class ObjectManagerFailingStub(object):
1088
1121
def GetManagedObjects(self):
1089
1122
dbus_logger.error("Test")
1090
1123
raise dbus.exceptions.DBusException("Test")
1101
1134
try:
1102
1135
with self.assertLogs(log, logging.CRITICAL) as watcher:
1103
1136
with self.assertRaises(SystemExit) as e:
1104
get_managed_objects(MockObjectManagerFailing())
1137
get_managed_objects(ObjectManagerFailingStub())
1105
1138
finally:
1106
1139
dbus_logger.removeFilter(counting_handler)
1107
1140
1108
1141
# Make sure the dbus logger was suppressed
1109
self.assertEqual(counting_handler.count, 0)
1142
self.assertEqual(0, counting_handler.count)
1110
1143
1111
1144
# Test that the dbus_logger still works
1112
1145
with self.assertLogs(dbus_logger, logging.ERROR):
1113
1146
dbus_logger.error("Test")
1114
1147
1115
1148
if isinstance(e.exception.code, int):
1116
self.assertNotEqual(e.exception.code, 0)
1149
self.assertNotEqual(0, e.exception.code)
1117
1150
else:
1118
1151
self.assertIsNotNone(e.exception.code)
1119
1152
1124
1157
add_command_line_options(self.parser)
1125
1158
1126
1159
def test_is_enabled(self):
1127
self.assert_command_from_args(["--is-enabled", "foo"],
1128
IsEnabledCmd)
1160
self.assert_command_from_args(["--is-enabled", "client"],
1161
command.IsEnabled)
1129
1162
1130
1163
def assert_command_from_args(self, args, command_cls,
1131
1164
**cmd_attrs):
1134
1167
options = self.parser.parse_args(args)
1135
1168
check_option_syntax(self.parser, options)
1136
1169
commands = commands_from_options(options)
1137
self.assertEqual(len(commands), 1)
1170
self.assertEqual(1, len(commands))
1138
1171
command = commands[0]
1139
1172
self.assertIsInstance(command, command_cls)
1140
1173
for key, value in cmd_attrs.items():
1141
self.assertEqual(getattr(command, key), value)
1174
self.assertEqual(value, getattr(command, key))
1142
1175
1143
1176
def test_is_enabled_short(self):
1144
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1177
self.assert_command_from_args(["-V", "client"],
1178
command.IsEnabled)
1145
1179
1146
1180
def test_approve(self):
1147
self.assert_command_from_args(["--approve", "foo"],
1148
ApproveCmd)
1181
self.assert_command_from_args(["--approve", "client"],
1182
command.Approve)
1149
1183
1150
1184
def test_approve_short(self):
1151
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1185
self.assert_command_from_args(["-A", "client"],
1186
command.Approve)
1152
1187
1153
1188
def test_deny(self):
1154
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1189
self.assert_command_from_args(["--deny", "client"],
1190
command.Deny)
1155
1191
1156
1192
def test_deny_short(self):
1157
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1193
self.assert_command_from_args(["-D", "client"], command.Deny)
1158
1194
1159
1195
def test_remove(self):
1160
self.assert_command_from_args(["--remove", "foo"],
1161
RemoveCmd)
1196
self.assert_command_from_args(["--remove", "client"],
1197
command.Remove)
1162
1198
1163
1199
def test_deny_before_remove(self):
1164
1200
options = self.parser.parse_args(["--deny", "--remove",
1165
"foo"])
1201
"client"])
1166
1202
check_option_syntax(self.parser, options)
1167
1203
commands = commands_from_options(options)
1168
self.assertEqual(len(commands), 2)
1169
self.assertIsInstance(commands[0], DenyCmd)
1170
self.assertIsInstance(commands[1], RemoveCmd)
1204
self.assertEqual(2, len(commands))
1205
self.assertIsInstance(commands[0], command.Deny)
1206
self.assertIsInstance(commands[1], command.Remove)
1171
1207
1172
1208
def test_deny_before_remove_reversed(self):
1173
1209
options = self.parser.parse_args(["--remove", "--deny",
1174
1210
"--all"])
1175
1211
check_option_syntax(self.parser, options)
1176
1212
commands = commands_from_options(options)
1177
self.assertEqual(len(commands), 2)
1178
self.assertIsInstance(commands[0], DenyCmd)
1179
self.assertIsInstance(commands[1], RemoveCmd)
1213
self.assertEqual(2, len(commands))
1214
self.assertIsInstance(commands[0], command.Deny)
1215
self.assertIsInstance(commands[1], command.Remove)
1180
1216
1181
1217
def test_remove_short(self):
1182
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1218
self.assert_command_from_args(["-r", "client"],
1219
command.Remove)
1183
1220
1184
1221
def test_dump_json(self):
1185
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1222
self.assert_command_from_args(["--dump-json"],
1223
command.DumpJSON)
1186
1224
1187
1225
def test_enable(self):
1188
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1226
self.assert_command_from_args(["--enable", "client"],
1227
command.Enable)
1189
1228
1190
1229
def test_enable_short(self):
1191
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1230
self.assert_command_from_args(["-e", "client"],
1231
command.Enable)
1192
1232
1193
1233
def test_disable(self):
1194
self.assert_command_from_args(["--disable", "foo"],
1195
DisableCmd)
1234
self.assert_command_from_args(["--disable", "client"],
1235
command.Disable)
1196
1236
1197
1237
def test_disable_short(self):
1198
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1238
self.assert_command_from_args(["-d", "client"],
1239
command.Disable)
1199
1240
1200
1241
def test_bump_timeout(self):
1201
self.assert_command_from_args(["--bump-timeout", "foo"],
1202
BumpTimeoutCmd)
1242
self.assert_command_from_args(["--bump-timeout", "client"],
1243
command.BumpTimeout)
1203
1244
1204
1245
def test_bump_timeout_short(self):
1205
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1246
self.assert_command_from_args(["-b", "client"],
1247
command.BumpTimeout)
1206
1248
1207
1249
def test_start_checker(self):
1208
self.assert_command_from_args(["--start-checker", "foo"],
1209
StartCheckerCmd)
1250
self.assert_command_from_args(["--start-checker", "client"],
1251
command.StartChecker)
1210
1252
1211
1253
def test_stop_checker(self):
1212
self.assert_command_from_args(["--stop-checker", "foo"],
1213
StopCheckerCmd)
1254
self.assert_command_from_args(["--stop-checker", "client"],
1255
command.StopChecker)
1214
1256
1215
1257
def test_approve_by_default(self):
1216
self.assert_command_from_args(["--approve-by-default", "foo"],
1217
ApproveByDefaultCmd)
1258
self.assert_command_from_args(["--approve-by-default",
1259
"client"],
1260
command.ApproveByDefault)
1218
1261
1219
1262
def test_deny_by_default(self):
1220
self.assert_command_from_args(["--deny-by-default", "foo"],
1221
DenyByDefaultCmd)
1263
self.assert_command_from_args(["--deny-by-default", "client"],
1264
command.DenyByDefault)
1222
1265
1223
1266
def test_checker(self):
1224
self.assert_command_from_args(["--checker", ":", "foo"],
1225
SetCheckerCmd, value_to_set=":")
1267
self.assert_command_from_args(["--checker", ":", "client"],
1268
command.SetChecker,
1269
value_to_set=":")
1226
1270
1227
1271
def test_checker_empty(self):
1228
self.assert_command_from_args(["--checker", "", "foo"],
1229
SetCheckerCmd, value_to_set="")
1272
self.assert_command_from_args(["--checker", "", "client"],
1273
command.SetChecker,
1274
value_to_set="")
1230
1275
1231
1276
def test_checker_short(self):
1232
self.assert_command_from_args(["-c", ":", "foo"],
1233
SetCheckerCmd, value_to_set=":")
1277
self.assert_command_from_args(["-c", ":", "client"],
1278
command.SetChecker,
1279
value_to_set=":")
1234
1280
1235
1281
def test_host(self):
1236
self.assert_command_from_args(["--host", "foo.example.org",
1237
"foo"], SetHostCmd,
1238
value_to_set="foo.example.org")
1282
self.assert_command_from_args(
1283
["--host", "client.example.org", "client"],
1284
command.SetHost, value_to_set="client.example.org")
1239
1285
1240
1286
def test_host_short(self):
1241
self.assert_command_from_args(["-H", "foo.example.org",
1242
"foo"], SetHostCmd,
1243
value_to_set="foo.example.org")
1287
self.assert_command_from_args(
1288
["-H", "client.example.org", "client"], command.SetHost,
1289
value_to_set="client.example.org")
1244
1290
1245
1291
def test_secret_devnull(self):
1246
1292
self.assert_command_from_args(["--secret", os.path.devnull,
1247
"foo"], SetSecretCmd,
1293
"client"], command.SetSecret,
1248
1294
value_to_set=b"")
1249
1295
1250
1296
def test_secret_tempfile(self):
1253
1299
f.write(value)
1254
1300
f.seek(0)
1255
1301
self.assert_command_from_args(["--secret", f.name,
1256
"foo"], SetSecretCmd,
1302
"client"],
1303
command.SetSecret,
1257
1304
value_to_set=value)
1258
1305
1259
1306
def test_secret_devnull_short(self):
1260
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1261
SetSecretCmd, value_to_set=b"")
1307
self.assert_command_from_args(["-s", os.path.devnull,
1308
"client"], command.SetSecret,
1309
value_to_set=b"")
1262
1310
1263
1311
def test_secret_tempfile_short(self):
1264
1312
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1265
1313
value = b"secret\0xyzzy\nbar"
1266
1314
f.write(value)
1267
1315
f.seek(0)
1268
self.assert_command_from_args(["-s", f.name, "foo"],
1269
SetSecretCmd,
1316
self.assert_command_from_args(["-s", f.name, "client"],
1317
command.SetSecret,
1270
1318
value_to_set=value)
1271
1319
1272
1320
def test_timeout(self):
1273
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1274
SetTimeoutCmd,
1321
self.assert_command_from_args(["--timeout", "PT5M", "client"],
1322
command.SetTimeout,
1275
1323
value_to_set=300000)
1276
1324
1277
1325
def test_timeout_short(self):
1278
self.assert_command_from_args(["-t", "PT5M", "foo"],
1279
SetTimeoutCmd,
1326
self.assert_command_from_args(["-t", "PT5M", "client"],
1327
command.SetTimeout,
1280
1328
value_to_set=300000)
1281
1329
1282
1330
def test_extended_timeout(self):
1283
1331
self.assert_command_from_args(["--extended-timeout", "PT15M",
1284
"foo"],
1285
SetExtendedTimeoutCmd,
1332
"client"],
1333
command.SetExtendedTimeout,
1286
1334
value_to_set=900000)
1287
1335
1288
1336
def test_interval(self):
1289
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1290
SetIntervalCmd,
1337
self.assert_command_from_args(["--interval", "PT2M",
1338
"client"], command.SetInterval,
1291
1339
value_to_set=120000)
1292
1340
1293
1341
def test_interval_short(self):
1294
self.assert_command_from_args(["-i", "PT2M", "foo"],
1295
SetIntervalCmd,
1342
self.assert_command_from_args(["-i", "PT2M", "client"],
1343
command.SetInterval,
1296
1344
value_to_set=120000)
1297
1345
1298
1346
def test_approval_delay(self):
1299
1347
self.assert_command_from_args(["--approval-delay", "PT30S",
1300
"foo"], SetApprovalDelayCmd,
1348
"client"],
1349
command.SetApprovalDelay,
1301
1350
value_to_set=30000)
1302
1351
1303
1352
def test_approval_duration(self):
1304
1353
self.assert_command_from_args(["--approval-duration", "PT1S",
1305
"foo"], SetApprovalDurationCmd,
1354
"client"],
1355
command.SetApprovalDuration,
1306
1356
value_to_set=1000)
1307
1357
1308
1358
def test_print_table(self):
1309
self.assert_command_from_args([], PrintTableCmd,
1359
self.assert_command_from_args([], command.PrintTable,
1310
1360
verbose=False)
1311
1361
1312
1362
def test_print_table_verbose(self):
1313
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1363
self.assert_command_from_args(["--verbose"],
1364
command.PrintTable,
1314
1365
verbose=True)
1315
1366
1316
1367
def test_print_table_verbose_short(self):
1317
self.assert_command_from_args(["-v"], PrintTableCmd,
1368
self.assert_command_from_args(["-v"], command.PrintTable,
1318
1369
verbose=True)
1319
1370
1320
1371
1321
class TestCmd(unittest.TestCase):
1372
class TestCommand(unittest.TestCase):
1322
1373
"""Abstract class for tests of command classes"""
1323
1374
1324
1375
def setUp(self):
1330
1381
self.attributes["Name"] = name
1331
1382
self.calls = []
1332
1383
def Set(self, interface, propname, value, dbus_interface):
1333
testcase.assertEqual(interface, client_dbus_interface)
1334
testcase.assertEqual(dbus_interface,
1335
dbus.PROPERTIES_IFACE)
1384
testcase.assertEqual(client_dbus_interface, interface)
1385
testcase.assertEqual(dbus.PROPERTIES_IFACE,
1386
dbus_interface)
1336
1387
self.attributes[propname] = value
1337
def Get(self, interface, propname, dbus_interface):
1338
testcase.assertEqual(interface, client_dbus_interface)
1339
testcase.assertEqual(dbus_interface,
1340
dbus.PROPERTIES_IFACE)
1341
return self.attributes[propname]
1342
1388
def Approve(self, approve, dbus_interface):
1343
testcase.assertEqual(dbus_interface,
1344
client_dbus_interface)
1389
testcase.assertEqual(client_dbus_interface,
1390
dbus_interface)
1345
1391
self.calls.append(("Approve", (approve,
1346
1392
dbus_interface)))
1347
1393
self.client = MockClient(
1394
1440
LastCheckerStatus=-2)
1395
1441
self.clients = collections.OrderedDict(
1396
1442
[
1397
("/clients/foo", self.client.attributes),
1398
("/clients/barbar", self.other_client.attributes),
1443
(self.client.__dbus_object_path__,
1444
self.client.attributes),
1445
(self.other_client.__dbus_object_path__,
1446
self.other_client.attributes),
1399
1447
])
1400
self.one_client = {"/clients/foo": self.client.attributes}
1448
self.one_client = {self.client.__dbus_object_path__:
1449
self.client.attributes}
1401
1450
1402
1451
@property
1403
1452
def bus(self):
1404
class Bus(object):
1453
class MockBus(object):
1405
1454
@staticmethod
1406
1455
def get_object(client_bus_name, path):
1407
self.assertEqual(client_bus_name, dbus_busname)
1408
return {
1409
# Note: "self" here is the TestCmd instance, not
1410
# the Bus instance, since this is a static method!
1411
"/clients/foo": self.client,
1412
"/clients/barbar": self.other_client,
1413
}[path]
1414
return Bus()
1415
1416
1417
class TestIsEnabledCmd(TestCmd):
1418
def test_is_enabled(self):
1419
self.assertTrue(all(IsEnabledCmd().is_enabled(client,
1420
properties)
1421
for client, properties
1422
in self.clients.items()))
1423
1424
def test_is_enabled_run_exits_successfully(self):
1456
self.assertEqual(dbus_busname, client_bus_name)
1457
# Note: "self" here is the TestCmd instance, not the
1458
# MockBus instance, since this is a static method!
1459
if path == self.client.__dbus_object_path__:
1460
return self.client
1461
elif path == self.other_client.__dbus_object_path__:
1462
return self.other_client
1463
return MockBus()
1464
1465
1466
class TestBaseCommands(TestCommand):
1467
1468
def test_IsEnabled_exits_successfully(self):
1425
1469
with self.assertRaises(SystemExit) as e:
1426
IsEnabledCmd().run(self.one_client)
1470
command.IsEnabled().run(self.one_client)
1427
1471
if e.exception.code is not None:
1428
self.assertEqual(e.exception.code, 0)
1472
self.assertEqual(0, e.exception.code)
1429
1473
else:
1430
1474
self.assertIsNone(e.exception.code)
1431
1475
1432
def test_is_enabled_run_exits_with_failure(self):
1476
def test_IsEnabled_exits_with_failure(self):
1433
1477
self.client.attributes["Enabled"] = dbus.Boolean(False)
1434
1478
with self.assertRaises(SystemExit) as e:
1435
IsEnabledCmd().run(self.one_client)
1479
command.IsEnabled().run(self.one_client)
1436
1480
if isinstance(e.exception.code, int):
1437
self.assertNotEqual(e.exception.code, 0)
1481
self.assertNotEqual(0, e.exception.code)
1438
1482
else:
1439
1483
self.assertIsNotNone(e.exception.code)
1440
1484
1441
1442
class TestApproveCmd(TestCmd):
1443
def test_approve(self):
1444
ApproveCmd().run(self.clients, self.bus)
1485
def test_Approve(self):
1486
command.Approve().run(self.clients, self.bus)
1445
1487
for clientpath in self.clients:
1446
1488
client = self.bus.get_object(dbus_busname, clientpath)
1447
1489
self.assertIn(("Approve", (True, client_dbus_interface)),
1448
1490
client.calls)
1449
1491
1450
1451
class TestDenyCmd(TestCmd):
1452
def test_deny(self):
1453
DenyCmd().run(self.clients, self.bus)
1492
def test_Deny(self):
1493
command.Deny().run(self.clients, self.bus)
1454
1494
for clientpath in self.clients:
1455
1495
client = self.bus.get_object(dbus_busname, clientpath)
1456
1496
self.assertIn(("Approve", (False, client_dbus_interface)),
1457
1497
client.calls)
1458
1498
1459
1460
class TestRemoveCmd(TestCmd):
1461
def test_remove(self):
1462
class MockMandos(object):
1499
def test_Remove(self):
1500
class MandosSpy(object):
1463
1501
def __init__(self):
1464
1502
self.calls = []
1465
1503
def RemoveClient(self, dbus_path):
1466
1504
self.calls.append(("RemoveClient", (dbus_path,)))
1467
mandos = MockMandos()
1468
super(TestRemoveCmd, self).setUp()
1469
RemoveCmd().run(self.clients, self.bus, mandos)
1470
self.assertEqual(len(mandos.calls), 2)
1505
mandos = MandosSpy()
1506
command.Remove().run(self.clients, self.bus, mandos)
1471
1507
for clientpath in self.clients:
1472
1508
self.assertIn(("RemoveClient", (clientpath,)),
1473
1509
mandos.calls)
1474
1510
1475
1476
class TestDumpJSONCmd(TestCmd):
1477
def setUp(self):
1478
self.expected_json = {
1479
"foo": {
1480
"Name": "foo",
1481
"KeyID": ("92ed150794387c03ce684574b1139a65"
1482
"94a34f895daaaf09fd8ea90a27cddb12"),
1483
"Host": "foo.example.org",
1484
"Enabled": True,
1485
"Timeout": 300000,
1486
"LastCheckedOK": "2019-02-03T00:00:00",
1487
"Created": "2019-01-02T00:00:00",
1488
"Interval": 120000,
1489
"Fingerprint": ("778827225BA7DE539C5A"
1490
"7CFA59CFF7CDBD9A5920"),
1491
"CheckerRunning": False,
1492
"LastEnabled": "2019-01-03T00:00:00",
1493
"ApprovalPending": False,
1494
"ApprovedByDefault": True,
1495
"LastApprovalRequest": "",
1496
"ApprovalDelay": 0,
1497
"ApprovalDuration": 1000,
1498
"Checker": "fping -q -- %(host)s",
1499
"ExtendedTimeout": 900000,
1500
"Expires": "2019-02-04T00:00:00",
1501
"LastCheckerStatus": 0,
1502
},
1503
"barbar": {
1504
"Name": "barbar",
1505
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1506
"6ab612cff5ad227247e46c2b020f441c"),
1507
"Host": "192.0.2.3",
1508
"Enabled": True,
1509
"Timeout": 300000,
1510
"LastCheckedOK": "2019-02-04T00:00:00",
1511
"Created": "2019-01-03T00:00:00",
1512
"Interval": 120000,
1513
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1514
"F547B3A107558FCA3A27"),
1515
"CheckerRunning": True,
1516
"LastEnabled": "2019-01-04T00:00:00",
1517
"ApprovalPending": False,
1518
"ApprovedByDefault": False,
1519
"LastApprovalRequest": "2019-01-03T00:00:00",
1520
"ApprovalDelay": 30000,
1521
"ApprovalDuration": 93785000,
1522
"Checker": ":",
1523
"ExtendedTimeout": 900000,
1524
"Expires": "2019-02-05T00:00:00",
1525
"LastCheckerStatus": -2,
1526
},
1527
}
1528
return super(TestDumpJSONCmd, self).setUp()
1529
1530
def test_normal(self):
1531
output = DumpJSONCmd().output(self.clients.values())
1532
json_data = json.loads(output)
1533
self.assertDictEqual(json_data, self.expected_json)
1534
1535
def test_one_client(self):
1536
output = DumpJSONCmd().output(self.one_client.values())
1537
json_data = json.loads(output)
1511
expected_json = {
1512
"foo": {
1513
"Name": "foo",
1514
"KeyID": ("92ed150794387c03ce684574b1139a65"
1515
"94a34f895daaaf09fd8ea90a27cddb12"),
1516
"Host": "foo.example.org",
1517
"Enabled": True,
1518
"Timeout": 300000,
1519
"LastCheckedOK": "2019-02-03T00:00:00",
1520
"Created": "2019-01-02T00:00:00",
1521
"Interval": 120000,
1522
"Fingerprint": ("778827225BA7DE539C5A"
1523
"7CFA59CFF7CDBD9A5920"),
1524
"CheckerRunning": False,
1525
"LastEnabled": "2019-01-03T00:00:00",
1526
"ApprovalPending": False,
1527
"ApprovedByDefault": True,
1528
"LastApprovalRequest": "",
1529
"ApprovalDelay": 0,
1530
"ApprovalDuration": 1000,
1531
"Checker": "fping -q -- %(host)s",
1532
"ExtendedTimeout": 900000,
1533
"Expires": "2019-02-04T00:00:00",
1534
"LastCheckerStatus": 0,
1535
},
1536
"barbar": {
1537
"Name": "barbar",
1538
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1539
"6ab612cff5ad227247e46c2b020f441c"),
1540
"Host": "192.0.2.3",
1541
"Enabled": True,
1542
"Timeout": 300000,
1543
"LastCheckedOK": "2019-02-04T00:00:00",
1544
"Created": "2019-01-03T00:00:00",
1545
"Interval": 120000,
1546
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1547
"F547B3A107558FCA3A27"),
1548
"CheckerRunning": True,
1549
"LastEnabled": "2019-01-04T00:00:00",
1550
"ApprovalPending": False,
1551
"ApprovedByDefault": False,
1552
"LastApprovalRequest": "2019-01-03T00:00:00",
1553
"ApprovalDelay": 30000,
1554
"ApprovalDuration": 93785000,
1555
"Checker": ":",
1556
"ExtendedTimeout": 900000,
1557
"Expires": "2019-02-05T00:00:00",
1558
"LastCheckerStatus": -2,
1559
},
1560
}
1561
1562
def test_DumpJSON_normal(self):
1563
with self.capture_stdout_to_buffer() as buffer:
1564
command.DumpJSON().run(self.clients)
1565
json_data = json.loads(buffer.getvalue())
1566
self.assertDictEqual(self.expected_json, json_data)
1567
1568
@staticmethod
1569
@contextlib.contextmanager
1570
def capture_stdout_to_buffer():
1571
capture_buffer = io.StringIO()
1572
old_stdout = sys.stdout
1573
sys.stdout = capture_buffer
1574
try:
1575
yield capture_buffer
1576
finally:
1577
sys.stdout = old_stdout
1578
1579
def test_DumpJSON_one_client(self):
1580
with self.capture_stdout_to_buffer() as buffer:
1581
command.DumpJSON().run(self.one_client)
1582
json_data = json.loads(buffer.getvalue())
1538
1583
expected_json = {"foo": self.expected_json["foo"]}
1539
self.assertDictEqual(json_data, expected_json)
1540
1541
1542
class TestPrintTableCmd(TestCmd):
1543
def test_normal(self):
1544
output = PrintTableCmd().output(self.clients.values())
1584
self.assertDictEqual(expected_json, json_data)
1585
1586
def test_PrintTable_normal(self):
1587
with self.capture_stdout_to_buffer() as buffer:
1588
command.PrintTable().run(self.clients)
1545
1589
expected_output = "\n".join((
1546
1590
"Name Enabled Timeout Last Successful Check",
1547
1591
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1548
1592
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1549
))
1550
self.assertEqual(output, expected_output)
1593
)) + "\n"
1594
self.assertEqual(expected_output, buffer.getvalue())
1551
1595
1552
def test_verbose(self):
1553
output = PrintTableCmd(verbose=True).output(
1554
self.clients.values())
1596
def test_PrintTable_verbose(self):
1597
with self.capture_stdout_to_buffer() as buffer:
1598
command.PrintTable(verbose=True).run(self.clients)
1555
1599
columns = (
1556
1600
(
1557
1601
"Name ",
1639
1683
)
1640
1684
)
1641
1685
num_lines = max(len(rows) for rows in columns)
1642
expected_output = "\n".join("".join(rows[line]
1643
for rows in columns)
1644
for line in range(num_lines))
1645
self.assertEqual(output, expected_output)
1686
expected_output = ("\n".join("".join(rows[line]
1687
for rows in columns)
1688
for line in range(num_lines))
1689
+ "\n")
1690
self.assertEqual(expected_output, buffer.getvalue())
1646
1691
1647
def test_one_client(self):
1648
output = PrintTableCmd().output(self.one_client.values())
1692
def test_PrintTable_one_client(self):
1693
with self.capture_stdout_to_buffer() as buffer:
1694
command.PrintTable().run(self.one_client)
1649
1695
expected_output = "\n".join((
1650
1696
"Name Enabled Timeout Last Successful Check",
1651
1697
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1652
))
1653
self.assertEqual(output, expected_output)
1654
1655
1656
class TestPropertyCmd(TestCmd):
1657
"""Abstract class for tests of PropertyCmd classes"""
1698
)) + "\n"
1699
self.assertEqual(expected_output, buffer.getvalue())
1700
1701
1702
class TestPropertyCmd(TestCommand):
1703
"""Abstract class for tests of command.Property classes"""
1658
1704
def runTest(self):
1659
1705
if not hasattr(self, "command"):
1660
1706
return
1665
1711
for clientpath in self.clients:
1666
1712
client = self.bus.get_object(dbus_busname, clientpath)
1667
1713
old_value = client.attributes[self.propname]
1668
self.assertNotIsInstance(old_value, self.Unique)
1669
1714
client.attributes[self.propname] = self.Unique()
1670
1715
self.run_command(value_to_set, self.clients)
1671
1716
for clientpath in self.clients:
1672
1717
client = self.bus.get_object(dbus_busname, clientpath)
1673
1718
value = client.attributes[self.propname]
1674
1719
self.assertNotIsInstance(value, self.Unique)
1675
self.assertEqual(value, value_to_get)
1720
self.assertEqual(value_to_get, value)
1676
1721
1677
1722
class Unique(object):
1678
1723
"""Class for objects which exist only to be unique objects,
1683
1728
1684
1729
1685
1730
class TestEnableCmd(TestPropertyCmd):
1686
command = EnableCmd
1731
command = command.Enable
1687
1732
propname = "Enabled"
1688
1733
values_to_set = [dbus.Boolean(True)]
1689
1734
1690
1735
1691
1736
class TestDisableCmd(TestPropertyCmd):
1692
command = DisableCmd
1737
command = command.Disable
1693
1738
propname = "Enabled"
1694
1739
values_to_set = [dbus.Boolean(False)]
1695
1740
1696
1741
1697
1742
class TestBumpTimeoutCmd(TestPropertyCmd):
1698
command = BumpTimeoutCmd
1743
command = command.BumpTimeout
1699
1744
propname = "LastCheckedOK"
1700
1745
values_to_set = [""]
1701
1746
1702
1747
1703
1748
class TestStartCheckerCmd(TestPropertyCmd):
1704
command = StartCheckerCmd
1749
command = command.StartChecker
1705
1750
propname = "CheckerRunning"
1706
1751
values_to_set = [dbus.Boolean(True)]
1707
1752
1708
1753
1709
1754
class TestStopCheckerCmd(TestPropertyCmd):
1710
command = StopCheckerCmd
1755
command = command.StopChecker
1711
1756
propname = "CheckerRunning"
1712
1757
values_to_set = [dbus.Boolean(False)]
1713
1758
1714
1759
1715
1760
class TestApproveByDefaultCmd(TestPropertyCmd):
1716
command = ApproveByDefaultCmd
1761
command = command.ApproveByDefault
1717
1762
propname = "ApprovedByDefault"
1718
1763
values_to_set = [dbus.Boolean(True)]
1719
1764
1720
1765
1721
1766
class TestDenyByDefaultCmd(TestPropertyCmd):
1722
command = DenyByDefaultCmd
1767
command = command.DenyByDefault
1723
1768
propname = "ApprovedByDefault"
1724
1769
values_to_set = [dbus.Boolean(False)]
1725
1770
1737
1782
1738
1783
1739
1784
class TestSetCheckerCmd(TestPropertyValueCmd):
1740
command = SetCheckerCmd
1785
command = command.SetChecker
1741
1786
propname = "Checker"
1742
1787
values_to_set = ["", ":", "fping -q -- %s"]
1743
1788
1744
1789
1745
1790
class TestSetHostCmd(TestPropertyValueCmd):
1746
command = SetHostCmd
1791
command = command.SetHost
1747
1792
propname = "Host"
1748
values_to_set = ["192.0.2.3", "foo.example.org"]
1793
values_to_set = ["192.0.2.3", "client.example.org"]
1749
1794
1750
1795
1751
1796
class TestSetSecretCmd(TestPropertyValueCmd):
1752
command = SetSecretCmd
1797
command = command.SetSecret
1753
1798
propname = "Secret"
1754
1799
values_to_set = [io.BytesIO(b""),
1755
1800
io.BytesIO(b"secret\0xyzzy\nbar")]
1756
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1801
values_to_get = [f.getvalue() for f in values_to_set]
1757
1802
1758
1803
1759
1804
class TestSetTimeoutCmd(TestPropertyValueCmd):
1760
command = SetTimeoutCmd
1805
command = command.SetTimeout
1761
1806
propname = "Timeout"
1762
1807
values_to_set = [datetime.timedelta(),
1763
1808
datetime.timedelta(minutes=5),
1764
1809
datetime.timedelta(seconds=1),
1765
1810
datetime.timedelta(weeks=1),
1766
1811
datetime.timedelta(weeks=52)]
1767
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1812
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1768
1813
1769
1814
1770
1815
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1771
command = SetExtendedTimeoutCmd
1816
command = command.SetExtendedTimeout
1772
1817
propname = "ExtendedTimeout"
1773
1818
values_to_set = [datetime.timedelta(),
1774
1819
datetime.timedelta(minutes=5),
1775
1820
datetime.timedelta(seconds=1),
1776
1821
datetime.timedelta(weeks=1),
1777
1822
datetime.timedelta(weeks=52)]
1778
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1823
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1779
1824
1780
1825
1781
1826
class TestSetIntervalCmd(TestPropertyValueCmd):
1782
command = SetIntervalCmd
1827
command = command.SetInterval
1783
1828
propname = "Interval"
1784
1829
values_to_set = [datetime.timedelta(),
1785
1830
datetime.timedelta(minutes=5),
1786
1831
datetime.timedelta(seconds=1),
1787
1832
datetime.timedelta(weeks=1),
1788
1833
datetime.timedelta(weeks=52)]
1789
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1834
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1790
1835
1791
1836
1792
1837
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1793
command = SetApprovalDelayCmd
1838
command = command.SetApprovalDelay
1794
1839
propname = "ApprovalDelay"
1795
1840
values_to_set = [datetime.timedelta(),
1796
1841
datetime.timedelta(minutes=5),
1797
1842
datetime.timedelta(seconds=1),
1798
1843
datetime.timedelta(weeks=1),
1799
1844
datetime.timedelta(weeks=52)]
1800
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1845
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1801
1846
1802
1847
1803
1848
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1804
command = SetApprovalDurationCmd
1849
command = command.SetApprovalDuration
1805
1850
propname = "ApprovalDuration"
1806
1851
values_to_set = [datetime.timedelta(),
1807
1852
datetime.timedelta(minutes=5),
1808
1853
datetime.timedelta(seconds=1),
1809
1854
datetime.timedelta(weeks=1),
1810
1855
datetime.timedelta(weeks=52)]
1811
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1856
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1812
1857
1813
1858
1814
1859
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0