feat: Add Linux VLAN management functionality

Fixes #5

This PR implements VLAN management functionality for Linux network interfaces, providing comprehensive support for creating, listing, and removing VLANs.

# Changes Made #
Added get_vlan_ids(): Retrieves all VLAN IDs configured on a Linux network interface using ifconfig and regex parsing
Added get_vlan_id(): Returns the first VLAN ID found, or 0 if none exist
Added add_vlan(vlan_id): Creates a new VLAN interface using ip link add command
Added remove_vlan(vlan_id): Removes a VLAN interface using ip link del command, with automatic detection if no VLAN ID is specified

# Technical Details #
Uses ifconfig with regex pattern matching to discover existing VLANs
Leverages ip link commands for VLAN creation and deletion
Includes duplicate prevention in VLAN ID collection
Provides comprehensive debug logging for troubleshooting
Returns boolean success indicators for add/remove operations

# Testing #
All methods include validation by re-querying the system state to confirm successful operations.

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
+        """
+        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_set = set()
+        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_ids_set.add(int(vlan_id_str))
+            vlan_ids = list(vlan_ids_set)
+            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