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

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

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