source: main/waeup.kofa/branches/uli-py3/src/waeup/kofa/tests/test_reports.py @ 17650

Last change on this file since 17650 was 14373, checked in by Henrik Bettermann, 8 years ago

Show report number (job_id) on report pdf slips.
  • Property svn:keywords set to Id
File size: 15.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.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, 'JOB_ID')
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
213class FakeJobWithResult(FakeJob):
214
215    def __init__(self, args=[], kw={}):
216        self.result = Report()
217        self.result.args = args
218        self.result.kw = kw
219        return
220
221class ReportJobContainerTests(unittest.TestCase):
222    # Test ReportJobContainer
223
224    def setUp(self):
225        # register a suitable ICSVExporter as named utility
226        self.generator = FakeReportGenerator('report1')
227        self.job_manager = FakeJobManager()
228        self.gsm = getGlobalSiteManager()
229        self.gsm.registerUtility(
230            self.generator, IReportGenerator, name='report1')
231        self.gsm.registerUtility(
232            self.job_manager, IJobManager)
233
234    def tearDown(self):
235        self.gsm.unregisterUtility(
236            self.generator, IReportGenerator, name='report1')
237        self.gsm.unregisterUtility(self.job_manager, IJobManager)
238
239    def test_report_job_interfaces(self):
240        # the ExportJobContainer implements promised interfaces correctly...
241        container = ReportJobContainer()
242        verifyClass(IReportJobContainer, ReportJobContainer)
243        verifyObject(IReportJobContainer, container)
244        return
245
246    def test_start_report_job(self):
247        # we can start jobs
248        container = ReportJobContainer()
249        container.start_report_job('report3', 'bob')
250        result = self.job_manager._jobs.values()[0]
251        self.assertTrue(IJob.providedBy(result))
252        self.assertEqual(
253            container.running_report_jobs,
254            [('1', 'report3', 'bob')]
255            )
256        return
257
258    def test_get_running_report_jobs_all(self):
259        # we can get report jobs of all users
260        container = ReportJobContainer()
261        container.start_report_job('report3', 'bob')
262        container.start_report_job('report3', 'alice')
263        result = container.get_running_report_jobs()
264        self.assertEqual(
265            result,
266            [('1', 'report3', 'bob'),
267             ('2', 'report3', 'alice')]
268            )
269        return
270
271    def test_get_running_report_jobs_user(self):
272        # we can get the report jobs running for a certain user
273        container = ReportJobContainer()
274        container.start_report_job('report3', 'bob')
275        container.start_report_job('report3', 'alice')
276        result1 = container.get_running_report_jobs(user_id='alice')
277        result2 = container.get_running_report_jobs(user_id='foo')
278        self.assertEqual(
279            result1, [('2', 'report3', 'alice')])
280        self.assertEqual(
281            result2, [])
282        return
283
284    def test_get_running_report_jobs_only_if_exist(self):
285        # we get only jobs that are accessible through the job manager...
286        container = ReportJobContainer()
287        container.start_report_job('report3', 'bob')
288        container.start_report_job('report3', 'bob')
289        self.assertTrue(
290            ('2', 'report3', 'bob') in container.running_report_jobs)
291        # we remove the second entry from job manager
292        del self.job_manager._jobs['2']
293        result = container.get_running_report_jobs(user_id='bob')
294        self.assertEqual(
295            result, [('1', 'report3', 'bob')])
296        self.assertTrue(
297            ('2', 'report3', 'bob') not in container.running_report_jobs)
298        return
299
300    def test_get_report_job_status(self):
301        # we can get the stati of jobs...
302        container = ReportJobContainer()
303        container.start_report_job('report1', 'alice')
304        container.start_report_job('report1', 'bob')
305        container.start_report_job('report1', 'bob')
306        result = container.get_report_jobs_status(user_id='bob')
307        # we'll get the raw value, a translation and the title of the
308        # exporter
309        self.assertEqual(
310            result,
311            [('new', u'new', u'Report 1'),
312             ('completed', u'completed', u'Report 1')]
313            )
314        return
315
316    def test_delete_report_entry(self):
317        # we can remove report entries in local lists and the job
318        # manager as well...
319        container = ReportJobContainer()
320        container.start_report_job('report3', 'bob')
321        entry = container.running_report_jobs[0]
322        container.delete_report_entry(entry)
323        # both, running_report_jobs list and job manager are empty now
324        self.assertEqual(
325            container.running_report_jobs, [])
326        self.assertEqual(
327            self.job_manager._jobs, {})
328        return
329
330    def test_report_entry_from_job_id(self):
331        # we can get an report entry for a job_id if the id exists
332        container = ReportJobContainer()
333        entry = ('4', 'report3', 'bob')
334        container.running_report_jobs = [entry]
335        fake_job = FakeJobWithResult()
336        self.job_manager._jobs['4'] = fake_job
337        result1 = container.report_entry_from_job_id(None)
338        result2 = container.report_entry_from_job_id('4')
339        result3 = container.report_entry_from_job_id('23')
340        self.assertEqual(result1, None)
341        self.assertEqual(result2, ('4', 'report3', 'bob'))
342        self.assertEqual(result3, None)
343        return
344
345
346class ReportsContainerTests(unittest.TestCase):
347    # Tests for ReportsContainer
348
349    def test_iface(self):
350        # ReportsContainers really provide the promised interfaces
351        obj = ReportsContainer()
352        verifyClass(IReportsContainer, ReportsContainer)
353        verifyClass(IContainer, ReportsContainer)
354        verifyObject(IReportsContainer, obj)
355        verifyObject(IContainer, obj)
356        return
357
358class ReportsContainerPluginTests(unittest.TestCase):
359    # Tests for ReportsContainerPlugin
360
361    def create_logger(self):
362        # create a logger suitable for local tests.
363        test_logger = logging.getLogger('waeup.kofa.reports.testlogger')
364        log = StringIO()
365        handler = logging.StreamHandler(log)
366        handler.setLevel(logging.DEBUG)
367        test_logger.addHandler(handler)
368        test_logger.setLevel(logging.DEBUG)
369        self.logger = test_logger
370        self.log = log
371        self.handler = handler
372        return self.logger
373
374    def remove_logger(self):
375        del self.handler
376        del self.logger
377        del self.log
378        pass
379
380    def get_log(self):
381        self.log.seek(0)
382        return self.log.read()
383
384    def setUp(self):
385        self.create_logger()
386        return
387
388    def tearDown(self):
389        self.remove_logger()
390        return
391
392    def test_iface(self):
393        # make sure we fullfill the promised interfaces
394        obj = ReportsContainerPlugin()
395        verifyClass(IKofaPluggable, ReportsContainerPlugin)
396        verifyObject(IKofaPluggable, obj)
397        return
398
399    def test_get_as_utility(self):
400        # make sure we can get the plugin as utility
401        grok.testing.grok('waeup.kofa.reports')
402        util = getUtility(IKofaPluggable, name='reports')
403        self.assertTrue(util is not None)
404        return
405
406    def test_update_no_container(self):
407        # we can update an existing site
408        fake_site = grok.Container()
409        plugin = ReportsContainerPlugin()
410        plugin.update(fake_site, 'app', self.logger)
411        log = self.get_log()
412        self.assertEqual(
413            log, 'Added reports container for site "app"\n')
414        self.assertTrue('reports' in fake_site.keys())
415        self.assertTrue(IReportsContainer.providedBy(fake_site['reports']))
416        return
417
418    def test_update_uptodate_site(self):
419        # we leave already existing reports containers in place
420        fake_site = grok.Container()
421        plugin = ReportsContainerPlugin()
422        fake_site['reports'] = ReportsContainer()
423        plugin.update(fake_site, 'app', self.logger)
424        log = self.get_log()
425        self.assertEqual(log, '') # no log message
426        return
427
428    def test_setup_new_site(self):
429        # if we setup a site, we always install a fresh reports container
430        fake_site = grok.Container()
431        plugin = ReportsContainerPlugin()
432        plugin.setup(fake_site, 'app', self.logger)
433        log1 = self.get_log()
434        result1 = fake_site.get('reports', None)
435        plugin.setup(fake_site, 'app', self.logger) # replace old container
436        log2 = self.get_log()
437        result2 = fake_site.get('reports', None)
438        self.assertTrue(result1 is not result2)
439        self.assertEqual(log1,
440                         'Added reports container for site "app"\n')
441        self.assertEqual(log2,
442                         'Added reports container for site "app"\n'
443                         'Removed reports container for site "app"\n'
444                         'Added reports container for site "app"\n')
445        return
Note: See TracBrowser for help on using the repository browser.