Source

celery-paste / tests / test_run.py

import sys

from celerypaste import set_loader

CELERY_CONFIG = {"BROKER_HOST": "localhost",
                 "BROKER_PORT": 5672,
                 "BROKER_VHOST": "celeryTst",
                 "BROKER_USER": "celery",
                 "BROKER_PASSWORD": "celery",
                 "CELERY_BACKEND": "database",
                 "DATABASE_ENGINE": "sqlite3",
                 "DATABASE_NAME": "test.db",
                 "CELERY_IMPORTS": ("mytasks", )}


set_loader(CELERY_CONFIG)

from celery.task import Task


class AddTask(Task):

    def run(self, x, y):
        return x+y


def setup_db():
    from celery.loaders import current_loader
    loader = current_loader()
    conf = loader.read_configuration()
    from django.core.management import call_command, setup_environ
    sys.stderr.write("Creating database tables...\n")
    setup_environ(conf)
    call_command("syncdb")


def test_run():
    setup_db()
    result = AddTask.apply(args=[10, 10])
    assert result.get()==20, "didn't get expected result"