diff --git a/openwisp_firmware_upgrader/tests/test_admin.py b/openwisp_firmware_upgrader/tests/test_admin.py index 1ce34c8ca..79e551a04 100644 --- a/openwisp_firmware_upgrader/tests/test_admin.py +++ b/openwisp_firmware_upgrader/tests/test_admin.py @@ -652,438 +652,373 @@ def test_batch_upgrade_operation_filters(self, *args): ) -_mock_upgrade = "openwisp_firmware_upgrader.upgraders.openwrt.OpenWrt.upgrade" -_mock_connect = "openwisp_controller.connection.models.DeviceConnection.connect" - - -@mock.patch(_mock_upgrade, return_value=True) -@mock.patch(_mock_connect, return_value=True) class TestAdminTransaction( BaseTestAdmin, AdminActionPermTestMixin, TransactionTestCase ): - def test_upgrade_selected_action_perms(self, *args): - env = self._create_upgrade_env() - org = env["d1"].organization - self._create_firmwareless_device(organization=org) - user = self._create_user(is_staff=True) - self._create_org_user(user=user, organization=org, is_admin=True) - # The user is redirected to the BatchUpgradeOperation page after success operation. - # Thus, we need to add the permission to the user. - user.user_permissions.add( - Permission.objects.get( - codename=f"change_{BatchUpgradeOperation._meta.model_name}" - ) - ) - self._test_action_permission( - path=self.build_list_url, - action="upgrade_selected", - user=user, - obj=env["build1"], - message=( - "You can track the progress of this mass upgrade operation " - "in this page." - ), - required_perms=["change"], - extra_payload={ - "upgrade_all": "upgrade_all", - "upgrade_options": '{"c": true}', - }, - ) + _mock_upgrade = "openwisp_firmware_upgrader.upgraders.openwrt.OpenWrt.upgrade" + _mock_connect = "openwisp_controller.connection.models.DeviceConnection.connect" - def test_upgrade_related(self, *args): - self._login() - env = self._create_upgrade_env() - self._create_firmwareless_device(organization=env["d1"].organization) - # check state is good before proceeding - fw = DeviceFirmware.objects.filter( - image__build_id=env["build2"].pk - ).select_related("image") - self.assertEqual(Device.objects.count(), 3) - self.assertEqual(UpgradeOperation.objects.count(), 0) - self.assertEqual(fw.count(), 0) - - with self.subTest("Invalid upgrade_options"): - response = self.client.post( - self.build_list_url, - { - "action": "upgrade_selected", - "upgrade_related": "upgrade_related", - "upgrade_options": "invalid", - ACTION_CHECKBOX_NAME: (env["build2"].pk,), - }, - follow=True, - ) - id_attr = ( - ' id="id_upgrade_options_error"' if django.VERSION >= (5, 2) else "" - ) - self.assertContains( - response, - f'', + @mock.patch(_mock_upgrade, return_value=True) + def test_upgrade_selected_action_perms(self, *args): + with mock.patch(self._mock_connect, return_value=True): + env = self._create_upgrade_env() + org = env["d1"].organization + self._create_firmwareless_device(organization=org) + user = self._create_user(is_staff=True) + self._create_org_user(user=user, organization=org, is_admin=True) + # The user is redirected to the BatchUpgradeOperation page after success operation. + # Thus, we need to add the permission to the user. + user.user_permissions.add( + Permission.objects.get( + codename=f"change_{BatchUpgradeOperation._meta.model_name}" + ) ) - - with self.subTest("Test with valid upgrade_options"): - r = self.client.post( - self.build_list_url, - { - "action": "upgrade_selected", - "upgrade_related": "upgrade_related", + self._test_action_permission( + path=self.build_list_url, + action="upgrade_selected", + user=user, + obj=env["build1"], + message=( + "You can track the progress of this mass upgrade operation " + "in this page." + ), + required_perms=["change"], + extra_payload={ + "upgrade_all": "upgrade_all", "upgrade_options": '{"c": true}', - ACTION_CHECKBOX_NAME: (env["build2"].pk,), }, - follow=True, - ) - self.assertContains(r, '
  • ') - self.assertContains(r, "track the progress") - self.assertEqual( - UpgradeOperation.objects.filter(upgrade_options={"c": True}).count(), 2 ) - self.assertEqual(fw.count(), 2) + @mock.patch(_mock_upgrade, return_value=True) + def test_upgrade_related(self, *args): + with mock.patch(self._mock_connect, return_value=True): + self._login() + env = self._create_upgrade_env() + self._create_firmwareless_device(organization=env["d1"].organization) + # check state is good before proceeding + fw = DeviceFirmware.objects.filter( + image__build_id=env["build2"].pk + ).select_related("image") + self.assertEqual(Device.objects.count(), 3) + self.assertEqual(UpgradeOperation.objects.count(), 0) + self.assertEqual(fw.count(), 0) + + with self.subTest("Invalid upgrade_options"): + response = self.client.post( + self.build_list_url, + { + "action": "upgrade_selected", + "upgrade_related": "upgrade_related", + "upgrade_options": "invalid", + ACTION_CHECKBOX_NAME: (env["build2"].pk,), + }, + follow=True, + ) + id_attr = ( + ' id="id_upgrade_options_error"' if django.VERSION >= (5, 2) else "" + ) + self.assertContains( + response, + f'', + ) + + with self.subTest("Test with valid upgrade_options"): + r = self.client.post( + self.build_list_url, + { + "action": "upgrade_selected", + "upgrade_related": "upgrade_related", + "upgrade_options": '{"c": true}', + ACTION_CHECKBOX_NAME: (env["build2"].pk,), + }, + follow=True, + ) + self.assertContains(r, '
  • ') + self.assertContains(r, "track the progress") + self.assertEqual( + UpgradeOperation.objects.filter( + upgrade_options={"c": True} + ).count(), + 2, + ) + self.assertEqual(fw.count(), 2) + + @mock.patch(_mock_upgrade, return_value=True) def test_upgrade_all(self, *args): - self._login() - env = self._create_upgrade_env() - self._create_firmwareless_device(organization=env["d1"].organization) - # check state is good before proceeding - fw = DeviceFirmware.objects.filter( - image__build_id=env["build2"].pk - ).select_related("image") - self.assertEqual(Device.objects.count(), 3) - self.assertEqual(UpgradeOperation.objects.count(), 0) - self.assertEqual(fw.count(), 0) - - with self.subTest("Invalid upgrade_options"): - response = self.client.post( - self.build_list_url, - { - "action": "upgrade_selected", - "upgrade_all": "upgrade_all", - "upgrade_options": "invalid", - ACTION_CHECKBOX_NAME: (env["build2"].pk,), - }, - follow=True, - ) - self.assertEqual(response.status_code, 200) - id_attr = ( - ' id="id_upgrade_options_error"' if django.VERSION >= (5, 2) else "" + with mock.patch(self._mock_connect, return_value=True): + self._login() + env = self._create_upgrade_env() + self._create_firmwareless_device(organization=env["d1"].organization) + # check state is good before proceeding + fw = DeviceFirmware.objects.filter( + image__build_id=env["build2"].pk + ).select_related("image") + self.assertEqual(Device.objects.count(), 3) + self.assertEqual(UpgradeOperation.objects.count(), 0) + self.assertEqual(fw.count(), 0) + + with self.subTest("Invalid upgrade_options"): + response = self.client.post( + self.build_list_url, + { + "action": "upgrade_selected", + "upgrade_all": "upgrade_all", + "upgrade_options": "invalid", + ACTION_CHECKBOX_NAME: (env["build2"].pk,), + }, + follow=True, + ) + self.assertEqual(response.status_code, 200) + id_attr = ( + ' id="id_upgrade_options_error"' if django.VERSION >= (5, 2) else "" + ) + self.assertContains( + response, + f'', + ) + + with self.subTest("Test with valid upgrade_options"): + response = self.client.post( + self.build_list_url, + { + "action": "upgrade_selected", + "upgrade_all": "upgrade_all", + "upgrade_options": '{"c": true}', + ACTION_CHECKBOX_NAME: (env["build2"].pk,), + }, + follow=True, + ) + self.assertContains(response, '
  • ') + self.assertContains(response, "track the progress") + self.assertEqual( + UpgradeOperation.objects.filter( + upgrade_options={"c": True} + ).count(), + 3, + ) + self.assertEqual(fw.count(), 3) + self.assertContains( + response, + ( + '
    " + ), + html=True, + ) + + @mock.patch(_mock_upgrade, return_value=True) + def test_mass_upgrade_shared_image(self, *args): + with mock.patch(self._mock_connect, return_value=True): + self._login() + shared_image = self._create_firmware_image(organization=None) + shared_build = shared_image.build + self._create_device_with_connection( + organization=self._create_org(name="org1"), + model=shared_image.boards[0], ) - self.assertContains( - response, - f'', + self._create_device_with_connection( + organization=self._create_org(name="org2"), + model=shared_image.boards[0], ) + fw = DeviceFirmware.objects.filter( + image__build_id=shared_build.pk + ).select_related("image") + self.assertEqual(Device.objects.count(), 2) + self.assertEqual(UpgradeOperation.objects.count(), 0) + self.assertEqual(fw.count(), 0) - with self.subTest("Test with valid upgrade_options"): response = self.client.post( self.build_list_url, { "action": "upgrade_selected", "upgrade_all": "upgrade_all", "upgrade_options": '{"c": true}', - ACTION_CHECKBOX_NAME: (env["build2"].pk,), + ACTION_CHECKBOX_NAME: (shared_build.pk,), }, follow=True, ) self.assertContains(response, '
  • ') self.assertContains(response, "track the progress") self.assertEqual( - UpgradeOperation.objects.filter(upgrade_options={"c": True}).count(), 3 - ) - self.assertEqual(fw.count(), 3) - self.assertContains( - response, - ( - '
    " - ), - html=True, + UpgradeOperation.objects.filter(upgrade_options={"c": True}).count(), 2 ) + self.assertEqual(fw.count(), 2) - def test_mass_upgrade_shared_image(self, *args): - self._login() - shared_image = self._create_firmware_image(organization=None) - shared_build = shared_image.build - self._create_device_with_connection( - organization=self._create_org(name="org1"), - model=shared_image.boards[0], - ) - self._create_device_with_connection( - organization=self._create_org(name="org2"), - model=shared_image.boards[0], - ) - fw = DeviceFirmware.objects.filter( - image__build_id=shared_build.pk - ).select_related("image") - self.assertEqual(Device.objects.count(), 2) - self.assertEqual(UpgradeOperation.objects.count(), 0) - self.assertEqual(fw.count(), 0) - - response = self.client.post( - self.build_list_url, - { - "action": "upgrade_selected", - "upgrade_all": "upgrade_all", - "upgrade_options": '{"c": true}', - ACTION_CHECKBOX_NAME: (shared_build.pk,), - }, - follow=True, - ) - self.assertContains(response, '
  • ') - self.assertContains(response, "track the progress") - self.assertEqual( - UpgradeOperation.objects.filter(upgrade_options={"c": True}).count(), 2 - ) - self.assertEqual(fw.count(), 2) - + @mock.patch(_mock_upgrade, return_value=True) def test_massive_upgrade_operation_page(self, *args): - self.test_upgrade_all() - uo = UpgradeOperation.objects.first() - url = reverse( - f"admin:{self.app_label}_batchupgradeoperation_change", args=[uo.batch.pk] - ) - response = self.client.get(url) - self.assertContains(response, "Success rate") - self.assertContains(response, "Failure rate") - self.assertContains(response, "Abortion rate") + with mock.patch(self._mock_connect, return_value=True): + self.test_upgrade_all() + uo = UpgradeOperation.objects.first() + url = reverse( + f"admin:{self.app_label}_batchupgradeoperation_change", + args=[uo.batch.pk], + ) + response = self.client.get(url) + self.assertContains(response, "Success rate") + self.assertContains(response, "Failure rate") + self.assertContains(response, "Abortion rate") + @mock.patch(_mock_upgrade, return_value=True) def test_upgrade_operation_change_breadcrumb_with_batch(self, *args): - self.test_upgrade_all() - uo = UpgradeOperation.objects.first() - url = reverse(f"admin:{self.app_label}_upgradeoperation_change", args=[uo.pk]) - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - batch_changelist_url = reverse( - f"admin:{self.app_label}_batchupgradeoperation_changelist" - ) - batch_change_url = reverse( - f"admin:{self.app_label}_batchupgradeoperation_change", args=[uo.batch.pk] - ) - self.assertTrue(response.context["batch_has_view_permission"]) - self.assertEqual(response.context["batch"], uo.batch) - self.assertContains(response, batch_changelist_url) - self.assertContains(response, batch_change_url) - self.assertContains(response, str(uo.batch)) - generic_upgrade_changelist_url = reverse( - f"admin:{self.app_label}_upgradeoperation_changelist" - ) - self.assertNotContains(response, f'href="{generic_upgrade_changelist_url}"') + with mock.patch(self._mock_connect, return_value=True): + self.test_upgrade_all() + uo = UpgradeOperation.objects.first() + url = reverse(f"admin:{self.app_label}_upgradeoperation_change", args=[uo.pk]) + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + batch_changelist_url = reverse( + f"admin:{self.app_label}_batchupgradeoperation_changelist" + ) + batch_change_url = reverse( + f"admin:{self.app_label}_batchupgradeoperation_change", args=[uo.batch.pk] + ) + self.assertTrue(response.context["batch_has_view_permission"]) + self.assertEqual(response.context["batch"], uo.batch) + self.assertContains(response, batch_changelist_url) + self.assertContains(response, batch_change_url) + self.assertContains(response, str(uo.batch)) + generic_upgrade_changelist_url = reverse( + f"admin:{self.app_label}_upgradeoperation_changelist" + ) + self.assertNotContains(response, f'href="{generic_upgrade_changelist_url}"') + @mock.patch(_mock_upgrade, return_value=True) def test_upgrade_operation_change_breadcrumb_without_batch(self, *args): - self._login() - device_fw = self._create_device_firmware() - device_fw.save(upgrade=True) - uo = device_fw.device.upgradeoperation_set.first() - self.assertIsNone(uo.batch_id) - url = reverse(f"admin:{self.app_label}_upgradeoperation_change", args=[uo.pk]) - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - self.assertIsNone(response.context.get("batch")) - generic_upgrade_changelist_url = reverse( - f"admin:{self.app_label}_upgradeoperation_changelist" - ) - self.assertContains(response, f'href="{generic_upgrade_changelist_url}"') + with mock.patch(self._mock_connect, return_value=True): + self._login() + device_fw = self._create_device_firmware() + device_fw.save(upgrade=True) + uo = device_fw.device.upgradeoperation_set.first() + self.assertIsNone(uo.batch_id) + url = reverse(f"admin:{self.app_label}_upgradeoperation_change", args=[uo.pk]) + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + self.assertIsNone(response.context.get("batch")) + generic_upgrade_changelist_url = reverse( + f"admin:{self.app_label}_upgradeoperation_changelist" + ) + self.assertContains(response, f'href="{generic_upgrade_changelist_url}"') + @mock.patch(_mock_upgrade, return_value=True) def test_upgrade_operation_change_breadcrumb_with_batch_no_permission(self, *args): - self.test_upgrade_all() - uo = UpgradeOperation.objects.first() - url = reverse(f"admin:{self.app_label}_upgradeoperation_change", args=[uo.pk]) - with mock.patch( - "openwisp_firmware_upgrader.admin.BatchUpgradeOperationAdmin.has_view_permission", - return_value=False, - ): - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - batch_changelist_url = reverse( - f"admin:{self.app_label}_batchupgradeoperation_changelist" - ) - batch_change_url = reverse( - f"admin:{self.app_label}_batchupgradeoperation_change", args=[uo.batch.pk] - ) - self.assertFalse(response.context["batch_has_view_permission"]) - self.assertEqual(response.context["batch"], uo.batch) - breadcrumbs = ( - response.content.decode() - .split('
    ', 1)[1] - .split("
    ", 1)[0] - ) - self.assertNotIn(f'href="{batch_changelist_url}"', breadcrumbs) - self.assertNotIn(f'href="{batch_change_url}"', breadcrumbs) - generic_upgrade_changelist_url = reverse( - f"admin:{self.app_label}_upgradeoperation_changelist" - ) - self.assertNotIn(f'href="{generic_upgrade_changelist_url}"', breadcrumbs) - self.assertIn(str(uo.batch), breadcrumbs) + with mock.patch(self._mock_connect, return_value=True): + self.test_upgrade_all() + uo = UpgradeOperation.objects.first() + url = reverse(f"admin:{self.app_label}_upgradeoperation_change", args=[uo.pk]) + with mock.patch( + "openwisp_firmware_upgrader.admin.BatchUpgradeOperationAdmin.has_view_permission", + return_value=False, + ): + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + batch_changelist_url = reverse( + f"admin:{self.app_label}_batchupgradeoperation_changelist" + ) + batch_change_url = reverse( + f"admin:{self.app_label}_batchupgradeoperation_change", args=[uo.batch.pk] + ) + self.assertFalse(response.context["batch_has_view_permission"]) + self.assertEqual(response.context["batch"], uo.batch) + breadcrumbs = ( + response.content.decode() + .split('
    ', 1)[1] + .split("
    ", 1)[0] + ) + self.assertNotIn(f'href="{batch_changelist_url}"', breadcrumbs) + self.assertNotIn(f'href="{batch_change_url}"', breadcrumbs) + generic_upgrade_changelist_url = reverse( + f"admin:{self.app_label}_upgradeoperation_changelist" + ) + self.assertNotIn(f'href="{generic_upgrade_changelist_url}"', breadcrumbs) + self.assertIn(str(uo.batch), breadcrumbs) + @mock.patch(_mock_upgrade, return_value=True) def test_recent_upgrades(self, *args): - self._login() - env = self._create_upgrade_env() - url = reverse( - f"admin:{self.config_app_label}_device_change", args=[env["d2"].pk] - ) - r = self.client.get(url) - self.assertNotContains(r, "Recent Firmware Upgrades") - env["build2"].batch_upgrade(firmwareless=True) - r = self.client.get(url) - self.assertContains(r, "Recent Firmware Upgrades") + with mock.patch(self._mock_connect, return_value=True): + self._login() + env = self._create_upgrade_env() + url = reverse( + f"admin:{self.config_app_label}_device_change", args=[env["d2"].pk] + ) + r = self.client.get(url) + self.assertNotContains(r, "Recent Firmware Upgrades") + env["build2"].batch_upgrade(firmwareless=True) + r = self.client.get(url) + self.assertContains(r, "Recent Firmware Upgrades") + @mock.patch(_mock_upgrade, return_value=True) def test_upgrade_operation_inline(self, *args): - device_fw = self._create_device_firmware() - device_fw.save(upgrade=True) - device = device_fw.device - request = self.make_device_admin_request(device.pk) - request.user = User.objects.first() - deviceadmin = DeviceAdmin(model=Device, admin_site=admin.site) - self.assertNotIn( - DeviceUpgradeOperationInline, deviceadmin.get_inlines(request, obj=None) - ) - self.assertIn( - DeviceUpgradeOperationInline, deviceadmin.get_inlines(request, obj=device) - ) + with mock.patch(self._mock_connect, return_value=True): + device_fw = self._create_device_firmware() + device_fw.save(upgrade=True) + device = device_fw.device + request = self.make_device_admin_request(device.pk) + request.user = User.objects.first() + deviceadmin = DeviceAdmin(model=Device, admin_site=admin.site) + self.assertNotIn( + DeviceUpgradeOperationInline, deviceadmin.get_inlines(request, obj=None) + ) + self.assertIn( + DeviceUpgradeOperationInline, + deviceadmin.get_inlines(request, obj=device), + ) + @mock.patch(_mock_upgrade, return_value=True) def test_upgrade_operation_inline_queryset(self, *args): - device_fw = self._create_device_firmware() - device_fw.save(upgrade=True) - # expect only 1 - uo = device_fw.device.upgradeoperation_set.get() - device = device_fw.device - request = self.make_device_admin_request(device.pk) - request.user = User.objects.first() - inline = DeviceUpgradeOperationInline(Device, admin.site) - qs = inline.get_queryset(request) - self.assertEqual(qs.count(), 1) - self.assertIn(uo, qs) - uo.created = localtime() - timedelta(days=30) - uo.modified = uo.created - uo.save() - qs = inline.get_queryset(request) - self.assertEqual(qs.count(), 0) - + with mock.patch(self._mock_connect, return_value=True): + device_fw = self._create_device_firmware() + device_fw.save(upgrade=True) + # expect only 1 + uo = device_fw.device.upgradeoperation_set.get() + device = device_fw.device + request = self.make_device_admin_request(device.pk) + request.user = User.objects.first() + inline = DeviceUpgradeOperationInline(Device, admin.site) + qs = inline.get_queryset(request) + self.assertEqual(qs.count(), 1) + self.assertIn(uo, qs) + uo.created = localtime() - timedelta(days=30) + uo.modified = uo.created + uo.save() + qs = inline.get_queryset(request) + self.assertEqual(qs.count(), 0) + + @mock.patch(_mock_upgrade, return_value=True) def test_device_firmware_upgrade_options(self, *args): - self._login() - device_fw = self._create_device_firmware() - device = device_fw.device - device_conn = device.deviceconnection_set.first() - build = self._create_build(version="0.2") - image = self._create_firmware_image(build=build) - upgrade_options = { - "c": True, - "o": False, - "u": False, - "n": False, - "p": False, - "k": False, - "F": True, - } - device_params = self._get_device_params( - device, device_conn, image, device_fw, json.dumps(upgrade_options) - ) - response = self.client.post( - reverse(f"admin:{self.config_app_label}_device_change", args=[device.id]), - data=device_params, - follow=True, - ) - self.assertEqual(response.status_code, 200) - self.assertEqual(device.upgradeoperation_set.count(), 1) - upgrade_operation = device.upgradeoperation_set.first() - self.assertEqual(upgrade_operation.upgrade_options, upgrade_options) - self.assertContains( - response, - ( - '
    " - ), - html=True, - ) - - @mock.patch.object(OpenWisp1, "SCHEMA", None) - def test_using_upgrade_options_with_unsupported_upgrader(self, *args): - self._login() - device_fw = self._create_device_firmware() - device = device_fw.device - device.config.backend = "netjsonconfig.OpenWisp" - device.config.save() - device_conn = device.deviceconnection_set.first() - device_conn.update_strategy = conn_settings.DEFAULT_UPDATE_STRATEGIES[1][0] - device_conn.save() - build = self._create_build(version="0.2") - image = self._create_firmware_image(build=build) - upgrade_options = { - "c": True, - "o": False, - "u": False, - "n": False, - "p": False, - "k": False, - "F": True, - } - - device_params = self._get_device_params( - device, device_conn, image, device_fw, json.dumps(upgrade_options) - ) - device_params.update( - { - "model": device.model, - "devicefirmware-0-image": str(image.id), - "devicefirmware-0-id": str(device_fw.id), - "devicefirmware-0-upgrade_options": json.dumps(upgrade_options), - "organization": str(device.organization.id), - "config-0-id": str(device.config.pk), - "config-0-device": str(device.id), - "deviceconnection_set-0-credentials": str(device_conn.credentials_id), - "deviceconnection_set-0-id": str(device_conn.id), - "deviceconnection_set-0-update_strategy": ( - conn_settings.DEFAULT_UPDATE_STRATEGIES[1][0] - ), - "deviceconnection_set-0-enabled": True, - "devicefirmware-TOTAL_FORMS": 1, - "devicefirmware-INITIAL_FORMS": 1, - "upgradeoperation_set-TOTAL_FORMS": 0, - "upgradeoperation_set-INITIAL_FORMS": 0, - "upgradeoperation_set-MIN_NUM_FORMS": 0, - "upgradeoperation_set-MAX_NUM_FORMS": 0, - "_continue": True, + with mock.patch(self._mock_connect, return_value=True): + self._login() + device_fw = self._create_device_firmware() + device = device_fw.device + device_conn = device.deviceconnection_set.first() + build = self._create_build(version="0.2") + image = self._create_firmware_image(build=build) + upgrade_options = { + "c": True, + "o": False, + "u": False, + "n": False, + "p": False, + "k": False, + "F": True, } - ) - - with self.subTest("Test DeviceFirmwareInline does not have schema defined"): - response = self.client.get( - reverse( - f"admin:{self.config_app_label}_device_change", args=[device.id] - ) + device_params = self._get_device_params( + device, device_conn, image, device_fw, json.dumps(upgrade_options) ) - self.assertContains( - response, "" + ) + + with self.subTest("Test using upgrade options with unsupported upgrader"): + response = self.client.post( + reverse( + f"admin:{self.config_app_label}_device_change", args=[device.id] + ), + data=device_params, + follow=True, + ) + self.assertContains( + response, + ( + '" + ), + ) + + with self.subTest("Test upgrading without upgrade options"): + del device_params["devicefirmware-0-upgrade_options"] + response = self.client.post( + reverse( + f"admin:{self.config_app_label}_device_change", args=[device.id] + ), + data=device_params, + follow=True, + ) + self.assertContains( + response, + ( + '
    Upgrade options are ' + "not supported for this upgrader.
    " + ), + ) + + @mock.patch(_mock_upgrade, return_value=True) def test_batch_upgrade_operation_status_filter(self, *args): """Test status filtering in batch upgrade operation admin page""" - self._login() - env = self._create_upgrade_env() - env["category"].organization = None - env["category"].save() - batch = env["build2"].batch_upgrade(firmwareless=True) - # Create upgrade operations with different statuses - upgrade_ops = list(batch.upgradeoperation_set.all()) - if len(upgrade_ops) >= 2: - upgrade_ops[0].status = "success" - upgrade_ops[0].save() - upgrade_ops[1].status = "failed" - upgrade_ops[1].save() - url = reverse( - f"admin:{self.app_label}_batchupgradeoperation_change", args=[batch.pk] - ) + with mock.patch(self._mock_connect, return_value=True): + self._login() + env = self._create_upgrade_env() + env["category"].organization = None + env["category"].save() + batch = env["build2"].batch_upgrade(firmwareless=True) + # Create upgrade operations with different statuses + upgrade_ops = list(batch.upgradeoperation_set.all()) + if len(upgrade_ops) >= 2: + upgrade_ops[0].status = "success" + upgrade_ops[0].save() + upgrade_ops[1].status = "failed" + upgrade_ops[1].save() + url = reverse( + f"admin:{self.app_label}_batchupgradeoperation_change", args=[batch.pk] + ) - with self.subTest("Test no filter - shows all operations"): - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - self.assertContains(response, "ow-filter status") - self.assertContains(response, "By status") - self.assertContains(response, "By organization") + with self.subTest("Test no filter - shows all operations"): + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + self.assertContains(response, "ow-filter status") + self.assertContains(response, "By status") + self.assertContains(response, "By organization") - with self.subTest("Test status success filter"): - response = self.client.get(url + "?status=success") - self.assertEqual(response.status_code, 200) - success_ops = batch.upgradeoperation_set.filter(status="success") - for op in success_ops: - self.assertContains(response, op.device.name) + with self.subTest("Test status success filter"): + response = self.client.get(url + "?status=success") + self.assertEqual(response.status_code, 200) + success_ops = batch.upgradeoperation_set.filter(status="success") + for op in success_ops: + self.assertContains(response, op.device.name) - with self.subTest("Test status failed filter"): - response = self.client.get(url + "?status=failed") - self.assertEqual(response.status_code, 200) - failed_ops = batch.upgradeoperation_set.filter(status="failed") - for op in failed_ops: - self.assertContains(response, op.device.name) + with self.subTest("Test status failed filter"): + response = self.client.get(url + "?status=failed") + self.assertEqual(response.status_code, 200) + failed_ops = batch.upgradeoperation_set.filter(status="failed") + for op in failed_ops: + self.assertContains(response, op.device.name) - with self.subTest("Test status idle filter"): - response = self.client.get(url + "?status=idle") - self.assertEqual(response.status_code, 200) - idle_ops = batch.upgradeoperation_set.filter(status="idle") - for op in idle_ops: - self.assertContains(response, op.device.name) + with self.subTest("Test status idle filter"): + response = self.client.get(url + "?status=idle") + self.assertEqual(response.status_code, 200) + idle_ops = batch.upgradeoperation_set.filter(status="idle") + for op in idle_ops: + self.assertContains(response, op.device.name) + @mock.patch(_mock_upgrade, return_value=True) def test_batch_upgrade_operation_organization_filter(self, *args): """Test organization filtering in batch upgrade operation admin page""" - self._login() - # Create devices from different organizations - org1 = self._create_org(name="Org1", slug="org1") - org2 = self._create_org(name="Org2", slug="org2") - device1 = self._create_device(organization=org1, name="device1-org-filter") - device2 = self._create_device(organization=org2, name="device2-org-filter") - self._create_config(device=device1) - self._create_config(device=device2) - cred1 = self._get_credentials(organization=org1) - cred2 = self._get_credentials(organization=org2) - self._create_device_connection(device=device1, credentials=cred1) - self._create_device_connection(device=device2, credentials=cred2) - shared_category = self._create_category( - organization=None, name="Shared Category" - ) - build = self._create_build(category=shared_category) - image = self._create_firmware_image(build=build) - self._create_device_firmware( - device=device1, image=image, device_connection=False - ) - self._create_device_firmware( - device=device2, image=image, device_connection=False - ) - batch = build.batch_upgrade(firmwareless=False) - url = reverse( - f"admin:{self.app_label}_batchupgradeoperation_change", args=[batch.pk] - ) + with mock.patch(self._mock_connect, return_value=True): + self._login() + # Create devices from different organizations + org1 = self._create_org(name="Org1", slug="org1") + org2 = self._create_org(name="Org2", slug="org2") + device1 = self._create_device(organization=org1, name="device1-org-filter") + device2 = self._create_device(organization=org2, name="device2-org-filter") + self._create_config(device=device1) + self._create_config(device=device2) + cred1 = self._get_credentials(organization=org1) + cred2 = self._get_credentials(organization=org2) + self._create_device_connection(device=device1, credentials=cred1) + self._create_device_connection(device=device2, credentials=cred2) + shared_category = self._create_category( + organization=None, name="Shared Category" + ) + build = self._create_build(category=shared_category) + image = self._create_firmware_image(build=build) + self._create_device_firmware( + device=device1, image=image, device_connection=False + ) + self._create_device_firmware( + device=device2, image=image, device_connection=False + ) + batch = build.batch_upgrade(firmwareless=False) + url = reverse( + f"admin:{self.app_label}_batchupgradeoperation_change", args=[batch.pk] + ) - with self.subTest("Test no organization filter - shows all operations"): - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - self.assertContains(response, device1.name) - self.assertContains(response, device2.name) - self.assertContains(response, "By organization") + with self.subTest("Test no organization filter - shows all operations"): + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + self.assertContains(response, device1.name) + self.assertContains(response, device2.name) + self.assertContains(response, "By organization") - with self.subTest("Test organization filter for org1"): - response = self.client.get(url + f"?organization={org1.id}") - self.assertEqual(response.status_code, 200) - self.assertContains(response, device1.name) - self.assertNotContains(response, device2.name) + with self.subTest("Test organization filter for org1"): + response = self.client.get(url + f"?organization={org1.id}") + self.assertEqual(response.status_code, 200) + self.assertContains(response, device1.name) + self.assertNotContains(response, device2.name) - with self.subTest("Test organization filter for org2"): - response = self.client.get(url + f"?organization={org2.id}") - self.assertEqual(response.status_code, 200) - self.assertNotContains(response, device1.name) - self.assertContains(response, device2.name) + with self.subTest("Test organization filter for org2"): + response = self.client.get(url + f"?organization={org2.id}") + self.assertEqual(response.status_code, 200) + self.assertNotContains(response, device1.name) + self.assertContains(response, device2.name) + @mock.patch(_mock_upgrade, return_value=True) def test_batch_upgrade_operation_combined_filters(self, *args): """Test combining status and organization filters""" - self._login() - # Create devices from different organizations - org1 = self._create_org(name="Org1", slug="org1") - org2 = self._create_org(name="Org2", slug="org2") - device1 = self._create_device(organization=org1, name="device1-combined-filter") - device2 = self._create_device(organization=org2, name="device2-combined-filter") - self._create_config(device=device1) - self._create_config(device=device2) - cred1 = self._get_credentials(organization=org1) - cred2 = self._get_credentials(organization=org2) - self._create_device_connection(device=device1, credentials=cred1) - self._create_device_connection(device=device2, credentials=cred2) - - # Create shared build and batch upgrade that works with any organization - shared_category = self._create_category( - organization=None, name="Shared Category" - ) - build = self._create_build(category=shared_category) - image = self._create_firmware_image(build=build) - self._create_device_firmware( - device=device1, image=image, device_connection=False - ) - self._create_device_firmware( - device=device2, image=image, device_connection=False - ) - batch = build.batch_upgrade(firmwareless=False) - # Set different statuses for devices from different orgs - upgrade_ops = list(batch.upgradeoperation_set.all()) - org1_op = ( - upgrade_ops[0] - if upgrade_ops[0].device.organization == org1 - else upgrade_ops[1] - ) - org2_op = ( - upgrade_ops[1] - if upgrade_ops[1].device.organization == org2 - else upgrade_ops[0] - ) - org1_op.status = "success" - org1_op.save() - org2_op.status = "failed" - org2_op.save() - url = reverse( - f"admin:{self.app_label}_batchupgradeoperation_change", args=[batch.pk] - ) + with mock.patch(self._mock_connect, return_value=True): + self._login() + # Create devices from different organizations + org1 = self._create_org(name="Org1", slug="org1") + org2 = self._create_org(name="Org2", slug="org2") + device1 = self._create_device(organization=org1, name="device1-combined-filter") + device2 = self._create_device(organization=org2, name="device2-combined-filter") + self._create_config(device=device1) + self._create_config(device=device2) + cred1 = self._get_credentials(organization=org1) + cred2 = self._get_credentials(organization=org2) + self._create_device_connection(device=device1, credentials=cred1) + self._create_device_connection(device=device2, credentials=cred2) - with self.subTest("Test combined filter: org1 + success"): - response = self.client.get(url + f"?organization={org1.id}&status=success") - self.assertEqual(response.status_code, 200) - self.assertContains(response, org1_op.device.name) - self.assertNotContains(response, org2_op.device.name) + # Create shared build and batch upgrade that works with any organization + shared_category = self._create_category( + organization=None, name="Shared Category" + ) + build = self._create_build(category=shared_category) + image = self._create_firmware_image(build=build) + self._create_device_firmware( + device=device1, image=image, device_connection=False + ) + self._create_device_firmware( + device=device2, image=image, device_connection=False + ) + batch = build.batch_upgrade(firmwareless=False) + # Set different statuses for devices from different orgs + upgrade_ops = list(batch.upgradeoperation_set.all()) + org1_op = ( + upgrade_ops[0] + if upgrade_ops[0].device.organization == org1 + else upgrade_ops[1] + ) + org2_op = ( + upgrade_ops[1] + if upgrade_ops[1].device.organization == org2 + else upgrade_ops[0] + ) + org1_op.status = "success" + org1_op.save() + org2_op.status = "failed" + org2_op.save() + url = reverse( + f"admin:{self.app_label}_batchupgradeoperation_change", args=[batch.pk] + ) - with self.subTest("Test combined filter: org2 + failed"): - response = self.client.get(url + f"?organization={org2.id}&status=failed") - self.assertEqual(response.status_code, 200) - self.assertNotContains(response, org1_op.device.name) - self.assertContains(response, org2_op.device.name) + with self.subTest("Test combined filter: org1 + success"): + response = self.client.get(url + f"?organization={org1.id}&status=success") + self.assertEqual(response.status_code, 200) + self.assertContains(response, org1_op.device.name) + self.assertNotContains(response, org2_op.device.name) - with self.subTest("Test combined filter: org1 + failed (no results)"): - response = self.client.get(url + f"?organization={org1.id}&status=failed") - self.assertEqual(response.status_code, 200) - self.assertNotContains(response, org1_op.device.name) - self.assertNotContains(response, org2_op.device.name) + with self.subTest("Test combined filter: org2 + failed"): + response = self.client.get(url + f"?organization={org2.id}&status=failed") + self.assertEqual(response.status_code, 200) + self.assertNotContains(response, org1_op.device.name) + self.assertContains(response, org2_op.device.name) - with self.subTest("Combined filters preserve each other in generated links"): - response = self.client.get(url + f"?organization={org1.id}&status=success") - # Organization 'All' should keep status - self.assertContains( - response, - 'All', - html=True, - ) - # Status 'All' should keep organization - self.assertContains( - response, - f'All', - html=True, - ) + with self.subTest("Test combined filter: org1 + failed (no results)"): + response = self.client.get(url + f"?organization={org1.id}&status=failed") + self.assertEqual(response.status_code, 200) + self.assertNotContains(response, org1_op.device.name) + self.assertNotContains(response, org2_op.device.name) + + with self.subTest("Combined filters preserve each other in generated links"): + response = self.client.get(url + f"?organization={org1.id}&status=success") + # Organization 'All' should keep status + self.assertContains( + response, + 'All', + html=True, + ) + # Status 'All' should keep organization + self.assertContains( + response, + f'All', + html=True, + ) + @mock.patch(_mock_upgrade, return_value=True) def test_batch_upgrade_operation_filter_search_combination(self, *args): """Test combining search with filters""" - self._login() - env = self._create_upgrade_env() - batch = env["build2"].batch_upgrade(firmwareless=True) + with mock.patch(self._mock_connect, return_value=True): + self._login() + env = self._create_upgrade_env() + batch = env["build2"].batch_upgrade(firmwareless=True) - upgrade_op = batch.upgradeoperation_set.first() - upgrade_op.device.name = "unique-test-device" - upgrade_op.device.save() - upgrade_op.status = "success" - upgrade_op.save() + upgrade_op = batch.upgradeoperation_set.first() + upgrade_op.device.name = "unique-test-device" + upgrade_op.device.save() + upgrade_op.status = "success" + upgrade_op.save() - url = reverse( - f"admin:{self.app_label}_batchupgradeoperation_change", args=[batch.pk] - ) - with self.subTest("Test search + status filter"): - with self.assertNumQueries(25 if django.VERSION < (5, 2) else 23): - response = self.client.get(url + "?q=unique-test&status=success") - self.assertEqual(response.status_code, 200) - self.assertContains(response, "unique-test-device") - self.assertContains( - response, - 'All', - html=True, + url = reverse( + f"admin:{self.app_label}_batchupgradeoperation_change", args=[batch.pk] ) + with self.subTest("Test search + status filter"): + with self.assertNumQueries(25 if django.VERSION < (5, 2) else 23): + response = self.client.get(url + "?q=unique-test&status=success") + self.assertEqual(response.status_code, 200) + self.assertContains(response, "unique-test-device") + self.assertContains( + response, + 'All', + html=True, + ) - with self.subTest("Test search + status filter (no match)"): - response = self.client.get(url + "?q=unique-test&status=failed") - self.assertEqual(response.status_code, 200) - self.assertNotContains(response, "unique-test-device") + with self.subTest("Test search + status filter (no match)"): + response = self.client.get(url + "?q=unique-test&status=failed") + self.assertEqual(response.status_code, 200) + self.assertNotContains(response, "unique-test-device") + @mock.patch(_mock_upgrade, return_value=True) def test_batch_upgrade_confirmation_form_multitenancy(self, *args): """Test BatchUpgradeConfirmationForm multitenancy for organization admin vs superuser.""" - # Setup common objects - org1 = self._get_org() - org2 = self._create_org(name="Org 2", slug="org2") - group1 = self._create_device_group(name="Group Org1", organization=org1) - group2 = self._create_device_group(name="Group Org2", organization=org2) - location1 = Location.objects.create( - name="Location Org1", address="123 Test St", organization=org1 - ) - location2 = Location.objects.create( - name="Location Org2", address="456 Test St", organization=org2 - ) - category_org1 = self._create_category(organization=org1) - build_org1 = self._create_build(category=category_org1) - category_shared = self._create_category(organization=None) - build_shared = self._create_build(category=category_shared) - category_org2 = self._create_category(organization=org2) - build_org2 = self._create_build(category=category_org2) - superuser = self._get_admin() - org_admin = self._create_administrator(organizations=[org1]) - - with self.subTest("Superuser: Org build should shown related org objects"): - form = BatchUpgradeConfirmationForm( - initial={"build": build_org1}, user=superuser - ) - self.assertIn(group1, form.fields["group"].queryset) - self.assertNotIn(group2, form.fields["group"].queryset) - self.assertIn(location1, form.fields["location"].queryset) - self.assertNotIn(location2, form.fields["location"].queryset) - - with self.subTest("Superuser: Shared build should show all org objects"): - form = BatchUpgradeConfirmationForm( - initial={"build": build_shared}, user=superuser - ) - self.assertIn(group1, form.fields["group"].queryset) - self.assertIn(group2, form.fields["group"].queryset) - self.assertIn(location1, form.fields["location"].queryset) - self.assertIn(location2, form.fields["location"].queryset) - - with self.subTest( - "Org admin: Shared build should show only managed org objects" - ): - form = BatchUpgradeConfirmationForm( - initial={"build": build_shared}, user=org_admin - ) - self.assertIn(group1, form.fields["group"].queryset) - self.assertNotIn(group2, form.fields["group"].queryset) - self.assertIn(location1, form.fields["location"].queryset) - self.assertNotIn(location2, form.fields["location"].queryset) - - with self.subTest("Org admin: Org build should show only that org objects"): - form = BatchUpgradeConfirmationForm( - initial={"build": build_org1}, user=org_admin + with mock.patch(self._mock_connect, return_value=True): + # Setup common objects + org1 = self._get_org() + org2 = self._create_org(name="Org 2", slug="org2") + group1 = self._create_device_group(name="Group Org1", organization=org1) + group2 = self._create_device_group(name="Group Org2", organization=org2) + location1 = Location.objects.create( + name="Location Org1", address="123 Test St", organization=org1 ) - self.assertIn(group1, form.fields["group"].queryset) - self.assertNotIn(group2, form.fields["group"].queryset) - self.assertIn(location1, form.fields["location"].queryset) - self.assertNotIn(location2, form.fields["location"].queryset) - - with self.subTest("Org admin: Different org build should show no objects"): - form = BatchUpgradeConfirmationForm( - initial={"build": build_org2}, user=org_admin + location2 = Location.objects.create( + name="Location Org2", address="456 Test St", organization=org2 ) - self.assertEqual(form.fields["group"].queryset.count(), 0) - self.assertEqual(form.fields["location"].queryset.count(), 0) + category_org1 = self._create_category(organization=org1) + build_org1 = self._create_build(category=category_org1) + category_shared = self._create_category(organization=None) + build_shared = self._create_build(category=category_shared) + category_org2 = self._create_category(organization=org2) + build_org2 = self._create_build(category=category_org2) + superuser = self._get_admin() + org_admin = self._create_administrator(organizations=[org1]) + + with self.subTest("Superuser: Org build should shown related org objects"): + form = BatchUpgradeConfirmationForm( + initial={"build": build_org1}, user=superuser + ) + self.assertIn(group1, form.fields["group"].queryset) + self.assertNotIn(group2, form.fields["group"].queryset) + self.assertIn(location1, form.fields["location"].queryset) + self.assertNotIn(location2, form.fields["location"].queryset) + + with self.subTest("Superuser: Shared build should show all org objects"): + form = BatchUpgradeConfirmationForm( + initial={"build": build_shared}, user=superuser + ) + self.assertIn(group1, form.fields["group"].queryset) + self.assertIn(group2, form.fields["group"].queryset) + self.assertIn(location1, form.fields["location"].queryset) + self.assertIn(location2, form.fields["location"].queryset) + + with self.subTest( + "Org admin: Shared build should show only managed org objects" + ): + form = BatchUpgradeConfirmationForm( + initial={"build": build_shared}, user=org_admin + ) + self.assertIn(group1, form.fields["group"].queryset) + self.assertNotIn(group2, form.fields["group"].queryset) + self.assertIn(location1, form.fields["location"].queryset) + self.assertNotIn(location2, form.fields["location"].queryset) + + with self.subTest("Org admin: Org build should show only that org objects"): + form = BatchUpgradeConfirmationForm( + initial={"build": build_org1}, user=org_admin + ) + self.assertIn(group1, form.fields["group"].queryset) + self.assertNotIn(group2, form.fields["group"].queryset) + self.assertIn(location1, form.fields["location"].queryset) + self.assertNotIn(location2, form.fields["location"].queryset) + + with self.subTest("Org admin: Different org build should show no objects"): + form = BatchUpgradeConfirmationForm( + initial={"build": build_org2}, user=org_admin + ) + self.assertEqual(form.fields["group"].queryset.count(), 0) + self.assertEqual(form.fields["location"].queryset.count(), 0) - with self.subTest("Location field exists and is not required"): - form = BatchUpgradeConfirmationForm( - initial={"build": build_org1}, user=superuser - ) - self.assertIn("location", form.fields) - self.assertFalse(form.fields["location"].required) - self.assertIn("location", form.fields["location"].help_text) + with self.subTest("Location field exists and is not required"): + form = BatchUpgradeConfirmationForm( + initial={"build": build_org1}, user=superuser + ) + self.assertIn("location", form.fields) + self.assertFalse(form.fields["location"].required) + self.assertIn("location", form.fields["location"].help_text) + @mock.patch(_mock_upgrade, return_value=True) def test_batch_upgrade_with_location_admin_action(self, *args): """Test mass upgrade admin action with location filtering.""" - self._login() - org = self._get_org() - category = self._create_category(organization=org) - build = self._create_build(category=category) - image = self._create_firmware_image(build=build) - # Create location - location = Location.objects.create( - name="Test Location", address="123 Test St", organization=org - ) - # Create devices - device1 = self._create_device( - name="Device1-WithLocation", - organization=org, - model=image.boards[0], - mac_address="00:11:22:33:55:71", - ) - device2 = self._create_device( - name="Device2-NoLocation", - organization=org, - model=image.boards[0], - mac_address="00:11:22:33:55:72", - ) - # Set location for device1 only - DeviceLocation.objects.create(content_object=device1, location=location) - # Create configs and connections - self._create_config(device=device1) - self._create_config(device=device2) - cred1 = self._get_credentials(organization=org) - if not DeviceConnection.objects.filter( - device=device1, credentials=cred1 - ).exists(): - self._create_device_connection(device=device1, credentials=cred1) - if not DeviceConnection.objects.filter( - device=device2, credentials=cred1 - ).exists(): - self._create_device_connection(device=device2, credentials=cred1) - url = reverse(f"admin:{self.app_label}_build_changelist") - data = { - ACTION_CHECKBOX_NAME: [build.pk], - "action": "upgrade_selected", - "location": location.pk, - "upgrade_related": "on", - } - with self.subTest("Test upgrade confirmation page with location"): - response = self.client.post(url, data, follow=True) - self.assertEqual(response.status_code, 200) - self.assertContains(response, location.name) - # Submit the actual upgrade with location filter - data.update( - { - "upgrade_all": "on", + with mock.patch(self._mock_connect, return_value=True): + self._login() + org = self._get_org() + category = self._create_category(organization=org) + build = self._create_build(category=category) + image = self._create_firmware_image(build=build) + # Create location + location = Location.objects.create( + name="Test Location", address="123 Test St", organization=org + ) + # Create devices + device1 = self._create_device( + name="Device1-WithLocation", + organization=org, + model=image.boards[0], + mac_address="00:11:22:33:55:71", + ) + device2 = self._create_device( + name="Device2-NoLocation", + organization=org, + model=image.boards[0], + mac_address="00:11:22:33:55:72", + ) + # Set location for device1 only + DeviceLocation.objects.create(content_object=device1, location=location) + # Create configs and connections + self._create_config(device=device1) + self._create_config(device=device2) + cred1 = self._get_credentials(organization=org) + if not DeviceConnection.objects.filter( + device=device1, credentials=cred1 + ).exists(): + self._create_device_connection(device=device1, credentials=cred1) + if not DeviceConnection.objects.filter( + device=device2, credentials=cred1 + ).exists(): + self._create_device_connection(device=device2, credentials=cred1) + url = reverse(f"admin:{self.app_label}_build_changelist") + data = { + ACTION_CHECKBOX_NAME: [build.pk], + "action": "upgrade_selected", "location": location.pk, + "upgrade_related": "on", } - ) - with self.subTest("Test actual batch upgrade with location"): - with mock.patch("openwisp_firmware_upgrader.tasks.upgrade_firmware.delay"): + with self.subTest("Test upgrade confirmation page with location"): response = self.client.post(url, data, follow=True) self.assertEqual(response.status_code, 200) - # Check that batch was created with location - batch = BatchUpgradeOperation.objects.first() - self.assertIsNotNone(batch) - self.assertEqual(batch.location, location) - + self.assertContains(response, location.name) + # Submit the actual upgrade with location filter + data.update( + { + "upgrade_all": "on", + "location": location.pk, + } + ) + with self.subTest("Test actual batch upgrade with location"): + with mock.patch("openwisp_firmware_upgrader.tasks.upgrade_firmware.delay"): + response = self.client.post(url, data, follow=True) + self.assertEqual(response.status_code, 200) + # Check that batch was created with location + batch = BatchUpgradeOperation.objects.first() + self.assertIsNotNone(batch) + self.assertEqual(batch.location, location) + + @mock.patch(_mock_upgrade, return_value=True) def test_batch_upgrade_operation_admin_location_field(self, *args): """Test location field in BatchUpgradeOperationAdmin.""" - self._login() - org = self._get_org() - category = self._create_category(organization=org) - build = self._create_build(category=category) - location = Location.objects.create( - name="Test Location", address="123 Test St", organization=org - ) - batch = BatchUpgradeOperation.objects.create(build=build, location=location) - url = reverse( - f"admin:{self.app_label}_batchupgradeoperation_change", args=[batch.pk] - ) - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - self.assertContains(response, location.name) + with mock.patch(self._mock_connect, return_value=True): + self._login() + org = self._get_org() + category = self._create_category(organization=org) + build = self._create_build(category=category) + location = Location.objects.create( + name="Test Location", address="123 Test St", organization=org + ) + batch = BatchUpgradeOperation.objects.create(build=build, location=location) + url = reverse( + f"admin:{self.app_label}_batchupgradeoperation_change", args=[batch.pk] + ) + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + self.assertContains(response, location.name) + @mock.patch(_mock_upgrade, return_value=True) def test_batch_upgrade_no_devices_error_handling(self, *args): """Test admin error handling when filters don't match any devices.""" - self._login() - org = self._get_org() - category = self._create_category(organization=org) - build = self._create_build(category=category, version="error-test") - # Create location and group but no devices matching both - location = Location.objects.create( - name="Empty Location", address="456 Empty St", organization=org - ) - group = self._create_device_group(name="Empty Group", organization=org) - url = reverse(f"admin:{self.app_label}_build_changelist") - data = { - ACTION_CHECKBOX_NAME: [build.pk], - "action": "upgrade_selected", - "location": location.pk, - "group": group.pk, - "upgrade_all": "on", - } - with self.subTest("Test error message when no devices match filters"): - response = self.client.post(url, data, follow=True) - self.assertEqual(response.status_code, 200) - # Should stay on confirmation page with error message - self.assertContains(response, "No devices found matching") - self.assertContains(response, "adjust your group and/or location filters") - # No batch should be created - self.assertEqual(BatchUpgradeOperation.objects.count(), 0) + with mock.patch(self._mock_connect, return_value=True): + self._login() + org = self._get_org() + category = self._create_category(organization=org) + build = self._create_build(category=category, version="error-test") + # Create location and group but no devices matching both + location = Location.objects.create( + name="Empty Location", address="456 Empty St", organization=org + ) + group = self._create_device_group(name="Empty Group", organization=org) + url = reverse(f"admin:{self.app_label}_build_changelist") + data = { + ACTION_CHECKBOX_NAME: [build.pk], + "action": "upgrade_selected", + "location": location.pk, + "group": group.pk, + "upgrade_all": "on", + } + with self.subTest("Test error message when no devices match filters"): + response = self.client.post(url, data, follow=True) + self.assertEqual(response.status_code, 200) + # Should stay on confirmation page with error message + self.assertContains(response, "No devices found matching") + self.assertContains(response, "adjust your group and/or location filters") + # No batch should be created + self.assertEqual(BatchUpgradeOperation.objects.count(), 0) + @mock.patch(_mock_upgrade, return_value=True) def test_batch_upgrade_operation_list_location_filter(self, *args): """Test location filter in BatchUpgradeOperation list view.""" - self._login() - org = self._get_org() - category = self._create_category( - name="Location Filter Test Category", organization=org - ) - build = self._create_build(category=category, version="location-test-1.0") - location1 = Location.objects.create( - name="Location 1", address="123 Main St", organization=org - ) - location2 = Location.objects.create( - name="Location 2", address="456 Oak Ave", organization=org - ) - # Create batch operations with different locations - batch1 = BatchUpgradeOperation.objects.create(build=build, location=location1) - batch2 = BatchUpgradeOperation.objects.create(build=build, location=location2) - batch3 = BatchUpgradeOperation.objects.create( - build=build, location=None # No location - ) - url = reverse(f"admin:{self.app_label}_batchupgradeoperation_changelist") - with self.subTest("Test no location filter - shows all batches"): - with self.assertNumQueries(5): - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - self.assertContains(response, str(batch1.pk)) - self.assertContains(response, str(batch2.pk)) - self.assertContains(response, str(batch3.pk)) + with mock.patch(self._mock_connect, return_value=True): + self._login() + org = self._get_org() + category = self._create_category( + name="Location Filter Test Category", organization=org + ) + build = self._create_build(category=category, version="location-test-1.0") + location1 = Location.objects.create( + name="Location 1", address="123 Main St", organization=org + ) + location2 = Location.objects.create( + name="Location 2", address="456 Oak Ave", organization=org + ) + # Create batch operations with different locations + batch1 = BatchUpgradeOperation.objects.create(build=build, location=location1) + batch2 = BatchUpgradeOperation.objects.create(build=build, location=location2) + batch3 = BatchUpgradeOperation.objects.create( + build=build, location=None # No location + ) + url = reverse(f"admin:{self.app_label}_batchupgradeoperation_changelist") + with self.subTest("Test no location filter - shows all batches"): + with self.assertNumQueries(5): + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + self.assertContains(response, str(batch1.pk)) + self.assertContains(response, str(batch2.pk)) + self.assertContains(response, str(batch3.pk)) - with self.subTest("Test location1 filter"): - with self.assertNumQueries(4): - response = self.client.get(url + f"?location={location1.pk}") - self.assertEqual(response.status_code, 200) - self.assertContains(response, str(batch1.pk)) - self.assertNotContains(response, str(batch2.pk)) - self.assertNotContains(response, str(batch3.pk)) + with self.subTest("Test location1 filter"): + with self.assertNumQueries(4): + response = self.client.get(url + f"?location={location1.pk}") + self.assertEqual(response.status_code, 200) + self.assertContains(response, str(batch1.pk)) + self.assertNotContains(response, str(batch2.pk)) + self.assertNotContains(response, str(batch3.pk)) del TestConfigAdmin diff --git a/openwisp_firmware_upgrader/tests/test_models.py b/openwisp_firmware_upgrader/tests/test_models.py index f7cd85d07..e51cb03c2 100644 --- a/openwisp_firmware_upgrader/tests/test_models.py +++ b/openwisp_firmware_upgrader/tests/test_models.py @@ -560,69 +560,69 @@ class TestModelsTransaction(TestUpgraderMixin, TransactionTestCase): image_type = TestModels.image_type @mock.patch(_mock_updrade, return_value=True) - @mock.patch(_mock_connect, return_value=True) def test_dry_run(self, *args): - env = self._create_upgrade_env() - # check pending upgrades - result = BatchUpgradeOperation.dry_run(build=env["build1"]) - self.assertEqual( - list(result["device_firmwares"]), - list(DeviceFirmware.objects.all().order_by("-created")), - ) - self.assertEqual(list(result["devices"]), []) - # upgrade devices - env["build1"].batch_upgrade(firmwareless=True) - # check pending upgrades again - result = BatchUpgradeOperation.dry_run(build=env["build1"]) - self.assertEqual(list(result["device_firmwares"]), []) - self.assertEqual(list(result["devices"]), []) + with mock.patch(self._mock_connect, return_value=True): + env = self._create_upgrade_env() + # check pending upgrades + result = BatchUpgradeOperation.dry_run(build=env["build1"]) + self.assertEqual( + list(result["device_firmwares"]), + list(DeviceFirmware.objects.all().order_by("-created")), + ) + self.assertEqual(list(result["devices"]), []) + # upgrade devices + env["build1"].batch_upgrade(firmwareless=True) + # check pending upgrades again + result = BatchUpgradeOperation.dry_run(build=env["build1"]) + self.assertEqual(list(result["device_firmwares"]), []) + self.assertEqual(list(result["devices"]), []) @mock.patch(_mock_updrade, return_value=True) - @mock.patch(_mock_connect, return_value=True) def test_upgrade_related_devices(self, *args): - env = self._create_upgrade_env() - # check everything is as expected - self.assertEqual(UpgradeOperation.objects.count(), 0) - self.assertEqual(env["d1"].devicefirmware.image, env["image1a"]) - self.assertEqual(env["d2"].devicefirmware.image, env["image1b"]) - # upgrade all related - env["build2"].batch_upgrade(firmwareless=False) - # ensure image is changed - env["d1"].devicefirmware.refresh_from_db() - env["d2"].devicefirmware.refresh_from_db() - self.assertEqual(env["d1"].devicefirmware.image, env["image2a"]) - self.assertEqual(env["d2"].devicefirmware.image, env["image2b"]) - # ensure upgrade operation objects have been created - self.assertEqual(UpgradeOperation.objects.count(), 2) - batch_qs = BatchUpgradeOperation.objects.all() - self.assertEqual(batch_qs.count(), 1) - batch = batch_qs.first() - self.assertEqual(batch.upgradeoperation_set.count(), 2) - self.assertEqual(batch.build, env["build2"]) - self.assertEqual(batch.status, "success") + with mock.patch(self._mock_connect, return_value=True): + env = self._create_upgrade_env() + # check everything is as expected + self.assertEqual(UpgradeOperation.objects.count(), 0) + self.assertEqual(env["d1"].devicefirmware.image, env["image1a"]) + self.assertEqual(env["d2"].devicefirmware.image, env["image1b"]) + # upgrade all related + env["build2"].batch_upgrade(firmwareless=False) + # ensure image is changed + env["d1"].devicefirmware.refresh_from_db() + env["d2"].devicefirmware.refresh_from_db() + self.assertEqual(env["d1"].devicefirmware.image, env["image2a"]) + self.assertEqual(env["d2"].devicefirmware.image, env["image2b"]) + # ensure upgrade operation objects have been created + self.assertEqual(UpgradeOperation.objects.count(), 2) + batch_qs = BatchUpgradeOperation.objects.all() + self.assertEqual(batch_qs.count(), 1) + batch = batch_qs.first() + self.assertEqual(batch.upgradeoperation_set.count(), 2) + self.assertEqual(batch.build, env["build2"]) + self.assertEqual(batch.status, "success") @mock.patch(_mock_updrade, return_value=True) - @mock.patch(_mock_connect, return_value=True) def test_upgrade_firmwareless_devices(self, *args): - env = self._create_upgrade_env(device_firmware=False) - # check everything is as expected - self.assertEqual(UpgradeOperation.objects.count(), 0) - self.assertFalse(hasattr(env["d1"], "devicefirmware")) - self.assertFalse(hasattr(env["d2"], "devicefirmware")) - # upgrade all related - env["build2"].batch_upgrade(firmwareless=True) - env["d1"].refresh_from_db() - env["d2"].refresh_from_db() - self.assertEqual(env["d1"].devicefirmware.image, env["image2a"]) - self.assertEqual(env["d2"].devicefirmware.image, env["image2b"]) - # ensure upgrade operation objects have been created - self.assertEqual(UpgradeOperation.objects.count(), 2) - batch_qs = BatchUpgradeOperation.objects.all() - self.assertEqual(batch_qs.count(), 1) - batch = batch_qs.first() - self.assertEqual(batch.upgradeoperation_set.count(), 2) - self.assertEqual(batch.build, env["build2"]) - self.assertEqual(batch.status, "success") + with mock.patch(self._mock_connect, return_value=True): + env = self._create_upgrade_env(device_firmware=False) + # check everything is as expected + self.assertEqual(UpgradeOperation.objects.count(), 0) + self.assertFalse(hasattr(env["d1"], "devicefirmware")) + self.assertFalse(hasattr(env["d2"], "devicefirmware")) + # upgrade all related + env["build2"].batch_upgrade(firmwareless=True) + env["d1"].refresh_from_db() + env["d2"].refresh_from_db() + self.assertEqual(env["d1"].devicefirmware.image, env["image2a"]) + self.assertEqual(env["d2"].devicefirmware.image, env["image2b"]) + # ensure upgrade operation objects have been created + self.assertEqual(UpgradeOperation.objects.count(), 2) + batch_qs = BatchUpgradeOperation.objects.all() + self.assertEqual(batch_qs.count(), 1) + batch = batch_qs.first() + self.assertEqual(batch.upgradeoperation_set.count(), 2) + self.assertEqual(batch.build, env["build2"]) + self.assertEqual(batch.status, "success") @mock.patch.object(upgrade_firmware, "max_retries", 0) def test_batch_upgrade_failure(self): @@ -634,28 +634,28 @@ def test_batch_upgrade_failure(self): self.assertEqual(BatchUpgradeOperation.objects.count(), 1) @mock.patch(_mock_updrade, return_value=True) - @mock.patch(_mock_connect, return_value=True) def test_upgrade_related_devices_existing_fw(self, *args): - env = self._create_upgrade_env() - self.assertEqual(UpgradeOperation.objects.count(), 0) - self.assertEqual(env["d1"].devicefirmware.image, env["image1a"]) - self.assertEqual(env["d2"].devicefirmware.image, env["image1b"]) - env["d1"].devicefirmware.installed = False - env["d1"].devicefirmware.save(upgrade=False) - env["d2"].devicefirmware.installed = False - env["d2"].devicefirmware.save(upgrade=False) - env["build1"].batch_upgrade(firmwareless=False) - env["d1"].devicefirmware.refresh_from_db() - env["d2"].devicefirmware.refresh_from_db() - self.assertEqual(env["d1"].devicefirmware.image, env["image1a"]) - self.assertEqual(env["d2"].devicefirmware.image, env["image1b"]) - self.assertEqual(UpgradeOperation.objects.count(), 2) - batch_qs = BatchUpgradeOperation.objects.all() - self.assertEqual(batch_qs.count(), 1) - batch = batch_qs.first() - self.assertEqual(batch.upgradeoperation_set.count(), 2) - self.assertEqual(batch.build, env["build1"]) - self.assertEqual(batch.status, "success") + with mock.patch(self._mock_connect, return_value=True): + env = self._create_upgrade_env() + self.assertEqual(UpgradeOperation.objects.count(), 0) + self.assertEqual(env["d1"].devicefirmware.image, env["image1a"]) + self.assertEqual(env["d2"].devicefirmware.image, env["image1b"]) + env["d1"].devicefirmware.installed = False + env["d1"].devicefirmware.save(upgrade=False) + env["d2"].devicefirmware.installed = False + env["d2"].devicefirmware.save(upgrade=False) + env["build1"].batch_upgrade(firmwareless=False) + env["d1"].devicefirmware.refresh_from_db() + env["d2"].devicefirmware.refresh_from_db() + self.assertEqual(env["d1"].devicefirmware.image, env["image1a"]) + self.assertEqual(env["d2"].devicefirmware.image, env["image1b"]) + self.assertEqual(UpgradeOperation.objects.count(), 2) + batch_qs = BatchUpgradeOperation.objects.all() + self.assertEqual(batch_qs.count(), 1) + batch = batch_qs.first() + self.assertEqual(batch.upgradeoperation_set.count(), 2) + self.assertEqual(batch.build, env["build1"]) + self.assertEqual(batch.status, "success") def test_upgrade_retried(self): env = self._create_upgrade_env() diff --git a/openwisp_firmware_upgrader/tests/test_selenium.py b/openwisp_firmware_upgrader/tests/test_selenium.py index 6bb5aa800..8fdbf0d2a 100644 --- a/openwisp_firmware_upgrader/tests/test_selenium.py +++ b/openwisp_firmware_upgrader/tests/test_selenium.py @@ -43,6 +43,7 @@ class TestDeviceAdmin(TestUpgraderMixin, SeleniumTestMixin, StaticLiveServerTest firmware_app_label = Build._meta.app_label os = "OpenWrt 19.07-SNAPSHOT r11061-6ffd4d8a4d" image_type = REVERSE_FIRMWARE_IMAGE_MAP["YunCore XD3200"] + _mock_connect = "openwisp_controller.connection.models.DeviceConnection.connect" def _set_up_env(self): org = self._get_org() @@ -114,9 +115,10 @@ def test_restoring_deleted_device(self, *args): By.XPATH, '//*[@id="device_form"]/div/div[1]/input[1]' ).click() try: - WebDriverWait(self.web_driver, 5).until( - EC.url_to_be(f"{self.live_server_url}/admin/config/device/") + device_changelist_url = self.live_server_url + reverse( + f"admin:{self.config_app_label}_device_changelist" ) + WebDriverWait(self.web_driver, 5).until(EC.url_to_be(device_changelist_url)) except TimeoutException: self.fail("Deleted device was not restored") @@ -128,170 +130,18 @@ def test_restoring_deleted_device(self, *args): "openwisp_firmware_upgrader.upgraders.openwrt.OpenWrt.upgrade", return_value=True, ) - @patch( - "openwisp_controller.connection.models.DeviceConnection.connect", - return_value=True, - ) def test_device_firmware_upgrade_options(self, *args): - def save_device(): - self.find_element( - by=By.XPATH, value='//*[@id="device_form"]/div/div[1]/input[3]' - ).click() - self.wait_for_visibility(By.CSS_SELECTOR, "#devicefirmware-group") - self.hide_loading_overlay() + with patch(self._mock_connect, return_value=True): - _, _, _, _, _, image, device = self._set_up_env() - self.login() - self.open( - "{}#devicefirmware-group".format( - reverse( - f"admin:{self.config_app_label}_device_change", args=[device.id] - ) - ) - ) - self.hide_loading_overlay() - # JSONSchema Editor should not be rendered without a change in the image field - self.wait_for_invisibility( - By.CSS_SELECTOR, "#devicefirmware-group .jsoneditor-wrapper" - ) - image_select = self._get_device_firmware_dropdown_select() - image_select.select_by_value(str(image.pk)) - # JSONSchema configuration editor should not be rendered - self.wait_for_invisibility( - By.XPATH, - '//*[@id="id_devicefirmware-0-upgrade_options_jsoneditor"]/div/h3/span[4]/input', - ) - # Select "None" image should hide JSONSchema Editor - image_select.select_by_value("") - self.wait_for_invisibility( - By.CSS_SELECTOR, "#id_devicefirmware-0-upgrade_options_jsoneditor" - ) - - # Select "build2" image - image_select.select_by_value(str(image.pk)) - # Enable '-c' option - self.find_element( - by=By.XPATH, - value='//*[@id="id_devicefirmware-0-upgrade_options_jsoneditor"]' - "/div/div[2]/div/div/div[1]/div/div[1]/label/input", - ).click() - # Enable '-F' option - self.find_element( - by=By.XPATH, - value='//*[@id="id_devicefirmware-0-upgrade_options_jsoneditor"]' - "/div/div[2]/div/div/div[7]/div/div[1]/label/input", - ).click() - save_device() + def save_device(): + self.find_element( + by=By.XPATH, value='//*[@id="device_form"]/div/div[1]/input[3]' + ).click() + self.wait_for_visibility(By.CSS_SELECTOR, "#devicefirmware-group") + self.hide_loading_overlay() - # Delete DeviceFirmware - self.find_element(By.CSS_SELECTOR, "#id_devicefirmware-0-DELETE").click() - save_device() - - # When adding firmware to the device for the first time, - # JSONSchema editor should be rendered only when the image - # is selected - self.find_element( - by=By.XPATH, value='//*[@id="devicefirmware-group"]/fieldset/div[2]/a' - ).click() - # JSONSchema Editor should not be rendered without a change in the image field - self.wait_for_invisibility( - By.CSS_SELECTOR, "#devicefirmware-group .jsoneditor-wrapper" - ) - image_select = self._get_device_firmware_dropdown_select() - image_select.select_by_index(1) - self.wait_for_visibility( - By.CSS_SELECTOR, "#devicefirmware-group .jsoneditor-wrapper" - ) - save_device() - - @patch( - "openwisp_firmware_upgrader.upgraders.openwrt.OpenWrt.upgrade", - return_value=True, - ) - @patch( - "openwisp_controller.connection.models.DeviceConnection.connect", - return_value=True, - ) - def test_batch_upgrade_upgrade_options(self, *args): - _, _, _, build2, _, _, _ = self._set_up_env() - self.login() - self.open( - reverse(f"admin:{self.firmware_app_label}_build_change", args=[build2.id]) - ) - # Launch mass upgrade operation - self.find_element( - by=By.CSS_SELECTOR, - value='.title-wrapper .object-tools form button[type="submit"]', - ).click() - - # Ensure JSONSchema form is rendered - self.wait_for_visibility(By.CSS_SELECTOR, ".jsoneditor-wrapper") - # JSONSchema configuration editor should not be rendered - self.wait_for_invisibility( - By.XPATH, - '//*[@id="id_devicefirmware-0-upgrade_options_jsoneditor"]/div/h3/span[4]/input', - ) - # Disable -c flag - self.find_element( - by=By.XPATH, - value='//*[@id="id_upgrade_options_jsoneditor"]/div/div[2]/div/div/div[1]/div/div[1]/label/input', - ).click() - # Enable -n flag - self.find_element( - by=By.XPATH, - value='//*[@id="id_upgrade_options_jsoneditor"]/div/div[2]/div/div/div[3]/div/div[1]/label/input', - ).click() - # Upgrade all devices - self.find_element(by=By.CSS_SELECTOR, value='input[name="upgrade_all"]').click() - try: - WebDriverWait(self.web_driver, 5).until( - EC.url_contains("batchupgradeoperation") - ) - except TimeoutException: - self.fail("User was not redirected to Mass upgrade operations page") - self.assertEqual( - BatchUpgradeOperation.objects.filter( - upgrade_options={ - "c": False, - "o": False, - "n": True, - "u": False, - "p": False, - "k": False, - "F": False, - } - ).count(), - 1, - ) - self.assertEqual( - UpgradeOperation.objects.filter( - upgrade_options={ - "c": False, - "o": False, - "n": True, - "u": False, - "p": False, - "k": False, - "F": False, - } - ).count(), - 1, - ) - - @patch( - "openwisp_firmware_upgrader.upgraders.openwrt.OpenWrt.upgrade", - return_value=True, - ) - @patch( - "openwisp_controller.connection.models.DeviceConnection.connect", - return_value=True, - ) - @patch.object(OpenWrt, "SCHEMA", None) - def test_upgrader_with_unsupported_upgrade_options(self, *args): - _org, _category, _build1, build2, _image1, image2, device = self._set_up_env() - self.login() - - with self.subTest("Test DeviceFirmware"): + _, _, _, _, _, image, device = self._set_up_env() + self.login() self.open( "{}#devicefirmware-group".format( reverse( @@ -300,24 +150,68 @@ def test_upgrader_with_unsupported_upgrade_options(self, *args): ) ) self.hide_loading_overlay() - image_select = self._get_device_firmware_dropdown_select() - image_select.select_by_value(str(image2.pk)) - # Ensure JSONSchema editor is not rendered because - # the upgrader does not define a schema + # JSONSchema Editor should not be rendered without a change in the image field self.wait_for_invisibility( By.CSS_SELECTOR, "#devicefirmware-group .jsoneditor-wrapper" ) + image_select = self._get_device_firmware_dropdown_select() + image_select.select_by_value(str(image.pk)) + # JSONSchema configuration editor should not be rendered + self.wait_for_invisibility( + By.XPATH, + '//*[@id="id_devicefirmware-0-upgrade_options_jsoneditor"]/div/h3/span[4]/input', + ) + # Select "None" image should hide JSONSchema Editor + image_select.select_by_value("") + self.wait_for_invisibility( + By.CSS_SELECTOR, "#id_devicefirmware-0-upgrade_options_jsoneditor" + ) + + # Select "build2" image + image_select.select_by_value(str(image.pk)) + # Enable '-c' option self.find_element( - by=By.XPATH, value='//*[@id="device_form"]/div/div[1]/input[3]' + by=By.XPATH, + value='//*[@id="id_devicefirmware-0-upgrade_options_jsoneditor"]' + "/div/div[2]/div/div/div[1]/div/div[1]/label/input", ).click() - self.wait_for_visibility(By.CSS_SELECTOR, "#devicefirmware-group") - self.assertEqual( - UpgradeOperation.objects.filter(upgrade_options={}).count(), 1 + # Enable '-F' option + self.find_element( + by=By.XPATH, + value='//*[@id="id_devicefirmware-0-upgrade_options_jsoneditor"]' + "/div/div[2]/div/div/div[7]/div/div[1]/label/input", + ).click() + save_device() + + # Delete DeviceFirmware + self.find_element(By.CSS_SELECTOR, "#id_devicefirmware-0-DELETE").click() + save_device() + + # When adding firmware to the device for the first time, + # JSONSchema editor should be rendered only when the image + # is selected + self.find_element( + by=By.XPATH, value='//*[@id="devicefirmware-group"]/fieldset/div[2]/a' + ).click() + # JSONSchema Editor should not be rendered without a change in the image field + self.wait_for_invisibility( + By.CSS_SELECTOR, "#devicefirmware-group .jsoneditor-wrapper" ) - DeviceFirmware.objects.all().delete() - UpgradeOperation.objects.all().delete() + image_select = self._get_device_firmware_dropdown_select() + image_select.select_by_index(1) + self.wait_for_visibility( + By.CSS_SELECTOR, "#devicefirmware-group .jsoneditor-wrapper" + ) + save_device() - with self.subTest("Test BatchUpgradeOperation"): + @patch( + "openwisp_firmware_upgrader.upgraders.openwrt.OpenWrt.upgrade", + return_value=True, + ) + def test_batch_upgrade_upgrade_options(self, *args): + with patch(self._mock_connect, return_value=True): + _, _, _, build2, _, _, _ = self._set_up_env() + self.login() self.open( reverse( f"admin:{self.firmware_app_label}_build_change", args=[build2.id] @@ -328,23 +222,133 @@ def test_upgrader_with_unsupported_upgrade_options(self, *args): by=By.CSS_SELECTOR, value='.title-wrapper .object-tools form button[type="submit"]', ).click() - # Ensure JSONSchema editor is not rendered because - # the upgrader does not define a schema + + # Ensure JSONSchema form is rendered + self.wait_for_visibility(By.CSS_SELECTOR, ".jsoneditor-wrapper") + # JSONSchema configuration editor should not be rendered self.wait_for_invisibility( - By.CSS_SELECTOR, "#devicefirmware-group .jsoneditor-wrapper" + By.XPATH, + '//*[@id="id_devicefirmware-0-upgrade_options_jsoneditor"]/div/h3/span[4]/input', ) + # Disable -c flag + self.find_element( + by=By.XPATH, + value='//*[@id="id_upgrade_options_jsoneditor"]' + "/div/div[2]/div/div/div[1]/div/div[1]/label/input", + ).click() + # Enable -n flag + self.find_element( + by=By.XPATH, + value='//*[@id="id_upgrade_options_jsoneditor"]' + "/div/div[2]/div/div/div[3]/div/div[1]/label/input", + ).click() # Upgrade all devices self.find_element( by=By.CSS_SELECTOR, value='input[name="upgrade_all"]' ).click() - self.wait_for_presence(By.ID, "batchupgradeoperation_form") - self.wait_for_presence(By.CSS_SELECTOR, ".messagelist .success") + try: + WebDriverWait(self.web_driver, 5).until( + EC.url_contains("batchupgradeoperation") + ) + except TimeoutException: + self.fail("User was not redirected to Mass upgrade operations page") self.assertEqual( - UpgradeOperation.objects.filter(upgrade_options={}).count(), 1 + BatchUpgradeOperation.objects.filter( + upgrade_options={ + "c": False, + "o": False, + "n": True, + "u": False, + "p": False, + "k": False, + "F": False, + } + ).count(), + 1, ) self.assertEqual( - BatchUpgradeOperation.objects.filter(upgrade_options={}).count(), 1 + UpgradeOperation.objects.filter( + upgrade_options={ + "c": False, + "o": False, + "n": True, + "u": False, + "p": False, + "k": False, + "F": False, + } + ).count(), + 1, + ) + + @patch( + "openwisp_firmware_upgrader.upgraders.openwrt.OpenWrt.upgrade", + return_value=True, + ) + @patch.object(OpenWrt, "SCHEMA", None) + def test_upgrader_with_unsupported_upgrade_options(self, *args): + with patch(self._mock_connect, return_value=True): + _org, _category, _build1, build2, _image1, image2, device = ( + self._set_up_env() ) + self.login() + + with self.subTest("Test DeviceFirmware"): + self.open( + "{}#devicefirmware-group".format( + reverse( + f"admin:{self.config_app_label}_device_change", + args=[device.id], + ) + ) + ) + self.hide_loading_overlay() + image_select = self._get_device_firmware_dropdown_select() + image_select.select_by_value(str(image2.pk)) + # Ensure JSONSchema editor is not rendered because + # the upgrader does not define a schema + self.wait_for_invisibility( + By.CSS_SELECTOR, "#devicefirmware-group .jsoneditor-wrapper" + ) + self.find_element( + by=By.XPATH, value='//*[@id="device_form"]/div/div[1]/input[3]' + ).click() + self.wait_for_visibility(By.CSS_SELECTOR, "#devicefirmware-group") + self.assertEqual( + UpgradeOperation.objects.filter(upgrade_options={}).count(), 1 + ) + DeviceFirmware.objects.all().delete() + UpgradeOperation.objects.all().delete() + + with self.subTest("Test BatchUpgradeOperation"): + self.open( + reverse( + f"admin:{self.firmware_app_label}_build_change", + args=[build2.id] + ) + ) + # Launch mass upgrade operation + self.find_element( + by=By.CSS_SELECTOR, + value='.title-wrapper .object-tools form button[type="submit"]', + ).click() + # Ensure JSONSchema editor is not rendered because + # the upgrader does not define a schema + self.wait_for_invisibility( + By.CSS_SELECTOR, "#devicefirmware-group .jsoneditor-wrapper" + ) + # Upgrade all devices + self.find_element( + by=By.CSS_SELECTOR, value='input[name="upgrade_all"]' + ).click() + self.wait_for_presence(By.ID, "batchupgradeoperation_form") + self.wait_for_presence(By.CSS_SELECTOR, ".messagelist .success") + self.assertEqual( + UpgradeOperation.objects.filter(upgrade_options={}).count(), 1 + ) + self.assertEqual( + BatchUpgradeOperation.objects.filter(upgrade_options={}).count(), 1 + ) def test_upgrade_cancel_modal(self): """Test upgrade cancel modal functionality""" diff --git a/openwisp_firmware_upgrader/tests/test_tasks.py b/openwisp_firmware_upgrader/tests/test_tasks.py index 3908573cd..4d98d7cd2 100644 --- a/openwisp_firmware_upgrader/tests/test_tasks.py +++ b/openwisp_firmware_upgrader/tests/test_tasks.py @@ -18,51 +18,51 @@ class TestTasks(TestUpgraderMixin, TransactionTestCase): _mock_connect = "openwisp_controller.connection.models.DeviceConnection.connect" @mock.patch(_mock_upgrade, side_effect=SoftTimeLimitExceeded()) - @mock.patch(_mock_connect, return_value=True) @mock.patch( "openwisp_firmware_upgrader.base.models.AbstractUpgradeOperation.upgrade", side_effect=SoftTimeLimitExceeded(), ) @capture_any_output() def test_upgrade_firmware_timeout(self, *args): - device_fw = self._create_device_firmware(upgrade=True) - self.assertEqual(UpgradeOperation.objects.count(), 1) - uo = device_fw.image.upgradeoperation_set.first() - self.assertEqual(uo.status, "failed") - self.assertIn("Operation timed out.", uo.log) + with mock.patch(self._mock_connect, return_value=True): + device_fw = self._create_device_firmware(upgrade=True) + self.assertEqual(UpgradeOperation.objects.count(), 1) + uo = device_fw.image.upgradeoperation_set.first() + self.assertEqual(uo.status, "failed") + self.assertIn("Operation timed out.", uo.log) @mock.patch(_mock_upgrade, return_value=True) - @mock.patch(_mock_connect, return_value=True) @mock.patch( "openwisp_firmware_upgrader.base.models.AbstractDeviceFirmware.create_upgrade_operation", side_effect=SoftTimeLimitExceeded(), ) @capture_any_output() def test_batch_upgrade_timeout(self, *args): - env = self._create_upgrade_env() - batch = BatchUpgradeOperation.objects.create(build=env["build2"]) - # will be executed synchronously due to CELERY_IS_EAGER = True - tasks.batch_upgrade_operation.delay(batch_id=batch.pk, firmwareless=False) - self.assertEqual(BatchUpgradeOperation.objects.count(), 1) - batch = BatchUpgradeOperation.objects.first() - self.assertEqual(batch.status, "failed") + with mock.patch(self._mock_connect, return_value=True): + env = self._create_upgrade_env() + batch = BatchUpgradeOperation.objects.create(build=env["build2"]) + # will be executed synchronously due to CELERY_IS_EAGER = True + tasks.batch_upgrade_operation.delay(batch_id=batch.pk, firmwareless=False) + self.assertEqual(BatchUpgradeOperation.objects.count(), 1) + batch = BatchUpgradeOperation.objects.first() + self.assertEqual(batch.status, "failed") @mock.patch(_mock_upgrade, return_value=True) - @mock.patch(_mock_connect, return_value=True) @mock.patch("logging.Logger.warning") def test_upgrade_firmware_resilience(self, mocked_logger, *args): - upgrade_op_id = UpgradeOperation().id - tasks.upgrade_firmware.run(upgrade_op_id) - mocked_logger.assert_called_with( - f"The UpgradeOperation object with id {upgrade_op_id} has been deleted" - ) + with mock.patch(self._mock_connect, return_value=True): + upgrade_op_id = UpgradeOperation().id + tasks.upgrade_firmware.run(upgrade_op_id) + mocked_logger.assert_called_with( + f"The UpgradeOperation object with id {upgrade_op_id} has been deleted" + ) @mock.patch(_mock_upgrade, return_value=True) - @mock.patch(_mock_connect, return_value=True) @mock.patch("logging.Logger.warning") def test_batch_upgrade_operation_resilience(self, mocked_logger, *args): - batch_id = BatchUpgradeOperation().id - tasks.batch_upgrade_operation.run(batch_id=batch_id, firmwareless=False) - mocked_logger.assert_called_with( - f"The BatchUpgradeOperation object with id {batch_id} has been deleted" - ) + with mock.patch(self._mock_connect, return_value=True): + batch_id = BatchUpgradeOperation().id + tasks.batch_upgrade_operation.run(batch_id=batch_id, firmwareless=False) + mocked_logger.assert_called_with( + f"The BatchUpgradeOperation object with id {batch_id} has been deleted" + ) diff --git a/tests/media/.gitignore b/tests/media/.gitignore new file mode 100644 index 000000000..ae48598e8 --- /dev/null +++ b/tests/media/.gitignore @@ -0,0 +1,3 @@ +* +!.gitignore +!floorplan.jpg diff --git a/tests/media/floorplan.jpg b/tests/media/floorplan.jpg new file mode 100644 index 000000000..e69de29bb diff --git a/tests/openwisp2/sample_connection/__init__.py b/tests/openwisp2/sample_connection/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/openwisp2/sample_connection/admin.py b/tests/openwisp2/sample_connection/admin.py new file mode 100644 index 000000000..5e9199435 --- /dev/null +++ b/tests/openwisp2/sample_connection/admin.py @@ -0,0 +1 @@ +from openwisp_controller.connection import admin # noqa diff --git a/tests/openwisp2/sample_connection/api/__init__.py b/tests/openwisp2/sample_connection/api/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/openwisp2/sample_connection/api/views.py b/tests/openwisp2/sample_connection/api/views.py new file mode 100644 index 000000000..411cecd58 --- /dev/null +++ b/tests/openwisp2/sample_connection/api/views.py @@ -0,0 +1,50 @@ +from openwisp_controller.connection.api.views import ( + CommandDetailsView as BaseCommandDetailsView, +) +from openwisp_controller.connection.api.views import ( + CommandListCreateView as BaseCommandListCreateView, +) +from openwisp_controller.connection.api.views import ( + CredentialDetailView as BaseCredentialDetailView, +) +from openwisp_controller.connection.api.views import ( + CredentialListCreateView as BaseCredentialListCreateView, +) +from openwisp_controller.connection.api.views import ( + DeviceConnectionDetailView as BaseDeviceConnectionDetailView, +) +from openwisp_controller.connection.api.views import ( + DeviceConnenctionListCreateView as BaseDeviceConnenctionListCreateView, +) + + +class CommandDetailsView(BaseCommandDetailsView): + pass + + +class CommandListCreateView(BaseCommandListCreateView): + pass + + +class CredentialListCreateView(BaseCredentialListCreateView): + pass + + +class CredentialDetailView(BaseCredentialDetailView): + pass + + +class DeviceConnenctionListCreateView(BaseDeviceConnenctionListCreateView): + pass + + +class DeviceConnectionDetailView(BaseDeviceConnectionDetailView): + pass + + +command_list_create_view = CommandListCreateView.as_view() +command_details_view = CommandDetailsView.as_view() +credential_list_create_view = CredentialListCreateView.as_view() +credential_detail_view = CredentialDetailView.as_view() +deviceconnection_list_create_view = DeviceConnenctionListCreateView.as_view() +deviceconnection_details_view = DeviceConnectionDetailView.as_view() diff --git a/tests/openwisp2/sample_connection/apps.py b/tests/openwisp2/sample_connection/apps.py new file mode 100644 index 000000000..1cd087c9c --- /dev/null +++ b/tests/openwisp2/sample_connection/apps.py @@ -0,0 +1,9 @@ +from openwisp_controller.connection.apps import ConnectionConfig + + +class SampleConnectionConfig(ConnectionConfig): + name = "openwisp2.sample_connection" + label = "sample_connection" + + +del ConnectionConfig diff --git a/tests/openwisp2/sample_connection/migrations/0001_initial.py b/tests/openwisp2/sample_connection/migrations/0001_initial.py new file mode 100644 index 000000000..d84865ad3 --- /dev/null +++ b/tests/openwisp2/sample_connection/migrations/0001_initial.py @@ -0,0 +1,292 @@ +# Generated by Django 3.0.6 on 2020-05-10 18:11 + +import collections +import uuid + +import django +import django.db.models.deletion +import django.utils.timezone +import jsonfield.fields +import model_utils.fields +import swapper +from django.conf import settings +from django.db import migrations, models + +import openwisp_controller.connection.base.models +import openwisp_users.mixins +from openwisp_controller.connection import settings as connection_settings +from openwisp_controller.connection.commands import COMMAND_CHOICES, get_command_choices + + +class Migration(migrations.Migration): + initial = True + + dependencies = [ + ("config", "0001_squashed_0002_config_settings_uuid"), + swapper.dependency( + *swapper.split(settings.AUTH_USER_MODEL), version="0004_default_groups" + ), + swapper.dependency("config", "Device"), + ] + + operations = [ + migrations.CreateModel( + name="Credentials", + fields=[ + ( + "id", + models.UUIDField( + default=uuid.uuid4, + editable=False, + primary_key=True, + serialize=False, + ), + ), + ( + "created", + model_utils.fields.AutoCreatedField( + default=django.utils.timezone.now, + editable=False, + verbose_name="created", + ), + ), + ( + "modified", + model_utils.fields.AutoLastModifiedField( + default=django.utils.timezone.now, + editable=False, + verbose_name="modified", + ), + ), + ("name", models.CharField(db_index=True, max_length=64, unique=True)), + ( + "connector", + models.CharField( + choices=connection_settings.CONNECTORS, + db_index=True, + max_length=128, + verbose_name="connection type", + ), + ), + ( + "params", + jsonfield.fields.JSONField( + default=dict, + dump_kwargs={"indent": 4}, + help_text="global connection parameters", + load_kwargs={"object_pairs_hook": collections.OrderedDict}, + verbose_name="parameters", + ), + ), + ( + "auto_add", + models.BooleanField( + default=False, + help_text=( + "automatically add these credentials to the " + "devices of this organization; if no organization is " + "specified will be added to all the new devices" + ), + verbose_name="auto add", + ), + ), + ("details", models.CharField(blank=True, max_length=64, null=True)), + ( + "organization", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.CASCADE, + to=swapper.get_model_name("openwisp_users", "Organization"), + verbose_name="organization", + ), + ), + ], + options={ + "verbose_name": "Access credentials", + "verbose_name_plural": "Access credentials", + "abstract": False, + }, + bases=( + openwisp_controller.connection.base.models.ConnectorMixin, + openwisp_users.mixins.ValidateOrgMixin, + models.Model, + ), + ), + migrations.CreateModel( + name="DeviceConnection", + fields=[ + ( + "id", + models.UUIDField( + default=uuid.uuid4, + editable=False, + primary_key=True, + serialize=False, + ), + ), + ( + "created", + model_utils.fields.AutoCreatedField( + default=django.utils.timezone.now, + editable=False, + verbose_name="created", + ), + ), + ( + "modified", + model_utils.fields.AutoLastModifiedField( + default=django.utils.timezone.now, + editable=False, + verbose_name="modified", + ), + ), + ( + "update_strategy", + models.CharField( + blank=True, + choices=connection_settings.UPDATE_STRATEGIES, + db_index=True, + help_text="leave blank to determine automatically", + max_length=128, + verbose_name="update strategy", + ), + ), + ("enabled", models.BooleanField(db_index=True, default=True)), + ( + "params", + jsonfield.fields.JSONField( + blank=True, + default=dict, + dump_kwargs={"indent": 4}, + help_text=( + "local connection parameters (will override the " + "global parameters if specified)" + ), + load_kwargs={"object_pairs_hook": collections.OrderedDict}, + verbose_name="parameters", + ), + ), + ( + "is_working", + models.BooleanField(blank=True, default=None, null=True), + ), + ( + "failure_reason", + models.TextField(blank=True, verbose_name="reason of failure"), + ), + ("last_attempt", models.DateTimeField(blank=True, null=True)), + ("details", models.CharField(blank=True, max_length=64, null=True)), + ( + "credentials", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + to="sample_connection.Credentials", + ), + ), + ( + "device", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + to="config.Device", + ), + ), + ], + options={ + "verbose_name": "Device connection", + "verbose_name_plural": "Device connections", + "unique_together": {("device", "credentials")}, + "abstract": False, + }, + bases=( + openwisp_controller.connection.base.models.ConnectorMixin, + models.Model, + ), + ), + migrations.CreateModel( + name="Command", + fields=[ + ( + "id", + models.UUIDField( + default=uuid.uuid4, + editable=False, + primary_key=True, + serialize=False, + ), + ), + ( + "created", + model_utils.fields.AutoCreatedField( + default=django.utils.timezone.now, + editable=False, + verbose_name="created", + ), + ), + ( + "modified", + model_utils.fields.AutoLastModifiedField( + default=django.utils.timezone.now, + editable=False, + verbose_name="modified", + ), + ), + ( + "status", + models.CharField( + choices=[ + ("in-progress", "in progress"), + ("success", "success"), + ("failed", "failed"), + ], + default="in-progress", + max_length=12, + ), + ), + ( + "type", + models.CharField( + choices=( + COMMAND_CHOICES + if django.VERSION < (5, 0) + else get_command_choices + ), + max_length=16, + ), + ), + ( + "input", + jsonfield.fields.JSONField( + blank=True, + dump_kwargs={"indent": 4}, + load_kwargs={"object_pairs_hook": collections.OrderedDict}, + null=True, + ), + ), + ("output", models.TextField(blank=True)), + ( + "connection", + models.ForeignKey( + null=True, + blank=True, + on_delete=django.db.models.deletion.SET_NULL, + to=swapper.get_model_name("connection", "DeviceConnection"), + ), + ), + ( + "device", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + to=swapper.get_model_name("config", "Device"), + ), + ), + ], + options={ + "verbose_name": "Command", + "verbose_name_plural": "Commands", + "ordering": ("created",), + "abstract": False, + "swappable": swapper.swappable_setting("connection", "Command"), + }, + ), + ] diff --git a/tests/openwisp2/sample_connection/migrations/0002_default_group_permissions.py b/tests/openwisp2/sample_connection/migrations/0002_default_group_permissions.py new file mode 100644 index 000000000..469610b20 --- /dev/null +++ b/tests/openwisp2/sample_connection/migrations/0002_default_group_permissions.py @@ -0,0 +1,19 @@ +from django.db import migrations + +from openwisp_controller.connection.migrations import ( + assign_command_permissions_to_groups, + assign_permissions_to_groups, +) + + +class Migration(migrations.Migration): + dependencies = [("sample_connection", "0001_initial")] + + operations = [ + migrations.RunPython( + assign_permissions_to_groups, reverse_code=migrations.RunPython.noop + ), + migrations.RunPython( + assign_command_permissions_to_groups, reverse_code=migrations.RunPython.noop + ), + ] diff --git a/tests/openwisp2/sample_connection/migrations/0003_name_unique_per_organization.py b/tests/openwisp2/sample_connection/migrations/0003_name_unique_per_organization.py new file mode 100644 index 000000000..ff3b8dbed --- /dev/null +++ b/tests/openwisp2/sample_connection/migrations/0003_name_unique_per_organization.py @@ -0,0 +1,20 @@ +# Generated by Django 3.1.6 on 2021-02-11 22:55 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("sample_connection", "0002_default_group_permissions"), + ] + + operations = [ + migrations.AlterField( + model_name="credentials", + name="name", + field=models.CharField(db_index=True, max_length=64), + ), + migrations.AlterUniqueTogether( + name="credentials", unique_together={("name", "organization")} + ), + ] diff --git a/tests/openwisp2/sample_connection/migrations/__init__.py b/tests/openwisp2/sample_connection/migrations/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/openwisp2/sample_connection/models.py b/tests/openwisp2/sample_connection/models.py new file mode 100644 index 000000000..7964c5471 --- /dev/null +++ b/tests/openwisp2/sample_connection/models.py @@ -0,0 +1,29 @@ +from django.db import models + +from openwisp_controller.connection.base.models import ( + AbstractCommand, + AbstractCredentials, + AbstractDeviceConnection, +) + + +class DetailsModel(models.Model): + details = models.CharField(max_length=64, blank=True, null=True) + + class Meta: + abstract = True + + +class Credentials(DetailsModel, AbstractCredentials): + class Meta(AbstractCredentials.Meta): + abstract = False + + +class DeviceConnection(DetailsModel, AbstractDeviceConnection): + class Meta(AbstractDeviceConnection.Meta): + abstract = False + + +class Command(AbstractCommand): + class Meta(AbstractCommand.Meta): + abstract = False diff --git a/tests/openwisp2/sample_connection/pytest.py b/tests/openwisp2/sample_connection/pytest.py new file mode 100644 index 000000000..910268306 --- /dev/null +++ b/tests/openwisp2/sample_connection/pytest.py @@ -0,0 +1,10 @@ +from openwisp_controller.connection.tests.pytest import ( + TestCommandsConsumer as BaseTestCommandsConsumer, +) + + +class TestCommandsConsumer(BaseTestCommandsConsumer): + pass + + +del BaseTestCommandsConsumer diff --git a/tests/openwisp2/sample_connection/tests.py b/tests/openwisp2/sample_connection/tests.py new file mode 100644 index 000000000..946e207c5 --- /dev/null +++ b/tests/openwisp2/sample_connection/tests.py @@ -0,0 +1,92 @@ +from django.urls import reverse +from swapper import load_model + +from openwisp_controller.connection import settings as conn_settings +from openwisp_controller.connection.tests.test_admin import ( + TestCommandInlines as BaseTestCommandInlines, +) +from openwisp_controller.connection.tests.test_admin import ( + TestConnectionAdmin as BaseTestConnectionAdmin, +) +from openwisp_controller.connection.tests.test_api import ( + TestConnectionApi as BaseTestConnectionApi, +) +from openwisp_controller.connection.tests.test_models import ( + TestModels as BaseTestModels, +) +from openwisp_controller.connection.tests.test_models import ( + TestModelsTransaction as BaseTestModelsTransaction, +) +from openwisp_controller.connection.tests.test_notifications import ( + TestNotifications as BaseTestNotifications, +) +from openwisp_controller.connection.tests.test_notifications import ( + TestNotificationTransaction as BaseTestNotificationTransaction, +) +from openwisp_controller.connection.tests.test_ssh import TestSsh as BaseTestSsh +from openwisp_controller.connection.tests.test_tasks import TestTasks as BaseTestTasks + + +class TestConnectionAdmin(BaseTestConnectionAdmin): + config_app_label = "config" + app_label = "sample_connection" + + +class TestCommandInlines(BaseTestCommandInlines): + config_app_label = "config" + + +class TestModels(BaseTestModels): + app_label = "sample_connection" + + +class TestModelsTransaction(BaseTestModelsTransaction): + app_label = "sample_connection" + + +class TestTasks(BaseTestTasks): + pass + + +class TestSsh(BaseTestSsh): + pass + + +Notification = load_model("openwisp_notifications", "Notification") + + +class TestNotifications(BaseTestNotifications): + app_label = "sample_connection" + config_app_label = "config" + + +class TestNotificationTransaction(BaseTestNotificationTransaction): + app_label = "sample_connection" + config_app_label = "config" + + +class TestConnectionApi(BaseTestConnectionApi): + def test_post_deviceconnection_list(self): + d1 = self._create_device() + self._create_config(device=d1) + path = reverse("connection_api:deviceconnection_list", args=(d1.pk,)) + data = { + "credentials": self._get_credentials().pk, + "update_strategy": conn_settings.UPDATE_STRATEGIES[0][0], + "enabled": True, + "failure_reason": "", + } + with self.assertNumQueries(13): + response = self.client.post(path, data, content_type="application/json") + self.assertEqual(response.status_code, 201) + + +del BaseTestCommandInlines +del BaseTestConnectionAdmin +del BaseTestModels +del BaseTestModelsTransaction +del BaseTestSsh +del BaseTestTasks +del BaseTestNotifications +del BaseTestNotificationTransaction +del BaseTestConnectionApi diff --git a/tests/openwisp2/sample_firmware_upgrader/tests.py b/tests/openwisp2/sample_firmware_upgrader/tests.py index 5abbdbefb..07e90d14a 100644 --- a/tests/openwisp2/sample_firmware_upgrader/tests.py +++ b/tests/openwisp2/sample_firmware_upgrader/tests.py @@ -30,6 +30,9 @@ from openwisp_firmware_upgrader.tests.test_private_storage import ( TestPrivateStorage as BaseTestPrivateStorage, ) +from openwisp_firmware_upgrader.tests.test_selenium import ( + TestDeviceAdmin as BaseTestDeviceAdmin, +) from openwisp_firmware_upgrader.tests.test_tasks import TestTasks as BaseTestTasks BatchUpgradeOperation = load_model("BatchUpgradeOperation") @@ -42,6 +45,7 @@ class TestAdmin(BaseTestAdmin): app_label = "sample_firmware_upgrader" + config_app_label = "config" build_list_url = reverse(f"admin:{app_label}_build_changelist") def test_category_details(self): @@ -73,7 +77,9 @@ def test_firmware_image_details(self): def test_device_firmware_details(self): self._login() device_fw = self._create_device_firmware(details="sample device_fw details") - path = reverse("admin:config_device_change", args=[device_fw.device_id]) + path = reverse( + f"admin:{self.config_app_label}_device_change", args=[device_fw.device_id] + ) r = self.client.get(path) self.assertContains( r, @@ -101,13 +107,17 @@ def test_upgrede_operation_details(self): uo = UpgradeOperation.objects.first() uo.details = "Test Upgrade device details" uo.save() - url = reverse("admin:config_device_change", args=[device_fw.device.pk]) + url = reverse( + f"admin:{self.config_app_label}_device_change", args=[device_fw.device.pk] + ) r = self.client.get(url) self.assertContains(r, '
    Test Upgrade device details') class TestAdminTransaction(BaseTestAdminTransaction): app_label = "sample_firmware_upgrader" + config_app_label = "config" + _mock_connect = "openwisp2.sample_connection.models.DeviceConnection.connect" build_list_url = reverse(f"admin:{app_label}_build_changelist") @@ -116,6 +126,7 @@ class TestModels(BaseTestModels): class TestModelsTransaction(BaseTestModelsTransaction): + _mock_connect = "openwisp2.sample_connection.models.DeviceConnection.connect" pass @@ -127,7 +138,15 @@ class TestPrivateStorage(BaseTestPrivateStorage): pass +class TestDeviceAdmin(BaseTestDeviceAdmin): + config_app_label = "config" + firmware_app_label = "sample_firmware_upgrader" + _mock_connect = "openwisp2.sample_connection.models.DeviceConnection.connect" + pass + + class TestTasks(BaseTestTasks): + _mock_connect = "openwisp2.sample_connection.models.DeviceConnection.connect" pass @@ -158,6 +177,7 @@ class TestOrgAPIMixin(BaseTestOrgAPIMixin): del BaseTestModelsTransaction del BaseTestOpenwrtUpgrader del BaseTestPrivateStorage +del BaseTestDeviceAdmin del BaseTestTasks del BaseTestBuildViews del BaseTestCategoryViews diff --git a/tests/openwisp2/settings.py b/tests/openwisp2/settings.py index 96b318a90..00dce81e5 100644 --- a/tests/openwisp2/settings.py +++ b/tests/openwisp2/settings.py @@ -132,7 +132,6 @@ TEMPLATES = [ { "BACKEND": "django.template.backends.django.DjangoTemplates", - "DIRS": [os.path.join(os.path.dirname(BASE_DIR), "templates")], "OPTIONS": { "loaders": [ "django.template.loaders.filesystem.Loader", @@ -240,6 +239,19 @@ "sample_firmware_upgrader.UpgradeOperation" ) + # For controller extended apps: + # Replace Connection + connection_index = INSTALLED_APPS.index("openwisp_controller.connection") + INSTALLED_APPS.remove("openwisp_controller.connection") + INSTALLED_APPS.insert(connection_index, "openwisp2.sample_connection") + # Extended apps + EXTENDED_APPS.append("openwisp_controller.connection") + # Swapper + CONNECTION_CREDENTIALS_MODEL = "sample_connection.Credentials" + CONNECTION_DEVICECONNECTION_MODEL = "sample_connection.DeviceConnection" + CONNECTION_COMMAND_MODEL = "sample_connection.Command" + + TEST_RUNNER = "openwisp_utils.tests.TimeLoggingTestRunner" # local settings must be imported before test runner otherwise they'll be ignored diff --git a/tests/openwisp2/urls.py b/tests/openwisp2/urls.py index 11c91c14b..534d7fbe0 100644 --- a/tests/openwisp2/urls.py +++ b/tests/openwisp2/urls.py @@ -1,3 +1,5 @@ +import os + from django.conf import settings from django.conf.urls.static import static from django.contrib import admin @@ -5,11 +7,36 @@ from django.urls import include, path, reverse_lazy from django.views.generic import RedirectView +from openwisp_controller.connection.api.urls import ( + get_api_urls as get_connection_api_urls, +) from openwisp_users.api.urls import get_api_urls +from .sample_connection.api import views as connection_api_views + redirect_view = RedirectView.as_view(url=reverse_lazy("admin:index")) -urlpatterns = [ +urlpatterns = [] + +if os.environ.get("SAMPLE_APP", False): + urlpatterns += [ + path( + "", + include(("openwisp_controller.config.urls", "config"), namespace="config"), + ), + path( + "api/v1/", + include( + ( + get_connection_api_urls(connection_api_views), + "connection_api", + ), + namespace="connection_api", + ), + ), + ] + +urlpatterns += [ path("admin/", admin.site.urls), path("", redirect_view, name="index"), path("", include("openwisp_controller.urls")),