def run(self):
# the code until the while statement does NOT run atomicaly
# a thread while loop cycle is atomic
# thread safe locals: L = threading.local(), then L.foo="baz"
django.setup()
self.logger.info('Worker Starts')
while not self._stopevent.isSet():
if not self.worker_queue.empty():
try:
task = self.worker_queue.get()
self.run_task(task)
except Exception as e:
helpers.save_task_failed(task,e)
else:
helpers.save_task_success(task)
self.worker_queue = None
self.logger.warn('Worker stopped, %s tasks handled'%self.tasks_counter)
python类setup()的实例源码
def run(*args):
"""
Check and/or create dj-stripe Django migrations.
If --check is present in the arguments then migrations are checked only.
"""
if not settings.configured:
settings.configure(**DEFAULT_SETTINGS)
django.setup()
parent = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, parent)
if "--check" in args:
check_migrations()
else:
django.core.management.call_command("makemigrations", APP_NAME, *args)
def run(self):
from django.conf import settings
settings.configure(
INSTALLED_APPS=('collectionfield.tests',),
SECRET_KEY='AAA',
DATABASES={
'default': {
'NAME': ':memory:',
'ENGINE': 'django.db.backends.sqlite3'
}
},
MIDDLEWARE_CLASSES = ()
)
from django.core.management import call_command
import django
django.setup()
call_command('test', 'collectionfield')
setup.py 文件源码
项目:django-calaccess-processed-data
作者: california-civic-data-coalition
项目源码
文件源码
阅读 21
收藏 0
点赞 0
评论 0
def run(self):
from django.conf import settings
settings.configure(
DATABASES={
'default': {
'NAME': ':memory:',
'ENGINE': 'django.db.backends.sqlite3'
}
},
INSTALLED_APPS=('calaccess_processed',),
MIDDLEWARE_CLASSES=()
)
from django.core.management import call_command
import django
django.setup()
call_command('test', 'calaccess_processed')
def get_sanic_application():
"""
Sets up django and returns a Sanic application
"""
if sys.version_info < (3, 5):
raise RuntimeError("The SanicDjango Adaptor may only be used with python 3.5 and above.")
django.setup()
from django.conf import settings
DEBUG = getattr(settings, 'DEBUG', False)
INSTALLED_APPS = getattr(settings, 'INSTALLED_APPS', [])
do_static = DEBUG and 'django.contrib.staticfiles' in INSTALLED_APPS
app = Sanic(__name__)
if do_static:
static_url = getattr(settings, 'STATIC_URL', "/static/")
static_root = getattr(settings, 'STATIC_ROOT', "./static")
app.static(static_url, static_root)
app.handle_request = SanicHandler(app) # patch the app to use the django adaptor handler
return app
def pytest_configure():
from django.conf import settings
settings.configure(
DEBUG_PROPAGATE_EXCEPTIONS=True,
DATABASES={'default': {'ENGINE': 'django.db.backends.sqlite3',
'NAME': ':memory:'}},
SITE_ID=1,
SECRET_KEY='not very secret in tests',
USE_I18N=True,
USE_L10N=True,
INSTALLED_APPS=(
'gstorage.apps.GStorageTestConfig',
),
)
try:
import django
django.setup()
except AttributeError:
pass
def setUpClass(cls):
from django.test.utils import setup_test_environment
from django.core.management import call_command
from django.conf import settings
settings.configure(
INSTALLED_APPS=[
'django.contrib.auth',
'django.contrib.contenttypes',
'lifter.contrib.django',
],
DATABASES={
'default': {'NAME': ':memory:', 'ENGINE': 'django.db.backends.sqlite3'}
},
)
django.setup()
setup_test_environment()
super(DjangoTestCase, cls).setUpClass()
call_command('migrate')
def runtests(*test_args):
if not settings.configured:
settings.configure(**DEFAULT_SETTINGS)
django.setup()
parent = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, parent)
try:
from django.test.runner import DiscoverRunner
runner_class = DiscoverRunner
test_args = ["pinax.news.tests"]
except ImportError:
from django.test.simple import DjangoTestSuiteRunner
runner_class = DjangoTestSuiteRunner
test_args = ["tests"]
failures = runner_class(verbosity=1, interactive=True, failfast=False).run_tests(test_args)
sys.exit(failures)
def test_autopatching_twice_middleware(self):
ok_(django._datadog_patch)
# Call django.setup() twice and ensure we don't add a duplicate tracer
django.setup()
found_app = settings.INSTALLED_APPS.count('ddtrace.contrib.django')
eq_(found_app, 1)
eq_(settings.MIDDLEWARE[0], 'ddtrace.contrib.django.TraceMiddleware')
ok_('ddtrace.contrib.django.TraceMiddleware' not in settings.MIDDLEWARE_CLASSES)
eq_(settings.MIDDLEWARE[-1], 'ddtrace.contrib.django.TraceExceptionMiddleware')
ok_('ddtrace.contrib.django.TraceExceptionMiddleware' not in settings.MIDDLEWARE_CLASSES)
found_mw = settings.MIDDLEWARE.count('ddtrace.contrib.django.TraceMiddleware')
eq_(found_mw, 1)
found_mw = settings.MIDDLEWARE.count('ddtrace.contrib.django.TraceExceptionMiddleware')
eq_(found_mw, 1)
def run_test_suite():
settings.configure(
DATABASES={
"default": {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': os.path.join(TESTS_ROOT, 'db.sqlite3'),
"USER": "",
"PASSWORD": "",
"HOST": "",
"PORT": "",
},
},
INSTALLED_APPS=[
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.sessions",
"django.contrib.sites",
"rhouser",
],
)
django.setup()
execute_from_command_line(['manage.py', 'test'])
def run_tests(self):
import django
django.setup()
from django.conf import settings
from django.test.utils import get_runner
TestRunner = get_runner(settings, self.testrunner)
test_runner = TestRunner(
pattern=self.pattern,
top_level=self.top_level_directory,
verbosity=self.verbose,
interactive=(not self.noinput),
failfast=self.failfast,
keepdb=self.keepdb,
reverse=self.reverse,
debug_sql=self.debug_sql,
output_dir=self.output_dir)
failures = test_runner.run_tests(self.test_labels)
sys.exit(bool(failures))
def setup():
try:
from elasticsearch import Elasticsearch, ElasticsearchException
except ImportError:
raise unittest.SkipTest("elasticsearch-py not installed.")
es = Elasticsearch(settings.HAYSTACK_CONNECTIONS['elasticsearch']['URL'])
try:
es.info()
except ElasticsearchException as e:
raise unittest.SkipTest("elasticsearch not running on %r" % settings.HAYSTACK_CONNECTIONS['elasticsearch']['URL'], e)
global test_runner
global old_config
from django.test.runner import DiscoverRunner
test_runner = DiscoverRunner()
test_runner.setup_test_environment()
old_config = test_runner.setup_databases()
def runtests(*test_args):
if not settings.configured:
settings.configure(**DEFAULT_SETTINGS)
django.setup()
parent = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, parent)
try:
from django.test.runner import DiscoverRunner
runner_class = DiscoverRunner
test_args = ["pinax.cohorts.tests"]
except ImportError:
from django.test.simple import DjangoTestSuiteRunner
runner_class = DjangoTestSuiteRunner
test_args = ["tests"]
failures = runner_class(verbosity=1, interactive=True, failfast=False).run_tests(test_args)
sys.exit(failures)
def mock_django_setup(settings_module, disabled_features=None):
""" Must be called *AT IMPORT TIME* to pretend that Django is set up.
This is useful for running tests without using the Django test runner.
This must be called before any Django models are imported, or they will
complain. Call this from a module in the calling project at import time,
then be sure to import that module at the start of all mock test modules.
Another option is to call it from the test package's init file, so it runs
before all the test modules are imported.
:param settings_module: the module name of the Django settings file,
like 'myapp.settings'
:param disabled_features: a list of strings that should be marked as
*False* on the connection features list. All others will default
to True.
"""
if apps.ready:
# We're running in a real Django unit test, don't do anything.
return
if 'DJANGO_SETTINGS_MODULE' not in os.environ:
os.environ['DJANGO_SETTINGS_MODULE'] = settings_module
django.setup()
mock_django_connection(disabled_features)
def _tests_1_7(self):
"""
Fire up the Django test suite developed for version 1.7 and up
"""
test_settings = self.custom_settings
installed_apps = test_settings.pop('INSTALLED_APPS', ())
settings.configure(
DEBUG=True,
DATABASES=self.get_database(1.7),
MIDDLEWARE_CLASSES=('django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware'),
INSTALLED_APPS=tuple(self.INSTALLED_APPS + installed_apps + self.apps),
**test_settings
)
from django.test.simple import DjangoTestSuiteRunner
import django
django.setup()
failures = DjangoTestSuiteRunner().run_tests(self.apps, verbosity=1)
if failures:
sys.exit(failures)
def _tests_1_8(self):
"""
Fire up the Django test suite developed for version 1.8 and up
"""
test_settings = self.custom_settings
installed_apps = test_settings.pop('INSTALLED_APPS', ())
settings.configure(
DEBUG=True,
DATABASES=self.get_database(1.8),
MIDDLEWARE_CLASSES=('django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware'),
INSTALLED_APPS=tuple(self.INSTALLED_APPS + installed_apps + self.apps),
**test_settings
)
from django.test.runner import DiscoverRunner
import django
django.setup()
failures = DiscoverRunner().run_tests(self.apps, verbosity=1)
if failures:
sys.exit(failures)
def __init__(self, **kwargs):
kwargs['debug'] = kwargs['config'].debug
from {{cookiecutter.project_slug}}.router import create_router
router = create_router()
super().__init__(router=router, **kwargs, middlewares=[jsonify])
self['state'] = collections.Counter()
self.models = self.m = AppModels(self)
aiohttp_jinja2.setup(
self, loader=jinja2.FileSystemLoader(
dirs(self.config.path.templates, base_dir=BASE_DIR)),
)
cls = type(self)
self.on_startup.append(cls.startup_database)
self.on_startup.append(cls.startup_redis)
self.on_cleanup.append(cls.cleanup_database)
self.on_cleanup.append(cls.cleanup_redis)
def initialize_django(settings_py_path):
"""
Initializes the environment required by Django. Must be called **before** importing your models::
import fillmydb
fillmydb.initialize_django("path/to/settings.py")
from mydjangoproject.myapp.models import MyModel, MyOtherModel
...
:param settings_py_path: Path to the ``settings.py`` file from your Django project.
"""
if not django:
raise RuntimeError("Module 'django' could not be imported")
if IS_PY35:
spec = importlib.util.spec_from_file_location("django_settings", settings_py_path)
django_settings = importlib.util.module_from_spec(spec)
spec.loader.exec_module(django_settings)
else:
django_settings = importlib.machinery.SourceFileLoader("django_settings", settings_py_path).load_module()
os.environ["DJANGO_SETTINGS_MODULE"] = "django_settings"
django.setup()
def run_tests(base_dir=None, apps=None, verbosity=1, interavtive=False):
base_dir = base_dir or os.path.dirname(__file__)
os.environ['DJANGO_SETTINGS_MODULE'] = 'settings'
sys.path.insert(0, os.path.join(base_dir, 'src'))
sys.path.insert(0, os.path.join(base_dir, 'tests'))
import django
if django.VERSION >= (1,7):
django.setup()
from django.conf import settings
from django.test.utils import get_runner
TestRunner = get_runner(settings)
test_runner = TestRunner(verbosity=verbosity,
interavtive=interavtive,
failfast=False)
if apps:
app_tests = [x.strip() for x in apps if x]
else:
app_tests = [
'notification'
]
failures = test_runner.run_tests(app_tests)
sys.exit(bool(failures))
def runtests(args=None):
test_dir = os.path.dirname(__file__)
sys.path.insert(0, test_dir)
os.environ['DJANGO_SETTINGS_MODULE'] = 'tests.settings'
import django
from django.test.utils import get_runner
from django.conf import settings
django.setup()
TestRunner = get_runner(settings)
test_runner = TestRunner(verbosity=1, interactive=True)
args = args or ['.']
failures = test_runner.run_tests(args)
sys.exit(failures)
def main():
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "all_settings")
# Load django models. This is needed to populate the DB before using it.
django.setup()
if platform.system() == 'Windows':
try:
import win32file
win32file._setmaxstdio(2048)
except ImportError:
print "Cannot find package 'win32file'. Installing it is "\
"recommended before running the UT " \
"(you can do so using 'pip install pypiwin32')"
from django.core.management import execute_from_command_line
execute_from_command_line(sys.argv)
db_connect_sh_ver_w_threads.py 文件源码
项目:Network-Automation-with-Python-and-Ansible-class
作者: pmusolino-rms
项目源码
文件源码
阅读 24
收藏 0
点赞 0
评论 0
def main():
django.setup()
devices = NetworkDevice.objects.all()
starttime = datetime.now()
for dev in devices:
my_thread = threading.Thread(target=show_version, args=(dev,))
my_thread.start()
main_thread = threading.currentThread()
for thread in threading.enumerate():
if thread != main_thread:
print thread
thread.join()
totaltime = datetime.now() - starttime
print
print "Elapsed time " + str(totaltime)
print
link_credentials.py 文件源码
项目:Network-Automation-with-Python-and-Ansible-class
作者: pmusolino-rms
项目源码
文件源码
阅读 20
收藏 0
点赞 0
评论 0
def main():
"""
Main Process
"""
django.setup()
devices = NetworkDevice.objects.all()
creds = Credentials.objects.all()
std_cred = creds[0]
arista_cred = creds[1]
for dev in devices:
if 'pynet-sw' in dev.device_name:
dev.credentials = arista_cred
else:
dev.credentials = std_cred
dev.save()
for dev in devices:
print dev.device_name, dev.credentials
set_vendor.py 文件源码
项目:Network-Automation-with-Python-and-Ansible-class
作者: pmusolino-rms
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def main():
"""
Main function doing all the heavy lifting
"""
django.setup()
devices = NetworkDevice.objects.all()
for dev in devices:
if 'cisco' in dev.device_type:
dev.vendor = "Cisco"
elif 'juniper' in dev.device_type:
dev.vendor = "Juniper"
elif 'arista' in dev.device_type:
dev.vendor = "Arista"
dev.save()
for dev in devices:
print dev.device_name, dev.vendor
db_connect_sh_ver_w_process.py 文件源码
项目:Network-Automation-with-Python-and-Ansible-class
作者: pmusolino-rms
项目源码
文件源码
阅读 24
收藏 0
点赞 0
评论 0
def main():
django.setup()
devices = NetworkDevice.objects.all()
starttime = datetime.now()
procs = []
for dev in devices:
my_proc = Process(target=show_version, args=(dev,))
my_proc.start()
procs.append(my_proc)
for proc in procs:
print proc
proc.join()
totaltime = datetime.now() - starttime
print
print "Elapsed time " + str(totaltime)
print
def django_tests(verbosity, interactive, failfast, test_labels):
from django.conf import settings
state = setup(verbosity, test_labels)
extra_tests = []
# Run the test suite, including the extra validation tests.
from django.test.utils import get_runner
if not hasattr(settings, 'TEST_RUNNER'):
settings.TEST_RUNNER = 'django.test.simple.DjangoTestSuiteRunner'
TestRunner = get_runner(settings)
test_runner = TestRunner(verbosity=verbosity, interactive=interactive,
failfast=failfast)
failures = test_runner.run_tests(test_labels or get_test_modules(), extra_tests=extra_tests)
teardown(state)
return failures
def runtests(args=None):
test_dir = os.path.dirname(__file__)
sys.path.insert(0, test_dir)
os.environ['DJANGO_SETTINGS_MODULE'] = 'tests.settings'
import django
from django.test.utils import get_runner
from django.conf import settings
django.setup()
TestRunner = get_runner(settings)
test_runner = TestRunner(verbosity=1, interactive=True)
args = args or ['.']
failures = test_runner.run_tests(args)
sys.exit(failures)
def run(self, arguments, *args, **kwargs):
try:
os.environ.setdefault(
'DJANGO_SETTINGS_MODULE', arguments.django_settings)
import django
if hasattr(django, 'setup'):
django.setup()
except ImportError:
raise Py2SwaggerPluginException('Invalid django settings module')
from .introspectors.api import ApiIntrospector
from .urlparser import UrlParser
# Patch Djago REST Framework, to make inspection process slightly easier
from . import injection
apis = UrlParser().get_apis(filter_path=arguments.filter)
a = ApiIntrospector(apis)
return a.inspect()
def main():
"""Main controller"""
init_generic_logging(logfile=join(localstatedir, 'log', LOGFILE),
stderr=False)
django.setup()
config = NetbiosTrackerConfig()
start = time.time()
_logger.info('=== Starting netbiostracker ===')
addresses = tracker.get_addresses_to_scan(config.get_exceptions())
scanresult = tracker.scan(addresses)
parsed_results = tracker.parse(scanresult, config.get_encoding())
tracker.update_database(parsed_results)
_logger.info('Scanned %d addresses, got %d results in %.2f seconds',
len(addresses), len(parsed_results), time.time() - start)
_logger.info('Netbiostracker done')
def runtests(*test_args):
if not settings.configured:
settings.configure(**DEFAULT_SETTINGS)
django.setup()
parent = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, parent)
try:
from django.test.runner import DiscoverRunner
runner_class = DiscoverRunner
test_args = ["pinax.api.tests"]
except ImportError:
from django.test.simple import DjangoTestSuiteRunner
runner_class = DjangoTestSuiteRunner
test_args = ["tests"]
failures = runner_class(verbosity=1, interactive=True, failfast=False).run_tests(test_args)
sys.exit(failures)