jeans/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-11-26 19:18:10 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
1dc84d6eff67748a376d5b873974b05867362c91
supervisor/tests/api/test_host.py
Jan Čermák 0837e05cb2 Strip ANSI escape color sequences from /latest log responses (#6319) 
* Strip ANSI escape color sequences from /latest log responses

Strip ANSI sequences of CSI commands [1] used for log coloring from
/latest log endpoints. These endpoint were primarily designed for log
downloads and colors are mostly not wanted in those. Add optional
argument for stripping the colors from the logs and enable it for the
/latest endpoints.

[1] https://en.wikipedia.org/wiki/ANSI_escape_code#CSIsection

* Refactor advanced logs' tests to use fixture factory

Introduce `advanced_logs_tester` fixture to simplify testing of advanced logs
in the API tests, declaring all the needed fixture in a single place. # Please
enter the commit message for your changes. Lines starting
2025-11-19 09:39:24 +01:00

851 lines
29 KiB
Python
Raw Blame History

"""Test Host API."""
from collections.abc import AsyncGenerator
from unittest.mock import ANY, MagicMock, patch
from aiohttp.test_utils import TestClient
import pytest
from supervisor.coresys import CoreSys
from supervisor.dbus.resolved import Resolved
from supervisor.homeassistant.api import APIState
from supervisor.host.const import LogFormat, LogFormatter
from supervisor.host.control import SystemControl
from tests.dbus_service_mocks.base import DBusServiceMock
from tests.dbus_service_mocks.systemd import Systemd as SystemdService
DEFAULT_RANGE = "entries=:-99:100"
DEFAULT_RANGE_FOLLOW = "entries=:-99:18446744073709551615"
# pylint: disable=protected-access
@pytest.fixture(name="coresys_disk_info")
async def fixture_coresys_disk_info(coresys: CoreSys) -> AsyncGenerator[CoreSys]:
"""Mock basic disk information for host APIs."""
async def mock_disk_lifetime(_):
return 0
coresys.hardware.disk.get_disk_life_time = mock_disk_lifetime
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: TestClient, coresys_disk_info: CoreSys):
"""Test host info api."""
coresys = coresys_disk_info
await coresys.dbus.agent.connect(coresys.dbus.bus)
await coresys.dbus.agent.update()
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: TestClient, coresys_disk_info: CoreSys, dbus_is_connected
):
"""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
coresys.host.sys_dbus.udisks2.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"]
assert "disk" 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"]
coresys.host.sys_dbus.udisks2.is_connected = True
coresys.host.supported_features.cache_clear()
resp = await api_client.get("/host/info")
result = await resp.json()
assert "disk" in result["data"]["features"]
async def test_api_llmnr_mdns_info(api_client: TestClient, coresys_disk_info: CoreSys):
"""Test llmnr and mdns details in info."""
coresys = coresys_disk_info
# pylint: disable=protected-access
coresys.host.sys_dbus._resolved = Resolved()
# pylint: enable=protected-access
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
await coresys.dbus.resolved.connect(coresys.dbus.bus)
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"
async def test_api_boot_ids_info(api_client: TestClient, journald_logs: MagicMock):
"""Test getting boot IDs."""
resp = await api_client.get("/host/logs/boots")
result = await resp.json()
assert result["data"] == {"boots": {"0": "ccc", "-1": "bbb", "-2": "aaa"}}
async def test_api_identifiers_info(api_client: TestClient, journald_logs: MagicMock):
"""Test getting syslog identifiers."""
resp = await api_client.get("/host/logs/identifiers")
result = await resp.json()
assert result["data"] == {
"identifiers": ["hassio_supervisor", "hassos-config", "kernel"]
}
async def test_api_virtualization_info(
api_client: TestClient,
all_dbus_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
coresys_disk_info: CoreSys,
):
"""Test getting virtualization info."""
systemd_service: SystemdService = all_dbus_services["systemd"]
resp = await api_client.get("/host/info")
result = await resp.json()
assert result["data"]["virtualization"] == ""
systemd_service.virtualization = "vmware"
await coresys_disk_info.dbus.systemd.update()
resp = await api_client.get("/host/info")
result = await resp.json()
assert result["data"]["virtualization"] == "vmware"
async def test_advanced_logs(
api_client: TestClient, coresys: CoreSys, journald_logs: MagicMock
):
"""Test advanced logging API entries with identifier and custom boot."""
await api_client.get("/host/logs")
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
identifier = "dropbear"
await api_client.get(f"/host/logs/identifiers/{identifier}")
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": identifier},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
bootid = "798cc03bcd77465482b6a1c43dc6a5fc"
await api_client.get(f"/host/logs/boots/{bootid}")
journald_logs.assert_called_once_with(
params={
"_BOOT_ID": bootid,
"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers,
},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
await api_client.get(f"/host/logs/boots/{bootid}/identifiers/{identifier}")
journald_logs.assert_called_once_with(
params={"_BOOT_ID": bootid, "SYSLOG_IDENTIFIER": identifier},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
headers = {"Range": "entries=:-19:10"}
await api_client.get("/host/logs", headers=headers)
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers},
range_header=headers["Range"],
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
await api_client.get("/host/logs/follow")
journald_logs.assert_called_once_with(
params={
"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers,
"follow": "",
},
range_header=DEFAULT_RANGE_FOLLOW,
accept=LogFormat.JOURNAL,
)
# Host logs don't have a /latest endpoint
resp = await api_client.get("/host/logs/latest")
assert resp.status == 404
async def test_advaced_logs_query_parameters(
api_client: TestClient,
coresys: CoreSys,
journald_logs: MagicMock,
journal_logs_reader: MagicMock,
):
"""Test advanced logging API entries controlled by query parameters."""
# Check lines query parameter
await api_client.get("/host/logs?lines=53")
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers},
range_header="entries=:-52:53",
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
# Check verbose logs formatter via query parameter
await api_client.get("/host/logs?verbose")
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journal_logs_reader.assert_called_with(ANY, LogFormatter.VERBOSE, False)
journal_logs_reader.reset_mock()
journald_logs.reset_mock()
# Query parameters should take precedence over headers
await api_client.get(
"/host/logs?lines=53&verbose",
headers={
"Range": "entries=:-19:10",
"Accept": "text/plain",
},
)
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers},
range_header="entries=:-52:53",
accept=LogFormat.JOURNAL,
)
journal_logs_reader.assert_called_with(ANY, LogFormatter.VERBOSE, False)
async def test_advanced_logs_boot_id_offset(
api_client: TestClient, coresys: CoreSys, journald_logs: MagicMock
):
"""Test advanced logging API when using an offset as boot ID."""
await api_client.get("/host/logs/boots/0")
journald_logs.assert_called_once_with(
params={
"_BOOT_ID": "ccc",
"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers,
},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
await api_client.get("/host/logs/boots/-2")
journald_logs.assert_called_once_with(
params={
"_BOOT_ID": "aaa",
"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers,
},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
await api_client.get("/host/logs/boots/2")
journald_logs.assert_called_once_with(
params={
"_BOOT_ID": "bbb",
"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers,
},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
async def test_advanced_logs_formatters(
journald_gateway: MagicMock,
api_client: TestClient,
coresys: CoreSys,
journal_logs_reader: MagicMock,
):
"""Test advanced logs formatters varying on Accept header."""
await api_client.get("/host/logs")
journal_logs_reader.assert_called_once_with(ANY, LogFormatter.VERBOSE, False)
journal_logs_reader.reset_mock()
headers = {"Accept": "text/x-log"}
await api_client.get("/host/logs", headers=headers)
journal_logs_reader.assert_called_once_with(ANY, LogFormatter.VERBOSE, False)
journal_logs_reader.reset_mock()
await api_client.get("/host/logs/identifiers/test")
journal_logs_reader.assert_called_once_with(ANY, LogFormatter.PLAIN, False)
journal_logs_reader.reset_mock()
headers = {"Accept": "text/x-log"}
await api_client.get("/host/logs/identifiers/test", headers=headers)
journal_logs_reader.assert_called_once_with(ANY, LogFormatter.VERBOSE, False)
async def test_advanced_logs_errors(coresys: CoreSys, api_client: TestClient):
"""Test advanced logging API errors."""
with patch("supervisor.host.logs.SYSTEMD_JOURNAL_GATEWAYD_SOCKET") as socket:
socket.is_socket.return_value = False
await coresys.host.logs.post_init()
resp = await api_client.get("/host/logs")
assert resp.content_type == "text/plain"
assert resp.status == 400
content = await resp.text()
assert content == "No systemd-journal-gatewayd Unix socket available"
headers = {"Accept": "application/json"}
resp = await api_client.get("/host/logs", headers=headers)
assert resp.content_type == "text/plain"
assert resp.status == 400
content = await resp.text()
assert (
content
== "Invalid content type requested. Only text/plain and text/x-log supported for now."
)
async def test_disk_usage_api(api_client: TestClient, coresys: CoreSys):
"""Test disk usage API endpoint."""
# Mock the disk usage methods
with (
patch.object(coresys.hardware.disk, "disk_usage") as mock_disk_usage,
patch.object(coresys.hardware.disk, "get_dir_sizes") as mock_dir_sizes,
):
# Mock the main disk usage call
mock_disk_usage.return_value = (
1000000000,
500000000,
500000000,
) # 1GB total, 500MB used, 500MB free
# Mock the directory structure sizes for each path
mock_dir_sizes.return_value = [
{
"id": "addons_data",
"label": "Addons Data",
"used_bytes": 100000000,
"children": [
{"id": "addon1", "label": "addon1", "used_bytes": 50000000}
],
},
{
"id": "addons_config",
"label": "Addons Config",
"used_bytes": 200000000,
"children": [
{"id": "media1", "label": "media1", "used_bytes": 100000000}
],
},
{
"id": "media",
"label": "Media",
"used_bytes": 50000000,
"children": [
{"id": "share1", "label": "share1", "used_bytes": 25000000}
],
},
{
"id": "share",
"label": "Share",
"used_bytes": 300000000,
"children": [
{"id": "backup1", "label": "backup1", "used_bytes": 150000000}
],
},
{
"id": "backup",
"label": "Backup",
"used_bytes": 10000000,
"children": [{"id": "ssl1", "label": "ssl1", "used_bytes": 5000000}],
},
{
"id": "ssl",
"label": "SSL",
"used_bytes": 40000000,
"children": [
{
"id": "homeassistant1",
"label": "homeassistant1",
"used_bytes": 20000000,
}
],
},
{
"id": "homeassistant",
"label": "Home Assistant",
"used_bytes": 40000000,
"children": [
{
"id": "homeassistant1",
"label": "homeassistant1",
"used_bytes": 20000000,
}
],
},
]
# Test default max_depth=1
resp = await api_client.get("/host/disks/default/usage")
assert resp.status == 200
result = await resp.json()
assert result["data"]["id"] == "root"
assert result["data"]["label"] == "Root"
assert result["data"]["total_bytes"] == 1000000000
assert result["data"]["used_bytes"] == 500000000
assert "children" in result["data"]
children = result["data"]["children"]
# First child should be system
assert children[0]["id"] == "system"
assert children[0]["label"] == "System"
# Verify all expected directories are present in the remaining children
assert children[1]["id"] == "addons_data"
assert children[2]["id"] == "addons_config"
assert children[3]["id"] == "media"
assert children[4]["id"] == "share"
assert children[5]["id"] == "backup"
assert children[6]["id"] == "ssl"
assert children[7]["id"] == "homeassistant"
# Verify the sizes are correct
assert children[1]["used_bytes"] == 100000000
assert children[2]["used_bytes"] == 200000000
assert children[3]["used_bytes"] == 50000000
assert children[4]["used_bytes"] == 300000000
assert children[5]["used_bytes"] == 10000000
assert children[6]["used_bytes"] == 40000000
assert children[7]["used_bytes"] == 40000000
# Verify system space calculation (total used - sum of known paths)
total_known_space = (
100000000
+ 200000000
+ 50000000
+ 300000000
+ 10000000
+ 40000000
+ 40000000
)
expected_system_space = 500000000 - total_known_space
assert children[0]["used_bytes"] == expected_system_space
# Verify disk_usage was called with supervisor path
mock_disk_usage.assert_called_once_with(coresys.config.path_supervisor)
# Verify get_dir_sizes was called once with all paths
assert mock_dir_sizes.call_count == 1
call_args = mock_dir_sizes.call_args
assert call_args[0][1] == 1 # max_depth parameter
paths_dict = call_args[0][0] # paths dictionary
assert paths_dict["addons_data"] == coresys.config.path_addons_data
assert paths_dict["addons_config"] == coresys.config.path_addon_configs
assert paths_dict["media"] == coresys.config.path_media
assert paths_dict["share"] == coresys.config.path_share
assert paths_dict["backup"] == coresys.config.path_backup
assert paths_dict["ssl"] == coresys.config.path_ssl
assert paths_dict["homeassistant"] == coresys.config.path_homeassistant
async def test_disk_usage_api_with_custom_depth(
api_client: TestClient, coresys: CoreSys
):
"""Test disk usage API endpoint with custom max_depth parameter."""
with (
patch.object(coresys.hardware.disk, "disk_usage") as mock_disk_usage,
patch.object(coresys.hardware.disk, "get_dir_sizes") as mock_dir_sizes,
):
mock_disk_usage.return_value = (1000000000, 500000000, 500000000)
# Mock deeper directory structure
mock_dir_sizes.return_value = [
{
"id": "addons_data",
"label": "Addons Data",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "addons_config",
"label": "Addons Config",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "media",
"label": "Media",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "share",
"label": "Share",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "backup",
"label": "Backup",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "ssl",
"label": "SSL",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "homeassistant",
"label": "Home Assistant",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
]
# Test with custom max_depth=2
resp = await api_client.get("/host/disks/default/usage?max_depth=2")
assert resp.status == 200
result = await resp.json()
assert result["data"]["used_bytes"] == 500000000
assert result["data"]["children"]
# Verify max_depth=2 was passed to get_dir_sizes
assert mock_dir_sizes.call_count == 1
call_args = mock_dir_sizes.call_args
assert call_args[0][1] == 2 # max_depth parameter
async def test_disk_usage_api_invalid_depth(api_client: TestClient, coresys: CoreSys):
"""Test disk usage API endpoint with invalid max_depth parameter."""
with (
patch.object(coresys.hardware.disk, "disk_usage") as mock_disk_usage,
patch.object(coresys.hardware.disk, "get_dir_sizes") as mock_dir_sizes,
):
mock_disk_usage.return_value = (1000000000, 500000000, 500000000)
mock_dir_sizes.return_value = [
{
"id": "addons_data",
"label": "Addons Data",
"used_bytes": 100000000,
},
{
"id": "addons_config",
"label": "Addons Config",
"used_bytes": 100000000,
},
{
"id": "media",
"label": "Media",
"used_bytes": 100000000,
},
{
"id": "share",
"label": "Share",
"used_bytes": 100000000,
},
{
"id": "backup",
"label": "Backup",
"used_bytes": 100000000,
},
{
"id": "ssl",
"label": "SSL",
"used_bytes": 100000000,
},
{
"id": "homeassistant",
"label": "Home Assistant",
"used_bytes": 100000000,
},
]
# Test with invalid max_depth (non-integer)
resp = await api_client.get("/host/disks/default/usage?max_depth=invalid")
assert resp.status == 200
result = await resp.json()
assert result["data"]["used_bytes"] == 500000000
assert result["data"]["children"]
# Should default to max_depth=1 when invalid value is provided
assert mock_dir_sizes.call_count == 1
call_args = mock_dir_sizes.call_args
assert call_args[0][1] == 1 # Should default to 1
async def test_disk_usage_api_empty_directories(
api_client: TestClient, coresys: CoreSys
):
"""Test disk usage API endpoint with empty directories."""
with (
patch.object(coresys.hardware.disk, "disk_usage") as mock_disk_usage,
patch.object(coresys.hardware.disk, "get_dir_sizes") as mock_dir_sizes,
):
mock_disk_usage.return_value = (1000000000, 500000000, 500000000)
# Mock empty directory structures (no children)
mock_dir_sizes.return_value = [
{
"id": "addons_data",
"label": "Addons Data",
"used_bytes": 0,
},
{
"id": "addons_config",
"label": "Addons Config",
"used_bytes": 0,
},
{
"id": "media",
"label": "Media",
"used_bytes": 0,
},
{
"id": "share",
"label": "Share",
"used_bytes": 0,
},
{
"id": "backup",
"label": "Backup",
"used_bytes": 0,
},
{
"id": "ssl",
"label": "SSL",
"used_bytes": 0,
},
{
"id": "homeassistant",
"label": "Home Assistant",
"used_bytes": 0,
},
]
resp = await api_client.get("/host/disks/default/usage")
assert resp.status == 200
result = await resp.json()
assert result["data"]["used_bytes"] == 500000000
children = result["data"]["children"]
# First child should be system with all the space
assert children[0]["id"] == "system"
assert children[0]["used_bytes"] == 500000000
# All other directories should have size 0
for i in range(1, len(children)):
assert children[i]["used_bytes"] == 0
@pytest.mark.parametrize("action", ["reboot", "shutdown"])
async def test_migration_blocks_shutdown(
api_client: TestClient,
coresys: CoreSys,
action: str,
):
"""Test that an offline db migration in progress stops users from shuting down or rebooting system."""
coresys.homeassistant.api.get_api_state.return_value = APIState("NOT_RUNNING", True)
resp = await api_client.post(f"/host/{action}")
assert resp.status == 503
result = await resp.json()
assert (
result["message"]
== "Home Assistant offline database migration in progress, please wait until complete before shutting down host"
)
async def test_force_reboot_during_migration(api_client: TestClient, coresys: CoreSys):
"""Test force option reboots even during a migration."""
coresys.homeassistant.api.get_api_state.return_value = APIState("NOT_RUNNING", True)
with patch.object(SystemControl, "reboot") as reboot:
await api_client.post("/host/reboot", json={"force": True})
reboot.assert_called_once()
async def test_force_shutdown_during_migration(
api_client: TestClient, coresys: CoreSys
):
"""Test force option shutdown even during a migration."""
coresys.homeassistant.api.get_api_state.return_value = APIState("NOT_RUNNING", True)
with patch.object(SystemControl, "shutdown") as shutdown:
await api_client.post("/host/shutdown", json={"force": True})
shutdown.assert_called_once()
Reference in New Issue View Git Blame Copy Permalink