feat: Add VLAN management for network_interface object

Add VLAN management for network_interface object in Linux OS.
This will allow control over a VLAN interface inside Virtual Machine.

Refs: #5
Signed-off-by: Kacper Tokarzewski <[email protected]>

Author: KTokarze

mfd_network_adapter/network_interface/feature/vlan/linux.py

@@ -5,6 +5,8 @@
 import logging
 
 from mfd_common_libs import add_logging_level, log_levels
+from mfd_common_libs.log_levels import MFD_DEBUG
+
 from .base import BaseFeatureVLAN
 
 logger = logging.getLogger(__name__)
@@ -13,3 +15,91 @@
 
 class LinuxVLAN(BaseFeatureVLAN):
     """Linux class for VLAN feature."""
+
+    def get_vlan_ids(self) -> list[int]:
+        """
+        Get all VLAN IDs on Linux network interface via terminal command.
+
+        :return: List of VLAN IDs
+        """
+        logger.log(
+            level=MFD_DEBUG,
+            msg=f"Getting VLAN IDs on interface {self._interface().name}.",
+        )
+        result = self.owner._connection.execute_command(
+            f"ifconfig | grep -oP '{self._interface().name}.\\d+'",
+            expected_return_codes={0},
+        ).stdout
+
+        vlan_ids = []
+        if result:
+            lines = result.strip().splitlines()
+            for line in lines:
+                if f"{self._interface().name}." in line:
+                    vlan_id_str = line.replace(f"{self._interface().name}.", "")
+                    if vlan_id_str.isdigit():
+                        vlan_id = int(vlan_id_str)
+                        if vlan_id not in vlan_ids:  # Avoid duplicates
+                            vlan_ids.append(vlan_id)
+            logger.log(
+                level=MFD_DEBUG,
+                msg=f"VLAN IDs on interface {self._interface().name}: {vlan_ids}.",
+            )
+        else:
+            logger.log(level=MFD_DEBUG, msg=f"No VLANs set on {self._interface().name}.")
+
+        return vlan_ids
+
+    def get_vlan_id(self) -> int:
+        """
+        Get first VLAN ID on Linux network interface via terminal command.
+
+        :return: First VLAN ID or 0 if none found
+        """
+        vlan_ids = self.get_vlan_ids()
+        return vlan_ids[0] if vlan_ids else 0
+
+    def add_vlan(self, vlan_id: int) -> bool:
+        """
+        Add VLAN on Linux network interface via terminal command.
+
+        :param vlan_id: VLAN ID to create
+        :return: True if VLAN was added successfully, False otherwise
+        """
+        logger.log(
+            level=MFD_DEBUG,
+            msg=f"Adding VLAN {vlan_id} on interface {self._interface().name}.",
+        )
+        command = (
+            f"ip link add link {self._interface().name} name {self._interface().name}.{vlan_id} "
+            f"type vlan id {vlan_id}"
+        )
+
+        self.owner._connection.execute_command(command, expected_return_codes={0})
+
+        return vlan_id in self.get_vlan_ids()
+
+    def remove_vlan(self, vlan_id: int = 0) -> bool:
+        """
+        Remove VLAN on Linux network interface via terminal command.
+
+        :param vlan_id: VLAN ID to remove, if 0 then first collected VLAN ID will be used
+        :return: True if VLAN was removed successfully, False otherwise
+        """
+        if vlan_id == 0:
+            vlan_id = self.get_vlan_id()
+            if vlan_id == 0:
+                logger.log(
+                    level=MFD_DEBUG,
+                    msg=f"No VLANs found on interface {self._interface().name}.",
+                )
+                return False
+        logger.log(level=MFD_DEBUG, msg=f"Removing VLAN on interface {self._interface().name}.")
+        command = (
+            f"ip link del link {self._interface().name} name {self._interface().name}.{vlan_id} "
+            f"type vlan id {vlan_id}"
+        )
+
+        self.owner._connection.execute_command(command, expected_return_codes={0})
+
+        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

@@ -0,0 +1,117 @@
+# Copyright (C) 2025 Intel Corporation
+# SPDX-License-Identifier: MIT
+import pytest
+from mfd_connect import RPyCConnection
+from mfd_connect.base import ConnectionCompletedProcess
+from mfd_typing import PCIAddress, OSName
+from mfd_typing.network_interface import LinuxInterfaceInfo
+
+from mfd_network_adapter.network_adapter_owner.linux import LinuxNetworkAdapterOwner
+from mfd_network_adapter.network_interface.feature.vlan import LinuxVLAN
+from mfd_network_adapter.network_interface.linux import LinuxNetworkInterface
+
+
+class TestVlanLinux:
+    @pytest.fixture()
+    def vlan(self, mocker):
+        pci_address = PCIAddress(0, 0, 0, 0)
+        name = "eth1"
+        mock_connection = mocker.create_autospec(RPyCConnection)
+        mock_connection.get_os_name.return_value = OSName.LINUX
+
+        interface = LinuxNetworkInterface(
+            connection=mock_connection,
+            interface_info=LinuxInterfaceInfo(pci_address=pci_address, name=name),
+        )
+
+        mock_owner = mocker.create_autospec(LinuxNetworkAdapterOwner)
+        mock_owner._connection = mock_connection
+
+        vlan = LinuxVLAN(connection=mock_connection, interface=interface)
+        vlan.owner = mock_owner
+        vlan._interface = lambda: interface
+
+        return vlan
+
+    def test_get_vlan_id_vlan_exist(self, vlan):
+        vlan.owner._connection.execute_command.return_value = ConnectionCompletedProcess(
+            return_code=0, args="command", stdout="eth1.100\n", stderr="stderr"
+        )
+        result = vlan.get_vlan_id()
+        vlan.owner._connection.execute_command.assert_called_once_with(
+            "ifconfig | grep -oP 'eth1.\\d+'",
+            expected_return_codes={0},
+        )
+        assert result == 100
+
+    def test_get_vlan_id_vlan_does_not_exist(self, vlan):
+        vlan.owner._connection.execute_command.return_value = ConnectionCompletedProcess(
+            return_code=0, args="command", stdout="", stderr="stderr"
+        )
+        result = vlan.get_vlan_id()
+        vlan.owner._connection.execute_command.assert_called_once_with(
+            "ifconfig | grep -oP 'eth1.\\d+'",
+            expected_return_codes={0},
+        )
+        assert result == 0
+
+    def test_add_vlan_positive(self, vlan, mocker):
+        mocker.patch.object(vlan, "get_vlan_ids", return_value=[100])
+        result = vlan.add_vlan(100)
+        vlan.owner._connection.execute_command.assert_called_once_with(
+            "ip link add link eth1 name eth1.100 type vlan id 100",
+            expected_return_codes={0},
+        )
+        assert result
+
+    def test_add_vlan_negative(self, vlan, mocker):
+        mocker.patch.object(vlan, "get_vlan_ids", return_value=[])
+        result = vlan.add_vlan(100)
+        vlan.owner._connection.execute_command.assert_called_once_with(
+            "ip link add link eth1 name eth1.100 type vlan id 100",
+            expected_return_codes={0},
+        )
+        assert not result
+
+    def test_remove_vlan_positive(self, vlan, mocker):
+        mocker.patch.object(vlan, "get_vlan_id", return_value=100)
+        mocker.patch.object(vlan, "get_vlan_ids", return_value=[])
+        result = vlan.remove_vlan()
+        vlan.owner._connection.execute_command.assert_called_once_with(
+            "ip link del link eth1 name eth1.100 type vlan id 100",
+            expected_return_codes={0},
+        )
+        assert result
+
+    def test_remove_vlan_negative(self, vlan, mocker):
+        mocker.patch.object(vlan, "get_vlan_id", return_value=100)
+        mocker.patch.object(vlan, "get_vlan_ids", return_value=[100])
+        result = vlan.remove_vlan()
+        vlan.owner._connection.execute_command.assert_called_once_with(
+            "ip link del link eth1 name eth1.100 type vlan id 100",
+            expected_return_codes={0},
+        )
+        assert not result
+
+    def test_remove_vlan_negative_no_vlan(self, vlan, mocker):
+        mocker.patch.object(vlan, "get_vlan_id", return_value=0)
+        result = vlan.remove_vlan()
+        assert not result
+
+    def test_remove_specified_vlan_positive(self, vlan, mocker):
+        mocker.patch.object(vlan, "get_vlan_ids", return_value=[100])
+        result = vlan.remove_vlan(vlan_id=101)
+        vlan.owner._connection.execute_command.assert_called_once_with(
+            "ip link del link eth1 name eth1.101 type vlan id 101",
+            expected_return_codes={0},
+        )
+        assert result
+
+    def test_remove_specified_vlan_negative(self, vlan, mocker):
+        mocker.patch.object(vlan, "get_vlan_ids", return_value=[100, 101])
+        result = vlan.remove_vlan(vlan_id=101)
+        vlan.owner._connection.execute_command.assert_called_once_with(
+            "ip link del link eth1 name eth1.101 type vlan id 101",
+            expected_return_codes={0},
+        )
+        assert not result