From efca96570bb846328df105156d492c4f9547752b Mon Sep 17 00:00:00 2001 From: Ben Keith Date: Mon, 8 Sep 2025 08:22:08 -0400 Subject: [PATCH] Static Networking Config Add tools for CRUD operations on a host's static network config. Also add a tool to validate the static network nmstate YAML, and a tool to generate an initial nmstate YAML for the most common configurations used. The API handling is a bit awkward given the MAC to IP mapping but all of that complexity is hidden from the LLM. --- Dockerfile | 1 + pyproject.toml | 9 + server.py | 125 ++++++++++- static_net/__init__.py | 16 ++ static_net/config.py | 106 +++++++++ static_net/template.py | 223 ++++++++++++++++++ tests/test_static_net.py | 474 +++++++++++++++++++++++++++++++++++++++ uv.lock | 29 +++ 8 files changed, 981 insertions(+), 2 deletions(-) create mode 100644 static_net/__init__.py create mode 100644 static_net/config.py create mode 100644 static_net/template.py create mode 100644 tests/test_static_net.py diff --git a/Dockerfile b/Dockerfile index b591e9b..37a505d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -13,6 +13,7 @@ RUN uv sync COPY server.py . COPY service_client ./service_client/ +COPY static_net ./static_net/ COPY metrics ./metrics/ RUN chown -R 1001:0 ${APP_HOME} diff --git a/pyproject.toml b/pyproject.toml index 860d8f9..2fcb4ce 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,9 @@ dependencies = [ "retry>=0.9.2", "types-requests>=2.32.4.20250611", "prometheus_client>=0.22.1", + "pyyaml>=6", + "jinja2>=3.1", + "pydantic>=2", ] [dependency-groups] @@ -22,6 +25,7 @@ dev = [ "pylint>=3.3.7", "pyright>=1.1.402", "ruff>=0.12.1", + "types-pyyaml>=6", ] test = [ "pytest>=8.0.0", @@ -62,3 +66,8 @@ markers = [ "integration: marks tests as integration tests", ] asyncio_mode = "auto" + +[tool.pydocstyle] +add-ignore=[ + "D400" # Stop requiring periods after non-sentences. +] diff --git a/server.py b/server.py index d045209..316093e 100644 --- a/server.py +++ b/server.py @@ -10,16 +10,23 @@ import asyncio from typing import Any, Annotated +from jinja2 import TemplateError import requests import uvicorn from pydantic import Field from assisted_service_client import models from mcp.server.fastmcp import FastMCP - +from metrics import metrics, track_tool_usage, initiate_metrics from service_client import InventoryClient from service_client.logger import log -from metrics import metrics, track_tool_usage, initiate_metrics +from static_net import ( + NMStateTemplateParams, + add_or_replace_static_host_config_yaml, + generate_nmstate_from_template, + remove_static_host_config_by_index, + validate_and_parse_nmstate, +) transport_type = os.environ.get("TRANSPORT", "sse").lower() @@ -441,6 +448,120 @@ async def create_cluster( # pylint: disable=too-many-arguments,too-many-positio return cluster.id +@mcp.tool() +@track_tool_usage() +async def validate_nmstate_yaml(nmstate_yaml: str) -> str: + """ + Validate an nmstate yaml document. + + The yaml should always be validated before submitted to the cluster. + """ + validate_and_parse_nmstate(nmstate_yaml) + return "YAML is valid" + + +@mcp.tool( + description=f""" + Generate an initial nmstate yaml. + + You should call this after gathering information from the user to generate the initial nmstate + yaml. Then you can tweak it as needed. Do not generate nmstate yaml from scratch without calling + this tool. + + Returns: the generated nmstate yaml + + Input param schema: + {NMStateTemplateParams.model_json_schema()} +""" +) +@track_tool_usage() +async def generate_nmstate_yaml(params: NMStateTemplateParams) -> str: + """ + Generate an initial nmstate yaml. + + See the mcp.tool description for more details (we have to use it since we dynamically generate + the input schema for the params). + """ + log.info("Generate nmstate yaml with params: %s", params.model_dump_json(indent=2)) + try: + generated = generate_nmstate_from_template(params) + log.debug("Generated yaml: %s", generated) + return generated + except TemplateError as e: + log.error("Failed to render nmstate template", exc_info=e) + return "ERROR: Failed to generate nmstate yaml" + except Exception as e: + log.error("Exception generating nmstate yaml", exc_info=e) + return "ERROR: Unknown error" + + +@mcp.tool() +@track_tool_usage() +async def alter_static_network_config_nmstate_for_host( + cluster_id: str, index: int | None, new_nmstate_yaml: str | None +) -> str: + """ + Add, replace, or delete the nmstate yaml for a single host. + + Args: + cluster_id (str): The unique identifier of the cluster + index (int | None): The index of the host in the existing static config to replace, or None to append a new host to the end of the config. + new_nmstate_yaml (str): The new nmstate YAML for a host. Leave this None to delete the config at the given index. + + Returns: the updated infra env with the new static network config + """ + client = InventoryClient(get_access_token()) + infra_env_id = await _get_cluster_infra_env_id(client, cluster_id) + infra_env = await client.get_infra_env(infra_env_id) + + if new_nmstate_yaml is None: + if index is None: + raise ValueError("index cannot be null when removing a host yaml") + if not infra_env.static_network_config: + raise ValueError( + "cannot remove host yaml with empty existing static network config" + ) + static_network_config = remove_static_host_config_by_index( + existing_static_network_config=infra_env.static_network_config, index=index + ) + else: + static_network_config = add_or_replace_static_host_config_yaml( + existing_static_network_config=infra_env.static_network_config, + index=index, + new_nmstate_yaml=new_nmstate_yaml, + ) + + result = await client.update_infra_env( + infra_env_id, static_network_config=static_network_config + ) + return result.to_str() + + +@mcp.tool() +@track_tool_usage() +async def list_static_network_config(cluster_id: str) -> str: + """ + List all of the host static network config associated with the given cluster_id. + + Args: + cluster_id (str): The unique identifier of the cluster to configure. + + Returns: + str: A JSON-formatted list of static network config or an error string + """ + client = InventoryClient(get_access_token()) + infra_envs = await client.list_infra_envs(cluster_id) + log.info("Found %d InfraEnvs for cluster %s", len(infra_envs), cluster_id) + + if len(infra_envs) != 1: + log.warning( + "cluster %s has %d infra_envs, expected 1", cluster_id, len(infra_envs) + ) + return "ERROR: this cluster doesn't have exactly 1 infra env, cannot manage static network config" + + return json.dumps(infra_envs[0].get("static_network_config", [])) + + @mcp.tool() @track_tool_usage() async def set_cluster_vips( diff --git a/static_net/__init__.py b/static_net/__init__.py new file mode 100644 index 0000000..e402143 --- /dev/null +++ b/static_net/__init__.py @@ -0,0 +1,16 @@ +"""Static networking related functionality.""" + +from .config import ( + add_or_replace_static_host_config_yaml, + remove_static_host_config_by_index, + validate_and_parse_nmstate, +) +from .template import NMStateTemplateParams, generate_nmstate_from_template + +__all__ = [ + "NMStateTemplateParams", + "add_or_replace_static_host_config_yaml", + "generate_nmstate_from_template", + "remove_static_host_config_by_index", + "validate_and_parse_nmstate", +] diff --git a/static_net/config.py b/static_net/config.py new file mode 100644 index 0000000..516fea3 --- /dev/null +++ b/static_net/config.py @@ -0,0 +1,106 @@ +"""infra env static_network_config field handling""" + +import json +from typing import Any, TypedDict, cast + +import yaml + + +class MacInterfaceMap(TypedDict): + """Maps a NIC to a MAC Address.""" + + logical_nic_name: str + mac_address: str + + +class HostStaticNetworkConfig(TypedDict): + """Matches the structure in the Assisted Installer API.""" + + mac_interface_map: list[MacInterfaceMap] + network_yaml: str + + +def remove_static_host_config_by_index( + existing_static_network_config: str, index: int +) -> list[HostStaticNetworkConfig]: + """Remove a single host's config by index position.""" + config = [ + cast(HostStaticNetworkConfig, d) + for d in json.loads(existing_static_network_config) + ] + if index < 0: + raise IndexError("negative indexes are not allowed") + if index >= len(config): + raise ValueError( + f"static network config only has {len(config)} elements, cannot delete index {index}" + ) + del config[index] + return config + + +def add_or_replace_static_host_config_yaml( + existing_static_network_config: str | None, + index: int | None, + new_nmstate_yaml: str, +) -> list[HostStaticNetworkConfig]: + """Add/update a single host's config by index. + + Raises: + - IndexError: if the index is out of range for the existing config + """ + config: list[HostStaticNetworkConfig] = [] + if existing_static_network_config: + config = [ + cast(HostStaticNetworkConfig, d) + for d in json.loads(existing_static_network_config) + ] + + host_config = _generate_host_static_config(new_nmstate_yaml) + if index is None: + config.append(host_config) + else: + if index < 0: + raise IndexError("negative indexes are not allowed") + if index >= len(config): + raise IndexError( + f"static network config only has {len(config)} elements, cannot replace index {index}" + ) + config[index] = host_config + + return config + + +def _generate_host_static_config(nmstate_yaml: str) -> HostStaticNetworkConfig: + nmstate = validate_and_parse_nmstate(nmstate_yaml) + interfaces = nmstate.get("interfaces") + name_and_mac_list: list[MacInterfaceMap] = [ + { + "mac_address": i.get("mac-address"), + "logical_nic_name": i.get("name"), + } + for i in interfaces + if i.get("mac-address") + ] + if not name_and_mac_list: + raise ValueError("At least one interface must be associated to a MAC Address") + + new_host: HostStaticNetworkConfig = { + "mac_interface_map": name_and_mac_list, + "network_yaml": nmstate_yaml, + } + return new_host + + +def validate_and_parse_nmstate(nmstate_yaml: str) -> Any: + """Validate nmstate yaml and return it parsed. + + Raises: + - ValueError: If the yaml is invalid is some way + """ + # Eventually when nmstate 2.2.51 is released we should be able to validate the nmstate by doing: + # libnmstate.validate(nmstate_yaml) + # For now just make sure it is valid yaml + try: + return yaml.safe_load(nmstate_yaml) + except yaml.YAMLError as e: + raise ValueError("Invalid YAML") from e diff --git a/static_net/template.py b/static_net/template.py new file mode 100644 index 0000000..3f98153 --- /dev/null +++ b/static_net/template.py @@ -0,0 +1,223 @@ +"""nmstate yaml templating logic""" + +from typing import Any, Literal +from ipaddress import IPv4Address + +from pydantic import BaseModel, Field +from jinja2 import Template + + +class RouteParams(BaseModel): + """The routes config in nmstate yaml""" + + destination: str = Field( + "0.0.0.0/0", description="The destination addreses for which this route applies" + ) + next_hop_address: str = Field( + description="The IP address to which to route traffic for this route" + ) + next_hop_interface: str = Field( + description="The interface name over which traffic should be routed" + ) + table_id: int | None = Field( + 254, description="The routing table id to add this route to" + ) + + +class IPV4AddressWithSubnet(BaseModel): + """IPv4 address config for nmstate yaml""" + + address: IPv4Address + cidr_length: int = Field(ge=0, le=32) + + +class EthernetInterfaceParams(BaseModel): + """Ethernet interface config for nmstate yaml""" + + mac_address: str + name: str = Field( + description="Use a unique name like eth0, eth1, etc. if the user doesn't supply it" + ) + ipv4_address: IPV4AddressWithSubnet | None = None + + +class BondInterfaceParams(BaseModel): + """Bond interface config for nmstate yaml""" + + name: str = Field( + description="Use a unique name like bond0, bond1, etc. if the user doesn't supply it" + ) + ipv4_address: IPV4AddressWithSubnet | None = Field( + None, + description="The port interfaces should not have IP addresses configured on them, only the bond interface.", + ) + mode: Literal[ + "balance-rr", + "active-backup", + "balance-xor", + "broadcast", + "802.3ad", + "balance-tlb", + "balance-alb", + ] = "active-backup" + port_interface_names: list[str] = Field( + description="The interface names that are aggregated for this bond." + ) + options: dict[str, Any] | None = Field( + None, description="Link aggregation options for the bond interface" + ) + + +class VLANInterfaceParams(BaseModel): + """VLAN config for nmstate yaml""" + + name: str = Field( + description="Use a unique name like vlan0, vlan1, etc. if the user doesn't supply it" + ) + ipv4_address: IPV4AddressWithSubnet | None = Field( + None, + description="If the user supplies an IP address for the vlan interface, don't reuse that same address on the base ethernet interface", + ) + vlan_id: int + base_interface_name: str = Field( + description="If there is only one other ethernet interface configured for this host, use that interface name. Generally this base interface will not have an ip address configured, only the vlan interface." + ) + + +class DNSParams(BaseModel): + """DNS config for nmstate yaml""" + + dns_servers: list[str] = Field( + min_length=1, description="A list of DNS server IP addresses" + ) + dns_search_domains: list[str] | None = Field( + None, description="An optional list of DNS search domain names" + ) + + +class NMStateTemplateParams(BaseModel): + """Top level params for generating nmstate yaml""" + + dns: DNSParams | None = None + routes: list[RouteParams] | None = Field( + None, description="A list of route table rules" + ) + bond_ifaces: list[BondInterfaceParams] | None = Field( + None, description="Configuration for bonded interfaces" + ) + vlan_ifaces: list[VLANInterfaceParams] | None = Field( + None, description="Configuration of vlan interfaces" + ) + ethernet_ifaces: list[EthernetInterfaceParams] = Field( + min_length=1, + description="List of the ethernet interfaces on the machine, at least one is required.", + ) + + +def generate_nmstate_from_template(params: NMStateTemplateParams) -> str: + """Generate the nmstate yaml based on the params""" + return NMSTATE_TEMPLATE.render(params) + + +NMSTATE_TEMPLATE = Template( + """ +{% if dns %} +dns-resolver: + config: + server: + {% for s in dns.dns_servers %} + - {{ s }} + {% endfor %} + {% if dns.dns_search_domains %} + search: + {% for d in dns.dns_search_domains %} + - {{ d }} + {% endfor %} + {% endif %} +{% endif %} +{% if routes %} +routes: + config: + {% for r in routes %} + - destination: {{r.destination}} + next-hop-address: {{r.next_hop_address}} + next-hop-interface: {{r.next_hop_interface}} + {% if r.table_id is not none %} + table-id: {{r.table_id}} + {% endif %} + {% endfor %} +{% endif %} +interfaces: +{% for i in ethernet_ifaces %} +- name: {{i.name}} + type: ethernet + state: up + mac-address: {{i.mac_address}} + ipv4: + {% if i.ipv4_address %} + address: + - ip: {{i.ipv4_address.address}} + prefix-length: {{i.ipv4_address.cidr_length}} + enabled: true + {% else %} + enabled: false + {% endif %} + dhcp: false + ipv6: + enabled: false +{% endfor %} +{% if vlan_ifaces %} +{% for i in vlan_ifaces %} +- name: {{i.name}} + type: vlan + state: up + ipv4: + {% if i.ipv4_address %} + address: + - ip: {{i.ipv4_address.address}} + prefix-length: {{i.ipv4_address.cidr_length}} + enabled: true + {% else %} + enabled: false + {% endif %} + dhcp: false + ipv6: + enabled: false + vlan: + base-iface: {{i.base_interface_name}} + id: {{i.vlan_id}} +{% endfor %} +{% endif %} +{% if bond_ifaces %} +{% for i in bond_ifaces %} +- name: {{i.name}} + type: bond + state: up + ipv4: + {% if i.ipv4_address %} + address: + - ip: {{i.ipv4_address.address}} + prefix-length: {{i.ipv4_address.cidr_length}} + enabled: true + {% else %} + enabled: false + {% endif %} + dhcp: false + ipv6: + enabled: false + link-aggregation: + mode: {{i.mode}} + port: + {% for p in i.port_interface_names %} + - {{p}} + {% endfor %} + {% if i.options %} + options: + {% for k, v in i.options.items() %} + {{k}}: {{v}} + {% endfor %} + {% endif %} +{% endfor %} +{% endif %} +""" +) diff --git a/tests/test_static_net.py b/tests/test_static_net.py new file mode 100644 index 0000000..25e67c0 --- /dev/null +++ b/tests/test_static_net.py @@ -0,0 +1,474 @@ +# type: ignore +""" +Unit tests for the static_net module. +""" + +import json +from ipaddress import IPv4Address + +import pytest +import yaml + +from static_net import ( + remove_static_host_config_by_index, + add_or_replace_static_host_config_yaml, + validate_and_parse_nmstate, + generate_nmstate_from_template, + NMStateTemplateParams, +) + +from static_net.template import ( + EthernetInterfaceParams, + IPV4AddressWithSubnet, + RouteParams, + DNSParams, + BondInterfaceParams, + VLANInterfaceParams, +) + + +class TestRemoveStaticHostConfigByIndex: + """Test the remove_static_host_config_by_index function.""" + + def test_remove_valid_index(self): + """Test removing a host config at a valid index.""" + config_data = [ + { + "mac_interface_map": [ + {"logical_nic_name": "eth0", "mac_address": "00:11:22:33:44:55"} + ], + "network_yaml": "interfaces:\n- name: eth0\n type: ethernet", + }, + { + "mac_interface_map": [ + {"logical_nic_name": "eth1", "mac_address": "00:11:22:33:44:66"} + ], + "network_yaml": "interfaces:\n- name: eth1\n type: ethernet", + }, + ] + existing_config = json.dumps(config_data) + + result = remove_static_host_config_by_index(existing_config, 0) + + assert len(result) == 1 + assert result[0]["mac_interface_map"][0]["logical_nic_name"] == "eth1" + + def test_remove_last_item(self): + """Test removing the last item in the config.""" + config_data = [ + { + "mac_interface_map": [ + {"logical_nic_name": "eth0", "mac_address": "00:11:22:33:44:55"} + ], + "network_yaml": "interfaces:\n- name: eth0\n type: ethernet", + } + ] + existing_config = json.dumps(config_data) + + result = remove_static_host_config_by_index(existing_config, 0) + + assert len(result) == 0 + + def test_remove_invalid_index_too_high(self): + """Test removing with an index that's too high.""" + config_data = [ + { + "mac_interface_map": [ + {"logical_nic_name": "eth0", "mac_address": "00:11:22:33:44:55"} + ], + "network_yaml": "interfaces:\n- name: eth0\n type: ethernet", + } + ] + existing_config = json.dumps(config_data) + + with pytest.raises( + ValueError, + match="static network config only has 1 elements, cannot delete index 5", + ): + remove_static_host_config_by_index(existing_config, 5) + + def test_remove_from_empty_config(self): + """Test removing from an empty config.""" + existing_config = json.dumps([]) + + with pytest.raises( + ValueError, + match="static network config only has 0 elements, cannot delete index 0", + ): + remove_static_host_config_by_index(existing_config, 0) + + +class TestAddOrReplaceStaticHostConfigYaml: + """Test the add_or_replace_static_host_config_yaml function.""" + + @pytest.fixture + def valid_nmstate_yaml(self): + """Fixture providing a valid nmstate YAML configuration.""" + return """ + interfaces: + - name: eth0 + type: ethernet + state: up + mac-address: "00:11:22:33:44:55" + ipv4: + enabled: true + address: + - ip: 192.168.1.10 + prefix-length: 24 + """ + + def test_add_to_empty_config(self, valid_nmstate_yaml): + """Test adding a config to an empty existing config.""" + result = add_or_replace_static_host_config_yaml(None, None, valid_nmstate_yaml) + + assert len(result) == 1 + assert result[0]["mac_interface_map"][0]["logical_nic_name"] == "eth0" + assert result[0]["mac_interface_map"][0]["mac_address"] == "00:11:22:33:44:55" + assert result[0]["network_yaml"] == valid_nmstate_yaml + + def test_add_to_existing_config(self, valid_nmstate_yaml): + """Test adding a config to an existing config.""" + existing_data = [ + { + "mac_interface_map": [ + {"logical_nic_name": "eth1", "mac_address": "00:11:22:33:44:66"} + ], + "network_yaml": "interfaces:\n- name: eth1\n type: ethernet", + } + ] + existing_config = json.dumps(existing_data) + + result = add_or_replace_static_host_config_yaml( + existing_config, None, valid_nmstate_yaml + ) + + assert len(result) == 2 + assert result[0]["mac_interface_map"][0]["logical_nic_name"] == "eth1" + assert result[1]["mac_interface_map"][0]["logical_nic_name"] == "eth0" + + def test_replace_at_index(self, valid_nmstate_yaml): + """Test replacing a config at a specific index.""" + existing_data = [ + { + "mac_interface_map": [ + {"logical_nic_name": "eth1", "mac_address": "00:11:22:33:44:66"} + ], + "network_yaml": "interfaces:\n- name: eth1\n type: ethernet", + } + ] + existing_config = json.dumps(existing_data) + + result = add_or_replace_static_host_config_yaml( + existing_config, 0, valid_nmstate_yaml + ) + + assert len(result) == 1 + assert result[0]["mac_interface_map"][0]["logical_nic_name"] == "eth0" + assert result[0]["mac_interface_map"][0]["mac_address"] == "00:11:22:33:44:55" + + def test_replace_invalid_index(self, valid_nmstate_yaml): + """Test replacing with an invalid index.""" + existing_data = [ + { + "mac_interface_map": [ + {"logical_nic_name": "eth1", "mac_address": "00:11:22:33:44:66"} + ], + "network_yaml": "interfaces:\n- name: eth1\n type: ethernet", + } + ] + existing_config = json.dumps(existing_data) + + with pytest.raises(IndexError): + add_or_replace_static_host_config_yaml( + existing_config, 5, valid_nmstate_yaml + ) + + def test_nmstate_without_mac_addresses(self): + """Test with nmstate YAML that has no MAC addresses.""" + nmstate_yaml = """ + interfaces: + - name: eth0 + type: ethernet + state: up + ipv4: + enabled: true + address: + - ip: 192.168.1.10 + prefix-length: 24 + """ + + with pytest.raises( + ValueError, + match="At least one interface must be associated to a MAC Address", + ): + add_or_replace_static_host_config_yaml(None, None, nmstate_yaml) + + +class TestValidateAndParseNmstate: + """Test the validate_and_parse_nmstate function.""" + + def test_valid_yaml(self): + """Test parsing valid YAML.""" + valid_yaml = """ + interfaces: + - name: eth0 + type: ethernet + state: up + """ + result = validate_and_parse_nmstate(valid_yaml) + + assert isinstance(result, dict) + assert "interfaces" in result + assert result["interfaces"][0]["name"] == "eth0" + + def test_invalid_yaml(self): + """Test parsing invalid YAML.""" + invalid_yaml = """ + interfaces: + - name: eth0 + type: ethernet + state: up + invalid_indentation: + """ + with pytest.raises(ValueError): + validate_and_parse_nmstate(invalid_yaml) + + def test_empty_yaml(self): + """Test parsing empty YAML.""" + result = validate_and_parse_nmstate("") + assert result is None + + def test_yaml_with_complex_structure(self): + """Test parsing YAML with complex nested structure.""" + complex_yaml = """ + dns-resolver: + config: + server: + - 8.8.8.8 + - 1.1.1.1 + search: + - example.com + routes: + config: + - destination: 0.0.0.0/0 + next-hop-address: 192.168.1.1 + next-hop-interface: eth0 + table-id: 254 + interfaces: + - name: eth0 + type: ethernet + state: up + mac-address: "00:11:22:33:44:55" + """ + result = validate_and_parse_nmstate(complex_yaml) + + assert "dns-resolver" in result + assert "routes" in result + assert "interfaces" in result + assert len(result["dns-resolver"]["config"]["server"]) == 2 + assert result["routes"]["config"][0]["destination"] == "0.0.0.0/0" + + +class TestGenerateNmstateFromTemplate: + """Test the generate_nmstate_from_template function.""" + + def test_minimal_ethernet_config(self): + """Test generating nmstate with minimal ethernet configuration.""" + params = NMStateTemplateParams( + ethernet_ifaces=[ + EthernetInterfaceParams( + name="eth0", + mac_address="00:11:22:33:44:55", + ipv4_address=IPV4AddressWithSubnet( + address=IPv4Address("192.168.1.10"), cidr_length=24 + ), + ) + ] + ) + + result = generate_nmstate_from_template(params) + + assert "interfaces:" in result + assert "name: eth0" in result + assert "type: ethernet" in result + assert "mac-address: 00:11:22:33:44:55" in result + assert "ip: 192.168.1.10" in result + assert "prefix-length: 24" in result + + # Verify the generated YAML is valid + yaml.safe_load(result) + + def test_ethernet_without_ip(self): + """Test generating nmstate with ethernet interface without IP.""" + params = NMStateTemplateParams( + ethernet_ifaces=[ + EthernetInterfaceParams( + name="eth0", mac_address="00:11:22:33:44:55", ipv4_address=None + ) + ] + ) + + result = generate_nmstate_from_template(params) + + assert "enabled: false" in result + assert "dhcp: false" in result + + # Verify the generated YAML is valid + yaml.safe_load(result) + + def test_with_dns_config(self): + """Test generating nmstate with DNS configuration.""" + params = NMStateTemplateParams( + dns=DNSParams( + dns_servers=["8.8.8.8", "1.1.1.1"], + dns_search_domains=["example.com", "test.com"], + ), + ethernet_ifaces=[ + EthernetInterfaceParams( + name="eth0", mac_address="00:11:22:33:44:55", ipv4_address=None + ) + ], + ) + + result = generate_nmstate_from_template(params) + + assert "dns-resolver:" in result + assert "- 8.8.8.8" in result + assert "- 1.1.1.1" in result + assert "- example.com" in result + assert "- test.com" in result + + def test_with_routes(self): + """Test generating nmstate with routing configuration.""" + params = NMStateTemplateParams( + routes=[ + RouteParams( + destination="0.0.0.0/0", + next_hop_address="192.168.1.1", + next_hop_interface="eth0", + table_id=254, + ) + ], + ethernet_ifaces=[ + EthernetInterfaceParams( + name="eth0", mac_address="00:11:22:33:44:55", ipv4_address=None + ) + ], + ) + + result = generate_nmstate_from_template(params) + + assert "routes:" in result + assert "destination: 0.0.0.0/0" in result + assert "next-hop-address: 192.168.1.1" in result + assert "next-hop-interface: eth0" in result + assert "table-id: 254" in result + + # Verify the generated YAML is valid + yaml.safe_load(result) + + def test_with_vlan_interfaces(self): + """Test generating nmstate with VLAN interfaces.""" + params = NMStateTemplateParams( + ethernet_ifaces=[ + EthernetInterfaceParams( + name="eth0", mac_address="00:11:22:33:44:55", ipv4_address=None + ) + ], + vlan_ifaces=[ + VLANInterfaceParams( + name="vlan100", + vlan_id=100, + base_interface_name="eth0", + ipv4_address=IPV4AddressWithSubnet( + address=IPv4Address("192.168.100.10"), cidr_length=24 + ), + ) + ], + ) + + result = generate_nmstate_from_template(params) + + assert "name: vlan100" in result + assert "type: vlan" in result + assert "base-iface: eth0" in result + assert "id: 100" in result + assert "ip: 192.168.100.10" in result + + # Verify the generated YAML is valid + yaml.safe_load(result) + + def test_with_bond_interfaces(self): + """Test generating nmstate with bond interfaces.""" + params = NMStateTemplateParams( + ethernet_ifaces=[ + EthernetInterfaceParams( + name="eth0", mac_address="00:11:22:33:44:55", ipv4_address=None + ), + EthernetInterfaceParams( + name="eth1", mac_address="00:11:22:33:44:66", ipv4_address=None + ), + ], + bond_ifaces=[ + BondInterfaceParams( + name="bond0", + mode="active-backup", + port_interface_names=["eth0", "eth1"], + ) + ], + ) + + result = generate_nmstate_from_template(params) + + assert "name: bond0" in result + assert "type: bond" in result + assert "mode: active-backup" in result + assert "- eth0" in result + assert "- eth1" in result + + # Verify the generated YAML is valid + yaml.safe_load(result) + + def test_complex_configuration(self): + """Test generating nmstate with all components.""" + params = NMStateTemplateParams( + dns=DNSParams(dns_servers=["8.8.8.8"], dns_search_domains=["example.com"]), + routes=[ + RouteParams( + destination="10.0.0.0/8", + next_hop_address="192.168.1.1", + next_hop_interface="eth0", + ) + ], + ethernet_ifaces=[ + EthernetInterfaceParams( + name="eth0", + mac_address="00:11:22:33:44:55", + ipv4_address=IPV4AddressWithSubnet( + address=IPv4Address("192.168.1.10"), cidr_length=24 + ), + ) + ], + vlan_ifaces=[ + VLANInterfaceParams( + name="vlan200", + vlan_id=200, + base_interface_name="eth0", + ipv4_address=IPV4AddressWithSubnet( + address=IPv4Address("192.168.200.10"), cidr_length=24 + ), + ) + ], + ) + + result = generate_nmstate_from_template(params) + + # Verify all components are present + assert "dns-resolver:" in result + assert "routes:" in result + assert "interfaces:" in result + assert "name: eth0" in result + assert "name: vlan200" in result + + # Verify the generated YAML is valid + yaml.safe_load(result) diff --git a/uv.lock b/uv.lock index 1bba6ce..d6c6cc2 100644 --- a/uv.lock +++ b/uv.lock @@ -46,8 +46,11 @@ source = { virtual = "." } dependencies = [ { name = "assisted-service-client" }, { name = "fastmcp" }, + { name = "jinja2" }, { name = "netaddr" }, { name = "prometheus-client" }, + { name = "pydantic" }, + { name = "pyyaml" }, { name = "requests" }, { name = "retry" }, { name = "types-requests" }, @@ -61,6 +64,7 @@ dev = [ { name = "pylint" }, { name = "pyright" }, { name = "ruff" }, + { name = "types-pyyaml" }, ] test = [ { name = "pytest" }, @@ -73,8 +77,11 @@ test = [ requires-dist = [ { name = "assisted-service-client", specifier = ">=2.41.0.post3" }, { name = "fastmcp", specifier = ">=2.8.0" }, + { name = "jinja2", specifier = ">=3.1" }, { name = "netaddr", specifier = ">=1.3.0" }, { name = "prometheus-client", specifier = ">=0.22.1" }, + { name = "pydantic", specifier = ">=2" }, + { name = "pyyaml", specifier = ">=6" }, { name = "requests", specifier = ">=2.32.3" }, { name = "retry", specifier = ">=0.9.2" }, { name = "types-requests", specifier = ">=2.32.4.20250611" }, @@ -88,6 +95,7 @@ dev = [ { name = "pylint", specifier = ">=3.3.7" }, { name = "pyright", specifier = ">=1.1.402" }, { name = "ruff", specifier = ">=0.12.1" }, + { name = "types-pyyaml", specifier = ">=6" }, ] test = [ { name = "pytest", specifier = ">=8.0.0" }, @@ -526,6 +534,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c1/11/114d0a5f4dabbdcedc1125dee0888514c3c3b16d3e9facad87ed96fad97c/isort-6.0.1-py3-none-any.whl", hash = "sha256:2dc5d7f65c9678d94c88dfc29161a320eec67328bc97aad576874cb4be1e9615", size = 94186, upload-time = "2025-02-26T21:13:14.911Z" }, ] +[[package]] +name = "jinja2" +version = "3.1.6" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markupsafe" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/df/bf/f7da0350254c0ed7c72f3e33cef02e048281fec7ecec5f032d4aac52226b/jinja2-3.1.6.tar.gz", hash = "sha256:0137fb05990d35f1275a587e9aee6d56da821fc83491a0fb838183be43f66d6d", size = 245115, upload-time = "2025-03-05T20:05:02.478Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/62/a1/3d680cbfd5f4b8f15abc1d571870c5fc3e594bb582bc3b64ea099db13e56/jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67", size = 134899, upload-time = "2025-03-05T20:05:00.369Z" }, +] + [[package]] name = "jsonschema" version = "4.25.1" @@ -1343,6 +1363,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/bd/75/8539d011f6be8e29f339c42e633aae3cb73bffa95dd0f9adec09b9c58e85/tomlkit-0.13.3-py3-none-any.whl", hash = "sha256:c89c649d79ee40629a9fda55f8ace8c6a1b42deb912b2a8fd8d942ddadb606b0", size = 38901, upload-time = "2025-06-05T07:13:43.546Z" }, ] +[[package]] +name = "types-pyyaml" +version = "6.0.12.20250915" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/7e/69/3c51b36d04da19b92f9e815be12753125bd8bc247ba0470a982e6979e71c/types_pyyaml-6.0.12.20250915.tar.gz", hash = "sha256:0f8b54a528c303f0e6f7165687dd33fafa81c807fcac23f632b63aa624ced1d3", size = 17522, upload-time = "2025-09-15T03:01:00.728Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/bd/e0/1eed384f02555dde685fff1a1ac805c1c7dcb6dd019c916fe659b1c1f9ec/types_pyyaml-6.0.12.20250915-py3-none-any.whl", hash = "sha256:e7d4d9e064e89a3b3cae120b4990cd370874d2bf12fa5f46c97018dd5d3c9ab6", size = 20338, upload-time = "2025-09-15T03:00:59.218Z" }, +] + [[package]] name = "types-requests" version = "2.32.4.20250913"