test_project_stats.py 文件源码

def test_create_dict_job(self):
        """Test JOB create dict job works."""
        user = UserFactory.create(pro=True)
        project = ProjectFactory.create(owner=user)
        from sqlalchemy.sql import text
        from pybossa.core import db
        sql = text('''SELECT project.id, project.short_name FROM project, "user"
                   WHERE project.owner_id="user".id AND "user".pro=True;''')
        results = db.slave_session.execute(sql)
        jobs_generator = create_dict_jobs(results, get_project_stats, (10 * 60))
        jobs = []
        for job in jobs_generator:
            jobs.append(job)

        err_msg = "There should be only one job"
        assert len(jobs) == 1, err_msg

        job = jobs[0]
        assert 'get_project_stats' in job['name'].__name__
        assert job['args'] == [project.id, project.short_name]