Changeset 9511 for main


Ignore:
Timestamp:
2 Nov 2012, 13:40:10 (12 years ago)
Author:
uli
Message:

Update tests.

File:
1 edited

Legend:

Unmodified
Added
Removed

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

    r9344 r9511  
    11import datetime
    22import grok
     3import logging
    34import pytz
    45import unittest
     6from cStringIO import StringIO
     7from grok.interfaces import IContainer
    58from zc.async.interfaces import IJob
    6 from zope.component import getGlobalSiteManager
     9from zc.async.testing import wait_for_result
     10from zope.component import getGlobalSiteManager, getUtility
     11from zope.component.hooks import setSite
    712from zope.interface.verify import verifyClass, verifyObject
    8 from waeup.kofa.interfaces import IJobManager
     13from waeup.kofa.interfaces import IJobManager, IKofaPluggable
    914from waeup.kofa.reports import (
    10     IReport, IReportGenerator, IReportJob, IReportJobContainer,)
     15    IReport, IReportGenerator, IReportJob, IReportJobContainer,
     16    IReportsContainer,)
    1117from waeup.kofa.reports import (
    1218    Report, ReportGenerator, get_generators, report_job, AsyncReportJob,
    13     ReportJobContainer,)
    14 from waeup.kofa.testing import FakeJob, FakeJobManager
     19    ReportJobContainer, ReportsContainer, ReportsContainerPlugin)
     20from waeup.kofa.testing import FakeJob, FakeJobManager, FunctionalLayer
     21from waeup.kofa.tests.test_async import FunctionalAsyncTestCase
    1522
    1623class FakeReportGenerator(ReportGenerator):
    … …  
    1825
    1926    def __init__(self, name=None, perm_create=None, perm_view=None):
    20         self.name = name
     27        self.name = 'Report 1'
    2128        self.perm_create = perm_create
    2229        self.perm_view = perm_view
    … …  
    2431
    2532    def __eq__(self, obj):
     33        if getattr(obj, 'name', None) is None:
     34            return False
    2635        return self.name == obj.name
    2736
    28     def generate(self, site):
    29         return Report()
     37    def generate(self, site, args=[], kw={}):
     38        result = Report()
     39        result.args = args
     40        result.kw = kw
     41        return result
    3042
    3143class ReportTests(unittest.TestCase):
    … …  
    4759        return
    4860
     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
    4977class ReportGeneratorTest(unittest.TestCase):
    5078    def setUp(self):
    … …  
    5886        return
    5987
    60 
    61 class HelpersTests(unittest.TestCase):
    62     # Tests for helper functions
    63 
    64     def setUp(self):
    65         grok.testing.grok('waeup.kofa.reports') # register utils
    66         self.registered = [] # report utils registrations
     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 = []
    6799
    68100    def tearDown(self):
    … …  
    81113        return generator
    82114
     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
    83123    def test_get_generators_none(self):
    84124        # we get no generators if none was registered
    … …  
    106146        return
    107147
    108 class ReportJobTests(HelpersTests):
     148
     149class ReportJobTests(FunctionalAsyncTestCase, GeneratorRegistrar):
    109150    # 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()
    110158
    111159    def test_report_job_func(self):
    … …  
    125173        return
    126174
     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
    127258
    128259class FakeJobWithResult(FakeJob):
    129260
    130     def __init__(self):
     261    def __init__(self, args=[], kw={}):
    131262        #self.dir_path = tempfile.mkdtemp()
    132263        #self.result = os.path.join(self.dir_path, 'fake.csv')
    133264        #open(self.result, 'wb').write('a fake result')
    134265        self.result = Report()
     266        self.result.args = args
     267        self.result.kw = kw
    135268        return
    136269
    … …  
    225358        self.assertEqual(
    226359            result,
    227             [('new', u'new', u'unnamed'),
    228              ('completed', u'completed', u'unnamed')]
     360            [('new', u'new', u'Unnamed Report'),
     361             ('completed', u'completed', u'Unnamed Report')]
    229362            )
    230363        return
    … …  
    257390        self.assertEqual(result2, ('4', 'report3', 'bob'))
    258391        self.assertEqual(result3, None)
    259         #shutil.rmtree(fake_job.dir_path)
    260         return
     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 TracChangeset for help on using the changeset viewer.