[Checkins] SVN: Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/ remove functional testing
Godefroid Chapelle
gotcha at bubblenet.be
Mon Mar 8 06:41:57 EST 2010
Log message for revision 109825:
remove functional testing
Changed:
D Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/TESTING.txt
D Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/ftesting.zcml
D Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/ftests.py
-=-
Deleted: Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/TESTING.txt
===================================================================
--- Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/TESTING.txt 2010-03-08 11:40:43 UTC (rev 109824)
+++ Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/TESTING.txt 2010-03-08 11:41:57 UTC (rev 109825)
@@ -1,129 +0,0 @@
-=====================
-Remote Task Execution
-=====================
-
-This package provides an implementation of a remote task execution Web service
-that allows to execute pre-defined tasks on another server. See more info
-about the TaskService in README.txt. This test will test the TaskServiceStub
-implementation. The only different is, that the TaskServiceStub will handle
-task implementation providing ITaskStub interfaces rather then ITask. This way
-we can register stub tasks for a testing setup.
-
-Let's now start by creating a task service stub:
-
- >>> from z3c import taskqueue
- >>> from z3c.taskqueue import testing
- >>> service = testing.TaskServiceStub()
-
-We can discover the available tasks:
-
- >>> service.getAvailableTasks()
- {}
-
-This list is initially empty, because we have not registered any tasks. Let's
-now define a task that simply echos an input string:
-
- >>> def echo(input):
- ... return input
-
- >>> import z3c.taskqueue.task
- >>> echoTask = taskqueue.task.SimpleTask(echo)
-
-The only API requirement on the converter is to be callable. Now we make sure
-that the task works:
-
- >>> echoTask(service, 1, input={'foo': 'blah'})
- {'foo': 'blah'}
-
-Let's now register the task as a utility. Note that we need to register the
-echo utility used in the REAME.txt tests for the ITaskStub interface:
-
- >>> import zope.component
- >>> zope.component.provideUtility(echoTask, provides=testing.ITaskStub,
- ... name='echo')
-
-The echo task is now available in the service:
-
- >>> service.getAvailableTasks()
- {u'echo': <SimpleTask <function echo ...>>}
-
-
-Since the service cannot instantaneously complete a task, incoming jobs are
-managed by a queue. First we request the echo task to be executed:
-
- >>> jobid = service.add(u'echo', {'foo': 'bar'})
- >>> jobid
- 1
-
-Let's also see wat's happen if we add a non existent task:
-
- >>> service.add(u'undefined', {'foo': 'bar'})
- Traceback (most recent call last):
- ...
- ValueError: Task does not exist
-
-The ``add()`` function schedules the task called "echo" to be executed with
-the specified arguments. The method returns a job id with which we can inquire
-about the job.
-
- >>> service.getStatus(jobid)
- 'queued'
-
-Since the job has not been processed, the status is set to "queued". Further,
-there is no result available yet:
-
- >>> service.getResult(jobid) is None
- True
-
-As long as the job is not being processed, it can be cancelled:
-
- >>> service.cancel(jobid)
- >>> service.getStatus(jobid)
- 'cancelled'
-
-Let's also see wat's happen if we cancel a non existent task:
-
- >>> service.cancel(u'undefined')
-
-Let's now readd a job:
-
- >>> jobid = service.add(u'echo', {'foo': 'bar'})
-
-The jobs in the queue are processed by calling the service's ``process()``
-method:
-
- >>> service.process()
-
-This method is usually called by other application logic, but we have to call
-it manually here, since none of the other infrastructure is setup.
-
- >>> service.getStatus(jobid)
- 'completed'
- >>> service.getResult(jobid)
- {'foo': 'bar'}
-
-Now, let's define a new task that causes an error:
-
- >>> def error(input):
- ... raise taskqueue.task.TaskError('An error occurred.')
-
- >>> zope.component.provideUtility(
- ... taskqueue.task.SimpleTask(error), provides=testing.ITaskStub,
- ... name='error')
-
-Now add and execute it:
-
- >>> jobid = service.add(u'error')
- >>> service.process()
-
-Let's now see what happened:
-
- >>> service.getStatus(jobid)
- 'error'
- >>> service.getError(jobid)
- 'An error occurred.'
-
-For management purposes, the service also allows you to inspect all jobs:
-
- >>> dict(service.jobs)
- {1: <Job 1>, 2: <Job 2>, 3: <Job 3>}
Deleted: Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/ftesting.zcml
===================================================================
--- Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/ftesting.zcml 2010-03-08 11:40:43 UTC (rev 109824)
+++ Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/ftesting.zcml 2010-03-08 11:41:57 UTC (rev 109825)
@@ -1,46 +0,0 @@
-<configure xmlns="http://namespaces.zope.org/zope"
- xmlns:browser="http://namespaces.zope.org/browser"
- xmlns:zcml="http://namespaces.zope.org/zcml"
- i18n_domain="zope">
-
- <include package="zope.securitypolicy" file="meta.zcml" />
-
- <include
- zcml:condition="installed zope.app.zcmlfiles"
- package="zope.app.zcmlfiles"
- />
- <include
- zcml:condition="not-installed zope.app.zcmlfiles"
- package="zope.app"
- />
-
- <include package="zope.login" />
- <include package="zope.app.authentication" />
- <include package="zope.app.session" />
- <include package="z3c.taskqueue" />
-
- <utility
- factory="z3c.taskqueue.testing.ExceptionTask"
- name="exception" />
-
- <securityPolicy
- component="zope.securitypolicy.zopepolicy.ZopeSecurityPolicy" />
- <include package="zope.app.securitypolicy" />
- <role id="zope.Anonymous" title="Everybody"
- description="All users have this role implicitly" />
- <role id="zope.Manager" title="Site Manager" />
- <role id="zope.Member" title="Site Member" />
- <grantAll role="zope.Manager" />
-
- <principal
- id="zope.manager"
- title="Administrator"
- login="mgr"
- password="mgrpw" />
-
- <grant
- role="zope.Manager"
- principal="zope.manager"
- />
-
-</configure>
Deleted: Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/ftests.py
===================================================================
--- Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/ftests.py 2010-03-08 11:40:43 UTC (rev 109824)
+++ Sandbox/gotcha/z3c.taskqueue/src/z3c/taskqueue/ftests.py 2010-03-08 11:41:57 UTC (rev 109825)
@@ -1,44 +0,0 @@
-##############################################################################
-#
-# Copyright (c) 2006 Lovely Systems and Contributors.
-# All Rights Reserved.
-#
-# This software is subject to the provisions of the Zope Public License,
-# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
-# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
-# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
-# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
-# FOR A PARTICULAR PURPOSE.
-#
-##############################################################################
-"""
-
-$Id$
-"""
-__docformat__ = 'restructuredtext'
-import unittest
-from zope.app.testing import functional
-import os
-import random
-
-zcml = os.path.join(os.path.dirname(__file__), 'ftesting.zcml')
-
-functional.defineLayer('RemotetaskLayer', zcml, allow_teardown=True)
-
-
-def setUp(test):
- random.seed(27)
-
-
-def tearDown(test):
- random.seed()
-
-
-def test_suite():
- suite1 = functional.FunctionalDocFileSuite(
- 'browser/README.txt',
- setUp=setUp,
- tearDown=tearDown,
- )
- suite1.layer = RemotetaskLayer
- return unittest.TestSuite((suite1, ))
More information about the checkins
mailing list