Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
self._fixture_teardown()
self.worker1 = Worker.objects.create(
worker_key='worker-1',
name='worker-1',
)
self.worker2 = Worker.objects.create(
worker_key='worker-2',
name='worker-2',
)
self.client = django.test.Client()
def test_list_workers_enabled(self):
Worker.objects.create(
worker_key='enabled-worker',
name='enabled-worker',
enabled=True,
)
Worker.objects.create(
worker_key='disabled-worker',
name='disabled-worker',
enabled=False,
)
worker_list = client.list_workers(_make_request(), True)
self.assertEqual(len(worker_list), 1)
self.assertEqual(worker_list[0], 'enabled-worker')
Task.objects.create(
worker=worker1,
arch=self._arch,
channel=self._channel,
owner=self._user,
method='DummyTask',
state=TASK_STATES['OPEN'],
weight=100
)
worker1.save()
worker2 = Worker.objects.create(worker_key='worker-2', name='Worker 2', enabled=True)
workers = Worker.objects.ready()
self.assertEquals(len(workers), 1)
self.assertEquals(workers[0].id, worker2.id)
def test_enabled(self):
Worker.objects.create(worker_key='worker-disabled', name='Worker disabled', enabled=False)
worker = Worker.objects.create(worker_key='worker-enabled', name='Worker enabled', enabled=True)
workers = Worker.objects.enabled()
self.assertEquals(len(workers), 1)
self.assertEquals(workers[0].id, worker.id)
def test_assigned_tasks(self):
worker = Worker.objects.create(worker_key='worker', name='Worker')
Task.objects.create(
worker=worker,
arch=self._arch,
channel=self._channel,
owner=self._user,
method='DummyTask',
state=TASK_STATES['ASSIGNED'],
)
Task.objects.create(
worker=worker,
arch=self._arch,
channel=self._channel,
owner=self._user,
method='DummyTask',
def save(self, *args, **kwargs):
# precompute task count, current load and ready
tasks = Task.objects.opened().filter(worker=self)
self.task_count = tasks.count()
self.current_load = sum(( task.weight for task in tasks if not task.waiting ))
self.ready = self.enabled and (self.current_load < self.max_load and self.task_count < 3*self.max_load)
while not self.worker_key:
# if worker_key is empty, generate a new one
key = random_string(64)
if Worker.objects.filter(worker_key=key).count() == 0:
self.worker_key = key
super(self.__class__, self).save(*args, **kwargs)
from django.conf.urls import url
from django.utils.translation import ugettext_lazy as _
from kobo.django.views.generic import ExtraListView, ExtraDetailView
from kobo.hub.models import Worker
urlpatterns = [
url(r"^$", ExtraListView.as_view(
queryset=Worker.objects.order_by("name"),
template_name="worker/list.html",
context_object_name="worker_list",
title = _("Workers"),
), name="worker/list"),
url(r"^(?P\d+)/$", ExtraDetailView.as_view(
queryset=Worker.objects.select_related(),
template_name="worker/detail.html",
context_object_name="worker",
title=_("Worker detail"),
), name="worker/detail"),
]