File: test_modules.py

package info (click to toggle)
pyinfra 0.2.2%2Bgit20161227.ec708ef-1
  • links: PTS, VCS
  • area: main
  • in suites: stretch
  • size: 11,804 kB
  • ctags: 677
  • sloc: python: 5,944; sh: 71; makefile: 11
file content (121 lines) | stat: -rw-r--r-- 3,978 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
# pyinfra
# File: tests/test_modules.py
# Desc: generate tests for module operations

from __future__ import print_function

import json
from os import listdir, path
from unittest import TestCase
from importlib import import_module
from types import FunctionType

import six
from nose.tools import nottest
from jsontest import JsonTest

from pyinfra import pseudo_state, pseudo_host
from pyinfra.cli import json_encode
from pyinfra.api.util import unroll_generators

from .util import FakeState, create_host, patch_files


@nottest
def make_operation_tests(arg):
    # Get the operation we're testing against
    module_name, op_name = arg.split('.')
    module = import_module('pyinfra.modules.{0}'.format(module_name))
    op = getattr(module, op_name)

    # Generate a test class
    @six.add_metaclass(JsonTest)
    class TestTests(TestCase):
        jsontest_files = path.join('tests', 'operations', arg)

        @classmethod
        def setUpClass(cls):
            # Create a global fake state that attach to pseudo state
            cls.state = FakeState()
            pseudo_state.set(cls.state)

        def jsontest_function(self, test_name, test_data):
            # Create a host with this tests facts and attach to pseudo host
            host = create_host(facts=test_data.get('facts', {}))
            pseudo_host.set(host)

            allowed_exception = test_data.get('exception')

            with patch_files(test_data.get('files', [])):
                try:
                    output_commands = unroll_generators(op._pyinfra_op(
                        pseudo_state, pseudo_host,
                        *test_data.get('args', []), **test_data.get('kwargs', {})
                    )) or []
                except Exception as e:
                    if allowed_exception:
                        if e.__class__.__name__ != allowed_exception['name']:
                            print('Wrong execption raised!')
                            raise

                        self.assertEqual(e.args[0], allowed_exception['message'])
                        return

                    raise

            commands = []

            for command in output_commands:
                if isinstance(command, six.string_types):
                    commands.append(command.strip())

                elif isinstance(command, dict):
                    command['command'] = command['command'].strip()
                    commands.append(command)

                elif isinstance(command, tuple):
                    if isinstance(command[0], FunctionType):
                        commands.append(command)
                    else:
                        if hasattr(command[0], 'read'):
                            data = command[0].read()
                        else:
                            data = command[0]

                        commands.append([data, command[1]])

                else:
                    commands.append(command)

            try:
                self.assertEqual(commands, test_data['commands'])
            except AssertionError as e:
                print()
                print('--> GOT:\n', json.dumps(commands, indent=4, default=json_encode))
                print('--> WANT:', json.dumps(
                    test_data['commands'], indent=4, default=json_encode
                ))
                raise e

    # Convert the op name (module.op) to a class name ModuleOp
    test_name = (
        arg.replace('.', ' ')
        .title()
        .replace(' ', '')
    )

    # Set the name
    TestTests.__name__ = 'TestOperation{0}'.format(test_name)
    return TestTests


# Find available operation tests
operations = [
    filename
    for filename in listdir(path.join('tests', 'operations'))
    if path.isdir(path.join('tests', 'operations', filename))
]

# Generate the classes, attaching to locals so nose picks them up
for operation_name in operations:
    locals()[operation_name] = make_operation_tests(operation_name)