Skip to content

Commit d7fc649

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 0873e1d commit d7fc649

File tree

2 files changed

+315
-0
lines changed
  • mfd_network_adapter/network_interface/feature/vlan
  • tests/unit/test_mfd_network_adapter/test_network_interface/test_feature/test_vlan

2 files changed

+315
-0
lines changed

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

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,11 @@
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+
811
from .base import BaseFeatureVLAN
912

1013
logger = logging.getLogger(__name__)
@@ -13,3 +16,102 @@
1316

1417
class LinuxVLAN(BaseFeatureVLAN):
1518
"""Linux class for VLAN feature."""
19+
20+
def get_vlan_ids(self) -> list[int]:
21+
"""
22+
Get all VLAN IDs on Linux network interface via terminal command.
23+
24+
:return: List of VLAN IDs
25+
"""
26+
vlan_ids = []
27+
logger.log(
28+
level=MFD_DEBUG,
29+
msg=f"Getting VLAN IDs on interface {self._interface().name}.",
30+
)
31+
result = self.owner._connection.execute_command(
32+
"ifconfig",
33+
expected_return_codes={0},
34+
).stdout
35+
36+
pattern = str(self._interface().name) + r"\.\d+"
37+
matches = re.findall(pattern, result)
38+
39+
vlan_ids_set = set()
40+
if matches:
41+
for match in matches:
42+
parts = match.split(".")
43+
if len(parts) == 2 and parts[1].isdigit():
44+
vlan_ids_set.add(int(parts[1]))
45+
vlan_ids = list(vlan_ids_set)
46+
logger.log(
47+
level=MFD_DEBUG,
48+
msg=f"VLAN IDs on interface {self._interface().name}: {vlan_ids}.",
49+
)
50+
else:
51+
logger.log(level=MFD_DEBUG, msg=f"No VLANs set on {self._interface().name}.")
52+
53+
return sorted(vlan_ids)
54+
55+
def get_vlan_id(self) -> int:
56+
"""
57+
Get first VLAN ID on Linux network interface via terminal command.
58+
59+
:return: First VLAN ID or 0 if none found
60+
"""
61+
vlan_ids = self.get_vlan_ids()
62+
return vlan_ids[0] if vlan_ids else 0
63+
64+
def add_vlan(self, vlan_id: int) -> bool:
65+
"""
66+
Add VLAN on Linux network interface via terminal command.
67+
68+
:param vlan_id: VLAN ID to create
69+
:return: True if VLAN was added successfully, False otherwise
70+
"""
71+
if vlan_id in self.get_vlan_ids():
72+
logger.log(
73+
level=MFD_DEBUG,
74+
msg=f"VLAN {vlan_id} already exists on interface {self._interface().name}.",
75+
)
76+
return True
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+
return vlan_id in self.get_vlan_ids()
93+
94+
def remove_vlan(self, vlan_id: int = 0) -> bool:
95+
"""
96+
Remove VLAN on Linux network interface via terminal command.
97+
98+
:param vlan_id: VLAN ID to remove, if 0 then first collected VLAN ID will be used
99+
:return: True if VLAN was removed successfully, False otherwise
100+
"""
101+
if vlan_id == 0:
102+
vlan_id = self.get_vlan_id()
103+
if vlan_id == 0:
104+
logger.log(
105+
level=MFD_DEBUG,
106+
msg=f"No VLANs found on interface {self._interface().name}.",
107+
)
108+
return False
109+
logger.log(level=MFD_DEBUG, msg=f"Removing VLAN on interface {self._interface().name}.")
110+
command = (
111+
f"ip link del link {self._interface().name} name {self._interface().name}.{vlan_id} "
112+
f"type vlan id {vlan_id}"
113+
)
114+
115+
self.owner._connection.execute_command(command, expected_return_codes={0})
116+
117+
return vlan_id not in self.get_vlan_ids()

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

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

0 commit comments

Comments
 (0)