source: main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/tests/test_async.py @ 9083

# Tests for the async module of waeup.kofa
import re
import zc.async.configure
import zc.async.ftesting
import zc.async.queue
from zc.async.interfaces import IJob
from zope.component import getUtility, provideAdapter
from zope.container.interfaces import IContainer
from zope.interface import verify
from waeup.kofa.interfaces import (
    WAEUP_KEY, IProgressable, IJobManager, IJobContainer
    )
from waeup.kofa.testing import FunctionalTestCase, FunctionalLayer
from waeup.kofa.async import (
    get_job_id, AsyncJob, ProgressableJob, JobManager, JobContainer
    )

def fake_async_func1():
    return 23

def fake_async_func2():
    return 42

class FunctionalAsyncTestCase(FunctionalTestCase):
    """A functional test case that additionally sets up asynchronous
    machinery.

    These type of test case takes _much_ more time than regular
    functional tests! So, only use it if really neccessary.
    """

    def setUp(self):
        super(FunctionalAsyncTestCase, self).setUp()
        zc.async.configure.base()
        provideAdapter(zc.async.queue.getDefaultQueue)
        connection = self.getRootFolder()._p_jar
        zc.async.ftesting.setUp(connection=connection)
        return

    def tearDown(self):
        zc.async.ftesting.tearDown()
        super(FunctionalAsyncTestCase, self).tearDown()
        return

class CommonAsyncFunctionalTests(FunctionalTestCase):

    layer = FunctionalLayer

    def test_iface_async_job(self):
        # AsyncJob implements the promised interfaces
        job = AsyncJob(fake_async_func1)
        verify.verifyClass(IJob, AsyncJob)
        verify.verifyObject(IJob, job)
        return

    def test_iface_progressable_job(self):
        # ProgressableJob implements the promised interfaces
        job = ProgressableJob(fake_async_func1)
        verify.verifyClass(IJob, ProgressableJob)
        verify.verifyClass(IProgressable, ProgressableJob)
        verify.verifyObject(IJob, job)
        verify.verifyObject(IProgressable, job)
        return

    def test_get_job_id(self):
        # for persisted jobs we can get an oid as job id
        job = AsyncJob(fake_async_func1)
        self.getRootFolder()['myjob'] = job
        import transaction
        transaction.commit()
        oid = job.__repr__().split(' ')[2][:-1]
        result = get_job_id(job)
        self.assertEqual(result, oid)
        return

class JobContainerTests(FunctionalTestCase):

    layer = FunctionalLayer

    def test_iface_job_container(self):
        # JobContainers implement the promised interfaces
        container = JobContainer()
        verify.verifyClass(IContainer, JobContainer)
        verify.verifyObject(IContainer, container)
        verify.verifyClass(IJobContainer, JobContainer)
        verify.verifyObject(IJobContainer, container)
        return

class JobManagerTests(FunctionalAsyncTestCase):

    layer = FunctionalLayer

    def setUp(self):
        super(JobManagerTests, self).setUp()
        self.root_folder = self.getRootFolder()
        self.root = self.root_folder._p_jar.root()

    def test_iface_job_manager(self):
        # JobManager implements the promised interfaces
        manager = JobManager()
        verify.verifyClass(IJobManager, JobManager)
        verify.verifyObject(IJobManager, manager)
        return

    def test_get_job_manager_as_util(self):
        # we can get a job manager as global, unnamed utility
        manager = getUtility(IJobManager)
        self.assertTrue(isinstance(manager, JobManager))
        return

    def test_put(self):
        # we can put jobs into the manager
        myjob = AsyncJob(fake_async_func1)
        manager = getUtility(IJobManager)
        job_id = manager.put(myjob, site=self.root_folder)
        # the returned job_id should be a string with a bunch of numbers
        self.assertTrue(re.match('^\d+$', job_id))
        # the job is also in the (new) jobs container
        self.assertTrue(job_id in self.root[WAEUP_KEY].keys())
        # the job is really stored
        stored_job =self.root[WAEUP_KEY][job_id]
        self.assertEqual(myjob, stored_job)
        return

    def test_put_no_site(self):
        # if no site is around we cannot store jobs
        myjob = AsyncJob(fake_async_func1)
        manager = getUtility(IJobManager)
        self.assertRaises(LookupError, manager.put, myjob)
        return

    def test_get(self):
        # we can get a job back with a valid job_id
        manager = getUtility(IJobManager)
        myjob = AsyncJob(fake_async_func1)
        job_id = manager.put(myjob, site=self.root_folder)
        stored_job = manager.get(job_id, site=self.root_folder)
        self.assertEqual(myjob, stored_job)
        return

    def test_get_no_site(self):
        # if no site is around we cannot retrieve jobs
        manager = getUtility(IJobManager)
        self.assertRaises(LookupError, manager.get, 'some_id')
        return

    def test_get_invalid_jobid(self):
        # invalid job_ids will result in `None` (if site is valid)
        manager = getUtility(IJobManager)
        result = manager.get('not-a-valid-job-id', site=self.root_folder)
        self.assertEqual(result, None)
        return

    def test_jobs(self):
        # we can get all jobs contained
        manager = getUtility(IJobManager)
        myjob1 = AsyncJob(fake_async_func1)
        job_id1 = manager.put(myjob1, site=self.root_folder)
        myjob2 = AsyncJob(fake_async_func2)
        job_id2 = manager.put(myjob2, site=self.root_folder)
        result = sorted(list(manager.jobs(site=self.root_folder)))
        self.assertEqual(
            result,
            [(job_id1, myjob1), (job_id2, myjob2)]
            )
        return

    def test_remove(self):
        # we can remove jobs from the job container
        manager = getUtility(IJobManager)
        myjob = AsyncJob(fake_async_func1)
        job_id = manager.put(myjob, site=self.root_folder)
        container = self.root[WAEUP_KEY]
        self.assertEqual(len(container), 1)
        manager.remove(job_id, site=self.root_folder)
        self.assertEqual(len(container), 0)
        return