Fix: Apply fixes on tests

Author: nesitor

src/aleph/vm/conf.py

@@ -168,7 +168,7 @@ class Settings(BaseSettings):
         default=True,
         description="Enable IPv6 forwarding on the host. Required for IPv6 connectivity in VMs.",
     )
-    NFTABLES_CHAIN_PREFIX = "aleph"
+    NFTABLES_CHAIN_PREFIX: str = "aleph"
     USE_NDP_PROXY: bool = Field(
         default=True,
         description="Use the Neighbor Discovery Protocol Proxy to respond to Router Solicitation for instances on IPv6",
@@ -179,16 +179,16 @@ class Settings(BaseSettings):
         description="Method used to resolve the dns server if DNS_NAMESERVERS is not present.",
     )
     DNS_NAMESERVERS: list[str] | None = None
-    DNS_NAMESERVERS_IPV4: list[str] | None
-    DNS_NAMESERVERS_IPV6: list[str] | None
+    DNS_NAMESERVERS_IPV4: list[str] | None = None
+    DNS_NAMESERVERS_IPV6: list[str] | None = None
 
     FIRECRACKER_PATH: Path = Path("/opt/firecracker/firecracker")
     JAILER_PATH: Path = Path("/opt/firecracker/jailer")
     SEV_CTL_PATH: Path = Path("/opt/sevctl")
     LINUX_PATH: Path = Path("/opt/firecracker/vmlinux.bin")
     INIT_TIMEOUT: float = 20.0
 
-    CONNECTOR_URL = Url("http://localhost:4021")
+    CONNECTOR_URL: HttpUrl = HttpUrl("http://localhost:4021")
 
     CACHE_ROOT: Path = Path("/var/cache/aleph/vm")
     MESSAGE_CACHE: Path | None = Field(
@@ -311,8 +311,10 @@ class Settings(BaseSettings):
         description="Identifier used for the 'fake instance' message defined in "
         "examples/instance_message_from_aleph.json",
     )
-    FAKE_INSTANCE_MESSAGE = Path(abspath(join(__file__, "../../../../examples/instance_message_from_aleph.json")))
-    FAKE_INSTANCE_QEMU_MESSAGE = Path(abspath(join(__file__, "../../../../examples/qemu_message_from_aleph.json")))
+    FAKE_INSTANCE_MESSAGE: Path = Path(abspath(join(__file__, "../../../../examples/instance_message_from_aleph.json")))
+    FAKE_INSTANCE_QEMU_MESSAGE: Path = Path(
+        abspath(join(__file__, "../../../../examples/qemu_message_from_aleph.json"))
+    )
 
     CHECK_FASTAPI_VM_ID: str = "63faf8b5db1cf8d965e6a464a0cb8062af8e7df131729e48738342d956f29ace"
     LEGACY_CHECK_FASTAPI_VM_ID: str = "67705389842a0a1b95eaa408b009741027964edc805997475e95c505d642edd8"
@@ -348,7 +350,7 @@ def check(self):
         assert isfile(self.JAILER_PATH), f"File not found {self.JAILER_PATH}"
         assert isfile(self.LINUX_PATH), f"File not found {self.LINUX_PATH}"
         assert self.NETWORK_INTERFACE, "Network interface is not specified"
-        assert self.CONNECTOR_URL.startswith("http://") or self.CONNECTOR_URL.startswith("https://")
+        assert str(self.CONNECTOR_URL).startswith("http://") or str(self.CONNECTOR_URL).startswith("https://")
         if self.ALLOW_VM_NETWORKING:
             assert exists(
                 f"/sys/class/net/{self.NETWORK_INTERFACE}"

src/aleph/vm/controllers/__main__.py

@@ -26,7 +26,7 @@
 def configuration_from_file(path: Path):
     with open(path) as f:
         data = json.load(f)
-        return Configuration.parse_obj(data)
+        return Configuration.model_validate(data)
 
 
 def parse_args(args):

src/aleph/vm/hypervisors/firecracker/config.py

@@ -1,6 +1,6 @@
 from pathlib import Path
 
-from pydantic import BaseModel, ConfigDict, Field, PositiveInt
+from pydantic import BaseModel, ConfigDict, PositiveInt
 
 VSOCK_PATH = "/tmp/v.sock"
 

src/aleph/vm/hypervisors/firecracker/microvm.py

@@ -192,7 +192,7 @@ async def save_configuration_file(self, config: FirecrackerConfig) -> Path:
             if not self.use_jailer
             else open(f"{self.jailer_path}/tmp/config.json", "wb")
         ) as config_file:
-            config_file.write(config.json(by_alias=True, exclude_none=True, indent=4).encode())
+            config_file.write(config.model_dump_json(by_alias=True, exclude_none=True, indent=4).encode())
             config_file.flush()
             config_file_path = Path(config_file.name)
             config_file_path.chmod(0o644)

src/aleph/vm/models.py

@@ -226,7 +226,7 @@ def prepare_gpus(self, available_gpus: list[GpuDevice]) -> None:
         gpus: list[HostGPU] = []
         if self.message.requirements and self.message.requirements.gpu:
             for gpu in self.message.requirements.gpu:
-                gpu = GpuProperties.parse_obj(gpu)
+                gpu = GpuProperties.model_validate(gpu)
                 for available_gpu in available_gpus:
                     if available_gpu.device_id == gpu.device_id:
                         gpus.append(
@@ -453,8 +453,8 @@ async def save(self):
                     vcpus=self.vm.hardware_resources.vcpus,
                     memory=self.vm.hardware_resources.memory,
                     network_tap=self.vm.tap_interface.device_name if self.vm.tap_interface else "",
-                    message=self.message.json(),
-                    original_message=self.original.json(),
+                    message=self.message.model_dump_json(),
+                    original_message=self.original.model_dump_json(),
                     persistent=self.persistent,
                 )
             )
@@ -477,8 +477,8 @@ async def save(self):
                     io_write_bytes=None,
                     vcpus=self.vm.hardware_resources.vcpus,
                     memory=self.vm.hardware_resources.memory,
-                    message=self.message.json(),
-                    original_message=self.original.json(),
+                    message=self.message.model_dump_json(),
+                    original_message=self.original.model_dump_json(),
                     persistent=self.persistent,
                     gpus=json.dumps(self.gpus, default=pydantic_encoder),
                 )

src/aleph/vm/orchestrator/reactor.py

@@ -61,7 +61,7 @@ async def trigger(self, message: AlephMessage):
             for subscription in listener.content.on.message:
                 if subscription_matches(subscription, message):
                     vm_hash = listener.item_hash
-                    event = message.json()
+                    event = message.model_dump_json()
                     # Register the listener in the list of coroutines to run asynchronously:
                     coroutines.append(run_code_on_event(vm_hash, event, self.pubsub, pool=self.pool))
                     break

src/aleph/vm/orchestrator/run.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 import logging
 from typing import Any
 

src/aleph/vm/orchestrator/views/__init__.py

@@ -391,7 +391,7 @@ async def update_allocations(request: web.Request):
         allocation_lock = asyncio.Lock()
     try:
         data = await request.json()
-        allocation = Allocation.parse_obj(data)
+        allocation = Allocation.model_validate(data)
     except ValidationError as error:
         return web.json_response(text=error.json(), status=web.HTTPBadRequest.status_code)
 
@@ -486,7 +486,7 @@ async def notify_allocation(request: web.Request):
     await update_aggregate_settings()
     try:
         data = await request.json()
-        vm_notification = VMNotification.parse_obj(data)
+        vm_notification = VMNotification.model_validate(data)
     except JSONDecodeError:
         return web.HTTPBadRequest(reason="Body is not valid JSON")
     except ValidationError as error:
@@ -631,7 +631,7 @@ async def operate_reserve_resources(request: web.Request, authenticated_sender:
     pool: VmPool = request.app["vm_pool"]
     try:
         data = await request.json()
-        message = InstanceContent.parse_obj(data)
+        message = InstanceContent.model_validate(data)
     except JSONDecodeError:
         return web.HTTPBadRequest(reason="Body is not valid JSON")
     except ValidationError as error:

src/aleph/vm/orchestrator/views/authentication.py

@@ -24,6 +24,7 @@
 from nacl.exceptions import BadSignatureError
 from pydantic import BaseModel, ValidationError, field_validator, model_validator
 from solathon.utils import verify_signature
+from typing_extensions import Self
 
 from aleph.vm.conf import settings
 
@@ -103,7 +104,7 @@ def payload_must_be_hex(cls, v: bytes) -> bytes:
         return bytes.fromhex(v.decode())
 
     @model_validator(mode="after")
-    def check_expiry(values) -> SignedPubKeyHeader:
+    def check_expiry(values) -> Self:
         """Check that the token has not expired"""
         payload = values.payload
         content = SignedPubKeyPayload.model_validate_json(payload)
@@ -112,7 +113,7 @@ def check_expiry(values) -> SignedPubKeyHeader:
         return values
 
     @model_validator(mode="after")
-    def check_signature(values) -> SignedPubKeyHeader:
+    def check_signature(values) -> Self:
         """Check that the signature is valid"""
         signature = values.signature
         payload = values.payload

src/aleph/vm/resources.py

@@ -1,5 +1,6 @@
 import subprocess
 from enum import Enum
+from typing import Optional
 
 from aleph_message.models import HashableModel
 from pydantic import BaseModel, ConfigDict, Field
@@ -27,7 +28,7 @@ class GpuDevice(HashableModel):
     """GPU properties."""
 
     vendor: str = Field(description="GPU vendor name")
-    model: str | None = Field(description="GPU model name on Aleph Network")
+    model: Optional[str] = Field(description="GPU model name on Aleph Network", default=None)
     device_name: str = Field(description="GPU vendor card name")
     device_class: GpuDeviceClass = Field(
         description="GPU device class. Look at https://admin.pci-ids.ucw.cz/read/PD/03"

src/aleph/vm/storage.py

@@ -136,7 +136,7 @@ async def get_latest_amend(item_hash: str) -> str:
     if settings.FAKE_DATA_PROGRAM:
         return item_hash
     else:
-        url = f"{settings.CONNECTOR_URL}/compute/latest_amend/{item_hash}"
+        url = f"{settings.CONNECTOR_URL}compute/latest_amend/{item_hash}"
         async with aiohttp.ClientSession() as session:
             resp = await session.get(url)
             resp.raise_for_status()
@@ -154,7 +154,7 @@ async def get_message(ref: str) -> ProgramMessage | InstanceMessage:
         logger.debug("Using the fake data message")
     else:
         cache_path = (Path(settings.MESSAGE_CACHE) / ref).with_suffix(".json")
-        url = f"{settings.CONNECTOR_URL}/download/message/{ref}"
+        url = f"{settings.CONNECTOR_URL}download/message/{ref}"
         await download_file(url, cache_path)
 
     with open(cache_path) as cache_file:
@@ -190,7 +190,7 @@ async def get_code_path(ref: str) -> Path:
             raise ValueError(msg)
 
     cache_path = Path(settings.CODE_CACHE) / ref
-    url = f"{settings.CONNECTOR_URL}/download/code/{ref}"
+    url = f"{settings.CONNECTOR_URL}download/code/{ref}"
     await download_file(url, cache_path)
     return cache_path
 
@@ -202,7 +202,7 @@ async def get_data_path(ref: str) -> Path:
         return Path(f"{data_dir}.zip")
 
     cache_path = Path(settings.DATA_CACHE) / ref
-    url = f"{settings.CONNECTOR_URL}/download/data/{ref}"
+    url = f"{settings.CONNECTOR_URL}download/data/{ref}"
     await download_file(url, cache_path)
     return cache_path
 
@@ -223,7 +223,7 @@ async def get_runtime_path(ref: str) -> Path:
         return Path(settings.FAKE_DATA_RUNTIME)
 
     cache_path = Path(settings.RUNTIME_CACHE) / ref
-    url = f"{settings.CONNECTOR_URL}/download/runtime/{ref}"
+    url = f"{settings.CONNECTOR_URL}download/runtime/{ref}"
 
     if not cache_path.is_file():
         # File does not exist, download it
@@ -241,7 +241,7 @@ async def get_rootfs_base_path(ref: ItemHash) -> Path:
         return Path(settings.FAKE_INSTANCE_BASE)
 
     cache_path = Path(settings.RUNTIME_CACHE) / ref
-    url = f"{settings.CONNECTOR_URL}/download/runtime/{ref}"
+    url = f"{settings.CONNECTOR_URL}download/runtime/{ref}"
     await download_file(url, cache_path)
     await chown_to_jailman(cache_path)
     return cache_path

src/aleph/vm/utils/__init__.py

@@ -24,9 +24,9 @@
 
 def get_message_executable_content(message_dict: dict) -> ExecutableContent:
     try:
-        return ProgramContent.parse_obj(message_dict)
+        return ProgramContent.model_validate(message_dict)
     except ValueError:
-        return InstanceContent.parse_obj(message_dict)
+        return InstanceContent.model_validate(message_dict)
 
 
 def cors_allow_all(function):

tests/supervisor/test_checkpayment.py

@@ -63,7 +63,7 @@ async def compute_required_flow(executions):
         return 500 * len(executions)
 
     mocker.patch("aleph.vm.orchestrator.tasks.compute_required_flow", compute_required_flow)
-    message = InstanceContent.parse_obj(fake_instance_content)
+    message = InstanceContent.model_validate(fake_instance_content)
 
     hash = "decadecadecadecadecadecadecadecadecadecadecadecadecadecadecadeca"
 
@@ -117,7 +117,7 @@ async def compute_required_flow(executions):
         return 500 * len(executions)
 
     mocker.patch("aleph.vm.orchestrator.tasks.compute_required_flow", compute_required_flow)
-    message = InstanceContent.parse_obj(fake_instance_content)
+    message = InstanceContent.model_validate(fake_instance_content)
 
     hash = "decadecadecadecadecadecadecadecadecadecadecadecadecadecadecadeca"
 
@@ -157,7 +157,7 @@ async def test_not_enough_flow(mocker, fake_instance_content):
     mocker.patch("aleph.vm.orchestrator.tasks.get_stream", return_value=2, autospec=True)
     mocker.patch("aleph.vm.orchestrator.tasks.get_message_status", return_value=MessageStatus.PROCESSED)
     mocker.patch("aleph.vm.orchestrator.tasks.compute_required_flow", return_value=5)
-    message = InstanceContent.parse_obj(fake_instance_content)
+    message = InstanceContent.model_validate(fake_instance_content)
 
     mocker.patch.object(VmExecution, "is_running", new=True)
     mocker.patch.object(VmExecution, "stop", new=mocker.AsyncMock(return_value=False))
@@ -201,7 +201,7 @@ async def get_stream(sender, receiver, chain):
     mocker.patch("aleph.vm.orchestrator.tasks.get_community_wallet_address", return_value=mock_community_wallet_address)
     mocker.patch("aleph.vm.orchestrator.tasks.get_message_status", return_value=MessageStatus.PROCESSED)
     mocker.patch("aleph.vm.orchestrator.tasks.compute_required_flow", return_value=5)
-    message = InstanceContent.parse_obj(fake_instance_content)
+    message = InstanceContent.model_validate(fake_instance_content)
 
     mocker.patch.object(VmExecution, "is_running", new=True)
     mocker.patch.object(VmExecution, "stop", new=mocker.AsyncMock(return_value=False))

tests/supervisor/test_gpu_x_vga_support.py

@@ -4,7 +4,7 @@
 
 from aleph.vm.controllers.configuration import QemuGPU
 from aleph.vm.hypervisors.qemu.qemuvm import QemuVM
-from aleph.vm.resources import GpuDevice, GpuDeviceClass, HostGPU
+from aleph.vm.resources import GpuDevice, GpuDeviceClass
 
 
 class TestGpuXVgaSupport:

tests/supervisor/test_views.py

@@ -26,14 +26,17 @@ async def test_allocation_fails_on_invalid_item_hash(aiohttp_client):
         "/control/allocations", json={"persistent_vms": ["not-an-ItemHash"]}, headers={"X-Auth-Signature": "test"}
     )
     assert response.status == 400
-    assert await response.json() == [
+    response = await response.json()
+    for error in response:
+        error.pop("url", None)
+
+    assert response == [
         {
-            "loc": [
-                "persistent_vms",
-                0,
-            ],
-            "msg": "Could not determine hash type: 'not-an-ItemHash'",
-            "type": "value_error.unknownhash",
+            "loc": ["persistent_vms", 0],
+            "msg": "Value error, Could not determine hash type: 'not-an-ItemHash'",
+            "type": "value_error",
+            "ctx": {"error": "Could not determine hash type: 'not-an-ItemHash'"},
+            "input": "not-an-ItemHash",
         },
     ]
 
@@ -365,7 +368,7 @@ async def test_reserve_resources(aiohttp_client, mocker, mock_app_with_pool):
             "size_mib": 1000,
         },
     }
-    InstanceContent.parse_obj(instance_content)
+    InstanceContent.model_validate(instance_content)
 
     response: web.Response = await client.post("/control/reserve_resources", json=instance_content)
     assert response.status == 200, await response.text()
@@ -479,7 +482,7 @@ async def test_reserve_resources_double_fail(aiohttp_client, mocker, mock_app_wi
             "size_mib": 1000,
         },
     }
-    InstanceContent.parse_obj(instance_content)
+    InstanceContent.model_validate(instance_content)
 
     response: web.Response = await client.post("/control/reserve_resources", json=instance_content)
     assert response.status == 400, await response.text()