[fix] Fixed failures in update_config operation #205

The update_config operation will be executed only when the transaction
is committed to the database.

Also handle rare but possible error conditions.

Fixes #205

Author: NoumbissiValere

openwisp_controller/connection/apps.py

@@ -1,5 +1,6 @@
 from celery.task.control import inspect
 from django.apps import AppConfig
+from django.db import transaction
 from django.db.models.signals import post_save
 from django.utils.translation import ugettext_lazy as _
 from swapper import load_model
@@ -41,7 +42,7 @@ def config_modified_receiver(cls, **kwargs):
         # or update is already in progress, stop here
         if conn_count < 1 or cls._is_update_in_progress(d.id):
             return
-        update_config.delay(d.id)
+        transaction.on_commit(lambda: update_config.delay(d.id))
 
     @classmethod
     def _is_update_in_progress(cls, device_id):

openwisp_controller/connection/tasks.py

@@ -1,9 +1,12 @@
+import logging
 from time import sleep
 
 from celery import shared_task
+from django.core.exceptions import ObjectDoesNotExist
 from swapper import load_model
 
 Device = load_model('config', 'Device')
+logger = logging.getLogger(__name__)
 
 
 @shared_task
@@ -17,7 +20,11 @@ def update_config(device_id):
     sleep(2)
     # avoid repeating the operation multiple times
     device = Device.objects.select_related('config').get(pk=device_id)
-    if device.config.status == 'applied':
+    try:
+        if device.config.status == 'applied':
+            return
+    except ObjectDoesNotExist:
+        logger.warning(f'Config with device id: {device_id} does not exist')
         return
     qs = device.deviceconnection_set.filter(device_id=device_id, enabled=True)
     conn = qs.first()

openwisp_controller/connection/tests/test_models.py

@@ -11,6 +11,7 @@
 
 from .. import settings as app_settings
 from ..signals import is_working_changed
+from ..tasks import update_config
 from .utils import CreateConnectionsMixin
 
 Config = load_model('config', 'Config')
@@ -256,36 +257,6 @@ def _exec_command_return_value(
         stderr_.read().decode('utf8').strip.return_value = stderr
         return (stdin_, stdout_, stderr_)
 
-    @mock.patch(_connect_path)
-    def test_device_config_update(self, mocked_connect):
-        org1 = self._create_org(name='org1')
-        cred = self._create_credentials_with_key(
-            organization=org1, port=self.ssh_server.port
-        )
-        device = self._create_device(organization=org1)
-        update_strategy = app_settings.UPDATE_STRATEGIES[0][0]
-        c = self._create_config(device=device, status='applied')
-        self._create_device_connection(
-            device=device, credentials=cred, update_strategy=update_strategy
-        )
-        c.config = {
-            'interfaces': [
-                {
-                    'name': 'eth10',
-                    'type': 'ethernet',
-                    'addresses': [{'family': 'ipv4', 'proto': 'dhcp'}],
-                }
-            ]
-        }
-        c.full_clean()
-
-        with mock.patch(self._exec_command_path) as mocked:
-            mocked.return_value = self._exec_command_return_value()
-            c.save()
-            mocked.assert_called_once()
-        c.refresh_from_db()
-        self.assertEqual(c.status, 'applied')
-
     @mock.patch(_connect_path)
     def test_ssh_exec_exit_code(self, *args):
         ckey = self._create_credentials_with_key(port=self.ssh_server.port)
@@ -378,3 +349,9 @@ def test_device_connection_set_connector(self):
         self._create_config(device=dev2)
         dc2 = self._create_device_connection(device=dev2)
         self.assertFalse(hasattr(dc2.connector_instance, 'IS_MODIFIED'))
+
+    @mock.patch('logging.Logger.warning')
+    def test_update_config_task_resilient_to_failure(self, mocked):
+        pk = self._create_device().pk
+        update_config.delay(pk)
+        mocked.assert_called_with(f'Config with device id: {pk} does not exist')

openwisp_controller/connection/tests/test_transaction_blocks.py

@@ -0,0 +1,54 @@
+from unittest import mock
+
+from django.test import TransactionTestCase
+
+from .. import settings as app_settings
+from .utils import CreateConnectionsMixin
+
+
+class TestTransactionBlocks(CreateConnectionsMixin, TransactionTestCase):
+    _connect_path = 'paramiko.SSHClient.connect'
+    _exec_command_path = 'paramiko.SSHClient.exec_command'
+
+    def _exec_command_return_value(
+        self, stdin='', stdout='mocked', stderr='', exit_code=0
+    ):
+        stdin_ = mock.Mock()
+        stdout_ = mock.Mock()
+        stderr_ = mock.Mock()
+        stdin_.read().decode('utf8').strip.return_value = stdin
+        stdout_.read().decode('utf8').strip.return_value = stdout
+        stdout_.channel.recv_exit_status.return_value = exit_code
+        stderr_.read().decode('utf8').strip.return_value = stderr
+        return (stdin_, stdout_, stderr_)
+
+    @mock.patch(_connect_path)
+    def test_device_config_update(self, mocked_connect):
+        org1 = self._create_org(name='org1')
+        cred = self._create_credentials_with_key(
+            organization=org1, port=self.ssh_server.port
+        )
+        device = self._create_device(organization=org1)
+        update_strategy = app_settings.UPDATE_STRATEGIES[0][0]
+        c = self._create_config(device=device, status='applied')
+        self._create_device_connection(
+            device=device, credentials=cred, update_strategy=update_strategy
+        )
+        c.config = {
+            'interfaces': [
+                {
+                    'name': 'eth10',
+                    'type': 'ethernet',
+                    'addresses': [{'family': 'ipv4', 'proto': 'dhcp'}],
+                }
+            ]
+        }
+        c.full_clean()
+
+        with mock.patch(self._exec_command_path) as mocked:
+            mocked.return_value = self._exec_command_return_value()
+            c.save()
+            c.refresh_from_db()
+            mocked.assert_called_once()
+        c.refresh_from_db()
+        self.assertEqual(c.status, 'applied')