diff --git a/appsettings/settings.py b/appsettings/settings.py index d575f37e..c14fc3ab 100644 --- a/appsettings/settings.py +++ b/appsettings/settings.py @@ -1,3 +1,5 @@ +from django.db import DatabaseError + from .models import AppSettings @@ -10,9 +12,8 @@ app_settings = Settings() def get_settings(): try: - entries = AppSettings.objects.all() - except: - pass - + entries = list(AppSettings.objects.all()) + except DatabaseError: + return # keep the values already loaded; a failed read must not crash the request for entry in entries: setattr(app_settings, entry.key, entry.value) diff --git a/appsettings/test_get_settings.py b/appsettings/test_get_settings.py new file mode 100644 index 00000000..3e71c77a --- /dev/null +++ b/appsettings/test_get_settings.py @@ -0,0 +1,36 @@ +"""get_settings() runs on every request (middleware). A database error must not +leave the loop reading an unbound variable (the old bare except did), and must +not crash the request chain over a transient read.""" + +from unittest.mock import patch + +from django.db import DatabaseError +from django.test import TestCase + +from appsettings.models import AppSettings +from appsettings.settings import app_settings, get_settings + + +class GetSettingsTestCase(TestCase): + def test_a_database_error_does_not_raise(self): + app_settings.MARKER = "kept" + with patch("appsettings.settings.AppSettings.objects") as objects: + objects.all.side_effect = DatabaseError("db down") + get_settings() # must not raise UnboundLocalError or DatabaseError + self.assertEqual(app_settings.MARKER, "kept") + + def test_settings_are_loaded_when_the_database_works(self): + AppSettings.objects.update_or_create(key="TEST_MARKER_KEY", defaults={"name": "m", "value": "from-db"}) + try: + get_settings() + self.assertEqual(app_settings.TEST_MARKER_KEY, "from-db") + finally: + if hasattr(app_settings, "TEST_MARKER_KEY"): + delattr(app_settings, "TEST_MARKER_KEY") + + def test_a_query_that_fails_during_iteration_does_not_raise(self): + app_settings.MARKER2 = "kept" + with patch("appsettings.settings.AppSettings.objects") as objects: + objects.all.return_value.__iter__.side_effect = DatabaseError("cursor died") + get_settings() + self.assertEqual(app_settings.MARKER2, "kept") diff --git a/storages/tests.py b/storages/tests.py index 1ba8c046..5bbe057f 100644 --- a/storages/tests.py +++ b/storages/tests.py @@ -841,3 +841,32 @@ class StorageUploadViewTests(TestCase): }) self.assertEqual(res4.status_code, 400) self.assertIn("ISO image already exists", res4.json().get("error", "")) + + +class GetVolumesViewTests(TestCase): + """get_volumes must not answer 200 with an empty list when the pool cannot + be read: the caller could not tell that apart from a pool with no volumes.""" + + def setUp(self): + self.admin = get_user_model().objects.create_superuser( + username="vols_admin", password="password", email="vols_admin@example.com" + ) + self.client.login(request=RequestFactory().get("/"), username="vols_admin", password="password") + self.compute = Compute.objects.create(name="VolCompute", hostname="localhost", type=1, login="u", password="p") + self.url = reverse("volumes", args=[self.compute.id, "default"]) + + @patch("storages.views.wvmStorage") + def test_listing_failure_is_not_a_200(self, mock_wvm): + from libvirt import libvirtError + + mock_wvm.return_value.get_volumes.side_effect = libvirtError("pool is gone") + response = self.client.get(self.url) + self.assertEqual(response.status_code, 400) + self.assertIn("error", response.json()) + + @patch("storages.views.wvmStorage") + def test_listing_success_returns_the_volumes(self, mock_wvm): + mock_wvm.return_value.get_volumes.return_value = ["b.qcow2", "a.qcow2"] + response = self.client.get(self.url) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["vols"], ["a.qcow2", "b.qcow2"]) diff --git a/storages/views.py b/storages/views.py index 88d125d2..8f8dcd73 100644 --- a/storages/views.py +++ b/storages/views.py @@ -332,6 +332,8 @@ def get_volumes(request, compute_id, pool): ) conn.refresh() data["vols"] = sorted(conn.get_volumes()) - except libvirtError: - pass + except libvirtError as err: + # do not answer 200 with an empty list when the listing failed: the + # caller cannot tell "no volumes" from "could not read the pool" + return JsonResponse({"error": str(err)}, status=400) return JsonResponse(data) diff --git a/vrtManager/network.py b/vrtManager/network.py index 51a7d6b6..9c1ef874 100644 --- a/vrtManager/network.py +++ b/vrtManager/network.py @@ -397,7 +397,7 @@ class wvmNetwork(wvmConnect): self.leases = self.net.DHCPLeases() except Exception as e: self.leases = [] - raise f"Error getting {self} DHCP leases: {e}" from e + raise util.OperationError(f"Error getting {self} DHCP leases: {e}") from e def get_dhcp_leases(self): if self.leases is None: diff --git a/vrtManager/test_dhcp_leases.py b/vrtManager/test_dhcp_leases.py new file mode 100644 index 00000000..30ec0c71 --- /dev/null +++ b/vrtManager/test_dhcp_leases.py @@ -0,0 +1,36 @@ +"""refresh_dhcp_leases must raise a real exception (handled as a libvirt error), +not a bare string (which Python turns into TypeError, masking the cause).""" + +import unittest +from unittest.mock import Mock + +from django.conf import settings + +if not settings.configured: + settings.configure(MAC_OUI="52:54:10") + +from vrtManager import util +from vrtManager.network import wvmNetwork + + +def network_with_leases(side_effect): + net = wvmNetwork.__new__(wvmNetwork) + net.leases = None + net.net = Mock() + net.net.DHCPLeases.side_effect = side_effect + return net + + +class RefreshDhcpLeasesTestCase(unittest.TestCase): + def test_a_lease_error_raises_an_operation_error(self): + net = network_with_leases(Exception("no DHCP on this network")) + with self.assertRaises(util.OperationError): + net.refresh_dhcp_leases() + + def test_the_error_keeps_the_cause_and_clears_leases(self): + cause = Exception("boom") + net = network_with_leases(cause) + with self.assertRaises(util.OperationError) as caught: + net.refresh_dhcp_leases() + self.assertIs(caught.exception.__cause__, cause) + self.assertEqual(net.leases, [])