[tests] Added tests for WHOIS creation #1034

Added tests for checking if `fetch_whois_details` task
is called properly or not, and for checking creation
of WHOIS record for a device if the last_ip recorded
is public.
As now whois record is also fetched whenever fetching
a device, there is an increase in query count in some
of the tests.

Closes #1034

Signed-off-by: Aman Jagotra <[email protected]>

Author: AmanJagotra

openwisp_controller/config/base/device.py

@@ -308,7 +308,8 @@ def _check_changed_fields(self):
         self._get_initial_values_for_checked_fields()
         # Execute method for checked for each field in self._changed_checked_fields
         for field in self._changed_checked_fields:
-            getattr(self, f'_check_{field}_changed')()
+            if hasattr(self, f'_check_{field}_changed'):
+                getattr(self, f'_check_{field}_changed')()
 
     def _is_deferred(self, field):
         """
@@ -377,16 +378,20 @@ def trigger_whois_lookup(self):
         """Trigger WHOIS lookup if the last IP has changed and is public IP."""
         from ipaddress import ip_address
 
-        from .. import tasks
+        from ..tasks import fetch_whois_details
 
         if self._initial_last_ip == models.DEFERRED:
             return
         # Trigger fetch WHOIS lookup if it does not exist
         # or if the last IP has changed and is a public IP
         if (
-            not hasattr(self, 'whoisinfo') or self.last_ip != self._initial_last_ip
-        ) and ip_address(self.last_ip).is_global:
-            tasks.fetch_whois_details.delay(self.pk, self.last_ip)
+            (not hasattr(self, 'whois_info') or self.last_ip != self._initial_last_ip)
+            and self.last_ip
+            and ip_address(self.last_ip).is_global
+        ):
+            transaction.on_commit(
+                lambda: fetch_whois_details.delay(device_pk=self.pk, ip=self.last_ip)
+            )
 
         self._initial_last_ip = self.last_ip
 

openwisp_controller/config/tests/test_controller.py

@@ -195,7 +195,7 @@ def test_get_cached_checksum(self):
                     self.assertEqual(mocked_view_debug.call_count, 1)
 
         with self.subTest('update_last_ip updates the cache'):
-            with self.assertNumQueries(3):
+            with self.assertNumQueries(4):
                 with patch.object(
                     controller_views_logger, 'debug'
                 ) as mocked_view_debug:
@@ -1314,12 +1314,12 @@ def test_ip_fields_not_duplicated(self):
         c2 = self._create_config(device=d2)
         org2 = self._create_org(name='org2', shared_secret='123456')
         c3 = self._create_config(organization=org2)
-        with self.assertNumQueries(7):
+        with self.assertNumQueries(8):
             self.client.get(
                 reverse('controller:device_checksum', args=[c3.device.pk]),
                 {'key': c3.device.key, 'management_ip': '192.168.1.99'},
             )
-        with self.assertNumQueries(7):
+        with self.assertNumQueries(8):
             self.client.get(
                 reverse('controller:device_checksum', args=[c1.device.pk]),
                 {'key': c1.device.key, 'management_ip': '192.168.1.99'},
@@ -1332,7 +1332,7 @@ def test_ip_fields_not_duplicated(self):
             )
         # triggers more queries because devices with conflicting addresses
         # need to be updated, luckily it does not happen often
-        with self.assertNumQueries(9):
+        with self.assertNumQueries(11):
             self.client.get(
                 reverse('controller:device_checksum', args=[c2.device.pk]),
                 {'key': c2.device.key, 'management_ip': '192.168.1.99'},
@@ -1361,14 +1361,14 @@ def test_organization_shares_management_ip_address_space(self):
         org1_config = self._create_config(organization=org1)
         org2 = self._create_org(name='org2', shared_secret='org2')
         org2_config = self._create_config(organization=org2)
-        with self.assertNumQueries(7):
+        with self.assertNumQueries(8):
             self.client.get(
                 reverse('controller:device_checksum', args=[org1_config.device_id]),
                 {'key': org1_config.device.key, 'management_ip': '192.168.1.99'},
             )
         # Device from another organization sends conflicting management IP
         # Extra queries due to conflict resolution
-        with self.assertNumQueries(9):
+        with self.assertNumQueries(11):
             self.client.get(
                 reverse('controller:device_checksum', args=[org2_config.device_id]),
                 {'key': org2_config.device.key, 'management_ip': '192.168.1.99'},

openwisp_controller/config/tests/test_device.py

@@ -524,6 +524,8 @@ def test_device_field_changed_checks(self):
             device.management_ip = '10.0.0.1'
             device.group_id = device_group.id
             device.organization_id = self._create_org().id
+            # assigning a random ip to last_ip
+            device.last_ip = '172.217.22.14'
             # Another query is generated due to "config.set_status_modified"
             # on name change
             with self.assertNumQueries(3):

openwisp_controller/config/tests/test_whois.py

@@ -0,0 +1,64 @@
+from unittest import mock
+
+from django.test import TransactionTestCase
+from swapper import load_model
+
+from ..tasks import fetch_whois_details
+from .utils import CreateDeviceMixin
+
+Device = load_model('config', 'Device')
+WHOISInfo = load_model('config', 'WHOISInfo')
+
+
+class TestWHOISInfo(CreateDeviceMixin, TransactionTestCase):
+    @mock.patch.object(fetch_whois_details, 'delay')
+    def test_task_called(self, mocked_task):
+        with self.subTest('task called when last_ip is public'):
+            device = self._create_device(last_ip='172.217.22.14')
+            mocked_task.assert_called()
+        mocked_task.reset_mock()
+
+        with self.subTest('task not called when last_ip is private'):
+            device.last_ip = '10.0.0.1'
+            device.save()
+            mocked_task.assert_not_called()
+        mocked_task.reset_mock()
+
+        with self.subTest('task not called when last_ip is not changed'):
+            device.last_ip = '10.0.0.1'
+            device.save()
+            mocked_task.assert_not_called()
+        mocked_task.reset_mock()
+
+    def test_whois_info_creation(self):
+        # mocking the geoip2 client to return a mock response
+        with mock.patch(
+            'openwisp_controller.config.tasks.geoip2.webservice.Client'
+        ) as mock_client:
+            # mocking the response from the geoip2 client
+            mock_response = mock.MagicMock()
+            mock_response.city.name = 'Mountain View'
+            mock_response.country.name = 'United States'
+            mock_response.continent.name = 'North America'
+            mock_response.postal.code = '94043'
+            mock_response.traits.autonomous_system_organization = 'Google LLC'
+            mock_response.traits.autonomous_system_number = 15169
+            mock_response.traits.network = '172.217.22.0/24'
+            mock_response.location.time_zone = 'America/Los_Angeles'
+            mock_client.return_value.city.return_value = mock_response
+
+            # creating a device with a last public IP
+            device = self._create_device(last_ip='172.217.22.14')
+            device.refresh_from_db()
+
+            # fetching the WHOIS info for the device
+            whois_info = device.whois_info
+            self.assertEqual(whois_info.organization_name, 'Google LLC')
+            self.assertEqual(whois_info.asn, '15169')
+            self.assertEqual(whois_info.country, 'United States')
+            self.assertEqual(whois_info.timezone, 'America/Los_Angeles')
+            self.assertEqual(
+                whois_info.address, 'Mountain View, United States, North America, 94043'
+            )
+            self.assertEqual(whois_info.cidr, '172.217.22.0/24')
+            self.assertEqual(whois_info.last_public_ip, '172.217.22.14')