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

Last change on this file since 9554 was 9511, checked in by uli, 12 years ago

Update tests.
File size: 17.6 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.name = '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, 'name', None) is None:
34            return False
35        return self.name == obj.name
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            [('report1', FakeReportGenerator('report1')),
145             ('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', a=1, b=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 (a=1, b=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'Invalid 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"Invalid 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'Invalid Report Generator (a=1, b=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"Invalid Report Generator ('a', b=2)")
256        return
257
258
259class FakeJobWithResult(FakeJob):
260
261    def __init__(self, args=[], kw={}):
262        #self.dir_path = tempfile.mkdtemp()
263        #self.result = os.path.join(self.dir_path, 'fake.csv')
264        #open(self.result, 'wb').write('a fake result')
265        self.result = Report()
266        self.result.args = args
267        self.result.kw = kw
268        return
269
270class ReportJobContainerTests(unittest.TestCase):
271    # Test ReportJobContainer
272
273    def setUp(self):
274        # register a suitable ICSVExporter as named utility
275        self.generator = FakeReportGenerator('report1')
276        self.job_manager = FakeJobManager()
277        self.gsm = getGlobalSiteManager()
278        self.gsm.registerUtility(
279            self.generator, IReportGenerator, name='report1')
280        self.gsm.registerUtility(
281            self.job_manager, IJobManager)
282
283    def tearDown(self):
284        self.gsm.unregisterUtility(
285            self.generator, IReportGenerator, name='report1')
286        self.gsm.unregisterUtility(self.job_manager, IJobManager)
287
288    def test_report_job_interfaces(self):
289        # the ExportJobContainer implements promised interfaces correctly...
290        container = ReportJobContainer()
291        verifyClass(IReportJobContainer, ReportJobContainer)
292        verifyObject(IReportJobContainer, container)
293        return
294
295    def test_start_report_job(self):
296        # we can start jobs
297        container = ReportJobContainer()
298        container.start_report_job('report3', 'bob')
299        result = self.job_manager._jobs.values()[0]
300        self.assertTrue(IJob.providedBy(result))
301        self.assertEqual(
302            container.running_report_jobs,
303            [('1', 'report3', 'bob')]
304            )
305        return
306
307    def test_get_running_report_jobs_all(self):
308        # we can get report jobs of all users
309        container = ReportJobContainer()
310        container.start_report_job('report3', 'bob')
311        container.start_report_job('report3', 'alice')
312        result = container.get_running_report_jobs()
313        self.assertEqual(
314            result,
315            [('1', 'report3', 'bob'),
316             ('2', 'report3', 'alice')]
317            )
318        return
319
320    def test_get_running_report_jobs_user(self):
321        # we can get the report jobs running for a certain user
322        container = ReportJobContainer()
323        container.start_report_job('report3', 'bob')
324        container.start_report_job('report3', 'alice')
325        result1 = container.get_running_report_jobs(user_id='alice')
326        result2 = container.get_running_report_jobs(user_id='foo')
327        self.assertEqual(
328            result1, [('2', 'report3', 'alice')])
329        self.assertEqual(
330            result2, [])
331        return
332
333    def test_get_running_report_jobs_only_if_exist(self):
334        # we get only jobs that are accessible through the job manager...
335        container = ReportJobContainer()
336        container.start_report_job('report3', 'bob')
337        container.start_report_job('report3', 'bob')
338        self.assertTrue(
339            ('2', 'report3', 'bob') in container.running_report_jobs)
340        # we remove the second entry from job manager
341        del self.job_manager._jobs['2']
342        result = container.get_running_report_jobs(user_id='bob')
343        self.assertEqual(
344            result, [('1', 'report3', 'bob')])
345        self.assertTrue(
346            ('2', 'report3', 'bob') not in container.running_report_jobs)
347        return
348
349    def test_get_report_job_status(self):
350        # we can get the stati of jobs...
351        container = ReportJobContainer()
352        container.start_report_job('report1', 'alice')
353        container.start_report_job('report1', 'bob')
354        container.start_report_job('report1', 'bob')
355        result = container.get_report_jobs_status(user_id='bob')
356        # we'll get the raw value, a translation and the title of the
357        # exporter
358        self.assertEqual(
359            result,
360            [('new', u'new', u'Unnamed Report'),
361             ('completed', u'completed', u'Unnamed Report')]
362            )
363        return
364
365    def test_delete_report_entry(self):
366        # we can remove report entries in local lists and the job
367        # manager as well...
368        container = ReportJobContainer()
369        container.start_report_job('report3', 'bob')
370        entry = container.running_report_jobs[0]
371        container.delete_report_entry(entry)
372        # both, running_report_jobs list and job manager are empty now
373        self.assertEqual(
374            container.running_report_jobs, [])
375        self.assertEqual(
376            self.job_manager._jobs, {})
377        return
378
379    def test_report_entry_from_job_id(self):
380        # we can get an report entry for a job_id if the id exists
381        container = ReportJobContainer()
382        entry = ('4', 'report3', 'bob')
383        container.running_report_jobs = [entry]
384        fake_job = FakeJobWithResult()
385        self.job_manager._jobs['4'] = fake_job
386        result1 = container.report_entry_from_job_id(None)
387        result2 = container.report_entry_from_job_id('4')
388        result3 = container.report_entry_from_job_id('23')
389        self.assertEqual(result1, None)
390        self.assertEqual(result2, ('4', 'report3', 'bob'))
391        self.assertEqual(result3, None)
392        return
393
394
395class ReportsContainerTests(unittest.TestCase):
396    # Tests for ReportsContainer
397
398    def test_iface(self):
399        # ReportsContainers really provide the promised interfaces
400        obj = ReportsContainer()
401        verifyClass(IReportsContainer, ReportsContainer)
402        verifyClass(IContainer, ReportsContainer)
403        verifyObject(IReportsContainer, obj)
404        verifyObject(IContainer, obj)
405        return
406
407class ReportsContainerPluginTests(unittest.TestCase):
408    # Tests for ReportsContainerPlugin
409
410    def create_logger(self):
411        # create a logger suitable for local tests.
412        test_logger = logging.getLogger('waeup.kofa.reports.testlogger')
413        log = StringIO()
414        handler = logging.StreamHandler(log)
415        handler.setLevel(logging.DEBUG)
416        test_logger.addHandler(handler)
417        test_logger.setLevel(logging.DEBUG)
418        self.logger = test_logger
419        self.log = log
420        self.handler = handler
421        return self.logger
422
423    def remove_logger(self):
424        del self.handler
425        del self.logger
426        del self.log
427        pass
428
429    def get_log(self):
430        self.log.seek(0)
431        return self.log.read()
432
433    def setUp(self):
434        self.create_logger()
435        return
436
437    def tearDown(self):
438        self.remove_logger()
439        return
440
441    def test_iface(self):
442        # make sure we fullfill the promised interfaces
443        obj = ReportsContainerPlugin()
444        verifyClass(IKofaPluggable, ReportsContainerPlugin)
445        verifyObject(IKofaPluggable, obj)
446        return
447
448    def test_get_as_utility(self):
449        # make sure we can get the plugin as utility
450        grok.testing.grok('waeup.kofa.reports')
451        util = getUtility(IKofaPluggable, name='reports')
452        self.assertTrue(util is not None)
453        return
454
455    def test_update_no_container(self):
456        # we can update an existing site
457        fake_site = grok.Container()
458        plugin = ReportsContainerPlugin()
459        plugin.update(fake_site, 'app', self.logger)
460        log = self.get_log()
461        self.assertEqual(
462            log, 'Added reports container for site "app"\n')
463        self.assertTrue('reports' in fake_site.keys())
464        self.assertTrue(IReportsContainer.providedBy(fake_site['reports']))
465        return
466
467    def test_update_uptodate_site(self):
468        # we leave already existing reports containers in place
469        fake_site = grok.Container()
470        plugin = ReportsContainerPlugin()
471        fake_site['reports'] = ReportsContainer()
472        plugin.update(fake_site, 'app', self.logger)
473        log = self.get_log()
474        self.assertEqual(log, '') # no log message
475        return
476
477    def test_setup_new_site(self):
478        # if we setup a site, we always install a fresh reports container
479        fake_site = grok.Container()
480        plugin = ReportsContainerPlugin()
481        plugin.setup(fake_site, 'app', self.logger)
482        log1 = self.get_log()
483        result1 = fake_site.get('reports', None)
484        plugin.setup(fake_site, 'app', self.logger) # replace old container
485        log2 = self.get_log()
486        result2 = fake_site.get('reports', None)
487        self.assertTrue(result1 is not result2)
488        self.assertEqual(log1,
489                         'Added reports container for site "app"\n')
490        self.assertEqual(log2,
491                         'Added reports container for site "app"\n'
492                         'Removed reports container for site "app"\n'
493                         'Added reports container for site "app"\n')
494        return
Note: See TracBrowser for help on using the repository browser.