python3 fixes

circus/plugins/watchdog.py

@@ -136,6 +136,7 @@ def _discover_monitored_pids(self):
         self.pid_status = dict()
         all_watchers = self.call("list")
         for watcher_name in all_watchers['watchers']:
+            # TODO: Do not discover watchdog
             if self._match_watcher_name(watcher_name):
                 processes = self.call("list", name=watcher_name)
                 if 'pids' in processes:
@@ -182,7 +183,12 @@ def _decode_received_udp_message(self, data):
         :return: decoded message
         :rtype: dict or None
         """
-        result = re.match(self.msg_regex, data)
+        """regex is created with string,
+        but bytes are received from the socket,
+        that cause exception.
+        Probably this can be done better,
+        that just decode data"""
+        result = re.match(self.msg_regex, data.decode())
         if result is not None:
             return result.groupdict()
 
@@ -217,6 +223,10 @@ def look_after(self):
 
         max_timeout = self.loop_rate * self.max_count
         too_old_time = time.time() - max_timeout
+        """Instead of del in loop, that will cause exception in Python3,
+        add to list and del from pid_status after loop.
+        """
+        pids_to_del = list()
         for pid, detail in self.pid_status.items():
             if detail['last_activity'] < too_old_time:
                 logger.info("watcher:%s, pid:%s is not responding. Kill it !",
@@ -233,4 +243,6 @@ def look_after(self):
 
                 # Trusting watcher to eventually stop the process after
                 # graceful timeout
-                del self.pid_status[pid]
+                pids_to_del.append(pid)
+        for pid in pids_to_del:
+            del self.pid_status[pid]

circus/tests/test_plugin_watchdog.py

@@ -32,11 +32,26 @@ def run_dummy_watchdogged(test_file):
     return 1
 
 
+class WatchDoggedWithoutSend(Process):
+    def run(self):
+        self._write('STARTWD')
+        for _ in range(5):
+            time.sleep(0.5)
+        self._write('STOPWD')
+
+
+def run_watchdog_without_send(test_file):
+    process = WatchDoggedWithoutSend(test_file)
+    process.run()
+    return 1
+
+
 def get_pid_status(queue, plugin):
     queue.put(plugin.pid_status)
 
 
 fqn = 'circus.tests.test_plugin_watchdog.run_dummy_watchdogged'
+dqn = 'circus.tests.test_plugin_watchdog.run_watchdog_without_send'
 
 
 class TestPluginWatchDog(TestCircus):
@@ -72,4 +87,38 @@ def test_watchdog_discovery_not_found(self):
         self.assertEqual(len(pid_status), 0, pid_status)
         yield self.stop_arbiter()
 
+    @gen_test
+    def test_watchdog_kill_processes(self):
+        yield self.start_arbiter(dqn)
+        async_poll_for(self.test_file, 'START')
+        pubsub = self.arbiter.pubsub_endpoint
+
+        config = {'loop_rate': 0.1}
+        with warnings.catch_warnings():
+            pid_status = yield arp(WatchDog, config,
+                                   get_pid_status,
+                                   endpoint=self.arbiter.endpoint,
+                                   pubsub_endpoint=pubsub,
+                                   duration=1000)
+        self.assertEqual(len(pid_status), 0, pid_status)
+        yield self.stop_arbiter()
+
+    """
+    @gen_test
+    def test_watchdog_handle_hearthbeat(self):
+        yield self.start_arbiter(fqn)
+        async_poll_for(self.test_file, 'START')
+        pubsub = self.arbiter.pubsub_endpoint
+
+        config = {'loop_rate': 0.1}
+        with warnings.catch_warnings():
+            pid_status = yield arp(WatchDog, config,
+                                   get_pid_status,
+                                   endpoint=self.arbiter.endpoint,
+                                   pubsub_endpoint=pubsub,
+                                   duration=600)
+        self.assertEqual(len(pid_status), 1, pid_status)
+        yield self.stop_arbiter()
+    """
+
 test_suite = EasyTestSuite(__name__)