Changeset 9084

Timestamp:

6 Aug 2012, 17:10:53 (12 years ago)

Author:

uli

Message:

Bring test coverage to maximum value we can get (as asynchronous tasks
running in a different thread cannot be seen by the coverage
detector).

File:

• main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/tests/test_async.py (modified) (4 diffs)

main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/tests/test_async.py

@@ -1 +1 @@
 # Tests for the async module of waeup.kofa
+import os
 import re
+import shutil
+import tempfile
+import transaction
 import zc.async.configure
 import zc.async.ftesting
 import zc.async.queue
 from zc.async.interfaces import IJob
+from ZODB import DB, FileStorage
 from zope.component import getUtility, provideAdapter
+from zope.component.hooks import setSite
 from zope.container.interfaces import IContainer
 from zope.interface import verify
@@ -13 +19 @@
 from waeup.kofa.testing import FunctionalTestCase, FunctionalLayer
 from waeup.kofa.async import (
-    get_job_id, AsyncJob, ProgressableJob, JobManager, JobContainer
+    get_job_id, AsyncJob, ProgressableJob, JobManager, JobContainer,
+    JobManagerTraverser, JobContainerInstaller,
     )
 
@@ -67 +74 @@
         job = AsyncJob(fake_async_func1)
         self.getRootFolder()['myjob'] = job
-        import transaction
         transaction.commit()
         oid = job.__repr__().split(' ')[2][:-1]
@@ -176 +182 @@
         self.assertEqual(len(container), 0)
         return
+
+    def test_start_test_job(self):
+        # we can start a test job
+        manager = getUtility(IJobManager)
+        job_id = manager.start_test_job(duration=1, site=self.root_folder)
+        container = self.root[WAEUP_KEY]
+        self.assertEqual(len(container), 1)
+        self.assertTrue(job_id is not None)
+        return
+
+class JobManagerTraverserTests(FunctionalAsyncTestCase):
+
+    layer = FunctionalLayer
+
+    def setUp(self):
+        super(JobManagerTraverserTests, self).setUp()
+        self.root_folder = self.getRootFolder()
+        self.root = self.root_folder._p_jar.root()
+        self.manager = getUtility(IJobManager)
+        setSite(self.getRootFolder())
+
+    def tearDown(self):
+        setSite(None)
+        super(JobManagerTraverserTests, self).tearDown()
+
+    def test_no_jobs(self):
+        # the traverser returns None if no job is available
+        traverser = JobManagerTraverser(self.manager, None)
+        result = traverser.traverse('123')
+        self.assertTrue(result is None)
+        return
+
+    def test_valid_job(self):
+        # the traverser returns the appropriate job if available
+        myjob = AsyncJob(fake_async_func1)
+        job_id = self.manager.put(myjob, site=self.root_folder)
+        traverser = JobManagerTraverser(self.manager, None)
+        result = traverser.traverse(job_id)
+        self.assertEqual(myjob, result)
+        return
+
+class FakeEvent(object):
+    # A faked DatabaseOpenedEvent as triggered at instance startup.
+    # The only attribute we need here is `database` which should
+    # return a valid ZODB.
+
+    _storage = None
+    _db = None
+
+    def __init__(self, workdir, db_name=None):
+        # workdir is the place to create a ZODB in, db_name its name.
+        self.workdir = workdir
+        self.db_name = db_name
+
+    @property
+    def database(self):
+        if self._storage is None:
+            # storage not created yet. Do it now.
+            path = os.path.join(self.workdir, 'testdb.fs')
+            self._storage = FileStorage.FileStorage(path)
+            self._db = DB(self._storage, database_name=self.db_name)
+        return self._db
+
+class FakeDB(object):
+
+    closed = False
+
+    def open(self, transaction_manager=None):
+        return self
+
+    def close(self):
+        self.closed = True
+        return
+
+class JobContainerInstallerTests(FunctionalTestCase):
+
+    layer = FunctionalLayer
+
+    def setUp(self):
+        super(JobContainerInstallerTests, self).setUp()
+        # setup two ZODBs, one unnamed, one named...
+        self.workdir1 = tempfile.mkdtemp()
+        self.workdir2 = tempfile.mkdtemp()
+        self.fake_event1 = FakeEvent(self.workdir1)
+        self.fake_event2 = FakeEvent(self.workdir2, 'mytestdb')
+        self.tm1 = transaction.TransactionManager()
+        self.tm2 = transaction.TransactionManager()
+        self.db1 = self.fake_event1.database
+        self.db2 = self.fake_event2.database
+        self.conn1 = self.db1.open(self.tm1)
+        self.conn2 = self.db2.open(self.tm2)
+
+    def tearDown(self):
+        self.tm1.abort()
+        self.tm2.abort()
+        shutil.rmtree(self.workdir1)
+        shutil.rmtree(self.workdir2)
+        super(JobContainerInstallerTests, self).tearDown()
+
+    def reopen(self):
+        # reopen connections to the ZODBs (for retrieving fresh data)...
+        self.conn1.close()
+        self.conn2.close()
+        self.conn1 = self.db1.open(self.tm1)
+        self.conn2 = self.db2.open(self.tm2)
+        return
+
+    @property
+    def root1(self):
+        # get root of first (unnamed) database
+        self.reopen()
+        return self.conn1.root()
+
+    @property
+    def root2(self):
+        # get root of second (named 'mytestdb') database
+        self.reopen()
+        return self.conn2.root()
+
+    def test_new_container(self):
+        # we can install a job container if none is available yet
+        installer = JobContainerInstaller()
+        self.assertTrue(WAEUP_KEY not in self.root1)
+        installer(self.fake_event1)
+        self.assertTrue(WAEUP_KEY in self.root1)
+        self.assertTrue(IJobContainer.providedBy(self.root1[WAEUP_KEY]))
+        return
+
+    def test_existing_container(self):
+        # we won't try to install a new container if one is already present
+        installer = JobContainerInstaller()
+        self.assertTrue(WAEUP_KEY not in self.root1)
+        installer(self.fake_event1)
+        self.assertTrue(WAEUP_KEY in self.root1)
+        self.assertTrue(IJobContainer.providedBy(self.root1[WAEUP_KEY]))
+        container = self.root1[WAEUP_KEY]
+        installer(self.fake_event1)
+        self.assertTrue(WAEUP_KEY in self.root1)
+        self.assertTrue(self.root1[WAEUP_KEY] is container)
+        return
+
+    def test_named_db(self):
+        # we can also install containers in named dbs
+        installer = JobContainerInstaller(db_name='mytestdb')
+        self.assertTrue(WAEUP_KEY not in self.root1)
+        self.assertTrue(WAEUP_KEY not in self.root2)
+        installer(self.fake_event2)
+        # now the job container is in database2 (but not database1)
+        self.assertTrue(WAEUP_KEY not in self.root1)
+        self.assertTrue(WAEUP_KEY in self.root2)
+        self.assertTrue(IJobContainer.providedBy(self.root2[WAEUP_KEY]))
+        return
+
+    def test_broken_db(self):
+        # if writing to db fails, we close the connection anyway...
+        installer = JobContainerInstaller()
+        self.conn1.close()
+        # create a fake db
+        db = FakeDB()
+        self.fake_event1._storage = 'dummy'
+        self.fake_event1._db = db
+        self.assertEqual(db.closed, False)
+        self.assertRaises(AttributeError, installer, self.fake_event1)
+        self.assertEqual(db.closed, True)
+        return