/mandos/trunk : revision 1227

1

#!/usr/bin/python3 -bbI

2

# -*- coding: utf-8; lexical-binding: t -*-

2

# -*- 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 -*-

3

#

4

# Mandos Control - Control or query the Mandos server

5

#

6

7

6

7

8

#

9

# This file is part of Mandos.

10

#

23

#

24

# Contact the authors at <mandos@recompile.se>.

25

#

26

27

from __future__ import (division, absolute_import, print_function,

27

28

unicode_literals)

28

29

32

33

pass

33

34

35

import sys

35

import unittest

36

import argparse

37

import logging

38

import os

39

37

import locale

40

38

import datetime

41

39

import re

40

import os

42

41

import collections

43

42

import json

43

import unittest

44

import logging

44

45

import io

45

46

import tempfile

46

47

import contextlib

48

49

if sys.version_info.major == 2:

49

50

__metaclass__ = type

50

51

str = unicode

51

input = raw_input

52

53

class gi:

54

"""Dummy gi module, for the tests"""

77

import warnings

78

warnings.simplefilter("default")

79

80

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

80

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

81

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

82

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

83

89

90

locale.setlocale(locale.LC_ALL, "")

91

92

version = "1.8.15"

92

version = "1.8.12"

93

94

95

def main():

102

clientnames = options.client

103

104

if options.debug:

105

logging.getLogger("").setLevel(logging.DEBUG)

105

log.setLevel(logging.DEBUG)

106

107

if dbussy is not None and ravel is not None:

108

bus = dbussy_adapter.CachingBus(dbussy, ravel)

256

return rfc3339_duration_to_delta(interval)

257

except ValueError as e:

258

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

259

" ".join(e.args))

259

' '.join(e.args))

260

return parse_pre_1_6_1_interval(interval)

261

262

395

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

396

and return a datetime.timedelta

397

398

>>> parse_pre_1_6_1_interval("7d") == datetime.timedelta(days=7)

399

True

400

>>> parse_pre_1_6_1_interval("60s") == datetime.timedelta(0, 60)

401

True

402

>>> parse_pre_1_6_1_interval("60m") == datetime.timedelta(hours=1)

403

True

404

>>> parse_pre_1_6_1_interval("24h") == datetime.timedelta(days=1)

405

True

406

>>> parse_pre_1_6_1_interval("1w") == datetime.timedelta(days=7)

407

True

408

>>> parse_pre_1_6_1_interval("5m 30s") == datetime.timedelta(0, 330)

409

True

410

>>> parse_pre_1_6_1_interval("") == datetime.timedelta(0)

398

>>> parse_pre_1_6_1_interval('7d') == datetime.timedelta(days=7)

399

True

400

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

401

True

402

>>> parse_pre_1_6_1_interval('60m') == datetime.timedelta(hours=1)

403

True

404

>>> parse_pre_1_6_1_interval('24h') == datetime.timedelta(days=1)

405

True

406

>>> parse_pre_1_6_1_interval('1w') == datetime.timedelta(days=7)

407

True

408

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

409

True

410

>>> parse_pre_1_6_1_interval('') == datetime.timedelta(0)

411

True

412

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

413

>>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") == datetime.timedelta(2, 480, 18000)

413

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)

414

True

415

416

"""

876

{key: properties[key]

877

for key in self.all_keywords}

878

for properties in clients.values()}

879

print(json.dumps(data, indent=4, separators=(",", ": ")))

879

print(json.dumps(data, indent=4, separators=(',', ': ')))

880

881

882

class PrintTable(Output):

2439

busname = "se.recompile.Mandos"

2440

client_interface = "se.recompile.Mandos.Client"

2441

command.Approve().run(self.bus.clients, self.bus)

2442

self.assertTrue(self.bus.clients)

2443

2442

for clientpath in self.bus.clients:

2444

2443

self.assertIn(("Approve", busname, clientpath,

2445

2444

client_interface, (True,)), self.bus.calls)

2448

2447

busname = "se.recompile.Mandos"

2449

2448

client_interface = "se.recompile.Mandos.Client"

2450

2449

command.Deny().run(self.bus.clients, self.bus)

2451

self.assertTrue(self.bus.clients)

2452

2450

for clientpath in self.bus.clients:

2453

2451

self.assertIn(("Approve", busname, clientpath,

2454

2452

client_interface, (False,)),

2455

2453

self.bus.calls)

2456

2454

2457

2455

def test_Remove(self):

2458

busname = "se.recompile.Mandos"

2459

server_path = "/"

2460

server_interface = "se.recompile.Mandos"

2461

orig_clients = self.bus.clients.copy()

2462

2456

command.Remove().run(self.bus.clients, self.bus)

2463

self.assertFalse(self.bus.clients)

2464

for clientpath in orig_clients:

2465

self.assertIn(("RemoveClient", busname,

2466

server_path, server_interface,

2457

for clientpath in self.bus.clients:

2458

self.assertIn(("RemoveClient", dbus_busname,

2459

dbus_server_path, dbus_server_interface,

2467

2460

(clientpath,)), self.bus.calls)

2468

2461

2469

2462

expected_json = {

2671

2664

else:

2672

2665

cmd_args = [() for x in range(len(self.values_to_get))]

2673

2666

values_to_get = self.values_to_get

2674

self.assertTrue(values_to_get)

2675

2667

for value_to_get, cmd_arg in zip(values_to_get, cmd_args):

2676

2668

for clientpath in self.bus.clients:

2677

2669

self.bus.clients[clientpath][self.propname] = (

2678

2670

Unique())

2679

2671

self.command(*cmd_arg).run(self.bus.clients, self.bus)

2680

self.assertTrue(self.bus.clients)

2681

2672

for clientpath in self.bus.clients:

2682

2673

value = (self.bus.clients[clientpath]

2683

2674

[self.propname])

2742

2733

class TestSetSecretCmd(TestPropertySetterCmd):

2743

2734

command = command.SetSecret

2744

2735

propname = "Secret"

2745

def __init__(self, *args, **kwargs):

2746

self.values_to_set = [io.BytesIO(b""),

2747

io.BytesIO(b"secret\0xyzzy\nbar")]

2748

self.values_to_get = [f.getvalue() for f in

2749

self.values_to_set]

2750

super(TestSetSecretCmd, self).__init__(*args, **kwargs)

2736

values_to_set = [io.BytesIO(b""),

2737

io.BytesIO(b"secret\0xyzzy\nbar")]

2738

values_to_get = [f.getvalue() for f in values_to_set]

2751

2739

2752

2740

2753

2741

class TestSetTimeoutCmd(TestPropertySetterCmd):

2806

2794

2807

2795

2808

2796

2809

def parse_test_args():

2810

# type: () -> argparse.Namespace

2797

def should_only_run_tests():

2811

2798

parser = argparse.ArgumentParser(add_help=False)

2812

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

2813

parser.add_argument("--prefix", )

2799

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

2814

2800

args, unknown_args = parser.parse_known_args()

2815

if args.check:

2816

# Remove test options from sys.argv

2801

run_tests = args.check

2802

if run_tests:

2803

# Remove --check argument from sys.argv

2817

2804

sys.argv[1:] = unknown_args

2818

return args

2805

return run_tests

2819

2806

2820

2807

# Add all tests from doctest strings

2821

2808

def load_tests(loader, tests, none):

2824

2811

return tests

2825

2812

2826

2813

if __name__ == "__main__":

2827

options = parse_test_args()

2828

2814

try:

2829

if options.check:

2830

extra_test_prefix = options.prefix

2831

if extra_test_prefix is not None:

2832

if not (unittest.main(argv=[""], exit=False)

2833

.result.wasSuccessful()):

2834

sys.exit(1)

2835

class ExtraTestLoader(unittest.TestLoader):

2836

testMethodPrefix = extra_test_prefix

2837

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

2838

unittest.main(argv=[""], testLoader=ExtraTestLoader())

2839

else:

2840

unittest.main(argv=[""])

2815

if should_only_run_tests():

2816

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

2817

unittest.main()

2841

2818

else:

2842

2819

main()

2843

2820

finally:

2844

2821

logging.shutdown()

2845

2846

# Local Variables:

2847

# run-tests:

2848

# (lambda (&optional extra)

2849

# (if (not (funcall run-tests-in-test-buffer default-directory

2850

# extra))

2851

# (funcall show-test-buffer-in-test-window)

2852

# (funcall remove-test-window)

2853

# (if extra (message "Extra tests run successfully!"))))

2854

# run-tests-in-test-buffer:

2855

# (lambda (dir &optional extra)

2856

# (with-current-buffer (get-buffer-create "*Test*")

2857

# (setq buffer-read-only nil

2858

# default-directory dir)

2859

# (erase-buffer)

2860

# (compilation-mode))

2861

# (let ((process-result

2862

# (let ((inhibit-read-only t))

2863

# (process-file-shell-command

2864

# (funcall get-command-line extra) nil "*Test*"))))

2865

# (and (numberp process-result)

2866

# (= process-result 0))))

2867

# get-command-line:

2868

# (lambda (&optional extra)

2869

# (let ((quoted-script

2870

# (shell-quote-argument (funcall get-script-name))))

2871

# (format

2872

# (concat "%s --check" (if extra " --prefix=atest" ""))

2873

# quoted-script)))

2874

# get-script-name:

2875

# (lambda ()

2876

# (if (fboundp 'file-local-name)

2877

# (file-local-name (buffer-file-name))

2878

# (or (file-remote-p (buffer-file-name) 'localname)

2879

# (buffer-file-name))))

2880

# remove-test-window:

2881

# (lambda ()

2882

# (let ((test-window (get-buffer-window "*Test*")))

2883

# (if test-window (delete-window test-window))))

2884

# show-test-buffer-in-test-window:

2885

# (lambda ()

2886

# (when (not (get-buffer-window-list "*Test*"))

2887

# (setq next-error-last-buffer (get-buffer "*Test*"))

2888

# (let* ((side (if (>= (window-width) 146) 'right 'bottom))

2889

# (display-buffer-overriding-action

2890

# `((display-buffer-in-side-window) (side . ,side)

2891

# (window-height . fit-window-to-buffer)

2892

# (window-width . fit-window-to-buffer))))

2893

# (display-buffer "*Test*"))))

2894

# eval:

2895

# (progn

2896

# (let* ((run-extra-tests (lambda () (interactive)

2897

# (funcall run-tests t)))

2898

# (inner-keymap `(keymap (116 . ,run-extra-tests))) ; t

2899

# (outer-keymap `(keymap (3 . ,inner-keymap)))) ; C-c

2900

# (setq minor-mode-overriding-map-alist

2901

# (cons `(run-tests . ,outer-keymap)

2902

# minor-mode-overriding-map-alist)))

2903

# (add-hook 'after-save-hook run-tests 90 t))

2904

# End: