[fix] One credential can be added to a device only once #795

Added unique_together constraint in DeviceConnection, device
and credentials should be unique.

Fixes #795

Author: pandafy

openwisp_controller/connection/base/models.py

@@ -254,6 +254,7 @@ class AbstractDeviceConnection(ConnectorMixin, TimeStampedEditableModel):
     class Meta:
         verbose_name = _('Device connection')
         verbose_name_plural = _('Device connections')
+        unique_together = (('device', 'credentials'),)
         abstract = True
 
     def __init__(self, *args, **kwargs):

openwisp_controller/connection/migrations/0008_alter_deviceconnection_unique_together.py

@@ -0,0 +1,19 @@
+# Generated by Django 3.2.20 on 2023-08-24 12:35
+
+from django.conf import settings
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        migrations.swappable_dependency(settings.CONFIG_DEVICE_MODEL),
+        ('connection', '0007_command'),
+    ]
+
+    operations = [
+        migrations.AlterUniqueTogether(
+            name='deviceconnection',
+            unique_together={('device', 'credentials')},
+        ),
+    ]

openwisp_controller/connection/tests/test_api.py

@@ -17,6 +17,7 @@
 from .utils import CreateCommandMixin, CreateConnectionsMixin
 
 Command = load_model('connection', 'Command')
+DeviceConnection = load_model('connection', 'DeviceConnection')
 command_qs = Command.objects.order_by('-created')
 OrganizationUser = load_model('openwisp_users', 'OrganizationUser')
 Group = load_model('openwisp_users', 'Group')
@@ -423,7 +424,7 @@ def test_post_deviceconnection_list(self):
             'enabled': True,
             'failure_reason': '',
         }
-        with self.assertNumQueries(11):
+        with self.assertNumQueries(12):
             response = self.client.post(path, data, content_type='application/json')
         self.assertEqual(response.status_code, 201)
 
@@ -436,7 +437,7 @@ def test_post_deviceconenction_with_no_config_device(self):
             'enabled': True,
             'failure_reason': '',
         }
-        with self.assertNumQueries(11):
+        with self.assertNumQueries(12):
             response = self.client.post(path, data, content_type='application/json')
         error_msg = '''
             the update strategy can be determined automatically only if
@@ -468,7 +469,7 @@ def test_put_devceconnection_detail(self):
             'enabled': False,
             'failure_reason': '',
         }
-        with self.assertNumQueries(13):
+        with self.assertNumQueries(14):
             response = self.client.put(path, data, content_type='application/json')
         self.assertEqual(response.status_code, 200)
         self.assertEqual(
@@ -482,7 +483,7 @@ def test_patch_deviceconnectoin_detail(self):
         path = reverse('connection_api:deviceconnection_detail', args=(d1, dc.pk))
         self.assertEqual(dc.update_strategy, app_settings.UPDATE_STRATEGIES[0][0])
         data = {'update_strategy': app_settings.UPDATE_STRATEGIES[1][0]}
-        with self.assertNumQueries(12):
+        with self.assertNumQueries(13):
             response = self.client.patch(path, data, content_type='application/json')
         self.assertEqual(response.status_code, 200)
         self.assertEqual(
@@ -501,8 +502,8 @@ def test_bearer_authentication(self):
         self.client.logout()
         token = self._obtain_auth_token(username='admin', password='tester')
         credentials = self._create_credentials(auto_add=True)
-        device_conn = self._create_device_connection(credentials=credentials)
-        device = device_conn.device
+        device = self._create_config(organization=credentials.organization).device
+        device_conn = device.deviceconnection_set.first()
 
         with self.subTest('Test CredentialListCreateView'):
             response = self.client.get(

openwisp_controller/connection/tests/test_models.py

@@ -328,6 +328,18 @@ def test_device_connection_credential_org_validation(self):
         else:
             self.fail('ValidationError not raised')
 
+    def test_device_connection_same_credential_twice(self):
+        device_conn = self._create_device_connection()
+        with self.assertRaises(ValidationError) as context_manager:
+            device_conn = DeviceConnection(
+                device=device_conn.device, credentials=device_conn.credentials
+            )
+            device_conn.full_clean()
+        self.assertEqual(
+            context_manager.exception.message_dict['__all__'][0],
+            'Device connection with this Device and Credentials already exists.',
+        )
+
     def test_auto_add_to_new_device(self):
         c = self._create_credentials(auto_add=True, organization=None)
         self._create_credentials(name='cred2', auto_add=False, organization=None)
@@ -1049,12 +1061,12 @@ def test_chunk_size(self):
                 organization=org, name='device3', mac_address='33:33:33:33:33:33'
             )
         )
-        with self.assertNumQueries(28):
+        with self.assertNumQueries(31):
             credential = self._create_credentials(auto_add=True, organization=org)
         self.assertEqual(credential.deviceconnection_set.count(), 3)
 
         with mock.patch.object(Credentials, 'chunk_size', 2):
-            with self.assertNumQueries(30):
+            with self.assertNumQueries(33):
                 credential = self._create_credentials(
                     name='Mocked Credential', auto_add=True, organization=org
                 )

tests/openwisp2/sample_connection/migrations/0001_initial.py

@@ -195,6 +195,7 @@ class Migration(migrations.Migration):
             options={
                 'verbose_name': 'Device connection',
                 'verbose_name_plural': 'Device connections',
+                'unique_together': {('device', 'credentials')},
                 'abstract': False,
             },
             bases=(