BVB Source Codes

rq Show fixtures.py Source code

Return Download rq: download fixtures.py Source code - Download rq Source code - Type:.py
  1. # -*- coding: utf-8 -*-
  2. """
  3. This file contains all jobs that are used in tests.  Each of these test
  4. fixtures has a slighty different characteristics.
  5. """
  6. from __future__ import (absolute_import, division, print_function,
  7.                         unicode_literals)
  8.  
  9. import os
  10. import time
  11. import sys
  12.  
  13. from rq import Connection, get_current_job, get_current_connection, Queue
  14. from rq.decorators import job
  15. from rq.compat import PY2
  16. from rq.worker import HerokuWorker
  17.  
  18.  
  19. def say_pid():
  20.     return os.getpid()
  21.  
  22.  
  23. def say_hello(name=None):
  24.     """A job with a single argument and a return value."""
  25.     if name is None:
  26.         name = 'Stranger'
  27.     return 'Hi there, %s!' % (name,)
  28.  
  29.  
  30. def do_nothing():
  31.     """The best job in the world."""
  32.     pass
  33.  
  34.  
  35. def div_by_zero(x):
  36.     """Prepare for a division-by-zero exception."""
  37.     return x / 0
  38.  
  39.  
  40. def some_calculation(x, y, z=1):
  41.     """Some arbitrary calculation with three numbers.  Choose z smartly if you
  42.    want a division by zero exception.
  43.    """
  44.     return x * y / z
  45.  
  46.  
  47. def create_file(path):
  48.     """Creates a file at the given path.  Actually, leaves evidence that the
  49.    job ran."""
  50.     with open(path, 'w') as f:
  51.         f.write('Just a sentinel.')
  52.  
  53.  
  54. def create_file_after_timeout(path, timeout):
  55.     time.sleep(timeout)
  56.     create_file(path)
  57.  
  58.  
  59. def access_self():
  60.     assert get_current_connection() is not None
  61.     assert get_current_job() is not None
  62.  
  63.  
  64. def echo(*args, **kwargs):
  65.     return (args, kwargs)
  66.  
  67.  
  68. class Number(object):
  69.     def __init__(self, value):
  70.         self.value = value
  71.  
  72.     @classmethod
  73.     def divide(cls, x, y):
  74.         return x * y
  75.  
  76.     def div(self, y):
  77.         return self.value / y
  78.  
  79.  
  80. class CallableObject(object):
  81.     def __call__(self):
  82.         return u"I'm callable"
  83.  
  84.  
  85. class UnicodeStringObject(object):
  86.     def __repr__(self):
  87.         if PY2:
  88.             return u'茅'.encode('utf-8')
  89.         else:
  90.             return u'茅'
  91.  
  92.  
  93. with Connection():
  94.     @job(queue='default')
  95.     def decorated_job(x, y):
  96.         return x + y
  97.  
  98.  
  99. def black_hole(job, *exc_info):
  100.     # Don't fall through to default behaviour (moving to failed queue)
  101.     return False
  102.  
  103.  
  104. def long_running_job(timeout=10):
  105.     time.sleep(timeout)
  106.     return 'Done sleeping...'
  107.  
  108.  
  109. def run_dummy_heroku_worker(sandbox, _imminent_shutdown_delay):
  110.     """
  111.    Run the work horse for a simplified heroku worker where perform_job just
  112.    creates two sentinel files 2 seconds apart.
  113.    :param sandbox: directory to create files in
  114.    :param _imminent_shutdown_delay: delay to use for HerokuWorker
  115.    """
  116.     sys.stderr = open(os.path.join(sandbox, 'stderr.log'), 'w')
  117.  
  118.     class TestHerokuWorker(HerokuWorker):
  119.         imminent_shutdown_delay = _imminent_shutdown_delay
  120.  
  121.         def perform_job(self, job, queue):
  122.             create_file(os.path.join(sandbox, 'started'))
  123.             # have to loop here rather than one sleep to avoid holding the GIL
  124.             # and preventing signals being received
  125.             for i in range(20):
  126.                 time.sleep(0.1)
  127.             create_file(os.path.join(sandbox, 'finished'))
  128.  
  129.     w = TestHerokuWorker(Queue('dummy'))
  130.     w.main_work_horse(None, None)
  131.  
downloadfixtures.py Source code - Download rq Source code
Related Source Codes/Software:
amazon-dsstne - Deep Scalable Sparse Tensor Network Engine (DSSTNE... 2017-01-08
webpack-demos - a collection of simple demos of Webpack 2017-01-08
Squire - HTML5 rich text editor. Try the demo integration a... 2017-01-08
thor - Thor is a toolkit for building powerful command-li... 2017-01-08
glide - Package Management for Golang h... 2017-01-08
emmet-vim - emmet for vim: http://emmet.io/ ... 2017-01-08
prose - A Content Editor for GitHub. ht... 2017-01-08
sshrc - ring your .bashrc, .vimrc, etc. with you when you ... 2017-01-08
typed.js - A jQuery typing animation script. ... 2017-01-08
find - High-precision indoor positioning framework for mo... 2017-01-08
zen - IOS, Swift, Objective - C 2017-04-25
neural-storyteller - A recurrent neural network for generating little s... 2017-04-25
CTAssetsPickerController - iOS control that allows picking multiple photos an... 2017-04-25
retire - A rich Ruby API and DSL for the Elasticsearch sear... 2017-04-25
jxcore - Evented IO for ChakraCore, SpiderMonkey & V8 J... 2017-04-25
node.io - 2017-04-25
PHPVerbalExpressions - PHP Regular expressions made easy 2017-04-25
Laravel-Administrator - An administrative interface package for Laravel ... 2017-04-25
elasticsearch-dump - Import and export tools for elasticsearch 2017-04-25
flyway - Flyway by Boxfuse Database Migrations Made Easy. ... 2017-04-25

 Back to top