Fix 199 ruff errors

src/aleph/vm/conf.py

@@ -9,15 +9,15 @@
 from os.path import abspath, exists, isdir, isfile, join
 from pathlib import Path
 from subprocess import CalledProcessError, check_output
-from typing import Any, Literal, NewType, Optional, Union
+from typing import Any, Literal, NewType
 
 from aleph_message.models import Chain
 from aleph_message.models.execution.environment import HypervisorType
 from pydantic import BaseSettings, Field, HttpUrl
 from pydantic.env_settings import DotenvType, env_file_sentinel
 from pydantic.typing import StrPath
 
-from aleph.vm.orchestrator.chain import STREAM_CHAINS, ChainInfo
+from aleph.vm.orchestrator.chain import STREAM_CHAINS
 from aleph.vm.utils import (
     check_amd_sev_es_supported,
     check_amd_sev_supported,
@@ -85,7 +85,7 @@ def resolvectl_dns_servers_ipv4(interface: str) -> Iterable[str]:
             yield server
 
 
-def get_default_interface() -> Optional[str]:
+def get_default_interface() -> str | None:
     """Returns the default network interface"""
     with open("/proc/net/route") as f:
         for line in f.readlines():
@@ -117,7 +117,8 @@ def obtain_dns_ips(dns_resolver: DnsResolver, network_interface: str) -> list[st
         return list(resolvectl_dns_servers_ipv4(interface=network_interface))
 
     else:
-        assert False, "No DNS resolve defined, this should never happen."
+        msg = "No DNS resolve defined, this should never happen."
+        raise AssertionError(msg)
 
 
 class Settings(BaseSettings):
@@ -150,7 +151,7 @@ class Settings(BaseSettings):
 
     # Networking does not work inside Docker/Podman
     ALLOW_VM_NETWORKING = True
-    NETWORK_INTERFACE: Optional[str] = None
+    NETWORK_INTERFACE: str | None = None
     IPV4_ADDRESS_POOL = Field(
         default="172.16.0.0/12",
         description="IPv4 address range used to provide networks to VMs.",
@@ -179,8 +180,8 @@ class Settings(BaseSettings):
         description="Use the Neighbor Discovery Protocol Proxy to respond to Router Solicitation for instances on IPv6",
     )
 
-    DNS_RESOLUTION: Optional[DnsResolver] = DnsResolver.detect
-    DNS_NAMESERVERS: Optional[list[str]] = None
+    DNS_RESOLUTION: DnsResolver | None = DnsResolver.detect
+    DNS_NAMESERVERS: list[str] | None = None
 
     FIRECRACKER_PATH = Path("/opt/firecracker/firecracker")
     JAILER_PATH = Path("/opt/firecracker/jailer")
@@ -259,7 +260,7 @@ class Settings(BaseSettings):
     ALLOCATION_TOKEN_HASH = "151ba92f2eb90bce67e912af2f7a5c17d8654b3d29895b042107ea312a7eebda"
 
     ENABLE_QEMU_SUPPORT: bool = Field(default=True)
-    INSTANCE_DEFAULT_HYPERVISOR: Optional[HypervisorType] = Field(
+    INSTANCE_DEFAULT_HYPERVISOR: HypervisorType | None = Field(
         default=HypervisorType.firecracker,  # User Firecracker
         description="Default hypervisor to use on running instances, can be Firecracker or QEmu",
     )
@@ -279,19 +280,17 @@ class Settings(BaseSettings):
 
     # Tests on programs
 
-    FAKE_DATA_PROGRAM: Optional[Path] = None
+    FAKE_DATA_PROGRAM: Path | None = None
     BENCHMARK_FAKE_DATA_PROGRAM = Path(abspath(join(__file__, "../../../../examples/example_fastapi")))
 
     FAKE_DATA_MESSAGE = Path(abspath(join(__file__, "../../../../examples/program_message_from_aleph.json")))
-    FAKE_DATA_DATA: Optional[Path] = Path(abspath(join(__file__, "../../../../examples/data/")))
+    FAKE_DATA_DATA: Path | None = Path(abspath(join(__file__, "../../../../examples/data/")))
     FAKE_DATA_RUNTIME = Path(abspath(join(__file__, "../../../../runtimes/aleph-debian-12-python/rootfs.squashfs")))
-    FAKE_DATA_VOLUME: Optional[Path] = Path(
-        abspath(join(__file__, "../../../../examples/volumes/volume-venv.squashfs"))
-    )
+    FAKE_DATA_VOLUME: Path | None = Path(abspath(join(__file__, "../../../../examples/volumes/volume-venv.squashfs")))
 
     # Tests on instances
 
-    TEST_INSTANCE_ID: Optional[str] = Field(
+    TEST_INSTANCE_ID: str | None = Field(
         default=None,  # TODO: Use a valid item_hash here
         description="Identifier of the instance message used when testing the launch of an instance from the network",
     )
@@ -312,11 +311,11 @@ class Settings(BaseSettings):
 
     # Developer options
 
-    SENTRY_DSN: Optional[str] = None
+    SENTRY_DSN: str | None = None
     SENTRY_TRACES_SAMPLE_RATE: float = Field(ge=0, le=1.0, default=0.1)
-    DEVELOPER_SSH_KEYS: Optional[list[str]] = []
+    DEVELOPER_SSH_KEYS: list[str] | None = []
     # Using an object here forces the value to come from Python code and not from an environment variable.
-    USE_DEVELOPER_SSH_KEYS: Union[Literal[False], object] = False
+    USE_DEVELOPER_SSH_KEYS: Literal[False] | object = False
 
     # Fields
     SENSITIVE_FIELDS: list[str] = Field(
@@ -461,10 +460,10 @@ def display(self) -> str:
 
     def __init__(
         self,
-        _env_file: Optional[DotenvType] = env_file_sentinel,
-        _env_file_encoding: Optional[str] = None,
-        _env_nested_delimiter: Optional[str] = None,
-        _secrets_dir: Optional[StrPath] = None,
+        _env_file: DotenvType | None = env_file_sentinel,
+        _env_file_encoding: str | None = None,
+        _env_nested_delimiter: str | None = None,
+        _secrets_dir: StrPath | None = None,
         **values: Any,
     ) -> None:
         super().__init__(_env_file, _env_file_encoding, _env_nested_delimiter, _secrets_dir, **values)

src/aleph/vm/controllers/__main__.py

@@ -6,7 +6,6 @@
 import sys
 from asyncio.subprocess import Process
 from pathlib import Path
-from typing import Union
 
 from aleph.vm.hypervisors.firecracker.microvm import MicroVM
 from aleph.vm.hypervisors.qemu.qemuvm import QemuVM
@@ -85,7 +84,7 @@ async def execute_persistent_vm(config: Configuration):
     return execution, process
 
 
-async def handle_persistent_vm(config: Configuration, execution: Union[MicroVM, QemuVM], process: Process):
+async def handle_persistent_vm(config: Configuration, execution: MicroVM | QemuVM, process: Process):
     # Catch the terminating signal and send a proper message to the vm to stop it so it close files properly
     loop = asyncio.get_event_loop()
 

src/aleph/vm/controllers/configuration.py

@@ -1,7 +1,6 @@
 import logging
 from enum import Enum
 from pathlib import Path
-from typing import List, Optional, Union
 
 from pydantic import BaseModel
 
@@ -26,26 +25,26 @@ class QemuVMHostVolume(BaseModel):
 
 class QemuVMConfiguration(BaseModel):
     qemu_bin_path: str
-    cloud_init_drive_path: Optional[str]
+    cloud_init_drive_path: str | None
     image_path: str
     monitor_socket_path: Path
     qmp_socket_path: Path
     vcpu_count: int
     mem_size_mb: int
-    interface_name: Optional[str]
-    host_volumes: List[QemuVMHostVolume]
+    interface_name: str | None
+    host_volumes: list[QemuVMHostVolume]
 
 
 class QemuConfidentialVMConfiguration(BaseModel):
     qemu_bin_path: str
-    cloud_init_drive_path: Optional[str]
+    cloud_init_drive_path: str | None
     image_path: str
     monitor_socket_path: Path
     qmp_socket_path: Path
     vcpu_count: int
     mem_size_mb: int
-    interface_name: Optional[str]
-    host_volumes: List[QemuVMHostVolume]
+    interface_name: str | None
+    host_volumes: list[QemuVMHostVolume]
     ovmf_path: Path
     sev_session_file: Path
     sev_dh_cert_file: Path
@@ -61,7 +60,7 @@ class Configuration(BaseModel):
     vm_id: int
     vm_hash: str
     settings: Settings
-    vm_configuration: Union[QemuConfidentialVMConfiguration, QemuVMConfiguration, VMConfiguration]
+    vm_configuration: QemuConfidentialVMConfiguration | QemuVMConfiguration | VMConfiguration
     hypervisor: HypervisorType = HypervisorType.firecracker
 
 

src/aleph/vm/controllers/firecracker/executable.py

@@ -8,7 +8,7 @@
 from multiprocessing import Process, set_start_method
 from os.path import exists, isfile
 from pathlib import Path
-from typing import Generic, Optional, TypeVar
+from typing import Generic, TypeVar
 
 from aiohttp import ClientResponseError
 from aleph_message.models import ExecutableContent, ItemHash
@@ -75,18 +75,18 @@ class HostVolume:
 @dataclass
 class BaseConfiguration:
     vm_hash: ItemHash
-    ip: Optional[str] = None
-    route: Optional[str] = None
+    ip: str | None = None
+    route: str | None = None
     dns_servers: list[str] = field(default_factory=list)
     volumes: list[Volume] = field(default_factory=list)
-    variables: Optional[dict[str, str]] = None
+    variables: dict[str, str] | None = None
 
 
 @dataclass
 class ConfigurationResponse:
     success: bool
-    error: Optional[str] = None
-    traceback: Optional[str] = None
+    error: str | None = None
+    traceback: str | None = None
 
 
 class AlephFirecrackerResources:
@@ -149,14 +149,14 @@ class AlephFirecrackerExecutable(Generic[ConfigurationType], AlephVmControllerIn
     enable_console: bool
     enable_networking: bool
     hardware_resources: MachineResources
-    tap_interface: Optional[TapInterface] = None
+    tap_interface: TapInterface | None = None
     fvm: MicroVM
-    vm_configuration: Optional[ConfigurationType]
-    guest_api_process: Optional[Process] = None
+    vm_configuration: ConfigurationType | None
+    guest_api_process: Process | None = None
     is_instance: bool
     persistent: bool
-    _firecracker_config: Optional[FirecrackerConfig] = None
-    controller_configuration: Optional[Configuration] = None
+    _firecracker_config: FirecrackerConfig | None = None
+    controller_configuration: Configuration | None = None
     support_snapshot: bool
 
     @property
@@ -169,9 +169,9 @@ def __init__(
         vm_hash: ItemHash,
         resources: AlephFirecrackerResources,
         enable_networking: bool = False,
-        enable_console: Optional[bool] = None,
-        hardware_resources: Optional[MachineResources] = None,
-        tap_interface: Optional[TapInterface] = None,
+        enable_console: bool | None = None,
+        hardware_resources: MachineResources | None = None,
+        tap_interface: TapInterface | None = None,
         persistent: bool = False,
         prepare_jailer: bool = True,
     ):

src/aleph/vm/controllers/firecracker/instance.py

@@ -4,7 +4,6 @@
 import logging
 from pathlib import Path
 from tempfile import NamedTemporaryFile
-from typing import Optional, Union
 
 import yaml
 from aleph_message.models import ItemHash
@@ -56,7 +55,7 @@ async def download_all(self):
 class AlephFirecrackerInstance(AlephFirecrackerExecutable):
     vm_configuration: BaseConfiguration
     resources: AlephInstanceResources
-    latest_snapshot: Optional[DiskVolumeSnapshot]
+    latest_snapshot: DiskVolumeSnapshot | None
     is_instance = True
     support_snapshot = False
 
@@ -66,9 +65,9 @@ def __init__(
         vm_hash: ItemHash,
         resources: AlephInstanceResources,
         enable_networking: bool = False,
-        enable_console: Optional[bool] = None,
-        hardware_resources: Optional[MachineResources] = None,
-        tap_interface: Optional[TapInterface] = None,
+        enable_console: bool | None = None,
+        hardware_resources: MachineResources | None = None,
+        tap_interface: TapInterface | None = None,
         prepare_jailer: bool = True,
     ):
         self.latest_snapshot = None
@@ -169,13 +168,13 @@ def _get_hostname(self) -> str:
     def _encode_user_data(self) -> bytes:
         """Creates user data configuration file for cloud-init tool"""
 
-        ssh_authorized_keys: Optional[list[str]]
+        ssh_authorized_keys: list[str] | None
         if settings.USE_DEVELOPER_SSH_KEYS:
             ssh_authorized_keys = settings.DEVELOPER_SSH_KEYS or []
         else:
             ssh_authorized_keys = self.resources.message_content.authorized_keys or []
 
-        config: dict[str, Union[str, bool, list[str]]] = {
+        config: dict[str, str | bool | list[str]] = {
             "hostname": self._get_hostname(),
             "disable_root": False,
             "ssh_pwauth": False,

src/aleph/vm/controllers/firecracker/snapshot_manager.py

@@ -2,7 +2,6 @@
 import logging
 import threading
 from time import sleep
-from typing import Optional
 
 from aleph_message.models import ItemHash
 from schedule import Job, Scheduler
@@ -95,7 +94,7 @@ def run_in_thread(self) -> None:
         )
         job_thread.start()
 
-    async def start_for(self, vm: AlephFirecrackerExecutable, frequency: Optional[int] = None) -> None:
+    async def start_for(self, vm: AlephFirecrackerExecutable, frequency: int | None = None) -> None:
         if not vm.support_snapshot:
             msg = "Snapshots are not implemented for programs."
             raise NotImplementedError(msg)

src/aleph/vm/controllers/firecracker/snapshots.py

@@ -1,6 +1,5 @@
 import logging
 from pathlib import Path
-from typing import Optional
 
 from aleph_message.models import ItemHash
 
@@ -35,7 +34,7 @@ async def upload(self) -> ItemHash:
 
 
 class DiskVolumeSnapshot(DiskVolumeFile):
-    compressed: Optional[CompressedDiskVolumeSnapshot]
+    compressed: CompressedDiskVolumeSnapshot | None
 
     def delete(self) -> None:
         if self.compressed:

src/aleph/vm/controllers/interface.py

@@ -2,8 +2,8 @@
 import logging
 from abc import ABC
 from asyncio.subprocess import Process
-from collections.abc import Coroutine
-from typing import Any, Callable, Optional
+from collections.abc import Callable, Coroutine
+from typing import Any
 
 from aleph_message.models import ItemHash
 from aleph_message.models.execution.environment import MachineResources
@@ -31,26 +31,26 @@ class AlephVmControllerInterface(ABC):
     hardware_resources: MachineResources
     support_snapshot: bool
     """Does this controller support snapshotting"""
-    guest_api_process: Optional[Process] = None
-    tap_interface: Optional[TapInterface] = None
+    guest_api_process: Process | None = None
+    tap_interface: TapInterface | None = None
     """Network interface used for this VM"""
 
-    def get_ip(self) -> Optional[str]:
+    def get_ip(self) -> str | None:
         if self.tap_interface:
             return self.tap_interface.guest_ip.with_prefixlen
         return None
 
-    def get_ip_route(self) -> Optional[str]:
+    def get_ip_route(self) -> str | None:
         if self.tap_interface:
             return str(self.tap_interface.host_ip).split("/", 1)[0]
         return None
 
-    def get_ipv6(self) -> Optional[str]:
+    def get_ipv6(self) -> str | None:
         if self.tap_interface:
             return self.tap_interface.guest_ipv6.with_prefixlen
         return None
 
-    def get_ipv6_gateway(self) -> Optional[str]:
+    def get_ipv6_gateway(self) -> str | None:
         if self.tap_interface:
             return str(self.tap_interface.host_ipv6.ip)
         return None

src/aleph/vm/controllers/qemu/client.py

@@ -16,7 +16,8 @@ class QemuVmClient:
     def __init__(self, vm):
         self.vm = vm
         if not (vm.qmp_socket_path and vm.qmp_socket_path.exists()):
-            raise Exception("VM is not running")
+            msg = "VM is not running"
+            raise Exception(msg)
         client = qmp.QEMUMonitorProtocol(str(vm.qmp_socket_path))
         client.connect()