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

Last change on this file since 6631 was 6581, checked in by uli, 13 years ago

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