source: main/waeup.kofa/trunk/src/waeup/kofa/tests/test_reports.py @ 9648

Last change on this file since 9648 was 9647, checked in by Henrik Bettermann, 12 years ago

Generate reports for students in a certain session and for groups of study modes.
File size: 18.9 KB
Line 
1import datetime
2import grok
3import logging
4import pytz
5import unittest
6from cStringIO import StringIO
7from grok.interfaces import IContainer
8from zc.async.interfaces import IJob
9from zc.async.testing import wait_for_result
10from zope.component import getGlobalSiteManager, getUtility
11from zope.component.hooks import setSite
12from zope.interface.verify import verifyClass, verifyObject
13from waeup.kofa.interfaces import IJobManager, IKofaPluggable
14from waeup.kofa.reports import (
15    IReport, IReportGenerator, IReportJob, IReportJobContainer,
16    IReportsContainer,)
17from waeup.kofa.reports import (
18    Report, ReportGenerator, get_generators, report_job, AsyncReportJob,
19    ReportJobContainer, ReportsContainer, ReportsContainerPlugin)
20from waeup.kofa.testing import FakeJob, FakeJobManager, FunctionalLayer
21from waeup.kofa.tests.test_async import FunctionalAsyncTestCase
22
23class FakeReportGenerator(ReportGenerator):
24    # fake report generator for tests.
25
26    def __init__(self, name=None, perm_create=None, perm_view=None):
27        self.title = 'Report 1'
28        self.perm_create = perm_create
29        self.perm_view = perm_view
30        return
31
32    def __eq__(self, obj):
33        if getattr(obj, 'title', None) is None:
34            return False
35        return self.title == obj.title
36
37    def generate(self, site, args=[], kw={}):
38        result = Report()
39        result.args = args
40        result.kw = kw
41        return result
42
43class ReportTests(unittest.TestCase):
44
45    def test_iface(self):
46        # make sure we fullfill the promised contracts
47        obj = Report()
48        verifyClass(IReport, Report)
49        verifyObject(IReport, obj)
50        return
51
52    def test_creation_dt(self):
53        # a report sets a datetime timestamp when created
54        report = Report()
55        self.assertTrue(hasattr(report, 'creation_dt'))
56        self.assertTrue(isinstance(report.creation_dt, datetime.datetime))
57        # the datetime is set with UTC timezone info
58        self.assertEqual(report.creation_dt.tzinfo, pytz.utc)
59        return
60
61    def test_args(self):
62        # a report stores the args and kw used for generation
63        report1 = Report()
64        report2 = Report(args=[1, 2], kwargs=dict(a=1, b=2))
65        self.assertEqual(report1.args, [])
66        self.assertEqual(report1.kwargs, dict())
67        self.assertEqual(report2.args, [1, 2])
68        self.assertEqual(report2.kwargs, dict(a=1, b=2))
69        return
70
71    def test_create_pdf(self):
72        # trying to create a pdf results in an error
73        report = Report()
74        self.assertRaises(NotImplementedError, report.create_pdf)
75        return
76
77class ReportGeneratorTest(unittest.TestCase):
78    def setUp(self):
79        grok.testing.grok('waeup.kofa.reports') # register utils
80
81    def test_iface(self):
82        # make sure we fullfill the promised contracts
83        obj = ReportGenerator()
84        verifyClass(IReportGenerator, ReportGenerator)
85        verifyObject(IReportGenerator, obj)
86        return
87
88    def test_generate(self):
89        # the base report generator delivers reports
90        obj = ReportGenerator()
91        result = obj.generate(None)
92        self.assertTrue(IReport.providedBy(result))
93        return
94
95class GeneratorRegistrar(object):
96    # Mix-in providing report generator registrations
97    def setUp(self):
98        self.registered = []
99
100    def tearDown(self):
101        # unregister any previously registered report types
102        gsm = getGlobalSiteManager()
103        for report, name in self.registered:
104            gsm.unregisterUtility(report, IReportGenerator, name=name)
105        return
106
107    def register_generator(self, name, perm_create=None, perm_view=None):
108        # helper to register report generators as utils
109        gsm = getGlobalSiteManager()
110        generator = FakeReportGenerator(name)
111        gsm.registerUtility(generator, provided=IReportGenerator, name=name)
112        self.registered.append((generator, name),)
113        return generator
114
115class HelpersTests(GeneratorRegistrar, unittest.TestCase):
116    # Tests for helper functions
117
118    def setUp(self):
119        grok.testing.grok('waeup.kofa.reports') # register utils
120        super(HelpersTests, self).setUp()
121        return
122
123    def test_get_generators_none(self):
124        # we get no generators if none was registered
125        result = list(get_generators())
126        self.assertEqual(result, [])
127        return
128
129    def test_get_generators_simple(self):
130        # we get a single generator if one is registered
131        self.register_generator('report1')
132        result = list(get_generators())
133        self.assertEqual(
134            result, [('report1', FakeReportGenerator('report1'))])
135        return
136
137    def test_get_generators_multiple(self):
138        # we also get multiple generators if available
139        self.register_generator('report1')
140        self.register_generator('report2')
141        result = list(get_generators())
142        self.assertEqual(
143            result,
144            [(u'report1', FakeReportGenerator('report1')),
145             (u'report2', FakeReportGenerator('report2'))])
146        return
147
148
149class ReportJobTests(FunctionalAsyncTestCase, GeneratorRegistrar):
150    # Test asynchronous report functionality (simple cases)
151
152    layer = FunctionalLayer
153
154    def setUp(self):
155        super(ReportJobTests, self).setUp()
156        GeneratorRegistrar.setUp(self)
157        self.root_folder = self.getRootFolder()
158
159    def test_report_job_func(self):
160        # the report_job func really creates reports...
161        self.register_generator('report1')
162        report = report_job(None, 'report1')
163        self.assertTrue(IReport.providedBy(report))
164        self.assertTrue(isinstance(report, Report))
165
166    def test_report_job_interfaces(self):
167        # the AsyncReportJob implements promised interfaces correctly...
168        job = AsyncReportJob(None, None)
169        verifyClass(IJob, AsyncReportJob)
170        verifyObject(IJob, job)
171        verifyClass(IReportJob, AsyncReportJob)
172        verifyObject(IReportJob, job)
173        return
174
175    def test_finished(self):
176        # AsyncReportJobs signal with a bool whether they`re  finished
177        job = AsyncReportJob(self.root_folder, None)
178        setSite(self.root_folder)
179        self.assertEqual(job.finished, False)
180        manager = getUtility(IJobManager)
181        manager.put(job)
182        wait_for_result(job)
183        self.assertEqual(job.finished, True)
184        return
185
186    def test_failed_true(self):
187        # We can test whether a job failed
188        job = AsyncReportJob(self.root_folder, None) # no report generator
189        setSite(self.root_folder)
190        # while a job is not finished, `failed` is ``None``
191        self.assertTrue(job.failed is None)
192        manager = getUtility(IJobManager)
193        manager.put(job)
194        wait_for_result(job)
195        # the finished job failed
196        self.assertEqual(job.failed, True)
197        return
198
199    def test_failed_false(self):
200        # We can test whether a job failed
201        self.register_generator('report1')
202        job = AsyncReportJob(self.root_folder, 'report1')
203        setSite(self.root_folder)
204        # while a job is not finished, `failed` is ``None``
205        self.assertTrue(job.failed is None)
206        manager = getUtility(IJobManager)
207        manager.put(job)
208        wait_for_result(job)
209        # the finished job failed
210        self.assertEqual(job.failed, False)
211        return
212
213    def test_description(self):
214        # IReportJobs provide a description of the started job
215        self.register_generator('report1')
216        args, kw = ['a', 'b'], dict(a=1, b=2)
217        job = AsyncReportJob(self.root_folder, 'report1', args=args, kw=kw)
218        self.assertEqual(
219            job.description,
220            "Report 1 ('a', 'b', 1, 2)")
221        # w/o args nor kwargs
222        job = AsyncReportJob(self.root_folder, 'report1')
223        self.assertEqual(
224            job.description,
225            'Report 1 ()')
226        # with args only
227        job = AsyncReportJob(self.root_folder, 'report1', args=args)
228        self.assertEqual(
229            job.description, "Report 1 ('a', 'b')")
230        # with keywords only
231        job = AsyncReportJob(self.root_folder, 'report1', kw=kw)
232        self.assertEqual(
233            job.description, "Report 1 (1, 2)")
234        return
235
236    def test_description_invalid_generator(self):
237        # We can get a description even with an invalid generator
238        job = AsyncReportJob(self.root_folder, 'NOT EXISTENT')
239        self.assertEqual(
240            job.description, u'Unregistered Report Generator ()')
241        # with args set (no kws)
242        job = AsyncReportJob(
243            self.root_folder, 'NOT EXISTENT', args=['a', 'b'])
244        self.assertEqual(
245            job.description, u"Unregistered Report Generator ('a', 'b')")
246        # with kw set (no args)
247        job = AsyncReportJob(
248            self.root_folder, 'NOT_EXISTENT', kw=dict(a=1, b=2))
249        self.assertEqual(
250            job.description, u'Unregistered Report Generator (1, 2)')
251        # with args and kws set
252        job = AsyncReportJob(
253            self.root_folder, 'NOT_EXISTENT', args=['a'], kw=dict(b=2))
254        self.assertEqual(
255            job.description, u"Unregistered Report Generator ('a', 2)")
256        return
257
258    def test_report_status(self):
259        # We can get a report status apart from the job status
260        job = AsyncReportJob(self.root_folder, 'report1')
261        self.assertEqual(job.report_status, u'running')
262        return
263
264    def test_report_status_failed(self):
265        # A failed job is reflected in report_status
266        job = AsyncReportJob(self.root_folder, None) # no report generator
267        setSite(self.root_folder)
268        self.assertTrue(job.failed is None)
269        manager = getUtility(IJobManager)
270        manager.put(job)
271        wait_for_result(job)
272        self.assertEqual(job.report_status, u'FAILED')
273        return
274
275    def test_report_status_finished(self):
276        # A finished report is reflected in report_status
277        job = AsyncReportJob(self.root_folder, 'report1')
278        setSite(self.root_folder)
279        manager = getUtility(IJobManager)
280        manager.put(job)
281        wait_for_result(job)
282        self.assertEqual(job.report_status, u'finished')
283        return
284
285class FakeJobWithResult(FakeJob):
286
287    def __init__(self, args=[], kw={}):
288        self.result = Report()
289        self.result.args = args
290        self.result.kw = kw
291        return
292
293class ReportJobContainerTests(unittest.TestCase):
294    # Test ReportJobContainer
295
296    def setUp(self):
297        # register a suitable ICSVExporter as named utility
298        self.generator = FakeReportGenerator('report1')
299        self.job_manager = FakeJobManager()
300        self.gsm = getGlobalSiteManager()
301        self.gsm.registerUtility(
302            self.generator, IReportGenerator, name='report1')
303        self.gsm.registerUtility(
304            self.job_manager, IJobManager)
305
306    def tearDown(self):
307        self.gsm.unregisterUtility(
308            self.generator, IReportGenerator, name='report1')
309        self.gsm.unregisterUtility(self.job_manager, IJobManager)
310
311    def test_report_job_interfaces(self):
312        # the ExportJobContainer implements promised interfaces correctly...
313        container = ReportJobContainer()
314        verifyClass(IReportJobContainer, ReportJobContainer)
315        verifyObject(IReportJobContainer, container)
316        return
317
318    def test_start_report_job(self):
319        # we can start jobs
320        container = ReportJobContainer()
321        container.start_report_job('report3', 'bob')
322        result = self.job_manager._jobs.values()[0]
323        self.assertTrue(IJob.providedBy(result))
324        self.assertEqual(
325            container.running_report_jobs,
326            [('1', 'report3', 'bob')]
327            )
328        return
329
330    def test_get_running_report_jobs_all(self):
331        # we can get report jobs of all users
332        container = ReportJobContainer()
333        container.start_report_job('report3', 'bob')
334        container.start_report_job('report3', 'alice')
335        result = container.get_running_report_jobs()
336        self.assertEqual(
337            result,
338            [('1', 'report3', 'bob'),
339             ('2', 'report3', 'alice')]
340            )
341        return
342
343    def test_get_running_report_jobs_user(self):
344        # we can get the report jobs running for a certain user
345        container = ReportJobContainer()
346        container.start_report_job('report3', 'bob')
347        container.start_report_job('report3', 'alice')
348        result1 = container.get_running_report_jobs(user_id='alice')
349        result2 = container.get_running_report_jobs(user_id='foo')
350        self.assertEqual(
351            result1, [('2', 'report3', 'alice')])
352        self.assertEqual(
353            result2, [])
354        return
355
356    def test_get_running_report_jobs_only_if_exist(self):
357        # we get only jobs that are accessible through the job manager...
358        container = ReportJobContainer()
359        container.start_report_job('report3', 'bob')
360        container.start_report_job('report3', 'bob')
361        self.assertTrue(
362            ('2', 'report3', 'bob') in container.running_report_jobs)
363        # we remove the second entry from job manager
364        del self.job_manager._jobs['2']
365        result = container.get_running_report_jobs(user_id='bob')
366        self.assertEqual(
367            result, [('1', 'report3', 'bob')])
368        self.assertTrue(
369            ('2', 'report3', 'bob') not in container.running_report_jobs)
370        return
371
372    def test_get_report_job_status(self):
373        # we can get the stati of jobs...
374        container = ReportJobContainer()
375        container.start_report_job('report1', 'alice')
376        container.start_report_job('report1', 'bob')
377        container.start_report_job('report1', 'bob')
378        result = container.get_report_jobs_status(user_id='bob')
379        # we'll get the raw value, a translation and the title of the
380        # exporter
381        self.assertEqual(
382            result,
383            [('new', u'new', u'Report 1'),
384             ('completed', u'completed', u'Report 1')]
385            )
386        return
387
388    def test_get_report_job_description(self):
389        # we can get the descriptions of jobs...
390        container = ReportJobContainer()
391        container.start_report_job('report1', 'alice')
392        container.start_report_job('report1', 'bob')
393        result = container.get_report_jobs_description(user_id='bob')
394        # we'll get the job id, a description and the status
395        self.assertEqual(
396            result,
397            [('2', 'Report 1 ()', u'running', False, False, None, 'bob'),]
398            )
399        return
400
401    def test_delete_report_entry(self):
402        # we can remove report entries in local lists and the job
403        # manager as well...
404        container = ReportJobContainer()
405        container.start_report_job('report3', 'bob')
406        entry = container.running_report_jobs[0]
407        container.delete_report_entry(entry)
408        # both, running_report_jobs list and job manager are empty now
409        self.assertEqual(
410            container.running_report_jobs, [])
411        self.assertEqual(
412            self.job_manager._jobs, {})
413        return
414
415    def test_report_entry_from_job_id(self):
416        # we can get an report entry for a job_id if the id exists
417        container = ReportJobContainer()
418        entry = ('4', 'report3', 'bob')
419        container.running_report_jobs = [entry]
420        fake_job = FakeJobWithResult()
421        self.job_manager._jobs['4'] = fake_job
422        result1 = container.report_entry_from_job_id(None)
423        result2 = container.report_entry_from_job_id('4')
424        result3 = container.report_entry_from_job_id('23')
425        self.assertEqual(result1, None)
426        self.assertEqual(result2, ('4', 'report3', 'bob'))
427        self.assertEqual(result3, None)
428        return
429
430
431class ReportsContainerTests(unittest.TestCase):
432    # Tests for ReportsContainer
433
434    def test_iface(self):
435        # ReportsContainers really provide the promised interfaces
436        obj = ReportsContainer()
437        verifyClass(IReportsContainer, ReportsContainer)
438        verifyClass(IContainer, ReportsContainer)
439        verifyObject(IReportsContainer, obj)
440        verifyObject(IContainer, obj)
441        return
442
443class ReportsContainerPluginTests(unittest.TestCase):
444    # Tests for ReportsContainerPlugin
445
446    def create_logger(self):
447        # create a logger suitable for local tests.
448        test_logger = logging.getLogger('waeup.kofa.reports.testlogger')
449        log = StringIO()
450        handler = logging.StreamHandler(log)
451        handler.setLevel(logging.DEBUG)
452        test_logger.addHandler(handler)
453        test_logger.setLevel(logging.DEBUG)
454        self.logger = test_logger
455        self.log = log
456        self.handler = handler
457        return self.logger
458
459    def remove_logger(self):
460        del self.handler
461        del self.logger
462        del self.log
463        pass
464
465    def get_log(self):
466        self.log.seek(0)
467        return self.log.read()
468
469    def setUp(self):
470        self.create_logger()
471        return
472
473    def tearDown(self):
474        self.remove_logger()
475        return
476
477    def test_iface(self):
478        # make sure we fullfill the promised interfaces
479        obj = ReportsContainerPlugin()
480        verifyClass(IKofaPluggable, ReportsContainerPlugin)
481        verifyObject(IKofaPluggable, obj)
482        return
483
484    def test_get_as_utility(self):
485        # make sure we can get the plugin as utility
486        grok.testing.grok('waeup.kofa.reports')
487        util = getUtility(IKofaPluggable, name='reports')
488        self.assertTrue(util is not None)
489        return
490
491    def test_update_no_container(self):
492        # we can update an existing site
493        fake_site = grok.Container()
494        plugin = ReportsContainerPlugin()
495        plugin.update(fake_site, 'app', self.logger)
496        log = self.get_log()
497        self.assertEqual(
498            log, 'Added reports container for site "app"\n')
499        self.assertTrue('reports' in fake_site.keys())
500        self.assertTrue(IReportsContainer.providedBy(fake_site['reports']))
501        return
502
503    def test_update_uptodate_site(self):
504        # we leave already existing reports containers in place
505        fake_site = grok.Container()
506        plugin = ReportsContainerPlugin()
507        fake_site['reports'] = ReportsContainer()
508        plugin.update(fake_site, 'app', self.logger)
509        log = self.get_log()
510        self.assertEqual(log, '') # no log message
511        return
512
513    def test_setup_new_site(self):
514        # if we setup a site, we always install a fresh reports container
515        fake_site = grok.Container()
516        plugin = ReportsContainerPlugin()
517        plugin.setup(fake_site, 'app', self.logger)
518        log1 = self.get_log()
519        result1 = fake_site.get('reports', None)
520        plugin.setup(fake_site, 'app', self.logger) # replace old container
521        log2 = self.get_log()
522        result2 = fake_site.get('reports', None)
523        self.assertTrue(result1 is not result2)
524        self.assertEqual(log1,
525                         'Added reports container for site "app"\n')
526        self.assertEqual(log2,
527                         'Added reports container for site "app"\n'
528                         'Removed reports container for site "app"\n'
529                         'Added reports container for site "app"\n')
530        return
Note: See TracBrowser for help on using the repository browser.