source: main/waeup.sirp/trunk/src/waeup/sirp/utils/tests/test_logger.py @ 7527

Last change on this file since 7527 was 7196, checked in by Henrik Bettermann, 13 years ago

More copyright adjustments (job finished).
  • Property svn:keywords set to Id
File size: 14.6 KB
Line 
1## $Id: test_logger.py 7196 2011-11-25 07:44:52Z henrik $
2##
3## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann
4## This program is free software; you can redistribute it and/or modify
5## it under the terms of the GNU General Public License as published by
6## the Free Software Foundation; either version 2 of the License, or
7## (at your option) any later version.
8##
9## This program is distributed in the hope that it will be useful,
10## but WITHOUT ANY WARRANTY; without even the implied warranty of
11## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12## GNU General Public License for more details.
13##
14## You should have received a copy of the GNU General Public License
15## along with this program; if not, write to the Free Software
16## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17##
18
19# Tests for waeup.sirp.utils.logger
20import logging
21import os
22import shutil
23import tempfile
24from zope.component import queryUtility
25from zope.component.hooks import setSite, clearSite
26from zope.interface.verify import verifyClass, verifyObject
27from waeup.sirp.app import University
28from waeup.sirp.testing import FunctionalLayer, FunctionalTestCase
29
30from waeup.sirp.utils.logger import (
31    Logger, MAX_BYTES, BACKUP_COUNT, ILoggerCollector, LoggerCollector,
32    ILogger)
33
34class FakeComponent(object):
35    # A component that can use the Logger mixin-class
36    def do_something(self):
37        self.logger.info('msg from do_something')
38
39class LoggingComponent(FakeComponent, Logger):
40    # A component that supports logging by mixing in Logger
41    logger_name = 'sample.${sitename}'
42    logger_filename = 'sample.log'
43
44class LoggersTests(FunctionalTestCase):
45
46    layer = FunctionalLayer
47
48    def setUp(self):
49        super(LoggersTests, self).setUp()
50        # Setup a sample site for each test
51        app = University()
52        self.dc_root = tempfile.mkdtemp()
53        self.dc_root2 = None
54        app['datacenter'].setStoragePath(self.dc_root)
55
56        # Prepopulate the ZODB...
57        self.getRootFolder()['app'] = app
58        self.app = self.getRootFolder()['app']
59        self.workdir = tempfile.mkdtemp()
60        return
61
62    def tearDown(self):
63        super(LoggersTests, self).tearDown()
64        shutil.rmtree(self.workdir)
65        shutil.rmtree(self.dc_root)
66        clearSite()
67        return
68
69    def clear_logger(self, logger):
70        del_list = [x for x in logger.handlers]
71        for handler in del_list:
72            logger.removeHandler(handler)
73        return
74
75    def test_iface(self):
76        setSite(self.app)
77        logger = LoggingComponent()
78        verifyClass(ILogger, Logger)
79        verifyObject(ILogger, logger)
80        return
81
82    def test_logger_prop_no_site(self):
83        # If not in a site, we will get a logger without handlers
84        logger = Logger().logger
85        self.assertTrue(isinstance(logger, logging.Logger))
86        self.assertEqual(len(logger.handlers), 0)
87        return
88
89    def test_site_no_site_name(self):
90        # If a site has not name we will get a logger anyway
91        setSite(self.app)
92        del self.app.__name__
93        logger = Logger().logger
94        self.assertTrue(logger is not None)
95        return
96
97    def test_get_logdir(self):
98        # We can get a logdir
99        setSite(self.app)
100        logger = Logger()
101        logdir = logger.logger_get_logdir()
102        expected_logdir = os.path.join(
103            self.app['datacenter'].storage, 'logs')
104        self.assertEqual(logdir, expected_logdir)
105        return
106
107    def test_get_logdir_no_site(self):
108        # We cannot get a logdir without a site
109        logger = Logger()
110        logdir = logger.logger_get_logdir()
111        self.assertTrue(logdir is None)
112        return
113
114    def test_get_logdir_no_datacenter(self):
115        # Having a site, is not enough. We also need a datacenter in it.
116        setSite(self.app)
117        del self.app['datacenter']
118        logger = Logger()
119        logdir = logger.logger_get_logdir()
120        self.assertTrue(logdir is None)
121        return
122
123    def test_get_logdir_create_dir(self):
124        # If the logdirectory does not exist, we will create it
125        setSite(self.app)
126        logger = Logger()
127        expected_logdir = os.path.join(
128            self.app['datacenter'].storage, 'logs')
129        if os.path.exists(expected_logdir):
130            shutil.rmtree(expected_logdir)
131        logdir = logger.logger_get_logdir()
132        exists_after = os.path.exists(expected_logdir)
133        self.assertEqual(logdir, expected_logdir)
134        self.assertTrue(exists_after is True)
135        return
136
137    def test_get_logfile(self):
138        # We can get a logfile
139        setSite(self.app)
140        logger = Logger()
141        expected_filepath = os.path.join(
142            self.app['datacenter'].storage, 'logs', 'main.log')
143        path = logger.logger_get_logfile_path()
144        self.assertEqual(expected_filepath, path)
145        return
146
147    def test_get_logfile_no_dir(self):
148        # We cannot get a logfilepath if there is no dir for it
149        setSite(self.app)
150        logger = Logger()
151        del self.app['datacenter']
152        path = logger.logger_get_logfile_path()
153        self.assertTrue(path is None)
154        return
155
156    def test_setup(self):
157        # We can setup a logger.
158        mylogger = logging.getLogger('test.sample')
159        setSite(self.app)
160        logger = Logger()
161        result = logger.logger_setup(mylogger)
162        self.assertEqual(len(result.handlers), 1)
163        handler = result.handlers[0]
164        self.assertEqual(handler.maxBytes, MAX_BYTES)
165        self.assertEqual(handler.backupCount, BACKUP_COUNT)
166        self.assertTrue(result.propagate is False)
167        return
168
169    def test_setup_no_site(self):
170        # Without a site we get no logger from setup
171        mylogger = logging.getLogger('test.sample')
172        logger = Logger()
173        result = logger.logger_setup(mylogger)
174        self.assertTrue(result is None)
175        return
176
177    def test_logfile_change(self):
178        # When the logfile location changes, logger_logfile_changed can react
179        setSite(self.app)
180        logger = Logger()
181        logger.logger.warn('Warning 1')  # Log something to old logfile
182        filename1 = logger.logger.handlers[0].baseFilename
183        content1 = open(filename1, 'r').read()
184        # Now move the logfile
185        self.dc_root2 = tempfile.mkdtemp()
186        self.app['datacenter'].setStoragePath(self.dc_root2)
187        logger.logger_logfile_changed()
188        logger.logger.warn('Warning 2')  # Log something to new logfile
189        filename2 = logger.logger.handlers[0].baseFilename
190        content2 = open(filename2, 'r').read()
191        self.assertTrue('Warning 1' in content1)
192        self.assertTrue('Warning 2' in content2)
193        self.assertTrue('Warning 1' not in content2)
194        self.assertTrue('Warning 2' not in content1)
195        self.assertTrue(filename1 != filename2)
196        return
197
198    def test_logger_additional_handlers(self):
199        # When we detect additional handlers in a logger, setup anew.
200        setSite(self.app)
201        logger = Logger()
202        old_py_logger = logger.logger
203        handler = logging.StreamHandler()
204        logger.logger.addHandler(handler)
205        old_handler_num = len(old_py_logger.handlers)
206        new_py_logger = logger.logger  # This should detect new handler
207        new_handler_num = len(new_py_logger.handlers)
208        self.assertEqual(new_handler_num, 1)
209        self.assertTrue(isinstance(
210                new_py_logger.handlers[0],
211                logging.handlers.RotatingFileHandler))
212
213class LoggersFunctionalTests(FunctionalTestCase):
214    # Check loggers with real components using them
215
216    layer = FunctionalLayer
217
218    def setUp(self):
219        super(LoggersFunctionalTests, self).setUp()
220
221        # Setup a sample site for each test
222        app = University()
223        self.dc_root = tempfile.mkdtemp()
224        app['datacenter'].setStoragePath(self.dc_root)
225
226        # Prepopulate the ZODB...
227        self.getRootFolder()['app'] = app
228        self.app = self.getRootFolder()['app']
229        self.workdir = tempfile.mkdtemp()
230
231        self.app['mycomponent'] = LoggingComponent()
232        self.component = self.app['mycomponent']
233        setSite(self.app)
234        return
235
236    def tearDown(self):
237        super(LoggersFunctionalTests, self).tearDown()
238        shutil.rmtree(self.workdir)
239        shutil.rmtree(self.dc_root)
240        clearSite()
241        return
242
243    def test_component_can_get_logger(self):
244        # Components can get a logger
245        logger = self.component.logger
246        self.assertTrue(logger is not None)
247        self.assertEqual(len(logger.handlers), 1)
248        self.assertEqual(logger.name, 'sample.app')
249        return
250
251    def test_component_handler_setup(self):
252        # handlers of components are setup
253        handler = self.component.logger.handlers[0]
254        self.assertTrue(isinstance(
255                handler, logging.handlers.RotatingFileHandler))
256        self.assertEqual(handler.maxBytes, MAX_BYTES)
257        self.assertEqual(handler.backupCount, BACKUP_COUNT)
258        self.assertTrue(handler.baseFilename.endswith('logs/sample.log'))
259        return
260
261    def test_component_can_log(self):
262        # a method of some 'loggerized' instance can log like this.
263        self.component.logger.setLevel(logging.INFO)
264        self.component.do_something()
265        handler = logging.getLogger('sample.app').handlers[0]
266        logfile = handler.baseFilename
267        content = open(logfile, 'rb').read()
268        self.assertTrue(content.endswith(' - INFO - msg from do_something\n'))
269        return
270
271class LoggerCollectorTests(FunctionalTestCase):
272
273    layer = FunctionalLayer
274
275    def setUp(self):
276        super(LoggerCollectorTests, self).setUp()
277        # Setup a sample site for each test
278        app = University()
279        self.dc_root = tempfile.mkdtemp()
280        app['datacenter'].setStoragePath(self.dc_root)
281
282        # Prepopulate the ZODB...
283        self.getRootFolder()['app'] = app
284        self.app = self.getRootFolder()['app']
285        self.workdir = tempfile.mkdtemp()
286
287        self.app['mycomponent'] = LoggingComponent()
288        self.component = self.app['mycomponent']
289        setSite(self.app)
290
291        self.component = LoggingComponent()
292
293
294        return
295
296    def tearDown(self):
297        super(LoggerCollectorTests, self).tearDown()
298        shutil.rmtree(self.workdir)
299        shutil.rmtree(self.dc_root)
300        clearSite()
301        return
302
303    def test_ifaces(self):
304        collector = LoggerCollector()
305        verifyClass(ILoggerCollector, LoggerCollector)
306        verifyObject(ILoggerCollector, collector)
307        return
308
309    def test_get_utility(self):
310        # We can get a logger collector via utility lookup
311        util1 = queryUtility(ILoggerCollector, default=None)
312        util2 = queryUtility(ILoggerCollector, default=None)
313        self.assertTrue(util1 is util2)
314        self.assertTrue(isinstance(util1, LoggerCollector))
315        return
316
317    def test_get_loggers(self):
318        # We can get loggers from a logger collector
319        collector = LoggerCollector()
320        result0 = collector.getLoggers(None)
321        result1 = collector.getLoggers('not-a-site')
322        result2 = collector.getLoggers(self.app)
323        collector.registerLogger(self.app, self.component)
324        result3 = collector.getLoggers(self.app)
325        self.assertEqual(result0, [])
326        self.assertEqual(result1, [])
327        self.assertEqual(result2, [])
328        self.assertEqual(result3, [self.component])
329        self.assertTrue(result3[0] is self.component)
330        return
331
332    def test_register_logger(self):
333        # We can register loggers by name
334        collector = LoggerCollector()
335        collector.registerLogger(None, self.component) # Should have no effect
336        collector.registerLogger(object(), self.component) # Same here
337        collector.registerLogger(self.app, self.component)
338        self.assertEqual(['app'], collector.keys())
339        return
340
341    def test_register_logger_double(self):
342        # Registering the same component twice gives one entry, not two
343        collector = LoggerCollector()
344        collector.registerLogger(self.app, self.component)
345        collector.registerLogger(self.app, self.component)
346        self.assertEqual(len(collector['app']), 1)
347        return
348
349    def test_unregister_logger(self):
350        # We can also unregister loggers
351        collector = LoggerCollector()
352        collector.registerLogger(self.app, self.component)
353        collector.unregisterLogger(self.app, self.component)
354        result = collector.getLoggers(self.app)
355        self.assertEqual(result, [])
356        return
357
358    def test_unregister_logger_invalid(self):
359        # We can cope with unregistration of non-existing entries.
360        collector = LoggerCollector()
361        collector.registerLogger(self.app, self.component)
362        collector.unregisterLogger(None, self.component)
363        collector.unregisterLogger(self.app, 'nonsense')
364        self.assertEqual(len(collector), 1)
365        return
366
367class LogfileChangeTests(FunctionalTestCase):
368
369    layer = FunctionalLayer
370
371    def setUp(self):
372        super(LogfileChangeTests, self).setUp()
373        # Setup a sample site for each test
374        app = University()
375        self.dc_root = tempfile.mkdtemp()
376        self.dc_root_new = None
377        app['datacenter'].setStoragePath(self.dc_root)
378
379        # Prepopulate the ZODB...
380        self.getRootFolder()['app'] = app
381        self.app = self.getRootFolder()['app']
382        self.workdir = tempfile.mkdtemp()
383
384        self.app['mycomponent'] = LoggingComponent()
385        self.component = self.app['mycomponent']
386        setSite(self.app)
387        self.component = LoggingComponent()
388        return
389
390    def tearDown(self):
391        super(LogfileChangeTests, self).tearDown()
392        shutil.rmtree(self.workdir)
393        shutil.rmtree(self.dc_root)
394        if self.dc_root_new is not None:
395            shutil.rmtree(self.dc_root_new)
396        clearSite()
397        return
398
399    def test_storage_move(self):
400        # When a datacenter storage (its path) changes, we have to
401        # move the logfiles as well.
402        filename1 = self.component.logger.handlers[0].baseFilename
403        self.dc_root_new = tempfile.mkdtemp()
404        self.app['datacenter'].setStoragePath(self.dc_root_new)
405        filename2 = self.component.logger.handlers[0].baseFilename
406        self.assertTrue(filename1 != filename2)
407        self.assertTrue(filename2.startswith(self.dc_root_new))
408        return
409
410    def test_storage_deletion(self):
411        # When a site is deleted, the loggers should go away as well
412        self.dc_root_new = tempfile.mkdtemp()
413        logger1 = self.component.logger.handlers # create entry in collector
414        collector = queryUtility(ILoggerCollector)
415        loggers1 = collector.keys()
416        root = self.app.__parent__
417        del root['app']
418        loggers2 = collector.keys()
419        self.assertEqual(loggers1, ['app'])
420        self.assertEqual(loggers2, [])
421        return
Note: See TracBrowser for help on using the repository browser.