Skip to content

Commit 7b8578f

Browse files
KTokarzeKTokarze
committed
feat: Add Linux VLAN management functionality
Add functions add_vlan, remove_vlan, get_vlan_id, get_vlan_ids for Linux VLAN management. Fixes #5 Signed-off-by: Kacper Tokarzewski <[email protected]>
1 parent fa86cad commit 7b8578f

File tree

4 files changed

+305
-2
lines changed

4 files changed

+305
-2
lines changed

‎README.md‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,7 @@ interfaces = owner.get_interfaces()
8888
## Exceptions raised by MFD-Network-Adapter module
8989
- related to module: `NetworkAdapterModuleException`
9090
- related to Network Interface: `InterfaceNameNotFound`, `IPException`, `IPAddressesNotFound`, `NetworkQueuesException`, `RDMADeviceNotFound`, `NumaNodeException`, `DriverInfoNotFound`, `FirmwareVersionNotFound`
91-
- related to NetworkInterface's features: `VirtualizationFeatureException`
91+
- related to NetworkInterface's features: `VirtualizationFeatureException`, `VlanNotFoundException`, `VlanAlreadyExistsException`
9292
-
9393
## Classes
9494

@@ -2335,6 +2335,12 @@ All functions are accesible by
23352335
- `interface.capture.pktcap`
23362336

23372337
#### VLAN
2338+
[Linux]
2339+
- `add_vlan(self, vlan_id: int) -> None:` - Add VLAN interface.
2340+
- `remove_vlan(self, vlan_id: int) -> None:` - Remove VLAN interface.
2341+
- `get_vlan_id(self) -> int:` - Get first collected VLAN ID of the interface.
2342+
- `get_vlan_ids(self) -> List[int]:` - Get all VLAN IDs of the interface.
2343+
23382344
[ESXi]
23392345
- `set_vlan_tpid(self, tpid: str) -> None:`: Set TPID used for VLAN tagging by VFs.
23402346

‎mfd_network_adapter/exceptions.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@
66
class NetworkAdapterModuleException(Exception):
77
"""Handle module exception."""
88

9-
109
class VlanNotFoundException(Exception):
1110
"""Handle errors while parsing VLANs."""
1211

@@ -17,3 +16,7 @@ class NetworkInterfaceIncomparableObject(Exception):
1716

1817
class VirtualFunctionCreationException(Exception):
1918
"""Exception raised when VF creation process fails."""
19+
20+
21+
class VlanAlreadyExistsException(Exception):
22+
"""Exception raised when attempting to create or assign a VLAN that already exists on the network interface."""

‎mfd_network_adapter/network_interface/feature/vlan/linux.py‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,12 @@
33
"""Module for VLAN feature for Linux."""
44

55
import logging
6+
import re
67

78
from mfd_common_libs import add_logging_level, log_levels
9+
from mfd_common_libs.log_levels import MFD_DEBUG
10+
11+
from mfd_network_adapter.exceptions import VlanAlreadyExistsException, VlanNotFoundException
812
from .base import BaseFeatureVLAN
913

1014
logger = logging.getLogger(__name__)
@@ -13,3 +17,93 @@
1317

1418
class LinuxVLAN(BaseFeatureVLAN):
1519
"""Linux class for VLAN feature."""
20+
21+
def get_vlan_ids(self) -> list[int]:
22+
"""
23+
Get all VLAN IDs on Linux network interface via terminal command.
24+
25+
:return: List of VLAN IDs
26+
"""
27+
vlan_ids = []
28+
logger.log(
29+
level=MFD_DEBUG,
30+
msg=f"Getting VLAN IDs on interface {self._interface().name}.",
31+
)
32+
result = self.owner._connection.execute_command(
33+
"ifconfig",
34+
expected_return_codes={0},
35+
).stdout
36+
37+
pattern = str(self._interface().name) + r"\.\d+"
38+
matches = re.findall(pattern, result)
39+
40+
vlan_ids_set = set()
41+
if matches:
42+
for match in matches:
43+
parts = match.split(".")
44+
if len(parts) == 2 and parts[1].isdigit():
45+
vlan_ids_set.add(int(parts[1]))
46+
vlan_ids = list(vlan_ids_set)
47+
logger.log(
48+
level=MFD_DEBUG,
49+
msg=f"VLAN IDs on interface {self._interface().name}: {vlan_ids}.",
50+
)
51+
else:
52+
logger.log(
53+
level=MFD_DEBUG,
54+
msg=f"No VLANs found on interface {self._interface().name}.",
55+
)
56+
57+
return sorted(vlan_ids)
58+
59+
def get_vlan_id(self) -> int:
60+
"""
61+
Get first VLAN ID on Linux network interface via terminal command.
62+
63+
:return: First VLAN ID or 0 if none found
64+
"""
65+
vlan_ids = self.get_vlan_ids()
66+
return vlan_ids[0] if vlan_ids else 0
67+
68+
def add_vlan(self, vlan_id: int) -> None:
69+
"""
70+
Add VLAN on Linux network interface via terminal command.
71+
72+
:param vlan_id: VLAN ID to create
73+
:raises VlanAlreadyExistsException: in case required VLAN interface already exists
74+
"""
75+
if vlan_id in self.get_vlan_ids():
76+
raise VlanAlreadyExistsException(f"VLAN {vlan_id} already exists on interface {self._interface().name}.")
77+
78+
logger.log(
79+
level=MFD_DEBUG,
80+
msg=f"Adding VLAN {vlan_id} on interface {self._interface().name}.",
81+
)
82+
command = (
83+
f"ip link add link {self._interface().name} name {self._interface().name}.{vlan_id} "
84+
f"type vlan id {vlan_id}"
85+
)
86+
self.owner._connection.execute_command(command, expected_return_codes={0})
87+
88+
logger.log(level=MFD_DEBUG, msg=f"Bring up interface {self._interface().name}.{vlan_id}.")
89+
command = f"ip link set dev {self._interface().name}.{vlan_id} up"
90+
self.owner._connection.execute_command(command, expected_return_codes={0})
91+
92+
def remove_vlan(self, vlan_id: int = 0) -> None:
93+
"""
94+
Remove VLAN on Linux network interface via terminal command.
95+
96+
:param vlan_id: VLAN ID to remove, if 0 then first collected VLAN ID will be used
97+
:raises VlanNotFoundException: in case collected VLAN ID is not found
98+
"""
99+
if vlan_id == 0:
100+
vlan_id = self.get_vlan_id()
101+
if vlan_id == 0:
102+
raise VlanNotFoundException(f"No VLANs found on interface {self._interface().name}.")
103+
104+
logger.log(level=MFD_DEBUG, msg=f"Removing VLAN on interface {self._interface().name}.")
105+
command = (
106+
f"ip link del link {self._interface().name} name {self._interface().name}.{vlan_id} "
107+
f"type vlan id {vlan_id}"
108+
)
109+
self.owner._connection.execute_command(command, expected_return_codes={0})

‎tests/unit/test_mfd_network_adapter/test_network_interface/test_feature/test_vlan/test_vlan_linux.py‎

Lines changed: 200 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,200 @@
1+
# Copyright (C) 2025 Intel Corporation
2+
# SPDX-License-Identifier: MIT
3+
from textwrap import dedent
4+
from unittest.mock import call
5+
6+
import pytest
7+
from mfd_connect import RPyCConnection
8+
from mfd_connect.base import ConnectionCompletedProcess
9+
from mfd_typing import PCIAddress, OSName
10+
from mfd_typing.network_interface import LinuxInterfaceInfo
11+
12+
from mfd_network_adapter.exceptions import VlanNotFoundException, VlanAlreadyExistsException
13+
from mfd_network_adapter.network_adapter_owner.linux import LinuxNetworkAdapterOwner
14+
from mfd_network_adapter.network_interface.feature.vlan import LinuxVLAN
15+
from mfd_network_adapter.network_interface.linux import LinuxNetworkInterface
16+
17+
18+
class TestVlanLinux:
19+
@pytest.fixture()
20+
def vlan(self, mocker):
21+
pci_address = PCIAddress(0, 0, 0, 0)
22+
name = "eth1"
23+
mock_connection = mocker.create_autospec(RPyCConnection)
24+
mock_connection.get_os_name.return_value = OSName.LINUX
25+
26+
interface = LinuxNetworkInterface(
27+
connection=mock_connection,
28+
interface_info=LinuxInterfaceInfo(pci_address=pci_address, name=name),
29+
)
30+
31+
mock_owner = mocker.create_autospec(LinuxNetworkAdapterOwner)
32+
mock_owner._connection = mock_connection
33+
34+
vlan = LinuxVLAN(connection=mock_connection, interface=interface)
35+
vlan.owner = mock_owner
36+
vlan._interface = lambda: interface
37+
38+
return vlan
39+
40+
def test_get_vlan_ids_multiple_vlans(self, vlan):
41+
vlan.owner._connection.execute_command.return_value = ConnectionCompletedProcess(
42+
return_code=0,
43+
args="command",
44+
stdout=dedent(
45+
"""
46+
eth0: flags=4163<UP,BROADCAST,RUNNING,MULTICAST> mtu 1500
47+
inet 10.91.218.28 netmask 255.255.252.0 broadcast 10.91.219.255
48+
ether 52:5a:00:5b:da:1c txqueuelen 1000 (Ethernet)
49+
RX packets 6402 bytes 814401 (795.3 KiB)
50+
RX errors 0 dropped 0 overruns 0 frame 0
51+
TX packets 3063 bytes 378535 (369.6 KiB)
52+
TX errors 0 dropped 0 overruns 0 carrier 0 collisions 0
53+
54+
eth1: flags=4163<UP,BROADCAST,RUNNING,MULTICAST> mtu 1500
55+
inet 1.1.10.1 netmask 255.0.0.0 broadcast 0.0.0.0
56+
ether 00:15:5d:24:32:2e txqueuelen 1000 (Ethernet)
57+
RX packets 435 bytes 42015 (41.0 KiB)
58+
RX errors 0 dropped 0 overruns 0 frame 0
59+
TX packets 18 bytes 1947 (1.9 KiB)
60+
TX errors 0 dropped 0 overruns 0 carrier 0 collisions 0
61+
62+
eth1.100: flags=4163<UP,BROADCAST,RUNNING,MULTICAST> mtu 1500
63+
inet6 fe80::215:5dff:fe24:322e prefixlen 64 scopeid 0x20<link>
64+
ether 00:15:5d:24:32:2e txqueuelen 1000 (Ethernet)
65+
RX packets 0 bytes 0 (0.0 B)
66+
RX errors 0 dropped 0 overruns 0 frame 0
67+
TX packets 1 bytes 90 (90.0 B)
68+
TX errors 0 dropped 0 overruns 0 carrier 0 collisions 0")
69+
70+
eth1.200: flags=4163<UP,BROADCAST,RUNNING,MULTICAST> mtu 1500
71+
inet6 fe80::215:5dff:fe24:322e prefixlen 64 scopeid 0x20<link>
72+
ether 00:15:5d:24:32:2e txqueuelen 1000 (Ethernet)
73+
RX packets 0 bytes 0 (0.0 B)
74+
RX errors 0 dropped 0 overruns 0 frame 0
75+
TX packets 1 bytes 90 (90.0 B)
76+
TX errors 0 dropped 0 overruns 0 carrier 0 collisions 0")
77+
78+
eth1.300: flags=4163<UP,BROADCAST,RUNNING,MULTICAST> mtu 1500
79+
inet6 fe80::215:5dff:fe24:322e prefixlen 64 scopeid 0x20<link>
80+
ether 00:15:5d:24:32:2e txqueuelen 1000 (Ethernet)
81+
RX packets 0 bytes 0 (0.0 B)
82+
RX errors 0 dropped 0 overruns 0 frame 0
83+
TX packets 1 bytes 90 (90.0 B)
84+
TX errors 0 dropped 0 overruns 0 carrier 0 collisions 0")
85+
"""
86+
),
87+
stderr="",
88+
)
89+
result = vlan.get_vlan_ids()
90+
vlan.owner._connection.execute_command.assert_called_once_with(
91+
"ifconfig",
92+
expected_return_codes={0},
93+
)
94+
assert result == [100, 200, 300]
95+
96+
def test_get_vlan_ids_no_vlans(self, vlan):
97+
vlan.owner._connection.execute_command.return_value = ConnectionCompletedProcess(
98+
return_code=0,
99+
args="command",
100+
stdout=dedent(
101+
"""
102+
eth0: flags=4163<UP,BROADCAST,RUNNING,MULTICAST> mtu 1500
103+
inet 10.91.218.28 netmask 255.255.252.0 broadcast 10.91.219.255
104+
ether 52:5a:00:5b:da:1c txqueuelen 1000 (Ethernet)
105+
RX packets 6402 bytes 814401 (795.3 KiB)
106+
RX errors 0 dropped 0 overruns 0 frame 0
107+
TX packets 3063 bytes 378535 (369.6 KiB)
108+
TX errors 0 dropped 0 overruns 0 carrier 0 collisions 0
109+
110+
eth1: flags=4163<UP,BROADCAST,RUNNING,MULTICAST> mtu 1500
111+
inet 1.1.10.1 netmask 255.0.0.0 broadcast 0.0.0.0
112+
ether 00:15:5d:24:32:2e txqueuelen 1000 (Ethernet)
113+
RX packets 435 bytes 42015 (41.0 KiB)
114+
RX errors 0 dropped 0 overruns 0 frame 0
115+
TX packets 18 bytes 1947 (1.9 KiB)
116+
TX errors 0 dropped 0 overruns 0 carrier 0 collisions 0
117+
"""
118+
),
119+
stderr="",
120+
)
121+
result = vlan.get_vlan_ids()
122+
assert result == []
123+
124+
def test_get_vlan_id_vlan_exist(self, vlan, mocker):
125+
mocker.patch.object(vlan, "get_vlan_ids", return_value=[100])
126+
result = vlan.get_vlan_id()
127+
assert result == 100
128+
129+
def test_get_vlan_id_vlan_does_not_exist(self, vlan, mocker):
130+
mocker.patch.object(vlan, "get_vlan_ids", return_value=[])
131+
result = vlan.get_vlan_id()
132+
assert result == 0
133+
134+
def test_add_vlan_success(self, vlan, mocker):
135+
vlan.owner._connection.execute_command.return_value = ConnectionCompletedProcess(
136+
return_code=0, args="command", stdout="", stderr=""
137+
)
138+
mocker.patch.object(vlan, "get_vlan_ids", side_effect=[[], [100]])
139+
140+
vlan.add_vlan(100)
141+
142+
expected_calls = [
143+
call(
144+
"ip link add link eth1 name eth1.100 type vlan id 100",
145+
expected_return_codes={0},
146+
),
147+
call("ip link set dev eth1.100 up", expected_return_codes={0}),
148+
]
149+
vlan.owner._connection.execute_command.assert_has_calls(expected_calls)
150+
151+
def test_add_vlan_failure(self, vlan, mocker):
152+
vlan.owner._connection.execute_command.return_value = ConnectionCompletedProcess(
153+
return_code=1, args="command", stdout="", stderr=""
154+
)
155+
vlan.add_vlan(100)
156+
157+
def test_add_vlan_already_exists(self, vlan, mocker):
158+
mocker.patch.object(vlan, "get_vlan_ids", return_value=[100])
159+
160+
with pytest.raises(VlanAlreadyExistsException, match="VLAN 100 already exists on interface eth1"):
161+
vlan.add_vlan(100)
162+
163+
def test_remove_vlan_success(self, vlan, mocker):
164+
mocker.patch.object(vlan, "get_vlan_id", return_value=100)
165+
vlan.owner._connection.execute_command.return_value = ConnectionCompletedProcess(
166+
return_code=0, args="command", stdout="", stderr=""
167+
)
168+
vlan.remove_vlan()
169+
vlan.owner._connection.execute_command.assert_called_once_with(
170+
"ip link del link eth1 name eth1.100 type vlan id 100",
171+
expected_return_codes={0},
172+
)
173+
174+
def test_remove_vlan_failure(self, vlan, mocker):
175+
vlan.owner._connection.execute_command.return_value = ConnectionCompletedProcess(
176+
return_code=1, args="command", stdout="", stderr=""
177+
)
178+
vlan.remove_vlan(101)
179+
180+
def test_remove_vlan_no_vlan_exists(self, vlan, mocker):
181+
mocker.patch.object(vlan, "get_vlan_id", return_value=0)
182+
with pytest.raises(VlanNotFoundException, match="No VLANs found on interface eth1"):
183+
vlan.remove_vlan()
184+
vlan.owner._connection.execute_command.assert_not_called()
185+
186+
def test_remove_specified_vlan_success(self, vlan, mocker):
187+
vlan.owner._connection.execute_command.return_value = ConnectionCompletedProcess(
188+
return_code=0, args="command", stdout="", stderr=""
189+
)
190+
vlan.remove_vlan(vlan_id=101)
191+
vlan.owner._connection.execute_command.assert_called_once_with(
192+
"ip link del link eth1 name eth1.101 type vlan id 101",
193+
expected_return_codes={0},
194+
)
195+
196+
def test_remove_specified_vlan_failure(self, vlan, mocker):
197+
vlan.owner._connection.execute_command.return_value = ConnectionCompletedProcess(
198+
return_code=1, args="command", stdout="", stderr=""
199+
)
200+
vlan.remove_vlan(vlan_id=101)

0 commit comments

Comments
 (0)