[tests] Optimize unit test runtime speed #69 (#134)

Co-authored-by: Federico Capoano <[email protected]>

.travis.yml

@@ -45,6 +45,8 @@ install:
   - pip install --upgrade https://github.com/openwisp/django-netjsonconfig/tarball/master
   # temporary: remove when openwisp-notifications is released
   - pip install -U https://github.com/openwisp/openwisp-notifications/tarball/master
+  # To be removed after new release of openwisp-utils
+  - pip install --upgrade https://github.com/openwisp/openwisp-utils/tarball/master
 
 script:
   - ./run-qa-checks

openwisp_monitoring/check/tests/__init__.py

@@ -0,0 +1,12 @@
+_FPING_REACHABLE = (
+    '',
+    bytes(
+        '10.40.0.1 : xmt/rcv/%loss = 5/5/0%, min/avg/max = 0.04/0.08/0.15',
+        encoding='utf8',
+    ),
+)
+
+_FPING_UNREACHABLE = (
+    '',
+    bytes('192.168.255.255 : xmt/rcv/%loss = 3/0/100%', encoding='utf8'),
+)

openwisp_monitoring/check/tests/test_ping.py

@@ -9,6 +9,7 @@
 from .. import settings
 from ..classes import Ping
 from ..exceptions import OperationalError
+from . import _FPING_REACHABLE, _FPING_UNREACHABLE
 
 Chart = load_model('monitoring', 'Chart')
 AlertSettings = load_model('monitoring', 'AlertSettings')
@@ -24,15 +25,9 @@ class TestPing(TestDeviceMonitoringMixin, TransactionTestCase):
         '',
         bytes("fping: option requires an argument -- 'z'", encoding='utf8'),
     )
-    _FPING_OUTPUT = (
-        '',
-        bytes(
-            '10.40.0.1 : xmt/rcv/%loss = 5/5/0%, ' 'min/avg/max = 0.04/0.08/0.15',
-            'utf8',
-        ),
-    )
 
-    def test_check_ping_no_params(self):
+    @patch.object(Ping, '_command', return_value=_FPING_REACHABLE)
+    def test_check_ping_no_params(self, mocked_method):
         device = self._create_device(organization=self._create_org())
         # will ping localhost
         device.management_ip = '127.0.0.1'
@@ -65,9 +60,9 @@ def test_check_ping_params(self):
         for key in self._RTT_KEYS:
             self.assertTrue(result[key] < 1)
 
-    def test_check_ping_unreachable(self):
+    @patch.object(Ping, '_command', return_value=_FPING_UNREACHABLE)
+    def test_check_ping_unreachable(self, mocked_method):
         device = self._create_device(organization=self._create_org())
-        # will hopefully ping an unexisting private address
         device.management_ip = '192.168.255.255'
         check = Check(
             name='Ping check',
@@ -98,7 +93,8 @@ def test_operational_error(self, _command):
         else:
             self.fail('OperationalError not raised')
 
-    def _check_no_ip_case(self, status, management_ip_only=False):
+    @patch.object(Ping, '_command', return_value=_FPING_REACHABLE)
+    def _check_no_ip_case(self, status, mocked_method, management_ip_only=False):
         device = self._create_device(
             organization=self._create_org(), last_ip='127.0.0.1'
         )
@@ -124,24 +120,29 @@ def _check_no_ip_case(self, status, management_ip_only=False):
         self.assertEqual(device.monitoring.status, expected_status)
         self.assertEqual(Metric.objects.count(), expected_metrics_count)
 
+    @patch.object(Ping, '_command', return_value=_FPING_REACHABLE)
     @patch('openwisp_monitoring.check.settings.MANAGEMENT_IP_ONLY', True)
-    def test_device_without_ip_unknown_status(self):
+    def test_device_without_ip_unknown_status(self, mocked_method):
         self._check_no_ip_case('unknown')
 
+    @patch.object(Ping, '_command', return_value=_FPING_REACHABLE)
     @patch('openwisp_monitoring.check.settings.MANAGEMENT_IP_ONLY', True)
-    def test_device_without_ip_ok_status(self):
+    def test_device_without_ip_ok_status(self, mocked_method):
         self._check_no_ip_case('ok')
 
+    @patch.object(Ping, '_command', return_value=_FPING_REACHABLE)
     @patch('openwisp_monitoring.check.settings.MANAGEMENT_IP_ONLY', True)
-    def test_device_without_ip_problem_status(self):
+    def test_device_without_ip_problem_status(self, mocked_method):
         self._check_no_ip_case('problem')
 
+    @patch.object(Ping, '_command', return_value=_FPING_REACHABLE)
     @patch('openwisp_monitoring.check.settings.MANAGEMENT_IP_ONLY', True)
-    def test_device_without_ip_critical_status(self):
+    def test_device_without_ip_critical_status(self, mocked_method):
         self._check_no_ip_case('critical')
 
+    @patch.object(Ping, '_command', return_value=_FPING_REACHABLE)
     @patch('openwisp_monitoring.check.settings.MANAGEMENT_IP_ONLY', False)
-    def test_device_with_last_ip_unknown_status(self):
+    def test_device_with_last_ip_unknown_status(self, mocked_method):
         self._check_no_ip_case('unknown', management_ip_only=True)
 
     def test_content_object_none(self):
@@ -193,7 +194,7 @@ def test_schema_violation(self):
             else:
                 self.fail('ValidationError not raised')
 
-    @patch.object(Ping, '_command', return_value=_FPING_OUTPUT)
+    @patch.object(Ping, '_command', return_value=_FPING_REACHABLE)
     def test_store_result(self, mocked_method):
         self.assertEqual(Check.objects.count(), 0)
         device = self._create_device(organization=self._create_org())
@@ -220,7 +221,7 @@ def test_store_result(self, mocked_method):
         self.assertEqual(points[0]['rtt_avg'], result['rtt_avg'])
         self.assertEqual(points[0]['rtt_max'], result['rtt_max'])
 
-    @patch.object(Ping, '_command', return_value=_FPING_OUTPUT)
+    @patch.object(Ping, '_command', return_value=_FPING_REACHABLE)
     @patch.object(monitoring_settings, 'AUTO_CHARTS', return_value=[])
     def test_auto_chart_disabled(self, *args):
         device = self._create_device(organization=self._create_org())

openwisp_monitoring/check/tests/test_utils.py

@@ -7,30 +7,24 @@
 from ..classes import Ping
 from ..settings import CHECK_CLASSES
 from ..utils import run_checks_async
+from . import _FPING_REACHABLE
 
 
 class TestUtils(TestDeviceMonitoringMixin, TransactionTestCase):
     _PING = CHECK_CLASSES[0][0]
-    _FPING_OUTPUT = (
-        '',
-        bytes(
-            '10.40.0.1 : xmt/rcv/%loss = 5/5/0%, ' 'min/avg/max = 0.04/0.08/0.15',
-            'utf8',
-        ),
-    )
 
     def _create_check(self):
         device = self._create_device(organization=self._create_org())
         device.last_ip = '10.40.0.1'
         device.save()
         # check is automatically created via django signal
 
-    @patch.object(Ping, '_command', return_value=_FPING_OUTPUT)
+    @patch.object(Ping, '_command', return_value=_FPING_REACHABLE)
     def test_run_checks_async_success(self, mocked_method):
         self._create_check()
         run_checks_async()
 
-    @patch.object(Ping, '_command', return_value=_FPING_OUTPUT)
+    @patch.object(Ping, '_command', return_value=_FPING_REACHABLE)
     def test_management_command(self, mocked_method):
         self._create_check()
         management.call_command('run_checks')

openwisp_monitoring/device/tests/__init__.py

@@ -1,4 +1,5 @@
 import json
+from copy import deepcopy
 
 from django.test import TestCase
 from django.urls import reverse
@@ -37,16 +38,24 @@ def _post_data(self, id, key, data):
         netjson = json.dumps(data)
         return self.client.post(url, netjson, content_type='application/json')
 
-    def create_test_adata(self):
+    def create_test_adata(self, no_resources=False):
         o = self._create_org()
         d = self._create_device(organization=o)
         data = self._data()
+        # creation of resources metrics can be avoided in tests not involving them
+        # this speeds up those tests by reducing requests made
+        if no_resources:
+            del data['resources']
         r = self._post_data(d.id, d.key, data)
         self.assertEqual(r.status_code, 200)
         dd = DeviceData(pk=d.pk)
         self.assertDictEqual(dd.data, data)
-        self.assertEqual(Metric.objects.count(), 9)
-        self.assertEqual(Chart.objects.count(), 7)
+        if no_resources:
+            metric_count, chart_count = 6, 4
+        else:
+            metric_count, chart_count = 9, 7
+        self.assertEqual(Metric.objects.count(), metric_count)
+        self.assertEqual(Chart.objects.count(), chart_count)
         if_dict = {'wlan0': data['interfaces'][0], 'wlan1': data['interfaces'][1]}
         for ifname in ['wlan0', 'wlan1']:
             iface = if_dict[ifname]
@@ -62,32 +71,42 @@ def create_test_adata(self):
             m = Metric.objects.get(key=ifname, field_name='clients', object_id=d.pk)
             points = m.read(limit=10, order='time DESC')
             self.assertEqual(len(points), len(iface['wireless']['clients']))
+        return dd
 
-    def _create_multiple_measurements(self, create=True):
+    def _create_multiple_measurements(self, create=True, no_resources=False, count=4):
         if create:
-            self.create_test_adata()
+            self.create_test_adata(no_resources=no_resources)
         self.assertEqual(self.device_model.objects.count(), 1)
         d = self.device_model.objects.first()
-        data2 = self._data()
+        dd = DeviceData(pk=d.pk)
+        data = self._data()
+        # creation of resources metrics can be avoided in tests not involving them
+        # this speeds up those tests by reducing requests made
+        if no_resources:
+            del data['resources']
+        data2 = deepcopy(data)
         data2['interfaces'][0]['statistics']['rx_bytes'] = 400000000
         data2['interfaces'][0]['statistics']['tx_bytes'] = 100000000
         data2['interfaces'][1]['statistics']['rx_bytes'] = 2000000000
         data2['interfaces'][1]['statistics']['tx_bytes'] = 1000000000
         r = self._post_data(d.id, d.key, data2)
-        data3 = self._data()
+        if count == 2:
+            return dd
+        data3 = deepcopy(data)
         data3['interfaces'][0]['statistics']['rx_bytes'] = 500000000
         data3['interfaces'][0]['statistics']['tx_bytes'] = 300000000
         data3['interfaces'][1]['statistics']['rx_bytes'] = 0
         data3['interfaces'][1]['statistics']['tx_bytes'] = 0
         r = self._post_data(d.id, d.key, data3)
-        data4 = self._data()
+        if count == 3:
+            return dd
+        data4 = deepcopy(data)
         data4['interfaces'][0]['statistics']['rx_bytes'] = 1200000000
         data4['interfaces'][0]['statistics']['tx_bytes'] = 600000000
         data4['interfaces'][1]['statistics']['rx_bytes'] = 1000000000
         data4['interfaces'][1]['statistics']['tx_bytes'] = 500000000
         r = self._post_data(d.id, d.key, data4)
         self.assertEqual(r.status_code, 200)
-        dd = DeviceData(pk=d.pk)
         self.assertDictEqual(dd.data, data4)
         return dd
 

openwisp_monitoring/device/tests/test_admin.py

@@ -20,7 +20,7 @@ def _login_admin(self):
         self.client.force_login(u)
 
     def test_device_admin(self):
-        dd = self._create_multiple_measurements()
+        dd = self.create_test_adata()
         url = reverse('admin:config_device_change', args=[dd.pk])
         self._login_admin()
         r = self.client.get(url)
@@ -51,17 +51,14 @@ def test_remove_invalid_interface(self):
 
     def test_wifi_clients_admin(self):
         self._login_admin()
-        d = self._create_device(organization=self._create_org())
-        data = self._data()
-        r = self._post_data(d.id, d.key, data)
-        url = reverse('admin:config_device_change', args=[d.id])
+        dd = self.create_test_adata(no_resources=True)
+        url = reverse('admin:config_device_change', args=[dd.id])
         r1 = self.client.get(url, follow=True)
-        self.assertEqual(r.status_code, 200)
         self.assertEqual(r1.status_code, 200)
         self.assertContains(r1, '00:ee:ad:34:f5:3b')
 
     def test_uuid_bug(self):
-        dd = self._create_multiple_measurements()
+        dd = self.create_test_adata(no_resources=True)
         uuid = str(dd.pk).replace('-', '')
         url = reverse('admin:config_device_change', args=[uuid])
         self._login_admin()