Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 18 additions & 3 deletions netbox_agent/network.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,23 @@
VIRTUAL_NET_FOLDER = Path("/sys/devices/virtual/net")


def get_vlan_id(interface, vlan_root=Path("/proc/net/vlan")):
vlan_path = vlan_root / interface
try:
if vlan_path.is_file():
match = re.search(r"\bVID:\s*(\d+)\b", vlan_path.read_text())
if match:
return int(match.group(1))
except OSError:
pass

parts = interface.split(".")
if len(parts) > 1 and parts[1].isdigit():
return int(parts[1])

return None


class Network(object):
def __init__(self, server, *args, **kwargs):
self.nics = []
Expand Down Expand Up @@ -104,9 +121,7 @@ def scan(self):
mac = mac.upper()

mtu = int(open("/sys/class/net/{}/mtu".format(interface), "r").read().strip())
vlan = None
if len(interface.split(".")) > 1:
vlan = int(interface.split(".")[1])
vlan = get_vlan_id(interface)

bonding = False
bonding_slaves = []
Expand Down
23 changes: 23 additions & 0 deletions tests/network.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from netbox_agent.lldp import LLDP
from netbox_agent.network import get_vlan_id
from tests.conftest import parametrize_with_fixtures


Expand Down Expand Up @@ -34,3 +35,25 @@ def test_lldp_parse_with_vlan(fixture):
lldp = LLDP(fixture)
assert lldp.get_switch_vlan("eth0") == {"300": {"pvid": True}}
assert lldp.get_switch_vlan("eth1") == {"300": {}}


def test_get_vlan_id_from_kernel_vlan_info(tmp_path):
(tmp_path / "p6p3.vlan9").write_text("p6p3.vlan9 VID: 9 REORDER_HDR: 1\n")
assert get_vlan_id("p6p3.vlan9", tmp_path) == 9


def test_get_vlan_id_from_numeric_suffix(tmp_path):
assert get_vlan_id("eth0.9", tmp_path) == 9
assert get_vlan_id("eno1.50", tmp_path) == 50
assert get_vlan_id("eth0.9.extra", tmp_path) == 9


def test_get_vlan_id_from_non_numeric_dotted_name(tmp_path):
assert get_vlan_id("foo.bar", tmp_path) is None


def test_get_vlan_id_with_unusable_kernel_data(tmp_path):
(tmp_path / "eth0.9").write_text("unexpected content\n")
(tmp_path / "p6p3.vlan9").write_text("unexpected content\n")
assert get_vlan_id("eth0.9", tmp_path) == 9
assert get_vlan_id("p6p3.vlan9", tmp_path) is None