source: nscp/scripts/python/test.py

Last change on this file was a629015, checked in by Michael Medin <michael@…>, 20 months ago
  • Started on NSCA unit-tests in python
  • Refactored channel API a bit to better suite NSCA (and how it actually works)
  • Changed how headers are created (nothice this might have broken NSCP stuff, havent checked)
  • Created helper lib for Python Unittests
  • Renamed NSCAAgent to NSCAClient (to be consistant)
  • Created a NSCAServer module (again for consistancy) This will be the defenition of the new "channels" so keep a lookout in the next few weeks
  • Property mode set to 100644
File size: 8.9 KB
Line 
1from NSCP import Settings, Registry, Core, log, status
2from test_helper import Callable, run_tests
3import plugin_pb2
4
5core = Core.get()
6
7prefix = 'py_'
8plugin_id = 0
9
10def get_help(arguments):
11        return (status.OK, 'help: Get help')
12
13class ChannelTest:
14        instance = None
15        channel = ''
16        reg = None
17       
18        last_channel = ''
19        last_command = ''
20        last_status = status.UNKNOWN
21        last_message = ''
22        last_perf = ''
23
24        class SingletonHelper:
25                def __call__( self, *args, **kw ) :
26                        if ChannelTest.instance is None :
27                                object = ChannelTest()
28                                ChannelTest.instance = object
29                        return ChannelTest.instance
30
31        getInstance = SingletonHelper()
32
33        def desc(self):
34                return 'Testing that channels work'
35
36        def test_submission_handler_001(channel, command, code, message, perf):
37                instance = ChannelTest.getInstance()
38                instance.last_channel = channel
39                instance.last_command = command
40                instance.last_status = code
41                instance.last_message = message
42                instance.last_perf = perf
43        test_submission_handler_001 = Callable(test_submission_handler_001)
44               
45        def test_command_handler_001(arguments):
46                instance = ChannelTest.getInstance()
47                return (instance.last_status, instance.last_message, instance.last_perf)
48        test_command_handler_001 = Callable(test_command_handler_001)
49
50        def setup(self, plugin_id, prefix):
51                self.channel = '_%stest_channel'%prefix
52                self.reg = Registry.get(plugin_id)
53                self.reg.simple_subscription('%s_001'%self.channel, ChannelTest.test_submission_handler_001)
54                self.reg.simple_function('%s_001'%self.channel, ChannelTest.test_command_handler_001, 'This is a sample command')
55               
56        def teardown(self):
57                None
58                #self.reg.unregister_simple_subscription('%s_001'%self.channel)
59                #self.reg.unregister_simple_function('%s_001'%self.channel)
60
61        def test_simple(self, channel, command, code, message, perf, tag):
62                core.simple_submit('%s'%channel, '%s'%command, code, '%s'%message, '%s'%perf)
63                (retcode, retmessage, retperf) = core.simple_query(channel, [])
64                isok = True
65                if retcode != code:
66                        log('FAILED - Test did not get the correct retuirn code: %s = %s (%s)'%(retcode, code, retmessage))
67                        isok = False
68                if retmessage != message:
69                        log('FAILED - Test did not get the correct retuirn code: %s = %s'%(retmessage, message))
70                        isok = False
71                if retperf != perf:
72                        log('FAILED - Test did not get the correct retuirn code: %s = %s'%(retperf, perf))
73                        isok = False
74                if isok:
75                        log('OK - Test successfull: %s'%tag)
76                        return 0
77                return 1
78               
79
80        def run_test(self):
81                count = 0
82                count += self.test_simple('%s_001'%self.channel, 'foobar', status.OK, 'qwerty', '', 'simple ok')
83                count += self.test_simple('%s_001'%self.channel, 'foobar', status.WARNING, 'foobar', '', 'simple warning')
84                count += self.test_simple('%s_001'%self.channel, 'foobar', status.CRITICAL, 'test', '', 'simple critical')
85                count += self.test_simple('%s_001'%self.channel, 'foobar', status.UNKNOWN, '1234567890', '', 'simple unknown')
86                count += self.test_simple('%s_001'%self.channel, 'foobar', status.OK, 'qwerty', "'foo'=5%", 'simple performance data 001')
87                count += self.test_simple('%s_001'%self.channel, 'foobar', status.OK, 'qwerty', "'foo'=5%;10", 'simple performance data 002')
88                count += self.test_simple('%s_001'%self.channel, 'foobar', status.OK, 'qwerty', "'foo'=5%;10;23", 'simple performance data 003')
89                count += self.test_simple('%s_001'%self.channel, 'foobar', status.OK, 'qwerty', "'foo'=5%;10;23;10;78", 'simple performance data 004')
90                count += self.test_simple('%s_001'%self.channel, 'foobar', status.OK, 'qwerty', "'foo'=5%;10;23;10;78 'bar'=1k;2;3", 'simple performance data 005')
91                if count > 0:
92                        log("ERROR: %d tests failed"%count)
93                else:
94                        log("OK: all tests successfull")
95                return (count, 9)
96
97
98class CommandTest:
99        instance = None
100        key = ''
101        reg = None
102       
103        class SingletonHelper:
104                def __call__( self, *args, **kw ) :
105                        if CommandTest.instance is None :
106                                object = CommandTest()
107                                CommandTest.instance = object
108                        return CommandTest.instance
109
110        getInstance = SingletonHelper()
111
112        def desc(self):
113                return 'Testing that channels work'
114
115        def test_command_handler_001(arguments):
116                if len(arguments) == 0:
117                        return (status.OK, 'no-arguments', '')
118                retcode = status.UNKNOWN
119                message = ''
120                perf = ''
121                if len(arguments) > 0:
122                        if arguments[0] == 'OK':
123                                retcode = status.OK
124                        elif arguments[0] == 'WARNING':
125                                retcode = status.WARNING
126                        elif arguments[0] == 'CRITICAL':
127                                retcode = status.CRITICAL
128                        elif arguments[0] == 'UNKNOWN':
129                                retcode = status.UNKNOWN
130                        message = 'arg-count: %d'%len(arguments)
131                return (retcode, message, perf)
132        test_command_handler_001 = Callable(test_command_handler_001)
133
134        def setup(self, plugin_id, prefix):
135                self.key = '_%stest_command'%prefix
136                self.reg = Registry.get(plugin_id)
137                self.reg.simple_function('%s_001'%self.key, CommandTest.test_command_handler_001, 'This is a sample command')
138               
139        def teardown(self):
140                None
141                #self.reg.unregister_simple_function('%s_001'%self.key)
142
143        def test_simple(self, key, args, code, message, perf, tag):
144                (retcode, retmessage, retperf) = core.simple_query(key, args)
145                isok = True
146                if code and retcode != code:
147                        log('FAILED - Test did not return correct values (code): %s = %s (%s)'%(retcode, code, retmessage))
148                        isok = False
149                if message and retmessage != message:
150                        log('FAILED - Test did not return correct values (message): %s = %s'%(retmessage, message))
151                        isok = False
152                if perf and retperf != perf:
153                        log('FAILED - Test did not return correct values (perf): %s = %s'%(retperf, perf))
154                        isok = False
155                if isok:
156                        log('OK - Test successfull: %s'%tag)
157                        return 0
158                return 1
159               
160
161        def run_test(self):
162                count = 0
163                key001 = '%s_001'%self.key
164                count += self.test_simple(key001, [], status.OK, 'no-arguments', '', 'simple check')
165                count += self.test_simple(key001, ['OK'], status.OK, 'arg-count: 1', None, 'simple check: Ok')
166                count += self.test_simple(key001, ['WARNING'], status.WARNING, 'arg-count: 1', None, 'simple check: Warn')
167                count += self.test_simple(key001, ['UNKNOWN'], status.UNKNOWN, 'arg-count: 1', None, 'simple check: Unknown')
168                count += self.test_simple(key001, ['CRITICAL'], status.CRITICAL, 'arg-count: 1', None, 'simple check: Crit')
169                if count > 0:
170                        log("ERROR: %d tests failed"%count)
171                else:
172                        log("OK: all tests successfull")
173                return (count, 9)
174
175def test(arguments):
176        global prefix
177        global plugin_id
178
179        (all_failed, all_count) = run_tests([ChannelTest, CommandTest])
180        if all_failed == 0:
181                return (status.OK, 'All tests ok: %d'%all_count)
182        else:
183                return (status.CRITICAL, 'Tests failed %d of %d'%(all_failed, all_count))
184
185def install_test(arguments):
186        log('-+---==(TEST INSTALLER)==---------------------------------------------------+-')
187        log(' | Setup nessecary configuration for running test                           |')
188        log(' | This includes: Loading the PythonScript module at startup                |')
189        log(' | To use this please run nsclient++ in "test mode" like so:                |')
190        log(' | nscp --test                                                              |')
191        log(' | Then start the pytest_test command by typing it and press enter like so: |')
192        log(' | pytest_test                                                              |')
193        log(' | Lastly exit by typing exit like so:                                      |')
194        log(' | exit                                                                     |')
195        log('-+--------------------------------------------------------==(DAS ENDE!)==---+-')
196        conf = Settings.get()
197        conf.set_string('/modules', 'pytest', 'PythonScript')
198        conf.set_string('/settings/pytest/scripts', 'pytest', 'test.py')
199        conf.save()
200
201def __main__():
202        install_test([])
203       
204def init(pid, plugin_alias, script_alias):
205        global prefix
206        global plugin_id
207        plugin_id = pid
208        if script_alias:
209                prefix = '%s_'%script_alias
210
211        conf = Settings.get()
212        #val = conf.get_string('/modules', 'PythonScript', 'foo')
213
214        #log('Got it: %s'%val)
215       
216        log('Testing to register a function')
217        reg = Registry.get(plugin_id)
218       
219        reg.simple_cmdline('help', get_help)
220        reg.simple_cmdline('install_python_test', install_test)
221
222        reg.simple_function('%stest'%prefix, test, 'Run python unittest')
223
224        #core.simple_submit('%stest'%prefix, 'test.py', status.WARNING, 'hello', '')
225        #core.simple_submit('test', 'test.py', status.WARNING, 'hello', '')
226       
227        #(ret, list) = core.simple_exec('%stest'%prefix, ['a', 'b', 'c'])
228        #for l in list:
229        #       log('-- %s --'%l)
230
231        #log('Testing to register settings keys')
232        #conf.register_path('hello', 'PYTHON SETTINGS', 'This is stuff for python')
233        #conf.register_key('hello', 'python', 'int', 'KEY', 'This is a key', '42')
234
235        #log('Testing to get key (nonexistant): %d' % conf.get_int('hello', 'python', -1))
236        #conf.set_int('hello', 'python', 4)
237        #log('Testing to get it (after setting it): %d' % conf.get_int('hello', 'python', -1))
238
239        #log('Saving configuration...')
240        #conf.save()
241
242def shutdown():
243        log('Unloading script...')
Note: See TracBrowser for help on using the repository browser.