/mandos/trunk

# -*- mode: python; coding: utf-8; after-save-hook: (lambda () (let ((command (if (and (boundp 'tramp-file-name-structure) (string-match (car tramp-file-name-structure) (buffer-file-name))) (tramp-file-name-localname (tramp-dissect-file-name (buffer-file-name))) (buffer-file-name)))) (if (= (shell-command (format "%s --check" (shell-quote-argument command)) "*Test*") 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w)) (kill-buffer "*Test*")) (display-buffer "*Test*")))); -*-

# Copyright © 2008-2019 Teddy Hogeborn

# Copyright © 2008-2019 Björn Påhlsson

import unittest

import logging

# Show warnings by default

if not sys.warnoptions:

    import warnings

    warnings.simplefilter("default")

 

log = logging.getLogger(sys.argv[0])

logging.basicConfig(level="INFO", # Show info level messages

                    format="%(message)s") # Show basic log messages

 

logging.captureWarnings(True)   # Show warnings via the logging system

 

version = "1.8.3"

    >>> rfc3339_duration_to_delta("P60M")

    datetime.timedelta(1680)

    >>> # Can not be empty:

    >>> rfc3339_duration_to_delta("")

    Traceback (most recent call last):

    ...

    ValueError: Invalid RFC 3339 duration: u''

    >>> # Must start with "P":

    >>> rfc3339_duration_to_delta("1D")

    Traceback (most recent call last):

    ...

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

    >>> # Must use correct order

    >>> rfc3339_duration_to_delta("PT1S2M")

    Traceback (most recent call last):

    ...

    ValueError: Invalid RFC 3339 duration: u'PT1S2M'

    >>> # Time needs time marker

    >>> rfc3339_duration_to_delta("P1H2S")

    Traceback (most recent call last):

    ...

    ValueError: Invalid RFC 3339 duration: u'P1H2S'

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

    >>> rfc3339_duration_to_delta("P1D2W")

    Traceback (most recent call last):

    ...

    ValueError: Invalid RFC 3339 duration: u'P1D2W'

    >>> rfc3339_duration_to_delta("P2W2H")

    Traceback (most recent call last):

    ...

    ValueError: Invalid RFC 3339 duration: u'P2W2H'

    except ValueError as e:

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

                    ' '.join(e.args))

    return parse_pre_1_6_1_interval(interval)

 

 

def parse_pre_1_6_1_interval(interval):

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

    return a datetime.timedelta

    >>> parse_pre_1_6_1_interval('7d')

    datetime.timedelta(7)

    >>> parse_pre_1_6_1_interval('60s')

    datetime.timedelta(0, 60)

    >>> parse_pre_1_6_1_interval('60m')

    datetime.timedelta(0, 3600)

    >>> parse_pre_1_6_1_interval('24h')

    datetime.timedelta(1)

    >>> parse_pre_1_6_1_interval('1w')

    datetime.timedelta(7)

    >>> parse_pre_1_6_1_interval('5m 30s')

    datetime.timedelta(0, 330)

    >>> parse_pre_1_6_1_interval('')

    datetime.timedelta(0)

    >>> # Ignore unknown characters, allow any order and repetitions

    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

    datetime.timedelta(2, 480, 18000)

 

    """

    print('\n'.join(TableOfClients(clients, keywords).rows()))

 

class TableOfClients(object):

    tablewords = {

        "Name": "Name",

        "Enabled": "Enabled",

        "Timeout": "Timeout",

        "LastCheckedOK": "Last Successful Check",

        "LastApprovalRequest": "Last Approval Request",

        "Created": "Created",

        "Interval": "Interval",

        "Host": "Host",

        "Fingerprint": "Fingerprint",

        "KeyID": "Key ID",

        "CheckerRunning": "Check Is Running",

        "LastEnabled": "Last Enabled",

        "ApprovalPending": "Approval Is Pending",

        "ApprovedByDefault": "Approved By Default",

        "ApprovalDelay": "Approval Delay",

        "ApprovalDuration": "Approval Duration",

        "Checker": "Checker",

        "ExtendedTimeout": "Extended Timeout",

        "Expires": "Expires",

        "LastCheckerStatus": "Last Checker Status",

    }

 

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

        self.clients = clients

        self.keywords = keywords

        if tablewords is not None:

            self.tablewords = tablewords

 

    @staticmethod

        if isinstance(value, dbus.Boolean):

    def rows(self):

        # Create format string to format table rows

        format_string = " ".join("{{{key}:{width}}}".format(

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

                      max(len(self.valuetostring(client[key], key))

                          for client in self.clients)),

            key=key)

                                 for key in self.keywords)

        # Start with header line

        rows = [format_string.format(**self.tablewords)]

        for client in self.clients:

            rows.append(format_string

                        .format(**{key: self.valuetostring(client[key], key)

                                   for key in self.keywords}))

        return rows

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

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

                     "\n%s", e)

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

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

                        "Fingerprint", "CheckerRunning",

                        "LastEnabled", "ApprovalPending",

                        "ApprovedByDefault", "LastApprovalRequest",

                        "ApprovalDelay", "ApprovalDuration",

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

 

class Test_milliseconds_to_string(unittest.TestCase):

    def test_all(self):

        self.assertEqual(milliseconds_to_string(93785000),

                         "1T02:03:05")

    def test_no_days(self):

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

    def test_all_zero(self):

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

    def test_no_fractional_seconds(self):

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

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

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

 

class Test_string_to_delta(unittest.TestCase):

    def test_handles_basic_rfc3339(self):

        self.assertEqual(string_to_delta("PT2H"),

                         datetime.timedelta(0, 7200))

    def test_falls_back_to_pre_1_6_1_with_warning(self):

        # assertLogs only exists in Python 3.4

        if hasattr(self, "assertLogs"):

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

                value = string_to_delta("2h")

        else:

            value = string_to_delta("2h")

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

 

class Test_TableOfClients(unittest.TestCase):

    def setUp(self):

        self.tablewords = {

            "Attr1": "X",

            "AttrTwo": "Yy",

            "AttrThree": "Zzz",

            "Bool": "A D-BUS Boolean",

            "NonDbusBoolean": "A Non-D-BUS Boolean",

            "Integer": "An Integer",

            "Timeout": "Timedelta 1",

            "Interval": "Timedelta 2",

            "ApprovalDelay": "Timedelta 3",

            "ApprovalDuration": "Timedelta 4",

            "ExtendedTimeout": "Timedelta 5",

            "String": "A String",

        }

        self.keywords = ["Attr1", "AttrTwo"]

        self.clients = [

            {

                "Attr1": "x1",

                "AttrTwo": "y1",

                "AttrThree": "z1",

                "Bool": dbus.Boolean(False),

                "NonDbusBoolean": False,

                "Integer": 0,

                "Timeout": 0,

                "Interval": 1000,

                "ApprovalDelay": 2000,

                "ApprovalDuration": 3000,

                "ExtendedTimeout": 4000,

                "String": "",

            },

            {

                "Attr1": "x2",

                "AttrTwo": "y2",

                "AttrThree": "z2",

                "Bool": dbus.Boolean(True),

                "NonDbusBoolean": True,

                "Integer": 1,

                "Timeout": 93785000,

                "Interval": 93786000,

                "ApprovalDelay": 93787000,

                "ApprovalDuration": 93788000,

                "ExtendedTimeout": 93789000,

                "String": "A huge string which will not fit," * 10,

            },

        ]

    def test_short_header(self):

        rows = TableOfClients(self.clients, self.keywords,

                              self.tablewords).rows()

        expected_rows = [

            "X  Yy",

            "x1 y1",

            "x2 y2"]

        self.assertEqual(rows, expected_rows)

    def test_booleans(self):

        keywords = ["Bool", "NonDbusBoolean"]

        rows = TableOfClients(self.clients, keywords,

                              self.tablewords).rows()

        expected_rows = [

            "A D-BUS Boolean A Non-D-BUS Boolean",

            "No              False              ",

            "Yes             True               ",

        ]

        self.assertEqual(rows, expected_rows)

    def test_milliseconds_detection(self):

        keywords = ["Integer", "Timeout", "Interval", "ApprovalDelay",

                    "ApprovalDuration", "ExtendedTimeout"]

        rows = TableOfClients(self.clients, keywords,

                              self.tablewords).rows()

        expected_rows = ("""

An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5

0          00:00:00    00:00:01    00:00:02    00:00:03    00:00:04   

1          1T02:03:05  1T02:03:06  1T02:03:07  1T02:03:08  1T02:03:09 

"""

        ).splitlines()[1:]

        self.assertEqual(rows, expected_rows)

    def test_empty_and_long_string_values(self):

        keywords = ["String"]

        rows = TableOfClients(self.clients, keywords,

                              self.tablewords).rows()

        expected_rows = ("""

A String                                                                                                                                                                                                                                                                                                                                  

                                                                                                                                                                                                                                                                                                                                          

A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,

"""

        ).splitlines()[1:]

        self.assertEqual(rows, expected_rows)

 

 

 

def should_only_run_tests():

    parser = argparse.ArgumentParser(add_help=False)

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

    args, unknown_args = parser.parse_known_args()

    run_tests = args.check

    if run_tests:

        # Remove --check argument from sys.argv

        sys.argv[1:] = unknown_args

    return run_tests

 

# Add all tests from doctest strings

def load_tests(loader, tests, none):

    import doctest

    tests.addTests(doctest.DocTestSuite())

    return tests

    if should_only_run_tests():

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

        unittest.main()

    else:

        main()