test ci/cd

SRV/services/ip_service.py

@@ -0,0 +1,126 @@
+from typing import Dict, Any
+from sdbus_async.networkmanager import (
+    NetworkManager,
+    NetworkManagerSettings,
+    NetworkConnectionSettings,
+    NetworkDeviceGeneric,
+)
+import sdbus
+from sdbus_async.networkmanager.settings import ConnectionProfile
+from services.arp_service import NetworkScanner
+
+
+class IPService:
+    """Сервис проверки доступности IP-адреса и применения настроек сети.
+
+    Предоставляет асинхронные методы для проверки занятости IP-адреса с использованием
+    ARP-сканирования и применения настроек через NetworkManager.
+    """
+
+    def __init__(self):
+        """Инициализирует сервис IP.
+
+        Создаёт объект ARP-сканера и устанавливает системную шину D-Bus как
+        по умолчанию.
+        """
+        self.arp_scanner = NetworkScanner()
+        self.bus = sdbus.sd_bus_open_system()
+        sdbus.set_default_bus(self.bus)
+
+    def _reset_bus(self):
+        """Перезапускает соединение с системной шиной D-Bus.
+
+        Закрывает текущее соединение и создаёт новое, устанавливая его как
+        по умолчанию.
+        """
+        new_bus = sdbus.sd_bus_open_system()
+        sdbus.set_default_bus(new_bus)
+        if hasattr(self, "bus") and self.bus is not None:
+            self.bus = None
+        self.bus = new_bus
+
+    async def check_ip_config(self, args: Dict[str, Any]) -> Dict[str, Any]:
+        """Проверяет доступность IP-адреса перед его применением.
+
+        Выполняет ARP-сканирование сети для проверки, занят ли указанный IP-адрес.
+        Перезапускает шину D-Bus перед выполнением.
+
+        Args:
+            args (Dict[str, Any]): Словарь с данными, включая "ipv4_updates"
+                с полем "ipv4.address-data" для нового IP-адреса.
+
+        Returns:
+            Dict[str, Any]: Словарь с ключами:
+                - "available" (bool): Доступен ли IP-адрес.
+                - "message" (str): Сообщение о результате проверки.
+
+        Raises:
+            Exception: Если произошла ошибка при сканировании ARP.
+        """
+        self._reset_bus()
+        clients = self.arp_scanner.scan_network()
+        new_ip = args["ipv4_updates"]["ipv4.address-data"][0]["address"]
+        if any(client["ip"] == new_ip for client in clients):
+            return {
+                "available": False,
+                "message": f"IP address {new_ip} уже занят."
+            }
+        return {
+            "available": True,
+            "message": "Network profile settings are available for use.",
+        }
+
+    async def apply_ip_config(self, args: Dict[str, Any]) -> None:
+        """Применяет проверенные настройки IP к указанному соединению.
+
+        Перезапускает шину D-Bus, обновляет настройки соединения через NetworkManager,
+        отключает и переподключает устройство к сети.
+
+        Args:
+            args (Dict[str, Any]): Словарь с данными, включая:
+                - "connection_id" (str): Идентификатор соединения.
+                - "ipv4_updates" (Dict): Обновления настроек IPv4.
+
+        Raises:
+            ValueError: Если соединение с указанным ID не найдено.
+            RuntimeError: Если устройство для интерфейса не найдено.
+
+        Note:
+            Требуются права для изменения настроек через NetworkManager.
+        """
+        self._reset_bus()
+        nm = NetworkManager()
+        settings = NetworkManagerSettings()
+        connection_paths = await settings.get_connections_by_id(
+            args["connection_id"])
+
+        if not connection_paths:
+            raise ValueError(f"Подключение {args['connection_id']} не найдено")
+
+        connection = NetworkConnectionSettings(connection_paths[0])
+        profile = await connection.get_profile()
+        settings_dict = profile.to_settings_dict()
+
+        for key_path, value in args["ipv4_updates"].items():
+            section, *props = key_path.split(".")
+            current = settings_dict[section]
+            for prop in props[:-1]:
+                current = current.setdefault(prop, {})
+            current[props[-1]] = value
+
+        new_profile = ConnectionProfile.from_settings_dict(settings_dict)
+        await connection.update(new_profile.to_dbus())
+        print(f"Обновлены настройки для {args['connection_id']}:")
+
+        interface_name = settings_dict["connection"]["interface-name"]
+        devices = await nm.get_devices()
+        for device_path in devices:
+            device = NetworkDeviceGeneric(device_path)
+            if await device.interface == interface_name:
+                await device.disconnect()
+                print(f"Отключён от {interface_name}")
+                await nm.activate_connection(connection=connection_paths[0],device=device_path)
+                print(f"Переподключён к {args['connection_id']}")
+                return
+
+        raise RuntimeError(f"Устройство для интерфейса {interface_name} не найдено")