4 files changed, 93 insertions, 6 deletions

diff --git a/eventmq/router.py b/eventmq/router.py
index d2b442a..56d7ebe 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -322,6 +322,9 @@ class Router(HeartbeatMixin):
         elif client_type == CLIENT_TYPE.scheduler:
             self.add_scheduler(sender)
             self.send_ack(self.frontend, sender, msgid)
+        else:
+            logger.error('Received invalid client type on INFORM ({}), '
+                         'ignoring'.format(client_type))
 
     def on_reply(self, sender, msgid, msg):
         """
diff --git a/eventmq/scheduler.py b/eventmq/scheduler.py
index 09f8db2..3d3f856 100644
--- a/eventmq/scheduler.py
+++ b/eventmq/scheduler.py
@@ -372,6 +372,9 @@ class Scheduler(HeartbeatMixin, EMQPService):
 
         schedule_hash = self.schedule_hash(message)
 
+        # If interval is negative, cron MUST be populated
+        interval_job = interval >= 0
+
         # Notify if this is updating existing, or new
         if (schedule_hash in self.cron_jobs or
                 schedule_hash in self.interval_jobs):
@@ -381,8 +384,7 @@ class Scheduler(HeartbeatMixin, EMQPService):
             logger.debug('Creating a new scheduled job with %s'
                          % schedule_hash)
 
-        # If interval is negative, cron MUST be populated
-        if interval >= 0:
+        if interval_job:
             inter_iter = IntervalIter(monotonic(), interval)
 
             self.interval_jobs[schedule_hash] = [
@@ -430,7 +432,8 @@ class Scheduler(HeartbeatMixin, EMQPService):
             if run_count > 0 or run_count == INFINITE_RUN_COUNT:
                 # Don't allow run_count to decrement below 0
                 if run_count > 0:
-                    self.interval_jobs[schedule_hash][4] -= 1
+                    if interval_job:
+                        self.interval_jobs[schedule_hash][4] -= 1
                 self.send_request(message[3], queue=queue)
 
     def get_run_count_from_headers(self, headers):
diff --git a/eventmq/tests/test_router.py b/eventmq/tests/test_router.py
index 0986f92..a737f89 100644
--- a/eventmq/tests/test_router.py
+++ b/eventmq/tests/test_router.py
@@ -75,9 +75,14 @@ class TestCase(unittest.TestCase):
 
             log_checker.check(
                 ('eventmq.router',
+                 'INFO',
+                 'Received INFORM request from {} (type: worker)'.format(
+                     sender_id)),
+
+                ('eventmq.router',
                  'ERROR',
-                 'Recieved INFORM message with no defined queues. Message '
-                 'was: {}'.format(msg))
+                 "Recieved INFORM message with no defined queues. Message "
+                 "was: {}".format(msg)),
             )
 
     def test_on_inform_invalid_queues(self):
@@ -93,9 +98,14 @@ class TestCase(unittest.TestCase):
 
             log_checker.check(
                 ('eventmq.router',
+                 'INFO',
+                 'Received INFORM request from {} (type: worker)'.format(
+                     sender_id)),
+
+                ('eventmq.router',
                  'ERROR',
                  'Received invalid queue names in INFORM. names:{} from:{} '
-                 'type:{}'.format(msg[0], sender_id, msg[1]))
+                 'type:{}'.format(msg[0], sender_id, msg[1])),
             )
 
     # @mock.patch('eventmq.router.Router.prioritize_queue_list')
diff --git a/eventmq/tests/test_scheduler.py b/eventmq/tests/test_scheduler.py
index ab0b51f..9f21354 100644
--- a/eventmq/tests/test_scheduler.py
+++ b/eventmq/tests/test_scheduler.py
@@ -66,6 +66,77 @@ class TestCase(unittest.TestCase):
         h2 = scheduler.Scheduler.schedule_hash(msg2)
         self.assertEqual('4658982cab9d32bf1ef9113a9d8bdec01775e2bc', h2)
 
+    def test_on_schedule(self):
+        override_settings = {}
+        sched = scheduler.Scheduler(override_settings=override_settings)
+
+        job_msg = json.dumps(['run', {
+            'path': 'test',
+            'args': [33, 'asdf'],
+            'kwargs': {'zeta': 'Z', 'alpha': 'α'},
+            'class_args': [0],
+            'class_kwargs': {
+                'donkey': True, 'apple': False},
+            'callable': 'do_the_thing'}])
+
+        msg = [
+            'default',
+            'run_count:3,guarantee',
+            '3',
+            job_msg,
+            None
+        ]
+
+        cron_msg = [
+            'default',
+            'run_count:3,guarantee',
+            '-1',
+            job_msg,
+            '* * * * *',
+        ]
+
+        sched.on_schedule('fake_msgid', msg)
+        self.assertEqual(1, len(sched.interval_jobs))
+        self.assertEqual(0, len(sched.cron_jobs))
+
+        self.assertEqual(1, len(json.loads(
+            sched.get_scheduled_jobs())['interval_jobs']))
+        self.assertEqual(0,
+                         len(json.loads(
+                             sched.get_scheduled_jobs())['cron_jobs']))
+
+        # Scheduling the same job as a cron should remove it from interval
+        sched.on_schedule('fake_msgid2', cron_msg)
+        self.assertEqual(0, len(sched.interval_jobs))
+        self.assertEqual(1, len(sched.cron_jobs))
+
+        self.assertEqual(0, len(json.loads(
+            sched.get_scheduled_jobs())['interval_jobs']))
+        self.assertEqual(1, len(json.loads(
+            sched.get_scheduled_jobs())['cron_jobs']))
+
+        # Change the job message and it should make new jobs
+        job_msg = json.dumps(['run', {
+            'path': 'test',
+            'args': [333, 'asdf'],
+            'kwargs': {'zeta': 'Z', 'alpha': 'α'},
+            'class_args': [0],
+            'class_kwargs': {
+                'donkey': True, 'apple': False},
+            'callable': 'do_the_thing'}])
+
+        msg[3] = job_msg
+
+        sched.on_schedule('fake_msgid3', msg)
+        self.assertEqual(1, len(sched.interval_jobs))
+        self.assertEqual(1, len(sched.cron_jobs))
+
+        self.assertEqual(1, len(json.loads(
+            sched.get_scheduled_jobs())['interval_jobs']))
+        self.assertEqual(1, len(json.loads(
+            sched.get_scheduled_jobs())['cron_jobs']))
+
+
 # EMQP Tests
     def test_reset(self):
         sched = scheduler.Scheduler()