mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-07-12 11:46:31 +00:00
Add MDNS and LLMNR status to API (#3545)
* Add mdns and llmnr status to API * Add broadcast info to host/info and move constants * Fix new test and isort error
This commit is contained in:
parent
6666637a77
commit
12da8a0c55
@ -1,16 +1,21 @@
|
||||
"""Const for API."""
|
||||
|
||||
ATTR_APPARMOR_VERSION = "apparmor_version"
|
||||
ATTR_AGENT_VERSION = "agent_version"
|
||||
ATTR_AVAILABLE_UPDATES = "available_updates"
|
||||
ATTR_BOOT_TIMESTAMP = "boot_timestamp"
|
||||
ATTR_BROADCAST_LLMNR = "broadcast_llmnr"
|
||||
ATTR_BROADCAST_MDNS = "broadcast_mdns"
|
||||
ATTR_DATA_DISK = "data_disk"
|
||||
ATTR_DEVICE = "device"
|
||||
ATTR_DT_SYNCHRONIZED = "dt_synchronized"
|
||||
ATTR_DT_UTC = "dt_utc"
|
||||
ATTR_LLMNR = "llmnr"
|
||||
ATTR_LLMNR_HOSTNAME = "llmnr_hostname"
|
||||
ATTR_MDNS = "mdns"
|
||||
ATTR_PANEL_PATH = "panel_path"
|
||||
ATTR_SIGNED = "signed"
|
||||
ATTR_STARTUP_TIME = "startup_time"
|
||||
ATTR_UPDATE_TYPE = "update_type"
|
||||
ATTR_USE_NTP = "use_ntp"
|
||||
ATTR_USE_RTC = "use_rtc"
|
||||
ATTR_APPARMOR_VERSION = "apparmor_version"
|
||||
ATTR_PANEL_PATH = "panel_path"
|
||||
ATTR_UPDATE_TYPE = "update_type"
|
||||
ATTR_AVAILABLE_UPDATES = "available_updates"
|
||||
ATTR_SIGNED = "signed"
|
||||
|
@ -26,6 +26,7 @@ from ..const import (
|
||||
from ..coresys import CoreSysAttributes
|
||||
from ..exceptions import APIError
|
||||
from ..validate import dns_server_list, version_tag
|
||||
from .const import ATTR_LLMNR, ATTR_MDNS
|
||||
from .utils import api_process, api_process_raw, api_validate
|
||||
|
||||
_LOGGER: logging.Logger = logging.getLogger(__name__)
|
||||
@ -49,6 +50,8 @@ class APICoreDNS(CoreSysAttributes):
|
||||
ATTR_HOST: str(self.sys_docker.network.dns),
|
||||
ATTR_SERVERS: self.sys_plugins.dns.servers,
|
||||
ATTR_LOCALS: self.sys_plugins.dns.locals,
|
||||
ATTR_MDNS: self.sys_plugins.dns.mdns,
|
||||
ATTR_LLMNR: self.sys_plugins.dns.llmnr,
|
||||
}
|
||||
|
||||
@api_process
|
||||
|
@ -29,8 +29,11 @@ from .const import (
|
||||
ATTR_AGENT_VERSION,
|
||||
ATTR_APPARMOR_VERSION,
|
||||
ATTR_BOOT_TIMESTAMP,
|
||||
ATTR_BROADCAST_LLMNR,
|
||||
ATTR_BROADCAST_MDNS,
|
||||
ATTR_DT_SYNCHRONIZED,
|
||||
ATTR_DT_UTC,
|
||||
ATTR_LLMNR_HOSTNAME,
|
||||
ATTR_STARTUP_TIME,
|
||||
ATTR_USE_NTP,
|
||||
ATTR_USE_RTC,
|
||||
@ -60,6 +63,7 @@ class APIHost(CoreSysAttributes):
|
||||
ATTR_DISK_LIFE_TIME: self.sys_host.info.disk_life_time,
|
||||
ATTR_FEATURES: self.sys_host.features,
|
||||
ATTR_HOSTNAME: self.sys_host.info.hostname,
|
||||
ATTR_LLMNR_HOSTNAME: self.sys_host.info.llmnr_hostname,
|
||||
ATTR_KERNEL: self.sys_host.info.kernel,
|
||||
ATTR_OPERATING_SYSTEM: self.sys_host.info.operating_system,
|
||||
ATTR_TIMEZONE: self.sys_host.info.timezone,
|
||||
@ -69,6 +73,8 @@ class APIHost(CoreSysAttributes):
|
||||
ATTR_USE_RTC: self.sys_host.info.use_rtc,
|
||||
ATTR_STARTUP_TIME: self.sys_host.info.startup_time,
|
||||
ATTR_BOOT_TIMESTAMP: self.sys_host.info.boot_timestamp,
|
||||
ATTR_BROADCAST_LLMNR: self.sys_host.info.broadcast_llmnr,
|
||||
ATTR_BROADCAST_MDNS: self.sys_host.info.broadcast_mdns,
|
||||
}
|
||||
|
||||
@api_process
|
||||
|
@ -42,6 +42,7 @@ class HostFeature(str, Enum):
|
||||
HOSTNAME = "hostname"
|
||||
NETWORK = "network"
|
||||
REBOOT = "reboot"
|
||||
RESOLVED = "resolved"
|
||||
SERVICES = "services"
|
||||
SHUTDOWN = "shutdown"
|
||||
OS_AGENT = "os_agent"
|
||||
|
@ -4,6 +4,8 @@ from datetime import datetime
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from supervisor.dbus.const import MulticastProtocolEnabled
|
||||
|
||||
from ..coresys import CoreSysAttributes
|
||||
from ..exceptions import DBusError, HostError
|
||||
|
||||
@ -22,6 +24,25 @@ class InfoCenter(CoreSysAttributes):
|
||||
"""Return local hostname."""
|
||||
return self.sys_dbus.hostname.hostname
|
||||
|
||||
@property
|
||||
def llmnr_hostname(self) -> Optional[str]:
|
||||
"""Return local llmnr hostname."""
|
||||
return self.sys_dbus.resolved.llmnr_hostname
|
||||
|
||||
@property
|
||||
def broadcast_llmnr(self) -> Optional[bool]:
|
||||
"""Host is broadcasting llmnr name."""
|
||||
if self.sys_dbus.resolved.llmnr:
|
||||
return self.sys_dbus.resolved.llmnr == MulticastProtocolEnabled.YES
|
||||
return None
|
||||
|
||||
@property
|
||||
def broadcast_mdns(self) -> Optional[bool]:
|
||||
"""Host is broadcasting mdns name."""
|
||||
if self.sys_dbus.resolved.multicast_dns:
|
||||
return self.sys_dbus.resolved.multicast_dns == MulticastProtocolEnabled.YES
|
||||
return None
|
||||
|
||||
@property
|
||||
def chassis(self) -> Optional[str]:
|
||||
"""Return local chassis type."""
|
||||
|
@ -93,6 +93,9 @@ class HostManager(CoreSysAttributes):
|
||||
if self.sys_os.available:
|
||||
features.append(HostFeature.HAOS)
|
||||
|
||||
if self.sys_dbus.resolved.is_connected:
|
||||
features.append(HostFeature.RESOLVED)
|
||||
|
||||
return features
|
||||
|
||||
async def reload(
|
||||
|
@ -14,6 +14,8 @@ from awesomeversion import AwesomeVersion
|
||||
import jinja2
|
||||
import voluptuous as vol
|
||||
|
||||
from supervisor.dbus.const import MulticastProtocolEnabled
|
||||
|
||||
from ..const import ATTR_SERVERS, DNS_SUFFIX, LogLevel
|
||||
from ..coresys import CoreSys
|
||||
from ..docker.dns import DockerDNS
|
||||
@ -98,6 +100,22 @@ class PluginDns(PluginBase):
|
||||
"""Return latest version of CoreDNS."""
|
||||
return self.sys_updater.version_dns
|
||||
|
||||
@property
|
||||
def mdns(self) -> bool:
|
||||
"""MDNS hostnames can be resolved."""
|
||||
return self.sys_dbus.resolved.multicast_dns in [
|
||||
MulticastProtocolEnabled.YES,
|
||||
MulticastProtocolEnabled.RESOLVE,
|
||||
]
|
||||
|
||||
@property
|
||||
def llmnr(self) -> bool:
|
||||
"""LLMNR hostnames can be resolved."""
|
||||
return self.sys_dbus.resolved.llmnr in [
|
||||
MulticastProtocolEnabled.YES,
|
||||
MulticastProtocolEnabled.RESOLVE,
|
||||
]
|
||||
|
||||
async def load(self) -> None:
|
||||
"""Load DNS setup."""
|
||||
# Initialize CoreDNS Template
|
||||
|
38
tests/api/test_dns.py
Normal file
38
tests/api/test_dns.py
Normal file
@ -0,0 +1,38 @@
|
||||
"""Test DNS API."""
|
||||
from unittest.mock import PropertyMock, patch
|
||||
|
||||
from supervisor.coresys import CoreSys
|
||||
from supervisor.dbus.const import MulticastProtocolEnabled
|
||||
|
||||
|
||||
async def test_llmnr_mdns_info(api_client, coresys: CoreSys):
|
||||
"""Test llmnr and mdns in info api."""
|
||||
coresys.host.sys_dbus.resolved.is_connected = False
|
||||
|
||||
resp = await api_client.get("/dns/info")
|
||||
result = await resp.json()
|
||||
assert result["data"]["llmnr"] is False
|
||||
assert result["data"]["mdns"] is False
|
||||
|
||||
coresys.host.sys_dbus.resolved.is_connected = True
|
||||
with patch.object(
|
||||
type(coresys.host.sys_dbus.resolved),
|
||||
"llmnr",
|
||||
PropertyMock(return_value=MulticastProtocolEnabled.NO),
|
||||
), patch.object(
|
||||
type(coresys.host.sys_dbus.resolved),
|
||||
"multicast_dns",
|
||||
PropertyMock(return_value=MulticastProtocolEnabled.NO),
|
||||
):
|
||||
resp = await api_client.get("/dns/info")
|
||||
result = await resp.json()
|
||||
assert result["data"]["llmnr"] is False
|
||||
assert result["data"]["mdns"] is False
|
||||
|
||||
await coresys.dbus.resolved.connect()
|
||||
await coresys.dbus.resolved.update()
|
||||
|
||||
resp = await api_client.get("/dns/info")
|
||||
result = await resp.json()
|
||||
assert result["data"]["llmnr"] is True
|
||||
assert result["data"]["mdns"] is True
|
@ -7,18 +7,110 @@ from supervisor.coresys import CoreSys
|
||||
# pylint: disable=protected-access
|
||||
|
||||
|
||||
@pytest.fixture(name="coresys_disk_info")
|
||||
async def fixture_coresys_disk_info(coresys: CoreSys) -> CoreSys:
|
||||
"""Mock basic disk information for host APIs."""
|
||||
coresys.hardware.disk.get_disk_life_time = lambda _: 0
|
||||
coresys.hardware.disk.get_disk_free_space = lambda _: 5000
|
||||
coresys.hardware.disk.get_disk_total_space = lambda _: 50000
|
||||
coresys.hardware.disk.get_disk_used_space = lambda _: 45000
|
||||
|
||||
yield coresys
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_host_info(api_client, tmp_path, coresys: CoreSys):
|
||||
async def test_api_host_info(api_client, coresys_disk_info: CoreSys):
|
||||
"""Test host info api."""
|
||||
coresys = coresys_disk_info
|
||||
|
||||
await coresys.dbus.agent.connect()
|
||||
await coresys.dbus.agent.update()
|
||||
|
||||
coresys.hardware.disk.get_disk_life_time = lambda x: 0
|
||||
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
||||
coresys.hardware.disk.get_disk_total_space = lambda x: 50000
|
||||
coresys.hardware.disk.get_disk_used_space = lambda x: 45000
|
||||
|
||||
resp = await api_client.get("/host/info")
|
||||
result = await resp.json()
|
||||
|
||||
assert result["data"]["apparmor_version"] == "2.13.2"
|
||||
|
||||
|
||||
async def test_api_host_features(api_client, coresys_disk_info: CoreSys):
|
||||
"""Test host info features."""
|
||||
coresys = coresys_disk_info
|
||||
|
||||
coresys.host.sys_dbus.systemd.is_connected = False
|
||||
coresys.host.sys_dbus.network.is_connected = False
|
||||
coresys.host.sys_dbus.hostname.is_connected = False
|
||||
coresys.host.sys_dbus.timedate.is_connected = False
|
||||
coresys.host.sys_dbus.agent.is_connected = False
|
||||
coresys.host.sys_dbus.resolved.is_connected = False
|
||||
|
||||
resp = await api_client.get("/host/info")
|
||||
result = await resp.json()
|
||||
assert "reboot" not in result["data"]["features"]
|
||||
assert "services" not in result["data"]["features"]
|
||||
assert "shutdown" not in result["data"]["features"]
|
||||
assert "network" not in result["data"]["features"]
|
||||
assert "hostname" not in result["data"]["features"]
|
||||
assert "timedate" not in result["data"]["features"]
|
||||
assert "os_agent" not in result["data"]["features"]
|
||||
assert "resolved" not in result["data"]["features"]
|
||||
|
||||
coresys.host.sys_dbus.systemd.is_connected = True
|
||||
coresys.host.supported_features.cache_clear()
|
||||
resp = await api_client.get("/host/info")
|
||||
result = await resp.json()
|
||||
assert "reboot" in result["data"]["features"]
|
||||
assert "services" in result["data"]["features"]
|
||||
assert "shutdown" in result["data"]["features"]
|
||||
|
||||
coresys.host.sys_dbus.network.is_connected = True
|
||||
coresys.host.supported_features.cache_clear()
|
||||
resp = await api_client.get("/host/info")
|
||||
result = await resp.json()
|
||||
assert "network" in result["data"]["features"]
|
||||
|
||||
coresys.host.sys_dbus.hostname.is_connected = True
|
||||
coresys.host.supported_features.cache_clear()
|
||||
resp = await api_client.get("/host/info")
|
||||
result = await resp.json()
|
||||
assert "hostname" in result["data"]["features"]
|
||||
|
||||
coresys.host.sys_dbus.timedate.is_connected = True
|
||||
coresys.host.supported_features.cache_clear()
|
||||
resp = await api_client.get("/host/info")
|
||||
result = await resp.json()
|
||||
assert "timedate" in result["data"]["features"]
|
||||
|
||||
coresys.host.sys_dbus.agent.is_connected = True
|
||||
coresys.host.supported_features.cache_clear()
|
||||
resp = await api_client.get("/host/info")
|
||||
result = await resp.json()
|
||||
assert "os_agent" in result["data"]["features"]
|
||||
|
||||
coresys.host.sys_dbus.resolved.is_connected = True
|
||||
coresys.host.supported_features.cache_clear()
|
||||
resp = await api_client.get("/host/info")
|
||||
result = await resp.json()
|
||||
assert "resolved" in result["data"]["features"]
|
||||
|
||||
|
||||
async def test_api_llmnr_mdns_info(api_client, coresys_disk_info: CoreSys):
|
||||
"""Test llmnr and mdns details in info."""
|
||||
coresys = coresys_disk_info
|
||||
|
||||
coresys.host.sys_dbus.resolved.is_connected = False
|
||||
|
||||
resp = await api_client.get("/host/info")
|
||||
result = await resp.json()
|
||||
assert result["data"]["broadcast_llmnr"] is None
|
||||
assert result["data"]["broadcast_mdns"] is None
|
||||
assert result["data"]["llmnr_hostname"] is None
|
||||
|
||||
coresys.host.sys_dbus.resolved.is_connected = True
|
||||
await coresys.dbus.resolved.connect()
|
||||
await coresys.dbus.resolved.update()
|
||||
|
||||
resp = await api_client.get("/host/info")
|
||||
result = await resp.json()
|
||||
assert result["data"]["broadcast_llmnr"] is True
|
||||
assert result["data"]["broadcast_mdns"] is False
|
||||
assert result["data"]["llmnr_hostname"] == "homeassistant"
|
||||
|
Loading…
x
Reference in New Issue
Block a user