1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-10 20:26:48 +00:00

feat(instances): the address search finds a VM by its MAC address too

The "Find IP" dialog takes a MAC address (aa:bb:..., aa-bb-... or
aabb...) as well as an IP address or a network. A MAC finds the VMs with
a network card of that MAC, shut-off ones too (from their definitions),
with that card's addresses when the VM runs: for a MAC seen on a switch
port or in an external DHCP server's log. A MAC on more than one VM is
marked as a possible conflict. Only the VMs with that MAC are read, and
the card's link-local address, which comes from the MAC, is left out.
This commit is contained in:
catborise 2026-10-24 22:38:52 +03:00
parent a80d093c94
commit ecb850739f
6 changed files with 112 additions and 19 deletions

View file

@ -102,6 +102,14 @@ class ReadAndMatchTests(unittest.TestCase):
with self.subTest(text=text), self.assertRaises(ValueError):
addresses.parse_query(text)
def test_a_mac_is_read_in_its_usual_spellings(self):
for text in ("52:54:00:AA:bb:01", " 52-54-00-aa-bb-01 ", "525400aabb01"):
with self.subTest(text=text):
self.assertEqual(addresses.parse_mac(text), "52:54:00:aa:bb:01")
for text in ("", "52:54:00:aa:bb", "192.0.2.1", "fe80::1", "52:54:00:aa:bb:zz"):
with self.subTest(text=text):
self.assertIsNone(addresses.parse_mac(text))
def test_the_nics_and_the_agent_come_from_the_definition(self):
xml = domain_xml(MAC1, MAC2.upper(), agent="connected")
self.assertEqual(addresses.nic_macs(xml), {MAC1, MAC2})
@ -304,3 +312,32 @@ class InventoryTests(unittest.TestCase):
self.assertEqual(result["unread"], 0)
finally:
hung.set()
class FindMacTests(unittest.TestCase):
"""A MAC finds the VMs with a NIC of that MAC, shut-off ones too, with that NIC's addresses."""
def test_a_running_vm_comes_with_the_nics_addresses(self):
# the NIC's link-local address comes from its MAC: not a finding
dom = FakeDomain({ARP: {"e": iface(MAC1, (0, "192.0.2.10", 24))},
AGENT: {"eth0": iface(MAC1, (0, "192.0.2.10", 24), (1, "fe80::5054:ff:fe00:1", 64)),
"eth1": iface(MAC2, (0, "198.51.100.2", 24))}},
xml=domain_xml(MAC1, MAC2, agent="connected"))
result = addresses.find_mac([addresses.Target(compute("c1", dom), None)], MAC1)
self.assertEqual([(m["mac"], m["address"], m["sources"]) for m in result["matches"]],
[(MAC1, "192.0.2.10", ["ARP", "agent"])])
def test_a_shut_off_vm_is_found_by_its_definition(self):
off = FakeDomain({}, uuid="off", name="vm-off", state=5, xml=domain_xml(MAC2))
(match,) = addresses.find_mac([addresses.Target(compute("c1", off), None)], MAC2)["matches"]
self.assertEqual((match["name"], match["address"], match["sources"], match["conflict"]), ("vm-off", "", [], False))
def test_one_mac_on_two_vms_is_a_conflict(self):
c = compute("c1", vm("a", MAC1, "192.0.2.10"), FakeDomain({}, uuid="b", name="vm-b", state=5, xml=domain_xml(MAC1)))
matches = addresses.find_mac([addresses.Target(c, None)], MAC1)["matches"]
self.assertEqual(sorted((m["uuid"], m["conflict"]) for m in matches), [("a", True), ("b", True)])
def test_only_vms_with_the_mac_are_read(self):
other = FakeDomain({AGENT: libvirtError("must not be asked")}, uuid="other", xml=domain_xml(MAC2, agent="connected"))
result = addresses.find_mac([addresses.Target(compute("c1", vm("a", MAC1, "192.0.2.10"), other), None)], MAC1)
self.assertEqual(([m["uuid"] for m in result["matches"]], result["unread"]), (["a"], 0))

View file

@ -82,12 +82,23 @@ class FindAddressTests(TestCase):
self.assertEqual([m["url"] for m in response.json()["matches"]],
[reverse("instances:instance", args=[i.id]) + "#network" for i in (self.a, twin)])
def test_a_mac_is_looked_up_by_its_nics(self):
self.client.force_login(self.admin)
answer = {"matches": [{**row(UUID_A, "vm-a", self.c1.id), "address": "", "sources": []}],
"checked": 1, "no_agent": 0, "unread": 0, "hosts_down": [], "hosts_unfinished": []}
with patch("instances.addresses.find_mac", return_value=answer) as find_mac, \
patch("instances.addresses.find") as find:
response = self.client.get(reverse("instances:find_address"), {"q": "52-54-00-00-00-01"})
self.assertEqual(find_mac.call_args.args[1], "52:54:00:00:00:01")
find.assert_not_called()
self.assertEqual([(m["vm"], m["address"]) for m in response.json()["matches"]], [("vm-a", "")])
def test_a_bad_query_is_refused_without_libvirt(self):
for q in ("", "abc", "192.0.2.300"):
with self.subTest(q=q):
response, find = self.search(self.admin, q=q)
self.assertEqual(response.status_code, 400)
self.assertIn("192.0.2.0/24", response.json()["error"])
self.assertIn("MAC", response.json()["error"])
find.assert_not_called()
def test_too_many_searches_answer_409(self):