feat(api): add discriminators to link iface, appliance version and template settings schemas

This commit is contained in:
Mark Paronyan committed 2026-10-07 14:22:18 +03:00
1 parent 2fb7056cb9
commit a4f20e0966
9 files changed
+202 -12

No files matched your search

@@ -101,3 +101,33 @@ class TestApplianceRoutes:
app.url_path_for("add_appliance_version", appliance_id=appliance_id), json=new_version
)
assert response.status_code == status.HTTP_409_CONFLICT
async def test_add_v8_version_appliance(self, app: FastAPI, client: AsyncClient) -> None:
appliance_id = "1cfdf900-7c30-4cb7-8f03-3f61d2581633" # Empty VM appliance
new_version = {"name": "v8-99G", "category": "guest", "images": {"hda_disk_image": "empty99G.qcow2"}}
response = await client.post(
app.url_path_for("add_appliance_version", appliance_id=appliance_id), json=new_version
)
assert response.status_code == status.HTTP_201_CREATED
assert new_version in response.json()["versions"]
async def test_add_version_appliance_with_registry_version(self, app: FastAPI, client: AsyncClient) -> None:
appliance_id = "1cfdf900-7c30-4cb7-8f03-3f61d2581633" # Empty VM appliance
stored_version = {"name": "explicit-99G", "images": {"hda_disk_image": "empty99G.qcow2"}}
response = await client.post(
app.url_path_for("add_appliance_version", appliance_id=appliance_id),
json=dict(stored_version, registry_version=3),
)
assert response.status_code == status.HTTP_201_CREATED
assert stored_version in response.json()["versions"]
async def test_add_version_appliance_unknown_registry_version(self, app: FastAPI, client: AsyncClient) -> None:
appliance_id = "1cfdf900-7c30-4cb7-8f03-3f61d2581633" # Empty VM appliance
response = await client.post(
app.url_path_for("add_appliance_version", appliance_id=appliance_id),
json={"name": "bad", "registry_version": 7, "images": {"hda_disk_image": "empty99G.qcow2"}},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@@ -376,6 +376,7 @@ class TestLinkRoutes:
assert result["rhost"] == "127.0.0.1"
assert result["rport"] == 30000
assert result["type"] == "udp"
assert result["kind"] == "udp"
async def test_get_ethernet_interface(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
"""
@@ -406,3 +407,4 @@ class TestLinkRoutes:
assert result["node_id"] == cloud_node.id
assert result["interface"] == "eth0"
assert result["type"] == "ethernet"
assert result["kind"] == "ethernet"
+65
View File
@@ -0,0 +1,65 @@
#!/usr/bin/env python
#
# Copyright (C) 2026 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import pytest
from fastapi import FastAPI
pytestmark = pytest.mark.asyncio
@pytest.fixture
def openapi(app: FastAPI) -> dict:
return app.openapi()
def _ref(name: str) -> str:
return f"#/components/schemas/{name}"
def _assert_discriminated(schema: dict, property_name: str, mapping: dict) -> None:
assert [item["$ref"] for item in schema["oneOf"]] == [_ref(name) for name in dict.fromkeys(mapping.values())]
assert schema["discriminator"]["propertyName"] == property_name
assert schema["discriminator"]["mapping"] == {key: _ref(name) for key, name in mapping.items()}
async def test_link_iface_response_is_discriminated(openapi: dict) -> None:
path = "/v3/projects/{project_id}/links/{link_id}/iface"
schema = openapi["paths"][path]["get"]["responses"]["200"]["content"]["application/json"]["schema"]
_assert_discriminated(schema, "kind", {"udp": "UDPPortInfo", "ethernet": "EthernetPortInfo"})
async def test_appliance_version_request_is_discriminated(openapi: dict) -> None:
schema = openapi["paths"]["/v3/appliances/{appliance_id}/version"]["post"]["requestBody"]["content"][
"application/json"
]["schema"]
mapping = {str(version): "ApplianceVersionCreateV1_6" for version in range(1, 7)}
mapping["8"] = "ApplianceVersionCreateV8"
_assert_discriminated(schema, "registry_version", mapping)
async def test_appliance_settings_are_discriminated(openapi: dict) -> None:
schema = openapi["components"]["schemas"]["ApplianceV8"]["properties"]["settings"]["items"]
_assert_discriminated(
schema,
"template_type",
{
"qemu": "QemuTemplateSetting",
"dynamips": "DynamipsTemplateSetting",
"iou": "IouTemplateSetting",
"docker": "DockerTemplateSetting",
},
)
+53 -1
View File
@@ -17,10 +17,21 @@
import pydantic
import pytest
from pydantic import TypeAdapter
from gns3server.controller.appliance import Appliance
from gns3server.controller.appliance_to_template import ApplianceToTemplate
from gns3server.schemas.controller.appliances import ApplianceModel
from gns3server.schemas.controller.appliances import (
ApplianceModel,
ApplianceVersionCreate,
ApplianceVersionCreateV1_6,
ApplianceVersionCreateV8,
DockerTemplateSetting,
DynamipsTemplateSetting,
IouTemplateSetting,
QemuTemplateSetting,
TemplateSetting,
)
# v8 mirror of the XRd Control Plane appliance shape (docker, custom_adapters, no versions)
XRD_V8 = {
@@ -242,3 +253,44 @@ def test_v8_netmiko_device_type_empty_string_clears():
appliance = dict(XRD_V8, netmiko_device_type="")
model = ApplianceModel.model_validate(appliance)
assert model.netmiko_device_type == ""
@pytest.mark.parametrize(
"template_type, properties, expected",
[
("qemu", {"ram": 512}, QemuTemplateSetting),
("dynamips", {"platform": "c7200", "ram": 256}, DynamipsTemplateSetting),
("iou", {"ram": 256}, IouTemplateSetting),
("docker", {"image": "alpine:latest"}, DockerTemplateSetting),
],
)
def test_template_setting_decoded_through_discriminator(template_type, properties, expected):
setting = TypeAdapter(TemplateSetting).validate_python(
{"name": "a", "template_type": template_type, "template_properties": properties}
)
assert type(setting) is expected
assert setting.template_type == template_type
def test_template_setting_unknown_template_type_rejected():
with pytest.raises(pydantic.ValidationError):
TypeAdapter(TemplateSetting).validate_python({"template_type": "vpcs", "template_properties": {}})
@pytest.mark.parametrize(
"data, expected",
[
({"name": "1", "registry_version": 1}, ApplianceVersionCreateV1_6),
({"name": "6", "registry_version": 6}, ApplianceVersionCreateV1_6),
({"name": "8", "registry_version": 8}, ApplianceVersionCreateV8),
({"name": "plain", "images": {"hda_disk_image": "a.qcow2"}}, ApplianceVersionCreateV1_6),
({"name": "v8", "category": "guest"}, ApplianceVersionCreateV8),
],
)
def test_appliance_version_decoded_through_discriminator(data, expected):
assert type(TypeAdapter(ApplianceVersionCreate).validate_python(data)) is expected
def test_appliance_version_unknown_registry_version_rejected():
with pytest.raises(pydantic.ValidationError):
TypeAdapter(ApplianceVersionCreate).validate_python({"name": "x", "registry_version": 7})