Timeout simulation, watchtest function intercept, fixed #44

app/models.py

@@ -2,6 +2,7 @@ import logging
 import os
 import shutil
 import zipfile
+import requests
 
 from django.contrib.auth.models import User
 from django.contrib.gis.gdal import GDALRaster
@@ -146,7 +147,9 @@ class Task(models.Model):
     pending_action = models.IntegerField(choices=PENDING_ACTIONS, db_index=True, null=True, blank=True, help_text="A requested action to be performed on the task. The selected action will be performed by the scheduler at the next iteration.")
 
     def __str__(self):
-        return 'Task ID: {}'.format(self.id)
+        name = self.name if self.name is not None else "unnamed"
+
+        return 'Task {} ({})'.format(name, self.id)
 
     def save(self, *args, **kwargs):
         # Autovalidate on save
@@ -343,11 +346,10 @@ class Task(models.Model):
                             self.save()
                     except ProcessingException as e:
                         self.set_failure(str(e))
-        except ConnectionRefusedError as e:
-            logger.warning("Task {} cannot communicate with processing node: {}".format(self, str(e)))
-
-            # In the future we might want to retry instead of just failing
-            #self.set_failure(str(e))
+        except (ConnectionRefusedError, ConnectionError) as e:
+            logger.warning("{} cannot communicate with processing node: {}".format(self, str(e)))
+        except requests.exceptions.ConnectTimeout as e:
+            logger.warning("{} timed out with error: {}. We'll try reprocessing at the next tick.".format(self, str(e)))
 
 
     def get_tile_path(self, z, x, y):
@@ -379,7 +381,7 @@ class Task(models.Model):
 
 
     def set_failure(self, error_message):
-        logger.error("{} ERROR: {}".format(self, error_message))
+        logger.error("FAILURE FOR {}: {}".format(self, error_message))
         self.last_error = error_message
         self.status = status_codes.FAILED
         self.save()

app/scheduler.py

@@ -7,6 +7,7 @@ from apscheduler.schedulers import SchedulerAlreadyRunningError, SchedulerNotRun
 from apscheduler.schedulers.background import BackgroundScheduler
 from django import db
 from django.db.models import Q, Count
+from webodm import settings
 
 from app.models import Task, Project
 from nodeodm import status_codes
@@ -51,14 +52,15 @@ def process_pending_tasks():
     def process(task):
         try:
             task.process()
-
+        except Exception as e:
+            logger.error("Uncaught error! This is potentially bad. Please report it to http://github.com/OpenDroneMap/WebODM/issues: {} {}".format(e, traceback.format_exc()))
+            if settings.TESTING: raise e
+        finally:
             # Might have been deleted
             if task.pk is not None:
                 task.processing_lock = False
                 task.save()
-        except Exception as e:
-            logger.error("Uncaught error: {} {}".format(e, traceback.format_exc()))
-        finally:
+
             db.connections.close_all()
 
     if tasks.count() > 0:

app/tests/test_api_task.py

@@ -5,6 +5,8 @@ import time
 import shutil
 
 import logging
+
+import requests
 from django.contrib.auth.models import User
 from rest_framework import status
 from rest_framework.test import APIClient
@@ -304,9 +306,21 @@ class TestApiTask(BootTransactionTestCase):
         self.assertTrue(task.status == status_codes.COMPLETED)
 
 
-        # TODO: timeout issues
+        # Test connection, timeout errors
+        res = client.post("/api/projects/{}/tasks/{}/restart/".format(project.id, task.id))
+        def connTimeout(*args, **kwargs):
+            raise requests.exceptions.ConnectTimeout("Simulated timeout")
+
+        testWatch.intercept("nodeodm.api_client.task_output", connTimeout)
+        scheduler.process_pending_tasks()
+
+        # Timeout errors should be handled by retrying again at a later time
+        # and not fail
+        task.refresh_from_db()
+        self.assertTrue(task.last_error is None)
 
         image1.close()
         image2.close()
+        node_odm.terminate()
 
 

app/tests/test_testwatch.py

@@ -36,5 +36,22 @@ class TestTestWatch(TestCase):
         test2(d)
         self.assertTrue(d['flag'])
 
+        # Test function replacement intercept
+        d = {
+            'a': False,
+            'b': False
+        }
+        @TestWatch.watch(testWatch=tw)
+        def test3(d):
+            d['a'] = True
+
+        def replacement(d):
+            d['b'] = True
+
+        tw.intercept("app.tests.test_testwatch.test3", replacement)
+        test3(d)
+        self.assertFalse(d['a'])
+        self.assertTrue(d['b'])
+
 
 

app/testwatch.py

@@ -17,8 +17,12 @@ class TestWatch:
     def func_to_name(f):
         return "{}.{}".format(f.__module__, f.__name__)
 
-    def intercept(self, fname):
-        self._intercept_list[fname] = True
+    def intercept(self, fname, f = None):
+        self._intercept_list[fname] = f if f is not None else True
+
+    def execute_intercept_function_replacement(self, fname, *args, **kwargs):
+        if fname in self._intercept_list and callable(self._intercept_list[fname]):
+            (self._intercept_list[fname])(*args, **kwargs)
 
     def should_prevent_execution(self, func):
         return TestWatch.func_to_name(func) in self._intercept_list
@@ -51,7 +55,9 @@ class TestWatch:
 
     def hook_pre(self, func, *args, **kwargs):
         if settings.TESTING and self.should_prevent_execution(func):
-            logger.info(func.__name__ + " intercepted")
+            fname = TestWatch.func_to_name(func)
+            logger.info(fname + " intercepted")
+            self.execute_intercept_function_replacement(fname, *args, **kwargs)
             self.log_call(func, *args, **kwargs)
             return True # Intercept
         return False # Do not intercept

nodeodm/api_client.py

@@ -7,6 +7,7 @@ import mimetypes
 import json
 import os
 from urllib.parse import urlunparse
+from app.testwatch import TestWatch
 
 TIMEOUT = 10
 
@@ -30,6 +31,7 @@ class ApiClient:
     def task_info(self, uuid):
         return requests.get(self.url('/task/{}/info').format(uuid), timeout=TIMEOUT).json()
 
+    @TestWatch.watch()
     def task_output(self, uuid, line = 0):
         return requests.get(self.url('/task/{}/output?line={}').format(uuid, line), timeout=TIMEOUT).json()
 

nodeodm/models.py

@@ -91,9 +91,9 @@ class ProcessingNode(models.Model):
         except requests.exceptions.ConnectionError as e:
             raise ProcessingException(e)
 
-        if 'uuid' in result:
+        if isinstance(result, dict) and 'uuid' in result:
             return result['uuid']
-        elif 'error' in result:
+        elif isinstance(result, dict) and 'error' in result:
             raise ProcessingException(result['error'])
         else:
             raise ProcessingException("Unexpected answer from server: {}".format(result))