/mandos/trunk

#!/usr/bin/python3 -bI

# -*- mode: python; after-save-hook: (lambda () (let ((command (if (fboundp 'file-local-name) (file-local-name (buffer-file-name)) (or (file-remote-p (buffer-file-name) 'localname) (buffer-file-name))))) (if (= (progn (if (get-buffer "*Test*") (kill-buffer "*Test*")) (process-file-shell-command (format "%s --check" (shell-quote-argument command)) nil "*Test*")) 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w))) (progn (with-current-buffer "*Test*" (compilation-mode)) (display-buffer "*Test*" '(display-buffer-in-side-window)))))); coding: utf-8 -*-

import unittest

import random

import shlex

import gi

if sys.version_info.major == 2:

    __metaclass__ = type

    str = unicode

 

# Add collections.abc.Callable if it does not exist

try:

    collections.abc.Callable

except AttributeError:

    class abc:

        Callable = collections.Callable

    collections.abc = abc

    del abc

 

# Add shlex.quote if it does not exist

try:

    shlex.quote

except AttributeError:

    shlex.quote = re.escape

 

# Show warnings by default

if not sys.warnoptions:

    import warnings

    warnings.simplefilter("default")

 

if sys.version_info < (3, 2):

    configparser.Configparser = configparser.SafeConfigParser

version = "1.8.11"

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

class PGPEngine:

                if name == b"gpg":

        if self.gpg == b"gpg" or self.gpg.endswith(b"/gpg"):

class avahi:

class AvahiService:

class gnutls:

    class Credentials:

    class ClientSession:

            if gnutls.check_version(b"3.5.6"):

    if check_version(b"3.6.4"):

class Client:

    checker: multiprocessing.Process(); a running checker process used

             to see if the client lives. 'None' if no process is

             running.

            random.randrange(int(self.interval.total_seconds() * 1000

                                 + 1)),

        if self.checker is not None:

            self.checker.join()

        self.checker_callback_tag = None

        self.checker = None

                attr: shlex.quote(str(getattr(self, attr)))

                GLib.IOChannel.unix_new(pipe[0].fileno()),

                GLib.PRIORITY_DEFAULT, GLib.IO_IN,

            value = dbus.ByteArray(bytes(value))

class ProxyClient:

                    fpr = b""

                    key_id = b""

class MultiprocessingMixIn:

class MultiprocessingMixInWithPipe(MultiprocessingMixIn):

                     socketserver.TCPServer):

            if self.server_address[1]:

                self.allow_reuse_address = True

            GLib.IOChannel.unix_new(parent_pipe.fileno()),

            GLib.PRIORITY_DEFAULT, GLib.IO_IN | GLib.IO_HUP,

                GLib.IOChannel.unix_new(parent_pipe.fileno()),

                GLib.PRIORITY_DEFAULT, GLib.IO_IN | GLib.IO_HUP,

                          collections.abc.Callable):

    >>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)

    True

    >>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)

    True

    >>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(0, 3600)

    True

    >>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)

    True

    >>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)

    True

    >>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)

    True

    >>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)

    True

    >>> string_to_delta('7d') == datetime.timedelta(7)

    True

    >>> string_to_delta('60s') == datetime.timedelta(0, 60)

    True

    >>> string_to_delta('60m') == datetime.timedelta(0, 3600)

    True

    >>> string_to_delta('24h') == datetime.timedelta(1)

    True

    >>> string_to_delta('1w') == datetime.timedelta(7)

    True

    >>> string_to_delta('5m 30s') == datetime.timedelta(0, 330)

    True

    server_config = configparser.ConfigParser(server_defaults)

    # Convert the ConfigParser object to a dict

    client_config = configparser.ConfigParser(Client.client_defaults)

    if gi.version_info < (3, 10, 2):

        # multiprocessing will use threads, so before we use GLib we

        # need to inform GLib that threads will be used.

        GLib.threads_init()

                        # .name, .host, and .checker_command

                        for k in ("name", "host", "checker_command"):

                        if "key_id" not in value:

                        elif "fingerprint" not in value:

                    # .host and .checker_command

                        for attribute in ("host", "checker_command"):

                            if isinstance(value[attribute], bytes):

                                value[attribute] = (value[attribute]

                                                    .decode("utf-8"))

        GLib.io_add_watch(

            GLib.IOChannel.unix_new(tcp_server.fileno()),

            GLib.PRIORITY_DEFAULT, GLib.IO_IN,

            lambda *args, **kwargs: (tcp_server.handle_request

                                     (*args[2:], **kwargs) or True))

 

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

    try:

        if should_only_run_tests():

            # Call using ./mandos --check [--verbose]

            unittest.main()

        else:

            main()

    finally:

        logging.shutdown()