mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-10 04:06:47 +00:00
5b8793ad4c
The Logs page now records every change an admin makes to a compute's libvirt objects, with the compute's name in the host column: - networks: add, start, stop, delete, autostart, DHCP range, fixed addresses, QoS and XML edits; - storage pools: add, start, stop, delete, autostart, from web and API; volumes: add and delete (web and API), clone and ISO upload; - host interfaces: add, start, stop, delete; - secrets: add, set value (never the value), delete; - network filters: add, clone, delete, XML edits, rules and references. A refused or failed change is a failed row with its reason (libvirt text through safe_text). The volume clone and upload jobs write their row when they end; a refused submission is a failed row at once. A volume row names the volume libvirt created (a dir pool adds .qcow2 or .img). Volume create or delete through the API on an inactive pool (409) is a failed row. Network filters: the rows read "Network filter f1: add" instead of "f1 network filter is created", and their host is the compute's name instead of its hostname. A duplicate name or UUID no longer raises a bare exception, and a filter XML that cannot be parsed no longer ends in UnboundLocalError.
93 lines
4.3 KiB
Python
93 lines
4.3 KiB
Python
"""Network changes write rows on the Logs page. Runs on a private libvirt test driver."""
|
|
|
|
import tempfile
|
|
from unittest.mock import patch
|
|
|
|
import libvirt
|
|
from computes.models import Compute
|
|
from django.contrib.auth import get_user_model
|
|
from django.contrib.messages import get_messages
|
|
from django.test import TestCase
|
|
from django.urls import reverse
|
|
|
|
from logs.models import Logs
|
|
|
|
NODE = """<node>
|
|
<network><name>lab</name><bridge name='virbr9'/>
|
|
<ip address='192.0.2.1' netmask='255.255.255.0'><dhcp><range start='192.0.2.10' end='192.0.2.20'/></dhcp></ip>
|
|
</network>
|
|
</node>"""
|
|
|
|
|
|
def rows():
|
|
return list(Logs.objects.order_by("id").values_list("message", "success"))
|
|
|
|
|
|
class NetworkRowTests(TestCase):
|
|
def setUp(self):
|
|
node = tempfile.NamedTemporaryFile("w", suffix=".xml")
|
|
self.addCleanup(node.close)
|
|
node.write(NODE)
|
|
node.flush()
|
|
self.conn = libvirt.open(f"test://{node.name}")
|
|
self.addCleanup(self.conn.close)
|
|
patcher = patch("vrtManager.connection.connection_manager.get_connection", return_value=self.conn)
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
self.compute = Compute.objects.create(name="c1", hostname="localhost", login="", password="", type=4)
|
|
self.client.force_login(get_user_model().objects.create_superuser("audit-admin", "a@example.com", "pw"))
|
|
Logs.objects.all().delete() # the login's row
|
|
self.url = reverse("network", args=[self.compute.id, "lab"])
|
|
|
|
def post(self, data):
|
|
return self.client.post(self.url, data)
|
|
|
|
def test_network_actions(self):
|
|
xml = self.conn.networkLookupByName("lab").XMLDesc(0)
|
|
self.post({"stop": ""})
|
|
self.post({"modify_dhcp_range": "", "range_start": "192.0.2.30", "range_end": "192.0.2.40", "family": "ipv4"})
|
|
self.post({"start": ""})
|
|
self.post({"set_autostart": ""})
|
|
self.post({"unset_autostart": ""})
|
|
self.post({"modify_fixed_address": "", "name": "web-1", "address": "192.0.2.31",
|
|
"mac": "52:54:00:00:00:31", "family": "ipv4"})
|
|
self.post({"delete_fixed_address": "", "address": "192.0.2.31", "family": "ipv4"})
|
|
self.post({"set_qos": "", "qos_direction": "inbound", "qos_average": "100"})
|
|
self.post({"unset_qos": "", "qos_direction": "inbound"})
|
|
self.post({"edit_network": "", "edit_xml": xml})
|
|
self.post({"stop": ""})
|
|
self.post({"delete": ""})
|
|
self.assertEqual(rows(), [
|
|
("Network lab: stop", True),
|
|
("Network lab: set DHCP range IPV4", True),
|
|
("Network lab: start", True),
|
|
("Network lab: autostart on", True),
|
|
("Network lab: autostart off", True),
|
|
("Network lab: set fixed address 192.0.2.31", True),
|
|
("Network lab: delete fixed address 192.0.2.31", True),
|
|
("Network lab: set QoS inbound", True),
|
|
("Network lab: unset QoS inbound", True),
|
|
("Network lab: change XML", True),
|
|
("Network lab: stop", True),
|
|
("Network lab: delete", True),
|
|
])
|
|
self.assertFalse(Logs.objects.filter(message__contains="<network").exists())
|
|
self.assertEqual(set(Logs.objects.values_list("host", flat=True)), {"c1"})
|
|
|
|
def test_a_refused_network_stop(self):
|
|
error = "Requested operation is not valid: network is in use"
|
|
with patch("vrtManager.network.wvmNetwork.stop", side_effect=libvirt.libvirtError(error)):
|
|
response = self.post({"stop": ""})
|
|
self.assertIn(error, [str(m) for m in get_messages(response.wsgi_request)])
|
|
self.assertEqual(rows(), [(f"Network lab: stop: {error}", False)])
|
|
|
|
def test_network_create(self):
|
|
url = reverse("networks", args=[self.compute.id])
|
|
self.client.post(url, {"create": "", "name": "audit-net", "forward": "nat", "subnet": "198.51.100.0/24",
|
|
"dhcp4": "on"})
|
|
self.client.post(url, {"create": "", "name": "bad name!", "forward": "nat"}) # invalid form
|
|
self.client.post(url, {"create": "", "name": "lab", "forward": "nat", "subnet": "203.0.113.0/24"})
|
|
[first, second] = rows()
|
|
self.assertEqual(first, ("Network audit-net: add", True))
|
|
self.assertFalse(second[1])
|
|
self.assertTrue(second[0].startswith("Network lab: add: "), second[0])
|