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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
#!/usr/bin/env python
"""This tool verifies that all clients known to the server build
without failures"""
import os
import sys
import signal
import fnmatch
import logging
import Bcfg2.Logger
import Bcfg2.Server.Core
from nose.core import TestProgram
from nose.suite import LazySuite
from unittest import TestCase
class ClientTest(TestCase):
"""
A test case representing the build of all of the configuration for
a single host. Checks that none of the build config entities has
had a failure when it is building. Optionally ignores some config
files that we know will cause errors (because they are private
files we don't have access to, for instance)
"""
__test__ = False # Do not collect
def __init__(self, bcfg2_core, client, ignore=None):
TestCase.__init__(self)
self.bcfg2_core = bcfg2_core
self.client = client
if ignore is None:
self.ignore = dict()
else:
self.ignore = ignore
def ignore_entry(self, tag, name):
""" return True if an error on a given entry should be ignored
"""
if tag in self.ignore:
if name in self.ignore[tag]:
return True
else:
# try wildcard matching
for pattern in self.ignore[tag]:
if fnmatch.fnmatch(name, pattern):
return True
return False
def shortDescription(self):
return "Building configuration for %s" % self.client
def runTest(self):
""" run this individual test """
config = self.bcfg2_core.BuildConfiguration(self.client)
assert len(config.findall("Bundle")) > 0, \
"%s has no content" % self.client
failures = []
msg = ["Failures:"]
for failure in config.xpath('//*[@failure]'):
if not self.ignore_entry(failure.tag, failure.get('name')):
failures.append(failure)
msg.append("%s:%s: %s" % (failure.tag, failure.get("name"),
failure.get("failure")))
assert len(failures) == 0, "\n".join(msg)
def __str__(self):
return "ClientTest(%s)" % self.client
id = __str__
def get_sigint_handler(core):
""" Get a function that handles SIGINT/Ctrl-C by shutting down the
core and exiting properly."""
def hdlr(sig, frame): # pylint: disable=W0613
""" Handle SIGINT/Ctrl-C by shutting down the core and exiting
properly. """
core.shutdown()
os._exit(1) # pylint: disable=W0212
return hdlr
def main():
optinfo = dict(noseopts=Bcfg2.Options.TEST_NOSEOPTS,
test_ignore=Bcfg2.Options.TEST_IGNORE,
validate=Bcfg2.Options.CFG_VALIDATION)
optinfo.update(Bcfg2.Options.CLI_COMMON_OPTIONS)
optinfo.update(Bcfg2.Options.SERVER_COMMON_OPTIONS)
setup = Bcfg2.Options.OptionParser(optinfo)
setup.hm = \
"bcfg2-test [options] [client] [client] [...]\nOptions:\n %s" % \
setup.buildHelpMessage()
setup.parse(sys.argv[1:])
if setup['debug']:
level = logging.DEBUG
elif setup['verbose']:
level = logging.INFO
else:
level = logging.WARNING
Bcfg2.Logger.setup_logging("bcfg2-test",
to_console=setup['verbose'] or setup['debug'],
to_syslog=False,
to_file=setup['logging'],
level=level)
if (setup['debug'] or setup['verbose']) and "-v" not in setup['noseopts']:
setup['noseopts'].append("-v")
core = Bcfg2.Server.Core.BaseCore(setup)
signal.signal(signal.SIGINT, get_sigint_handler(core))
ignore = dict()
for entry in setup['test_ignore']:
tag, name = entry.split(":")
try:
ignore[tag].append(name)
except KeyError:
ignore[tag] = [name]
core.fam.handle_events_in_interval(0.1)
if setup['args']:
clients = setup['args']
else:
clients = core.metadata.clients
def run_tests():
""" Run the test suite """
for client in clients:
yield ClientTest(core, client, ignore)
TestProgram(argv=sys.argv[0:1] + setup['noseopts'],
suite=LazySuite(run_tests))
core.shutdown()
os._exit(0) # pylint: disable=W0212
if __name__ == "__main__":
sys.exit(main())
|
