Merge pull request #17 from Koed00/dev

Adds management commands to the tests

django_q/management/commands/qcluster.py

@@ -1,5 +1,8 @@
+from optparse import make_option
+
 from django.core.management.base import BaseCommand
 from django.utils.translation import ugettext as _
+
 from django_q.cluster import Cluster
 
 
@@ -7,7 +10,16 @@ class Command(BaseCommand):
     # Translators: help text for qcluster management command
     help = _("Starts a Django Q Cluster.")
 
+    option_list = BaseCommand.option_list + (
+        make_option('--run-once',
+                    action='store_true',
+                    dest='run_once',
+                    default=False,
+                    help='Run once and then stop.'),
+    )
+
     def handle(self, *args, **options):
         q = Cluster()
         q.start()
-
+        if options.get('run_once', False):
+            q.stop()

django_q/management/commands/qmonitor.py

@@ -1,15 +1,23 @@
 # Django
+from optparse import make_option
+
 from django.core.management.base import BaseCommand
 from django.utils.translation import ugettext as _
 
-
 from django_q.monitor import monitor
 
+
 class Command(BaseCommand):
     # Translators: help text for qmonitor management command
     help = _("Monitors Q Cluster activity")
 
+    option_list = BaseCommand.option_list + (
+        make_option('--run-once',
+                    action='store_true',
+                    dest='run_once',
+                    default=False,
+                    help='Run once and then stop.'),
+    )
+
     def handle(self, *args, **options):
-        monitor()
+        monitor(run_once=options.get('run_once', False))
-
-

django_q/tests/test_commands.py

@@ -0,0 +1,11 @@
+import pytest
+from django.core.management import call_command
+
+
+@pytest.mark.django_db
+def test_qcluster():
+    call_command('qcluster', run_once=True)
+
+
+def test_qmonitor():
+    call_command('qmonitor', run_once=True)