jeans/supervisor
jeans/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-07-27 19:16:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use jinja for payload generation (#1985)

Browse Source
This commit is contained in:
Joakim Sørensen 2020-08-28 11:46:22 +02:00 committed by GitHub
parent 8d3694884d
commit 98785b00e2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 68 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
supervisor/dbus/network/interface.py
View File

@ -6,7 +6,7 @@ from ..const import (
DBUS_OBJECT_BASE,
InterfaceMethod,
)
from ..payloads.interface_update import interface_update_payload
from ..payloads.generate import interface_update_payload
from .connection import NetworkConnection

34
supervisor/dbus/payloads/generate.py Normal file
View File

@ -0,0 +1,34 @@
"""Payload generators for DBUS communication."""
from pathlib import Path
import jinja2
from ...const import ATTR_ADDRESS, ATTR_DNS, ATTR_METHOD, ATTR_PREFIX
from ..const import InterfaceMethod
from ..network.utils import ip2int
INTERFACE_UPDATE_TEMPLATE: Path = (
Path(__file__).parents[2].joinpath("dbus/payloads/interface_update.tmpl")
)
def interface_update_payload(interface, **kwargs) -> str:
"""Generate a payload for network interface update."""
template = jinja2.Template(INTERFACE_UPDATE_TEMPLATE.read_text())
if kwargs.get(ATTR_DNS):
kwargs[ATTR_DNS] = [ip2int(x.strip()) for x in kwargs[ATTR_DNS]]
if kwargs.get(ATTR_METHOD):
kwargs[ATTR_METHOD] = (
InterfaceMethod.MANUAL
if kwargs[ATTR_METHOD] == "static"
else InterfaceMethod.AUTO
)
if kwargs.get(ATTR_ADDRESS):
if "/" in kwargs[ATTR_ADDRESS]:
kwargs[ATTR_PREFIX] = kwargs[ATTR_ADDRESS].split("/")[-1]
kwargs[ATTR_ADDRESS] = kwargs[ATTR_ADDRESS].split("/")[0]
kwargs[ATTR_METHOD] = InterfaceMethod.MANUAL
return template.render(interface=interface, options=kwargs)

55
supervisor/dbus/payloads/interface_update.py
View File

@ -1,55 +0,0 @@
"""Payload generators for DBUS communication."""
from ...const import ATTR_ADDRESS, ATTR_DNS, ATTR_GATEWAY, ATTR_METHOD, ATTR_PREFIX
from ..const import InterfaceMethod
from ..network.utils import ip2int
def interface_update_payload(interface, **kwargs) -> str:
"""Generate a payload for network interface update."""
if kwargs.get(ATTR_DNS):
kwargs[ATTR_DNS] = [ip2int(x.strip()) for x in kwargs[ATTR_DNS]]
if kwargs.get(ATTR_METHOD):
kwargs[ATTR_METHOD] = (
InterfaceMethod.MANUAL
if kwargs[ATTR_METHOD] == "static"
else InterfaceMethod.AUTO
)
if kwargs.get(ATTR_ADDRESS):
if "/" in kwargs[ATTR_ADDRESS]:
kwargs[ATTR_PREFIX] = kwargs[ATTR_ADDRESS].split("/")[-1]
kwargs[ATTR_ADDRESS] = kwargs[ATTR_ADDRESS].split("/")[0]
kwargs[ATTR_METHOD] = InterfaceMethod.MANUAL
if kwargs.get(ATTR_METHOD) == "auto":
return f"""{{
'connection':
{{
'id': <'{interface.id}'>,
'type': <'{interface.type}'>
}},
'ipv4':
{{
'method': <'{InterfaceMethod.AUTO}'>
}}
}}"""
return f"""{{
'connection':
{{
'id': <'{interface.id}'>,
'type': <'{interface.type}'>
}},
'ipv4':
{{
'method': <'{InterfaceMethod.MANUAL}'>,
'dns': <[{",".join([f"uint32 {x}" for x in kwargs.get(ATTR_DNS, interface.nameservers)])}]>,
'address-data': <[
{{
'address': <'{kwargs.get(ATTR_ADDRESS, interface.ip_address)}'>,
'prefix': <uint32 {kwargs.get(ATTR_PREFIX, interface.prefix)}>
}}]>,
'gateway': <'{kwargs.get(ATTR_GATEWAY, interface.gateway)}'>
}}
}}"""

26
supervisor/dbus/payloads/interface_update.tmpl Normal file
View File

@ -0,0 +1,26 @@
{
'connection':
{
'id': <'{{interface.id}}'>,
'type': <'{{interface.type}}'>
},
{% if options.get("method") == "auto" %}
'ipv4':
{
'method': <'auto'>
}
{% else %}
'ipv4':
{
'method': <'manual'>,
'dns': <[uint32 {{ options.get("dns", interface.nameservers) | list | join(",") }}]>,
'address-data': <[
{
'address': <'{{ options.get("address", interface.ip_address) }}'>,
'prefix': <uint32 {{ options.get("prefix", interface.prefix) }}>
}]>,
'gateway': <'{{ options.get("gateway", interface.gateway) }}'>
}
{% endif %}
}

44
tests/dbus/payloads/test_interface_update_payload.py
View File

@ -1,49 +1,19 @@
"""Test interface update payload."""
import pytest
from supervisor.dbus.payloads.interface_update import interface_update_payload
from supervisor.dbus.payloads.generate import interface_update_payload
from supervisor.utils.gdbus import DBus
@pytest.mark.asyncio
async def test_interface_update_payload(network_interface):
"""Test interface update payload."""
assert (
interface_update_payload(network_interface, **{"method": "auto"})
== """{
'connection':
{
'id': <'Wired connection 1'>,
'type': <'802-3-ethernet'>
},
'ipv4':
{
'method': <'auto'>
}
}"""
)
data = interface_update_payload(network_interface, **{"method": "auto"})
assert DBus.parse_gvariant(data)["ipv4"]["method"] == "auto"
assert (
interface_update_payload(network_interface, **{})
== """{
'connection':
{
'id': <'Wired connection 1'>,
'type': <'802-3-ethernet'>
},
'ipv4':
{
'method': <'manual'>,
'dns': <[uint32 16951488]>,
'address-data': <[
{
'address': <'192.168.2.148'>,
'prefix': <uint32 24>
}]>,
'gateway': <'192.168.2.1'>
}
}"""
data = interface_update_payload(
network_interface, **{"address": "1.1.1.1", "dns": ["1.1.1.1", "1.0.0.1"]}
)
data = interface_update_payload(network_interface, **{"address": "1.1.1.1"})
assert DBus.parse_gvariant(data)["ipv4"]["method"] == "manual"
assert DBus.parse_gvariant(data)["ipv4"]["address-data"][0]["address"] == "1.1.1.1"
assert DBus.parse_gvariant(data)["ipv4"]["dns"] == [16843009, 16777217]