import datetime import grok import pytz import unittest from zc.async.interfaces import IJob from zope.component import getGlobalSiteManager from zope.interface.verify import verifyClass, verifyObject from waeup.kofa.interfaces import IJobManager from waeup.kofa.reports import ( IReport, IReportGenerator, IReportJob, IReportJobContainer,) from waeup.kofa.reports import ( Report, ReportGenerator, get_generators, report_job, AsyncReportJob, ReportJobContainer,) from waeup.kofa.testing import FakeJob, FakeJobManager class FakeReportGenerator(ReportGenerator): # fake report generator for tests. def __init__(self, name=None, perm_create=None, perm_view=None): self.name = name self.perm_create = perm_create self.perm_view = perm_view return def __eq__(self, obj): return self.name == obj.name def generate(self, site): return Report() class ReportTests(unittest.TestCase): def test_iface(self): # make sure we fullfill the promised contracts obj = Report() verifyClass(IReport, Report) verifyObject(IReport, obj) return def test_creation_dt(self): # a report sets a datetime timestamp when created report = Report() self.assertTrue(hasattr(report, 'creation_dt')) self.assertTrue(isinstance(report.creation_dt, datetime.datetime)) # the datetime is set with UTC timezone info self.assertEqual(report.creation_dt.tzinfo, pytz.utc) return class ReportGeneratorTest(unittest.TestCase): def setUp(self): grok.testing.grok('waeup.kofa.reports') # register utils def test_iface(self): # make sure we fullfill the promised contracts obj = ReportGenerator() verifyClass(IReportGenerator, ReportGenerator) verifyObject(IReportGenerator, obj) 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 tearDown(self): # unregister any previously registered report types gsm = getGlobalSiteManager() for report, name in self.registered: gsm.unregisterUtility(report, IReportGenerator, name=name) return def register_generator(self, name, perm_create=None, perm_view=None): # helper to register report generators as utils gsm = getGlobalSiteManager() generator = FakeReportGenerator(name) gsm.registerUtility(generator, provided=IReportGenerator, name=name) self.registered.append((generator, name),) return generator def test_get_generators_none(self): # we get no generators if none was registered result = list(get_generators()) self.assertEqual(result, []) return def test_get_generators_simple(self): # we get a single generator if one is registered self.register_generator('report1') result = list(get_generators()) self.assertEqual( result, [('report1', FakeReportGenerator('report1'))]) return def test_get_generators_multiple(self): # we also get multiple generators if available self.register_generator('report1') self.register_generator('report2') result = list(get_generators()) self.assertEqual( result, [('report1', FakeReportGenerator('report1')), ('report2', FakeReportGenerator('report2'))]) return class ReportJobTests(HelpersTests): # Test asynchronous report functionality (simple cases) def test_report_job_func(self): # the report_job func really creates reports... self.register_generator('report1') report = report_job(None, 'report1') self.assertTrue(IReport.providedBy(report)) self.assertTrue(isinstance(report, Report)) def test_report_job_interfaces(self): # the AsyncReportJob implements promised interfaces correctly... job = AsyncReportJob(None, None) verifyClass(IJob, AsyncReportJob) verifyObject(IJob, job) verifyClass(IReportJob, AsyncReportJob) verifyObject(IReportJob, job) return class FakeJobWithResult(FakeJob): def __init__(self): #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() return class ReportJobContainerTests(unittest.TestCase): # Test ReportJobContainer def setUp(self): # register a suitable ICSVExporter as named utility self.generator = FakeReportGenerator('report1') self.job_manager = FakeJobManager() self.gsm = getGlobalSiteManager() self.gsm.registerUtility( self.generator, IReportGenerator, name='report1') self.gsm.registerUtility( self.job_manager, IJobManager) def tearDown(self): self.gsm.unregisterUtility( self.generator, IReportGenerator, name='report1') self.gsm.unregisterUtility(self.job_manager, IJobManager) def test_report_job_interfaces(self): # the ExportJobContainer implements promised interfaces correctly... container = ReportJobContainer() verifyClass(IReportJobContainer, ReportJobContainer) verifyObject(IReportJobContainer, container) return def test_start_report_job(self): # we can start jobs container = ReportJobContainer() container.start_report_job('report3', 'bob') result = self.job_manager._jobs.values()[0] self.assertTrue(IJob.providedBy(result)) self.assertEqual( container.running_report_jobs, [('1', 'report3', 'bob')] ) return def test_get_running_report_jobs_all(self): # we can get report jobs of all users container = ReportJobContainer() container.start_report_job('report3', 'bob') container.start_report_job('report3', 'alice') result = container.get_running_report_jobs() self.assertEqual( result, [('1', 'report3', 'bob'), ('2', 'report3', 'alice')] ) return def test_get_running_report_jobs_user(self): # we can get the report jobs running for a certain user container = ReportJobContainer() container.start_report_job('report3', 'bob') container.start_report_job('report3', 'alice') result1 = container.get_running_report_jobs(user_id='alice') result2 = container.get_running_report_jobs(user_id='foo') self.assertEqual( result1, [('2', 'report3', 'alice')]) self.assertEqual( result2, []) return def test_get_running_report_jobs_only_if_exist(self): # we get only jobs that are accessible through the job manager... container = ReportJobContainer() container.start_report_job('report3', 'bob') container.start_report_job('report3', 'bob') self.assertTrue( ('2', 'report3', 'bob') in container.running_report_jobs) # we remove the second entry from job manager del self.job_manager._jobs['2'] result = container.get_running_report_jobs(user_id='bob') self.assertEqual( result, [('1', 'report3', 'bob')]) self.assertTrue( ('2', 'report3', 'bob') not in container.running_report_jobs) return def test_get_report_job_status(self): # we can get the stati of jobs... container = ReportJobContainer() container.start_report_job('report1', 'alice') container.start_report_job('report1', 'bob') container.start_report_job('report1', 'bob') result = container.get_report_jobs_status(user_id='bob') # we'll get the raw value, a translation and the title of the # exporter self.assertEqual( result, [('new', u'new', u'unnamed'), ('completed', u'completed', u'unnamed')] ) return def test_delete_report_entry(self): # we can remove report entries in local lists and the job # manager as well... container = ReportJobContainer() container.start_report_job('report3', 'bob') entry = container.running_report_jobs[0] container.delete_report_entry(entry) # both, running_report_jobs list and job manager are empty now self.assertEqual( container.running_report_jobs, []) self.assertEqual( self.job_manager._jobs, {}) return def test_report_entry_from_job_id(self): # we can get an report entry for a job_id if the id exists container = ReportJobContainer() entry = ('4', 'report3', 'bob') container.running_report_jobs = [entry] fake_job = FakeJobWithResult() self.job_manager._jobs['4'] = fake_job result1 = container.report_entry_from_job_id(None) result2 = container.report_entry_from_job_id('4') result3 = container.report_entry_from_job_id('23') self.assertEqual(result1, None) self.assertEqual(result2, ('4', 'report3', 'bob')) self.assertEqual(result3, None) #shutil.rmtree(fake_job.dir_path) return