Changeset 9511 for main/waeup.kofa/trunk

Timestamp:

2 Nov 2012, 13:40:10 (12 years ago)

Author:

uli

Message:

Update tests.

File:

• main/waeup.kofa/trunk/src/waeup/kofa/tests/test_reports.py (modified) (10 diffs)

main/waeup.kofa/trunk/src/waeup/kofa/tests/test_reports.py

@@ -1 +1 @@
 import datetime
 import grok
+import logging
 import pytz
 import unittest
+from cStringIO import StringIO
+from grok.interfaces import IContainer
 from zc.async.interfaces import IJob
-from zope.component import getGlobalSiteManager
+from zc.async.testing import wait_for_result
+from zope.component import getGlobalSiteManager, getUtility
+from zope.component.hooks import setSite
 from zope.interface.verify import verifyClass, verifyObject
-from waeup.kofa.interfaces import IJobManager
+from waeup.kofa.interfaces import IJobManager, IKofaPluggable
 from waeup.kofa.reports import (
-    IReport, IReportGenerator, IReportJob, IReportJobContainer,)
+    IReport, IReportGenerator, IReportJob, IReportJobContainer,
+    IReportsContainer,)
 from waeup.kofa.reports import (
     Report, ReportGenerator, get_generators, report_job, AsyncReportJob,
-    ReportJobContainer,)
-from waeup.kofa.testing import FakeJob, FakeJobManager
+    ReportJobContainer, ReportsContainer, ReportsContainerPlugin)
+from waeup.kofa.testing import FakeJob, FakeJobManager, FunctionalLayer
+from waeup.kofa.tests.test_async import FunctionalAsyncTestCase
 
 class FakeReportGenerator(ReportGenerator):
@@ -18 +25 @@
 
     def __init__(self, name=None, perm_create=None, perm_view=None):
-        self.name = name
+        self.name = 'Report 1'
         self.perm_create = perm_create
         self.perm_view = perm_view
@@ -24 +31 @@
 
     def __eq__(self, obj):
+        if getattr(obj, 'name', None) is None:
+            return False
         return self.name == obj.name
 
-    def generate(self, site):
-        return Report()
+    def generate(self, site, args=[], kw={}):
+        result = Report()
+        result.args = args
+        result.kw = kw
+        return result
 
 class ReportTests(unittest.TestCase):
@@ -47 +59 @@
         return
 
+    def test_args(self):
+        # a report stores the args and kw used for generation
+        report1 = Report()
+        report2 = Report(args=[1, 2], kwargs=dict(a=1, b=2))
+        self.assertEqual(report1.args, [])
+        self.assertEqual(report1.kwargs, dict())
+        self.assertEqual(report2.args, [1, 2])
+        self.assertEqual(report2.kwargs, dict(a=1, b=2))
+        return
+
+    def test_create_pdf(self):
+        # trying to create a pdf results in an error
+        report = Report()
+        self.assertRaises(NotImplementedError, report.create_pdf)
+        return
+
 class ReportGeneratorTest(unittest.TestCase):
     def setUp(self):
@@ -58 +86 @@
         return
 
-
-class HelpersTests(unittest.TestCase):
-    # Tests for helper functions
-
-    def setUp(self):
-        grok.testing.grok('waeup.kofa.reports') # register utils
-        self.registered = [] # report utils registrations
+    def test_generate(self):
+        # the base report generator delivers reports
+        obj = ReportGenerator()
+        result = obj.generate(None)
+        self.assertTrue(IReport.providedBy(result))
+        return
+
+class GeneratorRegistrar(object):
+    # Mix-in providing report generator registrations
+    def setUp(self):
+        self.registered = []
 
     def tearDown(self):
@@ -81 +113 @@
         return generator
 
+class HelpersTests(GeneratorRegistrar, unittest.TestCase):
+    # Tests for helper functions
+
+    def setUp(self):
+        grok.testing.grok('waeup.kofa.reports') # register utils
+        super(HelpersTests, self).setUp()
+        return
+
     def test_get_generators_none(self):
         # we get no generators if none was registered
@@ -106 +146 @@
         return
 
-class ReportJobTests(HelpersTests):
+
+class ReportJobTests(FunctionalAsyncTestCase, GeneratorRegistrar):
     # Test asynchronous report functionality (simple cases)
+
+    layer = FunctionalLayer
+
+    def setUp(self):
+        super(ReportJobTests, self).setUp()
+        GeneratorRegistrar.setUp(self)
+        self.root_folder = self.getRootFolder()
 
     def test_report_job_func(self):
@@ -125 +173 @@
         return
 
+    def test_finished(self):
+        # AsyncReportJobs signal with a bool whether they`re  finished
+        job = AsyncReportJob(self.root_folder, None)
+        setSite(self.root_folder)
+        self.assertEqual(job.finished, False)
+        manager = getUtility(IJobManager)
+        manager.put(job)
+        wait_for_result(job)
+        self.assertEqual(job.finished, True)
+        return
+
+    def test_failed_true(self):
+        # We can test whether a job failed
+        job = AsyncReportJob(self.root_folder, None) # no report generator
+        setSite(self.root_folder)
+        # while a job is not finished, `failed` is ``None``
+        self.assertTrue(job.failed is None)
+        manager = getUtility(IJobManager)
+        manager.put(job)
+        wait_for_result(job)
+        # the finished job failed
+        self.assertEqual(job.failed, True)
+        return
+
+    def test_failed_false(self):
+        # We can test whether a job failed
+        self.register_generator('report1')
+        job = AsyncReportJob(self.root_folder, 'report1')
+        setSite(self.root_folder)
+        # while a job is not finished, `failed` is ``None``
+        self.assertTrue(job.failed is None)
+        manager = getUtility(IJobManager)
+        manager.put(job)
+        wait_for_result(job)
+        # the finished job failed
+        self.assertEqual(job.failed, False)
+        return
+
+    def test_description(self):
+        # IReportJobs provide a description of the started job
+        self.register_generator('report1')
+        args, kw = ['a', 'b'], dict(a=1, b=2)
+        job = AsyncReportJob(self.root_folder, 'report1', args=args, kw=kw)
+        self.assertEqual(
+            job.description,
+            "Report 1 ('a', 'b', a=1, b=2)")
+        # w/o args nor kwargs
+        job = AsyncReportJob(self.root_folder, 'report1')
+        self.assertEqual(
+            job.description,
+            'Report 1 ()')
+        # with args only
+        job = AsyncReportJob(self.root_folder, 'report1', args=args)
+        self.assertEqual(
+            job.description, "Report 1 ('a', 'b')")
+        # with keywords only
+        job = AsyncReportJob(self.root_folder, 'report1', kw=kw)
+        self.assertEqual(
+            job.description, "Report 1 (a=1, b=2)")
+        return
+
+    def test_description_invalid_generator(self):
+        # We can get a description even with an invalid generator
+        job = AsyncReportJob(self.root_folder, 'NOT EXISTENT')
+        self.assertEqual(
+            job.description, u'Invalid Report Generator ()')
+        # with args set (no kws)
+        job = AsyncReportJob(
+            self.root_folder, 'NOT EXISTENT', args=['a', 'b'])
+        self.assertEqual(
+            job.description, u"Invalid Report Generator ('a', 'b')")
+        # with kw set (no args)
+        job = AsyncReportJob(
+            self.root_folder, 'NOT_EXISTENT', kw=dict(a=1, b=2))
+        self.assertEqual(
+            job.description, u'Invalid Report Generator (a=1, b=2)')
+        # with args and kws set
+        job = AsyncReportJob(
+            self.root_folder, 'NOT_EXISTENT', args=['a'], kw=dict(b=2))
+        self.assertEqual(
+            job.description, u"Invalid Report Generator ('a', b=2)")
+        return
+
 
 class FakeJobWithResult(FakeJob):
 
-    def __init__(self):
+    def __init__(self, args=[], kw={}):
         #self.dir_path = tempfile.mkdtemp()
         #self.result = os.path.join(self.dir_path, 'fake.csv')
         #open(self.result, 'wb').write('a fake result')
         self.result = Report()
+        self.result.args = args
+        self.result.kw = kw
         return
 
@@ -225 +358 @@
         self.assertEqual(
             result,
-            [('new', u'new', u'unnamed'),
-             ('completed', u'completed', u'unnamed')]
+            [('new', u'new', u'Unnamed Report'),
+             ('completed', u'completed', u'Unnamed Report')]
             )
         return
@@ -257 +390 @@
         self.assertEqual(result2, ('4', 'report3', 'bob'))
         self.assertEqual(result3, None)
-        #shutil.rmtree(fake_job.dir_path)
-        return
+        return
+
+
+class ReportsContainerTests(unittest.TestCase):
+    # Tests for ReportsContainer
+
+    def test_iface(self):
+        # ReportsContainers really provide the promised interfaces
+        obj = ReportsContainer()
+        verifyClass(IReportsContainer, ReportsContainer)
+        verifyClass(IContainer, ReportsContainer)
+        verifyObject(IReportsContainer, obj)
+        verifyObject(IContainer, obj)
+        return
+
+class ReportsContainerPluginTests(unittest.TestCase):
+    # Tests for ReportsContainerPlugin
+
+    def create_logger(self):
+        # create a logger suitable for local tests.
+        test_logger = logging.getLogger('waeup.kofa.reports.testlogger')
+        log = StringIO()
+        handler = logging.StreamHandler(log)
+        handler.setLevel(logging.DEBUG)
+        test_logger.addHandler(handler)
+        test_logger.setLevel(logging.DEBUG)
+        self.logger = test_logger
+        self.log = log
+        self.handler = handler
+        return self.logger
+
+    def remove_logger(self):
+        del self.handler
+        del self.logger
+        del self.log
+        pass
+
+    def get_log(self):
+        self.log.seek(0)
+        return self.log.read()
+
+    def setUp(self):
+        self.create_logger()
+        return
+
+    def tearDown(self):
+        self.remove_logger()
+        return
+
+    def test_iface(self):
+        # make sure we fullfill the promised interfaces
+        obj = ReportsContainerPlugin()
+        verifyClass(IKofaPluggable, ReportsContainerPlugin)
+        verifyObject(IKofaPluggable, obj)
+        return
+
+    def test_get_as_utility(self):
+        # make sure we can get the plugin as utility
+        grok.testing.grok('waeup.kofa.reports')
+        util = getUtility(IKofaPluggable, name='reports')
+        self.assertTrue(util is not None)
+        return
+
+    def test_update_no_container(self):
+        # we can update an existing site
+        fake_site = grok.Container()
+        plugin = ReportsContainerPlugin()
+        plugin.update(fake_site, 'app', self.logger)
+        log = self.get_log()
+        self.assertEqual(
+            log, 'Added reports container for site "app"\n')
+        self.assertTrue('reports' in fake_site.keys())
+        self.assertTrue(IReportsContainer.providedBy(fake_site['reports']))
+        return
+
+    def test_update_uptodate_site(self):
+        # we leave already existing reports containers in place
+        fake_site = grok.Container()
+        plugin = ReportsContainerPlugin()
+        fake_site['reports'] = ReportsContainer()
+        plugin.update(fake_site, 'app', self.logger)
+        log = self.get_log()
+        self.assertEqual(log, '') # no log message
+        return
+
+    def test_setup_new_site(self):
+        # if we setup a site, we always install a fresh reports container
+        fake_site = grok.Container()
+        plugin = ReportsContainerPlugin()
+        plugin.setup(fake_site, 'app', self.logger)
+        log1 = self.get_log()
+        result1 = fake_site.get('reports', None)
+        plugin.setup(fake_site, 'app', self.logger) # replace old container
+        log2 = self.get_log()
+        result2 = fake_site.get('reports', None)
+        self.assertTrue(result1 is not result2)
+        self.assertEqual(log1,
+                         'Added reports container for site "app"\n')
+        self.assertEqual(log2,
+                         'Added reports container for site "app"\n'
+                         'Removed reports container for site "app"\n'
+                         'Added reports container for site "app"\n')
+        return