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

feat: Add opt-in IPv6 for containers (#5879)

Browse Source 
Configurable and w/ migrations between IPv4-Only and Dual-Stack

Signed-off-by: David Rapan <david@rapan.cz>
Co-authored-by: Stefan Agner <stefan@agner.ch>
This commit is contained in:
David Rapan 2025-06-12 11:32:24 +02:00 committed by GitHub
parent 52b24e177f
commit d5b5a328d7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
17 changed files with 376 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
supervisor/api/__init__.py
View File

@ -8,7 +8,7 @@ from typing import Any
from aiohttp import hdrs, web
from ..const import AddonState
from ..const import SUPERVISOR_DOCKER_NAME, AddonState
from ..coresys import CoreSys, CoreSysAttributes
from ..exceptions import APIAddonNotInstalled, HostNotSupportedError
from ..utils.sentry import async_capture_exception
@ -426,7 +426,7 @@ class RestAPI(CoreSysAttributes):
async def get_supervisor_logs(*args, **kwargs):
try:
return await self._api_host.advanced_logs_handler(
*args, identifier="hassio_supervisor", **kwargs
*args, identifier=SUPERVISOR_DOCKER_NAME, **kwargs
)
except Exception as err: # pylint: disable=broad-exception-caught
# Supervisor logs are critical, so catch everything, log the exception
@ -789,6 +789,7 @@ class RestAPI(CoreSysAttributes):
self.webapp.add_routes(
[
web.get("/docker/info", api_docker.info),
web.post("/docker/options", api_docker.options),
web.get("/docker/registries", api_docker.registries),
web.post("/docker/registries", api_docker.create_registry),
web.delete("/docker/registries/{hostname}", api_docker.remove_registry),

45
supervisor/api/docker.py
View File

@ -7,6 +7,7 @@ from aiohttp import web
import voluptuous as vol
from ..const import (
ATTR_ENABLE_IPV6,
ATTR_HOSTNAME,
ATTR_LOGGING,
ATTR_PASSWORD,
@ -30,10 +31,39 @@ SCHEMA_DOCKER_REGISTRY = vol.Schema(
}
)
# pylint: disable=no-value-for-parameter
SCHEMA_OPTIONS = vol.Schema({vol.Optional(ATTR_ENABLE_IPV6): vol.Boolean()})
class APIDocker(CoreSysAttributes):
"""Handle RESTful API for Docker configuration."""
@api_process
async def info(self, request: web.Request):
"""Get docker info."""
data_registries = {}
for hostname, registry in self.sys_docker.config.registries.items():
data_registries[hostname] = {
ATTR_USERNAME: registry[ATTR_USERNAME],
}
return {
ATTR_VERSION: self.sys_docker.info.version,
ATTR_ENABLE_IPV6: self.sys_docker.config.enable_ipv6,
ATTR_STORAGE: self.sys_docker.info.storage,
ATTR_LOGGING: self.sys_docker.info.logging,
ATTR_REGISTRIES: data_registries,
}
@api_process
async def options(self, request: web.Request) -> None:
"""Set docker options."""
body = await api_validate(SCHEMA_OPTIONS, request)
if ATTR_ENABLE_IPV6 in body:
self.sys_docker.config.enable_ipv6 = body[ATTR_ENABLE_IPV6]
await self.sys_docker.config.save_data()
@api_process
async def registries(self, request) -> dict[str, Any]:
"""Return the list of registries."""
@ -64,18 +94,3 @@ class APIDocker(CoreSysAttributes):
del self.sys_docker.config.registries[hostname]
await self.sys_docker.config.save_data()
@api_process
async def info(self, request: web.Request):
"""Get docker info."""
data_registries = {}
for hostname, registry in self.sys_docker.config.registries.items():
data_registries[hostname] = {
ATTR_USERNAME: registry[ATTR_USERNAME],
}
return {
ATTR_VERSION: self.sys_docker.info.version,
ATTR_STORAGE: self.sys_docker.info.storage,
ATTR_LOGGING: self.sys_docker.info.logging,
ATTR_REGISTRIES: data_registries,
}

4
supervisor/api/network.py
View File

@ -40,8 +40,8 @@ from ..const import (
ATTR_TYPE,
ATTR_VLAN,
ATTR_WIFI,
DOCKER_IPV4_NETWORK_MASK,
DOCKER_NETWORK,
DOCKER_NETWORK_MASK,
)
from ..coresys import CoreSysAttributes
from ..exceptions import APIError, APINotFound, HostNetworkNotFound
@ -203,7 +203,7 @@ class APINetwork(CoreSysAttributes):
],
ATTR_DOCKER: {
ATTR_INTERFACE: DOCKER_NETWORK,
ATTR_ADDRESS: str(DOCKER_NETWORK_MASK),
ATTR_ADDRESS: str(DOCKER_IPV4_NETWORK_MASK),
ATTR_GATEWAY: str(self.sys_docker.network.gateway),
ATTR_DNS: str(self.sys_docker.network.dns),
},

13
supervisor/const.py
View File

@ -2,7 +2,7 @@
from dataclasses import dataclass
from enum import StrEnum
from ipaddress import IPv4Network
from ipaddress import IPv4Network, IPv6Network
from pathlib import Path
from sys import version_info as systemversion
from typing import Self
@ -12,6 +12,10 @@ from aiohttp import __version__ as aiohttpversion
SUPERVISOR_VERSION = "9999.09.9.dev9999"
SERVER_SOFTWARE = f"HomeAssistantSupervisor/{SUPERVISOR_VERSION} aiohttp/{aiohttpversion} Python/{systemversion[0]}.{systemversion[1]}"
DOCKER_PREFIX: str = "hassio"
OBSERVER_DOCKER_NAME: str = f"{DOCKER_PREFIX}_observer"
SUPERVISOR_DOCKER_NAME: str = f"{DOCKER_PREFIX}_supervisor"
URL_HASSIO_ADDONS = "https://github.com/home-assistant/addons"
URL_HASSIO_APPARMOR = "https://version.home-assistant.io/apparmor_{channel}.txt"
URL_HASSIO_VERSION = "https://version.home-assistant.io/{channel}.json"
@ -41,8 +45,10 @@ SYSTEMD_JOURNAL_PERSISTENT = Path("/var/log/journal")
SYSTEMD_JOURNAL_VOLATILE = Path("/run/log/journal")
DOCKER_NETWORK = "hassio"
DOCKER_NETWORK_MASK = IPv4Network("172.30.32.0/23")
DOCKER_NETWORK_RANGE = IPv4Network("172.30.33.0/24")
DOCKER_NETWORK_DRIVER = "bridge"
DOCKER_IPV6_NETWORK_MASK = IPv6Network("fd0c:ac1e:2100::/48")
DOCKER_IPV4_NETWORK_MASK = IPv4Network("172.30.32.0/23")
DOCKER_IPV4_NETWORK_RANGE = IPv4Network("172.30.33.0/24")
# This needs to match the dockerd --cpu-rt-runtime= argument.
DOCKER_CPU_RUNTIME_TOTAL = 950_000
@ -172,6 +178,7 @@ ATTR_DOCKER_API = "docker_api"
ATTR_DOCUMENTATION = "documentation"
ATTR_DOMAINS = "domains"
ATTR_ENABLE = "enable"
ATTR_ENABLE_IPV6 = "enable_ipv6"
ATTR_ENABLED = "enabled"
ATTR_ENVIRONMENT = "environment"
ATTR_EVENT = "event"

4
supervisor/dbus/network/connection.py
View File

@ -96,7 +96,7 @@ class NetworkConnection(DBusInterfaceProxy):
@ipv4.setter
def ipv4(self, ipv4: IpConfiguration | None) -> None:
"""Set ipv4 configuration."""
"""Set IPv4 configuration."""
if self._ipv4 and self._ipv4 is not ipv4:
self._ipv4.shutdown()
@ -109,7 +109,7 @@ class NetworkConnection(DBusInterfaceProxy):
@ipv6.setter
def ipv6(self, ipv6: IpConfiguration | None) -> None:
"""Set ipv6 configuration."""
"""Set IPv6 configuration."""
if self._ipv6 and self._ipv6 is not ipv6:
self._ipv6.shutdown()

4
supervisor/dbus/network/setting/__init__.py
View File

@ -152,12 +152,12 @@ class NetworkSetting(DBusInterface):
@property
def ipv4(self) -> Ip4Properties | None:
"""Return ipv4 properties if any."""
"""Return IPv4 properties if any."""
return self._ipv4
@property
def ipv6(self) -> Ip6Properties | None:
"""Return ipv6 properties if any."""
"""Return IPv6 properties if any."""
return self._ipv6
@property

15
supervisor/docker/manager.py
View File

@ -22,6 +22,7 @@ from docker.types.daemon import CancellableStream
import requests
from ..const import (
ATTR_ENABLE_IPV6,
ATTR_REGISTRIES,
DNS_SUFFIX,
DOCKER_NETWORK,
@ -93,6 +94,16 @@ class DockerConfig(FileConfiguration):
"""Initialize the JSON configuration."""
super().__init__(FILE_HASSIO_DOCKER, SCHEMA_DOCKER_CONFIG)
@property
def enable_ipv6(self) -> bool:
"""Return IPv6 configuration for docker network."""
return self._data.get(ATTR_ENABLE_IPV6, False)
@enable_ipv6.setter
def enable_ipv6(self, value: bool) -> None:
"""Set IPv6 configuration for docker network."""
self._data[ATTR_ENABLE_IPV6] = value
@property
def registries(self) -> dict[str, Any]:
"""Return credentials for docker registries."""
@ -124,9 +135,11 @@ class DockerAPI:
timeout=900,
),
)
self._network = DockerNetwork(self._docker)
self._info = DockerInfo.new(self.docker.info())
await self.config.read_data()
self._network = await DockerNetwork(self.docker).post_init(
self.config.enable_ipv6
)
return self
@property

207
supervisor/docker/network.py
View File

@ -1,17 +1,52 @@
"""Internal network manager for Supervisor."""
import asyncio
from contextlib import suppress
from ipaddress import IPv4Address
import logging
from typing import Self
import docker
import requests
from ..const import DOCKER_NETWORK, DOCKER_NETWORK_MASK, DOCKER_NETWORK_RANGE
from ..const import (
ATTR_AUDIO,
ATTR_CLI,
ATTR_DNS,
ATTR_ENABLE_IPV6,
ATTR_OBSERVER,
ATTR_SUPERVISOR,
DOCKER_IPV4_NETWORK_MASK,
DOCKER_IPV4_NETWORK_RANGE,
DOCKER_IPV6_NETWORK_MASK,
DOCKER_NETWORK,
DOCKER_NETWORK_DRIVER,
DOCKER_PREFIX,
OBSERVER_DOCKER_NAME,
SUPERVISOR_DOCKER_NAME,
)
from ..exceptions import DockerError
_LOGGER: logging.Logger = logging.getLogger(__name__)
DOCKER_ENABLEIPV6 = "EnableIPv6"
DOCKER_NETWORK_PARAMS = {
"name": DOCKER_NETWORK,
"driver": DOCKER_NETWORK_DRIVER,
"ipam": docker.types.IPAMConfig(
pool_configs=[
docker.types.IPAMPool(subnet=str(DOCKER_IPV6_NETWORK_MASK)),
docker.types.IPAMPool(
subnet=str(DOCKER_IPV4_NETWORK_MASK),
gateway=str(DOCKER_IPV4_NETWORK_MASK[1]),
iprange=str(DOCKER_IPV4_NETWORK_RANGE),
),
]
),
ATTR_ENABLE_IPV6: True,
"options": {"com.docker.network.bridge.name": DOCKER_NETWORK},
}
class DockerNetwork:
"""Internal Supervisor Network.
@ -22,7 +57,14 @@ class DockerNetwork:
def __init__(self, docker_client: docker.DockerClient):
"""Initialize internal Supervisor network."""
self.docker: docker.DockerClient = docker_client
self._network: docker.models.networks.Network = self._get_network()
self._network: docker.models.networks.Network | None = None
async def post_init(self, enable_ipv6: bool = False) -> Self:
"""Post init actions that must be done in event loop."""
self._network = await asyncio.get_running_loop().run_in_executor(
None, self._get_network, enable_ipv6
)
return self
@property
def name(self) -> str:
@ -42,55 +84,101 @@ class DockerNetwork:
@property
def gateway(self) -> IPv4Address:
"""Return gateway of the network."""
return DOCKER_NETWORK_MASK[1]
return DOCKER_IPV4_NETWORK_MASK[1]
@property
def supervisor(self) -> IPv4Address:
"""Return supervisor of the network."""
return DOCKER_NETWORK_MASK[2]
return DOCKER_IPV4_NETWORK_MASK[2]
@property
def dns(self) -> IPv4Address:
"""Return dns of the network."""
return DOCKER_NETWORK_MASK[3]
return DOCKER_IPV4_NETWORK_MASK[3]
@property
def audio(self) -> IPv4Address:
"""Return audio of the network."""
return DOCKER_NETWORK_MASK[4]
return DOCKER_IPV4_NETWORK_MASK[4]
@property
def cli(self) -> IPv4Address:
"""Return cli of the network."""
return DOCKER_NETWORK_MASK[5]
return DOCKER_IPV4_NETWORK_MASK[5]
@property
def observer(self) -> IPv4Address:
"""Return observer of the network."""
return DOCKER_NETWORK_MASK[6]
return DOCKER_IPV4_NETWORK_MASK[6]
def _get_network(self) -> docker.models.networks.Network:
def _get_network(self, enable_ipv6: bool = False) -> docker.models.networks.Network:
"""Get supervisor network."""
try:
return self.docker.networks.get(DOCKER_NETWORK)
if network := self.docker.networks.get(DOCKER_NETWORK):
if network.attrs.get(DOCKER_ENABLEIPV6) == enable_ipv6:
return network
_LOGGER.info(
"Migrating Supervisor network to %s",
"IPv4/IPv6 Dual-Stack" if enable_ipv6 else "IPv4-Only",
)
if (containers := network.containers) and (
containers_all := all(
container.name in (OBSERVER_DOCKER_NAME, SUPERVISOR_DOCKER_NAME)
for container in containers
)
):
for container in containers:
with suppress(
docker.errors.APIError,
docker.errors.DockerException,
requests.RequestException,
):
network.disconnect(container, force=True)
if not containers or containers_all:
try:
network.remove()
except docker.errors.APIError:
_LOGGER.warning("Failed to remove existing Supervisor network")
return network
else:
_LOGGER.warning(
"System appears to be running, "
"not applying Supervisor network change. "
"Reboot your system to apply the change."
)
return network
except docker.errors.NotFound:
_LOGGER.info("Can't find Supervisor network, creating a new network")
ipam_pool = docker.types.IPAMPool(
subnet=str(DOCKER_NETWORK_MASK),
gateway=str(self.gateway),
iprange=str(DOCKER_NETWORK_RANGE),
)
network_params = DOCKER_NETWORK_PARAMS.copy()
network_params[ATTR_ENABLE_IPV6] = enable_ipv6
ipam_config = docker.types.IPAMConfig(pool_configs=[ipam_pool])
try:
self._network = self.docker.networks.create(**network_params)
except docker.errors.APIError as err:
raise DockerError(
f"Can't create Supervisor network: {err}", _LOGGER.error
) from err
return self.docker.networks.create(
DOCKER_NETWORK,
driver="bridge",
ipam=ipam_config,
enable_ipv6=False,
options={"com.docker.network.bridge.name": DOCKER_NETWORK},
)
with suppress(DockerError):
self.attach_container_by_name(
SUPERVISOR_DOCKER_NAME, [ATTR_SUPERVISOR], self.supervisor
)
with suppress(DockerError):
self.attach_container_by_name(
OBSERVER_DOCKER_NAME, [ATTR_OBSERVER], self.observer
)
for name, ip in (
(ATTR_CLI, self.cli),
(ATTR_DNS, self.dns),
(ATTR_AUDIO, self.audio),
):
with suppress(DockerError):
self.attach_container_by_name(f"{DOCKER_PREFIX}_{name}", [name], ip)
return self._network
def attach_container(
self,
@ -102,8 +190,6 @@ class DockerNetwork:
Need run inside executor.
"""
ipv4_address = str(ipv4) if ipv4 else None
# Reload Network information
with suppress(docker.errors.DockerException, requests.RequestException):
self.network.reload()
@ -116,12 +202,43 @@ class DockerNetwork:
# Attach Network
try:
self.network.connect(container, aliases=alias, ipv4_address=ipv4_address)
except docker.errors.APIError as err:
self.network.connect(
container, aliases=alias, ipv4_address=str(ipv4) if ipv4 else None
)
except (
docker.errors.NotFound,
docker.errors.APIError,
docker.errors.DockerException,
requests.RequestException,
) as err:
raise DockerError(
f"Can't link container to hassio-net: {err}", _LOGGER.error
f"Can't connect {container.name} to Supervisor network: {err}",
_LOGGER.error,
) from err
def attach_container_by_name(
self,
name: str,
alias: list[str] | None = None,
ipv4: IPv4Address | None = None,
) -> None:
"""Attach container to Supervisor network.
Need run inside executor.
"""
try:
container = self.docker.containers.get(name)
except (
docker.errors.NotFound,
docker.errors.APIError,
docker.errors.DockerException,
requests.RequestException,
) as err:
raise DockerError(f"Can't find {name}: {err}", _LOGGER.error) from err
if container.id not in self.containers:
self.attach_container(container, alias, ipv4)
def detach_default_bridge(
self, container: docker.models.containers.Container
) -> None:
@ -130,25 +247,33 @@ class DockerNetwork:
Need run inside executor.
"""
try:
default_network = self.docker.networks.get("bridge")
default_network = self.docker.networks.get(DOCKER_NETWORK_DRIVER)
default_network.disconnect(container)
except docker.errors.NotFound:
return
except docker.errors.APIError as err:
pass
except (
docker.errors.APIError,
docker.errors.DockerException,
requests.RequestException,
) as err:
raise DockerError(
f"Can't disconnect container from default: {err}", _LOGGER.warning
f"Can't disconnect {container.name} from default network: {err}",
_LOGGER.warning,
) from err
def stale_cleanup(self, container_name: str):
"""Remove force a container from Network.
def stale_cleanup(self, name: str) -> None:
"""Force remove a container from Network.
Fix: https://github.com/moby/moby/issues/23302
"""
try:
self.network.disconnect(container_name, force=True)
except docker.errors.NotFound:
pass
except (docker.errors.DockerException, requests.RequestException) as err:
raise DockerError() from err
self.network.disconnect(name, force=True)
except (
docker.errors.APIError,
docker.errors.DockerException,
requests.RequestException,
) as err:
raise DockerError(
f"Can't disconnect {name} from Supervisor network: {err}",
_LOGGER.warning,
) from err

5
supervisor/docker/observer.py
View File

@ -2,7 +2,7 @@
import logging
from ..const import DOCKER_NETWORK_MASK
from ..const import DOCKER_IPV4_NETWORK_MASK, OBSERVER_DOCKER_NAME
from ..coresys import CoreSysAttributes
from ..exceptions import DockerJobError
from ..jobs.const import JobExecutionLimit
@ -12,7 +12,6 @@ from .interface import DockerInterface
_LOGGER: logging.Logger = logging.getLogger(__name__)
OBSERVER_DOCKER_NAME: str = "hassio_observer"
ENV_NETWORK_MASK: str = "NETWORK_MASK"
@ -49,7 +48,7 @@ class DockerObserver(DockerInterface, CoreSysAttributes):
environment={
ENV_TIME: self.sys_timezone,
ENV_TOKEN: self.sys_plugins.observer.supervisor_token,
ENV_NETWORK_MASK: DOCKER_NETWORK_MASK,
ENV_NETWORK_MASK: DOCKER_IPV4_NETWORK_MASK,
},
mounts=[MOUNT_DOCKER],
ports={"80/tcp": 4357},

4
supervisor/misc/filter.py
View File

@ -9,7 +9,7 @@ from aiohttp import hdrs
import attr
from sentry_sdk.types import Event, Hint
from ..const import DOCKER_NETWORK_MASK, HEADER_TOKEN, HEADER_TOKEN_OLD, CoreState
from ..const import DOCKER_IPV4_NETWORK_MASK, HEADER_TOKEN, HEADER_TOKEN_OLD, CoreState
from ..coresys import CoreSys
from ..exceptions import AddonConfigurationError
@ -21,7 +21,7 @@ def sanitize_host(host: str) -> str:
try:
# Allow internal URLs
ip = ipaddress.ip_address(host)
if ip in ipaddress.ip_network(DOCKER_NETWORK_MASK):
if ip in ipaddress.ip_network(DOCKER_IPV4_NETWORK_MASK):
return host
except ValueError:
pass

6
supervisor/validate.py
View File

@ -20,6 +20,7 @@ from .const import (
ATTR_DIAGNOSTICS,
ATTR_DISPLAYNAME,
ATTR_DNS,
ATTR_ENABLE_IPV6,
ATTR_FORCE_SECURITY,
ATTR_HASSOS,
ATTR_HOMEASSISTANT,
@ -81,7 +82,7 @@ def dns_url(url: str) -> str:
raise vol.Invalid("Doesn't start with dns://") from None
address: str = url[6:] # strip the dns:// off
try:
ip = ipaddress.ip_address(address) # matches ipv4 or ipv6 addresses
ip = ipaddress.ip_address(address) # matches IPv4 or IPv6 addresses
except ValueError:
raise vol.Invalid(f"Invalid DNS URL: {url}") from None
@ -180,7 +181,8 @@ SCHEMA_DOCKER_CONFIG = vol.Schema(
vol.Required(ATTR_PASSWORD): str,
}
}
)
),
vol.Optional(ATTR_ENABLE_IPV6): vol.Boolean(),
}
)

17
tests/api/test_docker.py
View File

@ -3,6 +3,8 @@
from aiohttp.test_utils import TestClient
import pytest
from supervisor.coresys import CoreSys
@pytest.mark.asyncio
async def test_api_docker_info(api_client: TestClient):
@ -15,6 +17,21 @@ async def test_api_docker_info(api_client: TestClient):
assert result["data"]["version"] == "1.0.0"
async def test_api_network_enable_ipv6(coresys: CoreSys, api_client: TestClient):
"""Test setting docker network for enabled IPv6."""
assert coresys.docker.config.enable_ipv6 is False
resp = await api_client.post("/docker/options", json={"enable_ipv6": True})
assert resp.status == 200
assert coresys.docker.config.enable_ipv6 is True
resp = await api_client.get("/docker/info")
assert resp.status == 200
body = await resp.json()
assert body["data"]["enable_ipv6"] is True
async def test_registry_not_found(api_client: TestClient):
"""Test registry not found error."""
resp = await api_client.delete("/docker/registries/bad")

4
tests/api/test_network.py
View File

@ -6,7 +6,7 @@ from aiohttp.test_utils import TestClient
from dbus_fast import Variant
import pytest
from supervisor.const import DOCKER_NETWORK, DOCKER_NETWORK_MASK
from supervisor.const import DOCKER_IPV4_NETWORK_MASK, DOCKER_NETWORK
from supervisor.coresys import CoreSys
from tests.const import (
@ -61,7 +61,7 @@ async def test_api_network_info(api_client: TestClient, coresys: CoreSys):
}
assert result["data"]["docker"]["interface"] == DOCKER_NETWORK
assert result["data"]["docker"]["address"] == str(DOCKER_NETWORK_MASK)
assert result["data"]["docker"]["address"] == str(DOCKER_IPV4_NETWORK_MASK)
assert result["data"]["docker"]["dns"] == str(coresys.docker.network.dns)
assert result["data"]["docker"]["gateway"] == str(coresys.docker.network.gateway)

4
tests/dbus/network/test_connection.py
View File

@ -56,7 +56,7 @@ async def test_active_connection(
async def test_old_ipv4_disconnect(
network_manager: NetworkManager, active_connection_service: ActiveConnectionService
):
"""Test old ipv4 disconnects on ipv4 change."""
"""Test old IPv4 disconnects on IPv4 change."""
connection = network_manager.get(TEST_INTERFACE_ETH_NAME).connection
ipv4 = connection.ipv4
assert ipv4.is_connected is True
@ -71,7 +71,7 @@ async def test_old_ipv4_disconnect(
async def test_old_ipv6_disconnect(
network_manager: NetworkManager, active_connection_service: ActiveConnectionService
):
"""Test old ipv6 disconnects on ipv6 change."""
"""Test old IPv6 disconnects on IPv6 change."""
connection = network_manager.get(TEST_INTERFACE_ETH_NAME).connection
ipv6 = connection.ipv6
assert ipv6.is_connected is True

4
tests/dbus/network/test_ip_configuration.py
View File

@ -31,7 +31,7 @@ async def fixture_ip6config_service(dbus_session_bus: MessageBus) -> IP4ConfigSe
async def test_ipv4_configuration(
ip4config_service: IP4ConfigService, dbus_session_bus: MessageBus
):
"""Test ipv4 configuration object."""
"""Test IPv4 configuration object."""
ip4 = IpConfiguration("/org/freedesktop/NetworkManager/IP4Config/1")
assert ip4.gateway is None
@ -55,7 +55,7 @@ async def test_ipv4_configuration(
async def test_ipv6_configuration(
ip6config_service: IP6ConfigService, dbus_session_bus: MessageBus
):
"""Test ipv4 configuration object."""
"""Test IPv6 configuration object."""
ip6 = IpConfiguration("/org/freedesktop/NetworkManager/IP6Config/1", ip4=False)
assert ip6.gateway is None

113
tests/docker/test_network.py Normal file
View File

@ -0,0 +1,113 @@
"""Test Internal network manager for Supervisor."""
from unittest.mock import MagicMock, PropertyMock, patch
import docker
import pytest
from supervisor.const import (
ATTR_ENABLE_IPV6,
DOCKER_NETWORK,
OBSERVER_DOCKER_NAME,
SUPERVISOR_DOCKER_NAME,
)
from supervisor.docker.network import (
DOCKER_ENABLEIPV6,
DOCKER_NETWORK_PARAMS,
DockerNetwork,
)
class MockContainer:
"""Mock implementation of a Docker container."""
def __init__(self, name: str) -> None:
"""Initialize a mock container."""
self.name = name
class MockNetwork:
"""Mock implementation of internal network."""
def __init__(
self, raise_error: bool, containers: list[str], enableIPv6: bool
) -> None:
"""Initialize a mock network."""
self.raise_error = raise_error
self.containers = [MockContainer(container) for container in containers or []]
self.attrs = {DOCKER_ENABLEIPV6: enableIPv6}
def remove(self) -> None:
"""Simulate a network removal."""
if self.raise_error:
raise docker.errors.APIError("Simulated removal error")
def reload(self, *args, **kwargs) -> None:
"""Simulate a network reload."""
def connect(self, *args, **kwargs) -> None:
"""Simulate a network connection."""
def disconnect(self, *args, **kwargs) -> None:
"""Simulate a network disconnection."""
@pytest.mark.parametrize(
("raise_error", "containers", f"old_{ATTR_ENABLE_IPV6}", f"new_{ATTR_ENABLE_IPV6}"),
[
(False, [OBSERVER_DOCKER_NAME, SUPERVISOR_DOCKER_NAME], False, True),
(False, [OBSERVER_DOCKER_NAME, SUPERVISOR_DOCKER_NAME], True, False),
(False, ["test_container"], False, True),
(False, None, False, True),
(True, None, False, True),
(False, None, True, True),
],
)
async def test_network_recreation(
raise_error: bool,
containers: list[str] | None,
old_enable_ipv6: bool,
new_enable_ipv6: bool,
):
"""Test network recreation with IPv6 enabled/disabled."""
with (
patch(
"supervisor.docker.network.DockerNetwork.docker",
new_callable=PropertyMock,
return_value=MagicMock(),
create=True,
),
patch(
"supervisor.docker.network.DockerNetwork.docker.networks",
new_callable=PropertyMock,
return_value=MagicMock(),
create=True,
),
patch(
"supervisor.docker.network.DockerNetwork.docker.networks.get",
return_value=MockNetwork(raise_error, containers, old_enable_ipv6),
) as mock_get,
patch(
"supervisor.docker.network.DockerNetwork.docker.networks.create",
return_value=MockNetwork(raise_error, containers, new_enable_ipv6),
) as mock_create,
):
network = (await DockerNetwork(MagicMock()).post_init(new_enable_ipv6)).network
mock_get.assert_called_with(DOCKER_NETWORK)
assert network is not None
assert network.attrs.get(DOCKER_ENABLEIPV6) is (
new_enable_ipv6
if not raise_error and (not containers or len(containers) > 1)
else old_enable_ipv6
)
if (
not raise_error and (not containers or len(containers) > 1)
) and old_enable_ipv6 != new_enable_ipv6:
network_params = DOCKER_NETWORK_PARAMS.copy()
network_params[ATTR_ENABLE_IPV6] = new_enable_ipv6
mock_create.assert_called_with(**network_params)

6
tests/test_validate.py
View File

@ -36,14 +36,14 @@ IMAGE_NAME_BAD = [
]
async def test_dns_url_v4_good():
"""Test the DNS validator with known-good ipv6 DNS URLs."""
def test_dns_url_v4_good():
"""Test the DNS validator with known-good IPv4 DNS URLs."""
for url in DNS_GOOD_V4:
assert validate.dns_url(url)
def test_dns_url_v6_good():
"""Test the DNS validator with known-good ipv6 DNS URLs."""
"""Test the DNS validator with known-good IPv6 DNS URLs."""
with pytest.raises(vol.error.Invalid):
for url in DNS_GOOD_V6:
assert validate.dns_url(url)