1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 11:16:47 +00:00

fix(appsettings,vrtManager,storages): error handlers that hid the real failure

Reviewing the error handlers outside instances/ turned up three that masked or
swallowed the failure they caught:

- appsettings/settings.get_settings() runs on every request (middleware) and
  loads app_settings from the database. Its bare `except: pass` left `entries`
  unbound on a database error, so the next line raised UnboundLocalError
  instead; it also caught KeyboardInterrupt/SystemExit. It now catches
  DatabaseError and returns, keeping the already-loaded values.
- vrtManager.network.refresh_dhcp_leases raised a formatted string
  (`raise f"..."`), which Python rejects with TypeError, hiding the real
  DHCPLeases failure. It now raises util.OperationError, which the API/HTML
  error handling already renders as a libvirt error, and clears self.leases.
- storages.views.get_volumes caught libvirtError and answered 200 with an empty
  body, so the caller could not tell a failed pool read from an empty pool. It
  now returns 400 with the error.

The other broad handlers reviewed (appsettings/nwfilters/virtsecrets views,
storages/upload cleanup and recovery paths, vrtManager display getters) report
the error or are best-effort and were left as they are.

Tests: get_settings() survives a database error during the query and during
iteration without raising and still loads a real value; refresh_dhcp_leases
raises OperationError (keeping the cause) not TypeError; get_volumes returns 400
on a listing failure and the sorted volumes on success. The tests fail on the
previous code.
This commit is contained in:
catborise 2026-10-07 10:54:21 +03:00
parent 80e11f5eba
commit e84db7542d
6 changed files with 111 additions and 7 deletions

View file

@ -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)

View file

@ -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")

View file

@ -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"])

View file

@ -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)

View file

@ -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:

View file

@ -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, [])