source: nscp/scripts/python/test_eventlog.py @ 465866c

0.4.10.4.2
Last change on this file since 465866c was 465866c, checked in by Michael Medin <michael@…>, 13 months ago

2012-06-05 MickeM

  • Tweaked all servers to use the new internals and added first testcase for NSCP socket

2012-05-24 MickeM

  • Reworked real time event log support to be a lot more flexible You can now specify all options on a "filter" level.
  • WARNING* Old syntax NOT supported (and will not upgrade) but hopefully not to many will be affected.
  • Added support for ipv6 allowed hosts validation

2012-05-21 MickeM

  • Sofia Born (My second daughter)
  • Property mode set to 100644
File size: 9.4 KB
Line 
1from NSCP import Settings, Registry, Core, log, status, sleep
2from test_helper import BasicTest, TestResult, Callable, setup_singleton, install_testcases, init_testcases, shutdown_testcases
3import plugin_pb2
4from types import *
5import socket
6import unicodedata
7
8class EventLogTest(BasicTest):
9        instance = None
10        key = ''
11        reg = None
12        got_response = False
13        last_source = None
14        last_command = None
15        last_status = None
16        last_message = None
17        last_perfdata = None
18        last_tag = []
19        got_simple_response = None
20        message_count = 0
21       
22        class SingletonHelper:
23                def __call__( self, *args, **kw ) :
24                        if EventLogTest.instance is None :
25                                object = EventLogTest()
26                                EventLogTest.instance = object
27                        return EventLogTest.instance
28
29        getInstance = SingletonHelper()
30
31        def desc(self):
32                return 'Testcase for eventlog'
33
34        def title(self):
35                return 'EventLog test'
36
37        def setup(self, plugin_id, prefix):
38                self.key = '_%stest_command'%prefix
39                self.reg = Registry.get(plugin_id)
40                self.reg.simple_subscription('pytest_evlog_01', EventLogTest.simple_inbox_handler_01)
41                self.reg.simple_subscription('pytest_evlog_02', EventLogTest.simple_inbox_handler_02)
42
43        def simple_inbox_handler_01(channel, source, command, code, message, perf):
44                instance = EventLogTest.getInstance()
45                return instance.simple_inbox_handler_wrapped(channel, source, command, code, message, perf, '001')
46        simple_inbox_handler_01 = Callable(simple_inbox_handler_01)
47
48        def simple_inbox_handler_02(channel, source, command, code, message, perf):
49                instance = EventLogTest.getInstance()
50                return instance.simple_inbox_handler_wrapped(channel, source, command, code, message, perf, '002')
51        simple_inbox_handler_02 = Callable(simple_inbox_handler_02)
52       
53        def simple_inbox_handler_wrapped(self, channel, source, command, status, message, perf, tag):
54                message = unicodedata.normalize('NFKD', message).encode('ascii','ignore')
55                log('Got simple message %s on %s'%(command, channel))
56                self.got_simple_response = True
57                self.last_source = source
58                self.last_command = command
59                self.last_status = status
60                self.last_message = message
61                if self.last_tag:
62                        self.last_tag.append(tag)
63                else:
64                        self.last_tag = [ tag ]
65                self.message_count = self.message_count + 1
66                self.last_perfdata = perf
67                return True
68
69        def teardown(self):
70                None
71
72        def test_create(self, source, id, level, severity, category, facility, arguments):
73                self.last_message = None
74                self.message_count = 0
75                args = ['--source', source,
76                                        '--id', id,
77                                        '--level', level,
78                                        '--severity', severity,
79                                        '--category', category,
80                                        '--facility', facility
81                                        ]
82                for f in arguments:
83                        args.append('--argument')
84                        args.append(f)
85                (ret, msg) = Core.get().simple_exec('any', 'insert-eventlog', args)
86                return ret == 0
87               
88               
89        def test_w_expected(self, filter, syntax, expected):
90                result = TestResult('Validating filter: %s'%filter)
91                (res, msg, perf) = Core.get().simple_query('CheckEventLog', ['file=Application', 'debug=true', 'warn=ne:%d'%expected, 'crit=ne:%d'%expected, 'filter=%s'%filter, 'syntax=%s'%syntax])
92                result.assert_equals(res, status.OK, "Validate status OK for %s"%filter)
93                (res, msg, perf) = Core.get().simple_query('CheckEventLog', ['file=Application', 'debug=true', 'warn=eq:%d'%expected, 'crit=ne:%d'%expected, 'filter=%s'%filter, 'syntax=%s'%syntax])
94                result.assert_equals(res, status.WARNING, "Validate status OK for %s"%filter)
95                (res, msg, perf) = Core.get().simple_query('CheckEventLog', ['file=Application', 'debug=true', 'warn=eq:%d'%expected, 'crit=eq:%d'%expected, 'filter=%s'%filter, 'syntax=%s'%syntax])
96                result.assert_equals(res, status.CRITICAL, "Validate status CRIT for %s"%filter)
97                return result
98
99        def test_syntax(self, filter, syntax, expected):
100                result = TestResult('Validating syntax: %s'%syntax)
101                (res, msg, perf) = Core.get().simple_query('CheckEventLog', ['file=Application', 'warn=ne:1', 'filter=%s'%filter, 'syntax=%s'%syntax, 'descriptions'])
102                result.assert_equals(msg, expected, "Validate message rendering syntax: %s"%msg)
103                return result
104               
105        def run_test(self):
106                result = TestResult('Checking CheckEventLog')
107                cache = TestResult('Checking CheckEventLog CACHE')
108               
109                (res, msg, perf) = Core.get().simple_query('CheckEventLogCACHE', ['warn=eq:1', 'crit=eq:2'])
110                cache.assert_equals(res, status.OK, "Validate cache is empty")
111                cache.assert_equals(msg, 'Eventlog check ok', "Validate cache is ok: %s"%msg)
112               
113               
114                a_list = ['a', 'a', 'a', 'a', 'a', 'a', 'a', 'a', 'a', 'a', 'a', 'a', 'a']
115                result.add_message(self.test_create('Application Error', 1000, 'error', 0, 0, 0, a_list), 'Testing to create a log message')
116                sleep(500)
117                result.assert_equals(self.last_message, 'error Application Error: ', 'Verify that message is sent through')
118                result.assert_equals(self.message_count, 1, 'Verify that only one message is sent through')
119                log('Got tags: %s'%self.last_tag)
120
121                result.add_message(self.test_create('Application Error', 1000, 'info', 2, 1, 5, a_list), 'Testing to create a log message')
122                sleep(500)
123                result.assert_equals(self.last_message, 'info Application Error: ', 'Verify that message is sent through')
124                result.assert_equals(self.message_count, 1, 'Verify that onlyt one message is sent through')
125
126                (res, msg, perf) = Core.get().simple_query('CheckEventLogCACHE', ['warn=eq:1', 'crit=eq:2'])
127                cache.assert_equals(res, status.CRITICAL, "Validate cache has items: %s"%msg)
128                cache.assert_equals(msg, 'error Application Error: , info Application Error: , eventlog: 2 = critical', "Validate cache is ok: %s"%msg)
129                cache.assert_equals(perf, "'eventlog'=2;1;2", "Validate cache is ok: %s"%msg)
130                (res, msg, perf) = Core.get().simple_query('CheckEventLogCACHE', ['warn=eq:1', 'crit=eq:2'])
131                cache.assert_equals(res, status.OK, "Validate cache is empty (again)")
132                cache.assert_equals(msg, 'Eventlog check ok', "Validate cache is ok: %s"%msg)
133               
134                result.add(cache)
135               
136                r = TestResult('Checking filters')
137                r.add(self.test_w_expected('id = 1000 and generated gt 1m', '%generated%', 0))
138                r.add(self.test_w_expected('id = 1000 and generated gt -1m', '%generated%', 2))
139                r.add(self.test_w_expected('id = 1000 and generated gt -1m and id = 1000', '%generated%: %id%, %category%', 2))
140                r.add(self.test_w_expected('id = 1000 and generated gt -1m and category = 1', '%category%', 1))
141                r.add(self.test_w_expected('id = 1000 and generated gt -1m and category = 0', '%category%', 1))
142                r.add(self.test_w_expected("id = 1000 and generated gt -1m and level = 'error'", '%level%', 1))
143                r.add(self.test_w_expected("id = 1000 and generated gt -1m and level = 'info'", '%level%', 1))
144                result.add(r)
145               
146                r = TestResult('Checking syntax')
147                r.add(self.test_syntax('id = 1000 and generated gt -2m and category = 0', '%source% - %type% - %category%', 'Application Error - error - 0'))
148                r.add(self.test_syntax('id = 1000 and generated gt -2m and category = 1', '%source% - %type% - %category%', 'Application Error - info - 1'))
149                r.add(self.test_syntax('id = 1000 and generated gt -2m and category = 0', '%facility% - %qualifier% - %customer%', '0 - 0 - 0'))
150                r.add(self.test_syntax('id = 1000 and generated gt -2m and category = 1', '%facility% - %qualifier% - %customer%', '5 - 5 - 0'))
151                r.add(self.test_syntax('id = 1000 and generated gt -2m and category = 0', '%rawid% - %severity% - %log%', '1000 - success - Application'))
152                r.add(self.test_syntax('id = 1000 and generated gt -2m and category = 1', '%rawid% - %severity% - %log%', '2147812328 - warning - Application'))
153                r.add(self.test_syntax('id = 1000 and generated gt -2m and category = 0', '%id% - %strings%', '1000 - a, a, a, a, a, a, a, a, a, a, a, a, a, '))
154                r.add(self.test_syntax('id = 1000 and generated gt -2m and category = 1', '%id% - %strings%', '1000 - a, a, a, a, a, a, a, a, a, a, a, a, a, '))
155                result.add(r)
156
157                return result
158
159        def install_filter(self, conf, path, target, filter):
160                conf.set_string(path, 'filter', filter)
161                conf.set_string(path, 'maximum age', '5s')
162                conf.set_string(path, 'destination', target)
163                conf.set_string(path, 'language', 'english')
164                conf.set_string(path, 'debug', 'true')
165       
166        def install(self, arguments):
167                conf = Settings.get()
168                conf.set_string('/modules', 'pytest_eventlog', 'CheckEventLog')
169                conf.set_string('/modules', 'pytest', 'PythonScript')
170
171                conf.set_string('/settings/pytest/scripts', 'test_eventlog', 'test_eventlog.py')
172               
173                conf.set_string('/settings/pytest_eventlog/real-time', 'enabled', 'true')
174               
175                self.install_filter(conf, '/settings/pytest_eventlog/real-time/filters/py_test_001', 'pytest_evlog_01', 'id = 1000 and category = 0')
176                self.install_filter(conf, '/settings/pytest_eventlog/real-time/filters/py_test_002', 'pytest_evlog_02', 'id = 1000 and category = 1')
177               
178                conf.set_string('/settings/pytest_eventlog/real-time', 'maximum age', '5s')
179                conf.set_string('/settings/pytest_eventlog/real-time', 'debug', 'true')
180                conf.set_string('/settings/pytest_eventlog/real-time', 'enable active', 'true')
181               
182                conf.save()
183       
184        def uninstall(self):
185                None
186
187        def help(self):
188                None
189
190        def init(self, plugin_id):
191                None
192                #reg = Registry.get(plugin_id)
193                #reg.simple_function('test_eventlog', test, 'Run python EventLog unit test suite')
194
195        def shutdown(self):
196                None
197
198setup_singleton(EventLogTest)
199
200all_tests = [EventLogTest]
201
202def __main__():
203        install_testcases(all_tests)
204       
205def init(plugin_id, plugin_alias, script_alias):
206        init_testcases(plugin_id, plugin_alias, script_alias, all_tests)
207
208def shutdown():
209        shutdown_testcases()
Note: See TracBrowser for help on using the repository browser.