mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-12-03 06:28:23 +00:00
684 lines
24 KiB
Python
684 lines
24 KiB
Python
"""Test addons api."""
|
|
|
|
import asyncio
|
|
from unittest.mock import MagicMock, PropertyMock, patch
|
|
|
|
from aiohttp import ClientResponse
|
|
from aiohttp.test_utils import TestClient
|
|
from docker.errors import DockerException
|
|
import pytest
|
|
|
|
from supervisor.addons.addon import Addon
|
|
from supervisor.addons.build import AddonBuild
|
|
from supervisor.arch import CpuArchManager
|
|
from supervisor.const import AddonState
|
|
from supervisor.coresys import CoreSys
|
|
from supervisor.docker.addon import DockerAddon
|
|
from supervisor.docker.const import ContainerState
|
|
from supervisor.docker.manager import CommandReturn
|
|
from supervisor.docker.monitor import DockerContainerStateEvent
|
|
from supervisor.exceptions import HassioError
|
|
from supervisor.store.repository import Repository
|
|
|
|
from ..const import TEST_ADDON_SLUG
|
|
|
|
|
|
def _create_test_event(name: str, state: ContainerState) -> DockerContainerStateEvent:
|
|
"""Create a container state event."""
|
|
return DockerContainerStateEvent(
|
|
name=name,
|
|
state=state,
|
|
id="abc123",
|
|
time=1,
|
|
)
|
|
|
|
|
|
async def test_addons_info(
|
|
api_client: TestClient, coresys: CoreSys, install_addon_ssh: Addon
|
|
):
|
|
"""Test getting addon info."""
|
|
install_addon_ssh.state = AddonState.STOPPED
|
|
install_addon_ssh.ingress_panel = True
|
|
install_addon_ssh.protected = True
|
|
install_addon_ssh.watchdog = False
|
|
|
|
resp = await api_client.get(f"/addons/{TEST_ADDON_SLUG}/info")
|
|
result = await resp.json()
|
|
assert result["data"]["version_latest"] == "9.2.1"
|
|
assert result["data"]["version"] == "9.2.1"
|
|
assert result["data"]["state"] == "stopped"
|
|
assert result["data"]["ingress_panel"] is True
|
|
assert result["data"]["protected"] is True
|
|
assert result["data"]["watchdog"] is False
|
|
|
|
|
|
# DEPRECATED - Remove with legacy routing logic on 1/2023
|
|
async def test_addons_info_not_installed(
|
|
api_client: TestClient, coresys: CoreSys, test_repository: Repository
|
|
):
|
|
"""Test getting addon info for not installed addon."""
|
|
resp = await api_client.get(f"/addons/{TEST_ADDON_SLUG}/info")
|
|
result = await resp.json()
|
|
assert result["data"]["version_latest"] == "9.2.1"
|
|
assert result["data"]["version"] is None
|
|
assert result["data"]["state"] == "unknown"
|
|
assert result["data"]["update_available"] is False
|
|
assert result["data"]["options"] == {
|
|
"authorized_keys": [],
|
|
"apks": [],
|
|
"password": "",
|
|
"server": {"tcp_forwarding": False},
|
|
}
|
|
|
|
|
|
async def test_api_addon_logs(
|
|
advanced_logs_tester,
|
|
install_addon_ssh: Addon,
|
|
):
|
|
"""Test addon logs."""
|
|
await advanced_logs_tester("/addons/local_ssh", "addon_local_ssh")
|
|
|
|
|
|
async def test_api_addon_logs_not_installed(api_client: TestClient):
|
|
"""Test error is returned for non-existing add-on."""
|
|
resp = await api_client.get("/addons/hic_sunt_leones/logs")
|
|
|
|
assert resp.status == 404
|
|
assert resp.content_type == "text/plain"
|
|
content = await resp.text()
|
|
assert content == "Addon hic_sunt_leones does not exist"
|
|
|
|
|
|
async def test_api_addon_logs_error(
|
|
api_client: TestClient,
|
|
journald_logs: MagicMock,
|
|
docker_logs: MagicMock,
|
|
install_addon_ssh: Addon,
|
|
):
|
|
"""Test errors are properly handled for add-on logs."""
|
|
journald_logs.side_effect = HassioError("Something bad happened!")
|
|
resp = await api_client.get("/addons/local_ssh/logs")
|
|
|
|
assert resp.status == 400
|
|
assert resp.content_type == "text/plain"
|
|
content = await resp.text()
|
|
assert content == "Something bad happened!"
|
|
|
|
|
|
async def test_api_addon_start_healthcheck(
|
|
api_client: TestClient,
|
|
coresys: CoreSys,
|
|
install_addon_ssh: Addon,
|
|
container: MagicMock,
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
):
|
|
"""Test starting an addon waits for healthy."""
|
|
install_addon_ssh.path_data.mkdir()
|
|
container.attrs["Config"] = {"Healthcheck": "exists"}
|
|
await install_addon_ssh.load()
|
|
await asyncio.sleep(0)
|
|
assert install_addon_ssh.state == AddonState.STOPPED
|
|
|
|
state_changes: list[AddonState] = []
|
|
_container_events_task: asyncio.Task | None = None
|
|
|
|
async def container_events():
|
|
nonlocal state_changes
|
|
await asyncio.sleep(0.01)
|
|
|
|
await install_addon_ssh.container_state_changed(
|
|
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.RUNNING)
|
|
)
|
|
state_changes.append(install_addon_ssh.state)
|
|
|
|
await install_addon_ssh.container_state_changed(
|
|
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.HEALTHY)
|
|
)
|
|
|
|
async def container_events_task(*args, **kwargs):
|
|
nonlocal _container_events_task
|
|
_container_events_task = asyncio.create_task(container_events())
|
|
|
|
with patch.object(DockerAddon, "run", new=container_events_task):
|
|
resp = await api_client.post("/addons/local_ssh/start")
|
|
|
|
assert state_changes == [AddonState.STARTUP]
|
|
assert install_addon_ssh.state == AddonState.STARTED
|
|
assert resp.status == 200
|
|
|
|
|
|
async def test_api_addon_restart_healthcheck(
|
|
api_client: TestClient,
|
|
coresys: CoreSys,
|
|
install_addon_ssh: Addon,
|
|
container: MagicMock,
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
):
|
|
"""Test restarting an addon waits for healthy."""
|
|
install_addon_ssh.path_data.mkdir()
|
|
container.attrs["Config"] = {"Healthcheck": "exists"}
|
|
await install_addon_ssh.load()
|
|
await asyncio.sleep(0)
|
|
assert install_addon_ssh.state == AddonState.STOPPED
|
|
|
|
state_changes: list[AddonState] = []
|
|
_container_events_task: asyncio.Task | None = None
|
|
|
|
async def container_events():
|
|
nonlocal state_changes
|
|
await asyncio.sleep(0.01)
|
|
|
|
await install_addon_ssh.container_state_changed(
|
|
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.RUNNING)
|
|
)
|
|
state_changes.append(install_addon_ssh.state)
|
|
|
|
await install_addon_ssh.container_state_changed(
|
|
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.HEALTHY)
|
|
)
|
|
|
|
async def container_events_task(*args, **kwargs):
|
|
nonlocal _container_events_task
|
|
_container_events_task = asyncio.create_task(container_events())
|
|
|
|
with patch.object(DockerAddon, "run", new=container_events_task):
|
|
resp = await api_client.post("/addons/local_ssh/restart")
|
|
|
|
assert state_changes == [AddonState.STARTUP]
|
|
assert install_addon_ssh.state == AddonState.STARTED
|
|
assert resp.status == 200
|
|
|
|
|
|
async def test_api_addon_rebuild_healthcheck(
|
|
api_client: TestClient,
|
|
coresys: CoreSys,
|
|
install_addon_ssh: Addon,
|
|
container: MagicMock,
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
):
|
|
"""Test rebuilding an addon waits for healthy."""
|
|
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
|
container.status = "running"
|
|
install_addon_ssh.path_data.mkdir()
|
|
container.attrs["Config"] = {"Healthcheck": "exists"}
|
|
await install_addon_ssh.load()
|
|
await asyncio.sleep(0)
|
|
assert install_addon_ssh.state == AddonState.STARTUP
|
|
|
|
state_changes: list[AddonState] = []
|
|
_container_events_task: asyncio.Task | None = None
|
|
|
|
async def container_events():
|
|
nonlocal state_changes
|
|
|
|
await install_addon_ssh.container_state_changed(
|
|
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.STOPPED)
|
|
)
|
|
state_changes.append(install_addon_ssh.state)
|
|
|
|
await install_addon_ssh.container_state_changed(
|
|
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.RUNNING)
|
|
)
|
|
state_changes.append(install_addon_ssh.state)
|
|
await asyncio.sleep(0)
|
|
|
|
await install_addon_ssh.container_state_changed(
|
|
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.HEALTHY)
|
|
)
|
|
|
|
async def container_events_task(*args, **kwargs):
|
|
nonlocal _container_events_task
|
|
_container_events_task = asyncio.create_task(container_events())
|
|
|
|
with (
|
|
patch.object(AddonBuild, "is_valid", return_value=True),
|
|
patch.object(DockerAddon, "is_running", return_value=False),
|
|
patch.object(Addon, "need_build", new=PropertyMock(return_value=True)),
|
|
patch.object(
|
|
CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"])
|
|
),
|
|
patch.object(DockerAddon, "run", new=container_events_task),
|
|
patch.object(
|
|
coresys.docker,
|
|
"run_command",
|
|
new=PropertyMock(return_value=CommandReturn(0, b"Build successful")),
|
|
),
|
|
patch.object(
|
|
DockerAddon, "healthcheck", new=PropertyMock(return_value={"exists": True})
|
|
),
|
|
patch.object(
|
|
type(coresys.config),
|
|
"local_to_extern_path",
|
|
return_value="/addon/path/on/host",
|
|
),
|
|
):
|
|
resp = await api_client.post("/addons/local_ssh/rebuild")
|
|
|
|
assert state_changes == [AddonState.STOPPED, AddonState.STARTUP]
|
|
assert install_addon_ssh.state == AddonState.STARTED
|
|
assert resp.status == 200
|
|
|
|
|
|
async def test_api_addon_rebuild_force(
|
|
api_client: TestClient,
|
|
coresys: CoreSys,
|
|
install_addon_ssh: Addon,
|
|
container: MagicMock,
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
):
|
|
"""Test rebuilding an image-based addon with force parameter."""
|
|
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
|
container.status = "running"
|
|
install_addon_ssh.path_data.mkdir()
|
|
container.attrs["Config"] = {"Healthcheck": "exists"}
|
|
await install_addon_ssh.load()
|
|
await asyncio.sleep(0)
|
|
assert install_addon_ssh.state == AddonState.STARTUP
|
|
|
|
state_changes: list[AddonState] = []
|
|
_container_events_task: asyncio.Task | None = None
|
|
|
|
async def container_events():
|
|
nonlocal state_changes
|
|
|
|
await install_addon_ssh.container_state_changed(
|
|
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.STOPPED)
|
|
)
|
|
state_changes.append(install_addon_ssh.state)
|
|
|
|
await install_addon_ssh.container_state_changed(
|
|
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.RUNNING)
|
|
)
|
|
state_changes.append(install_addon_ssh.state)
|
|
await asyncio.sleep(0)
|
|
|
|
await install_addon_ssh.container_state_changed(
|
|
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.HEALTHY)
|
|
)
|
|
|
|
async def container_events_task(*args, **kwargs):
|
|
nonlocal _container_events_task
|
|
_container_events_task = asyncio.create_task(container_events())
|
|
|
|
# Test 1: Without force, image-based addon should fail
|
|
with (
|
|
patch.object(AddonBuild, "is_valid", return_value=True),
|
|
patch.object(DockerAddon, "is_running", return_value=False),
|
|
patch.object(
|
|
Addon, "need_build", new=PropertyMock(return_value=False)
|
|
), # Image-based
|
|
patch.object(
|
|
CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"])
|
|
),
|
|
):
|
|
resp = await api_client.post("/addons/local_ssh/rebuild")
|
|
|
|
assert resp.status == 400
|
|
result = await resp.json()
|
|
assert "Can't rebuild a image based add-on" in result["message"]
|
|
|
|
# Reset state for next test
|
|
state_changes.clear()
|
|
|
|
# Test 2: With force=True, image-based addon should succeed
|
|
with (
|
|
patch.object(AddonBuild, "is_valid", return_value=True),
|
|
patch.object(DockerAddon, "is_running", return_value=False),
|
|
patch.object(
|
|
Addon, "need_build", new=PropertyMock(return_value=False)
|
|
), # Image-based
|
|
patch.object(
|
|
CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"])
|
|
),
|
|
patch.object(DockerAddon, "run", new=container_events_task),
|
|
patch.object(
|
|
coresys.docker,
|
|
"run_command",
|
|
new=PropertyMock(return_value=CommandReturn(0, b"Build successful")),
|
|
),
|
|
patch.object(
|
|
DockerAddon, "healthcheck", new=PropertyMock(return_value={"exists": True})
|
|
),
|
|
patch.object(
|
|
type(coresys.config),
|
|
"local_to_extern_path",
|
|
return_value="/addon/path/on/host",
|
|
),
|
|
):
|
|
resp = await api_client.post("/addons/local_ssh/rebuild", json={"force": True})
|
|
|
|
assert state_changes == [AddonState.STOPPED, AddonState.STARTUP]
|
|
assert install_addon_ssh.state == AddonState.STARTED
|
|
assert resp.status == 200
|
|
|
|
await _container_events_task
|
|
|
|
|
|
async def test_api_addon_uninstall(
|
|
api_client: TestClient,
|
|
coresys: CoreSys,
|
|
install_addon_example: Addon,
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
):
|
|
"""Test uninstall."""
|
|
install_addon_example.data["map"].append(
|
|
{"type": "addon_config", "read_only": False}
|
|
)
|
|
install_addon_example.path_config.mkdir()
|
|
(test_file := install_addon_example.path_config / "test.txt").touch()
|
|
|
|
resp = await api_client.post("/addons/local_example/uninstall")
|
|
assert resp.status == 200
|
|
assert not coresys.addons.get("local_example", local_only=True)
|
|
assert test_file.exists()
|
|
|
|
|
|
async def test_api_addon_uninstall_remove_config(
|
|
api_client: TestClient,
|
|
coresys: CoreSys,
|
|
install_addon_example: Addon,
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
):
|
|
"""Test uninstall and remove config."""
|
|
install_addon_example.data["map"].append(
|
|
{"type": "addon_config", "read_only": False}
|
|
)
|
|
(test_folder := install_addon_example.path_config).mkdir()
|
|
(install_addon_example.path_config / "test.txt").touch()
|
|
|
|
resp = await api_client.post(
|
|
"/addons/local_example/uninstall", json={"remove_config": True}
|
|
)
|
|
assert resp.status == 200
|
|
assert not coresys.addons.get("local_example", local_only=True)
|
|
assert not test_folder.exists()
|
|
|
|
|
|
async def test_api_addon_system_managed(
|
|
api_client: TestClient,
|
|
coresys: CoreSys,
|
|
install_addon_example: Addon,
|
|
caplog: pytest.LogCaptureFixture,
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
):
|
|
"""Test setting system managed for an addon."""
|
|
install_addon_example.data["ingress"] = False
|
|
|
|
# Not system managed
|
|
resp = await api_client.get("/addons")
|
|
body = await resp.json()
|
|
assert body["data"]["addons"][0]["slug"] == "local_example"
|
|
assert body["data"]["addons"][0]["system_managed"] is False
|
|
|
|
resp = await api_client.get("/addons/local_example/info")
|
|
body = await resp.json()
|
|
assert body["data"]["system_managed"] is False
|
|
assert body["data"]["system_managed_config_entry"] is None
|
|
|
|
# Mark as system managed
|
|
coresys.addons.data.save_data.reset_mock()
|
|
resp = await api_client.post(
|
|
"/addons/local_example/sys_options",
|
|
json={"system_managed": True, "system_managed_config_entry": "abc123"},
|
|
)
|
|
assert resp.status == 200
|
|
coresys.addons.data.save_data.assert_called_once()
|
|
|
|
resp = await api_client.get("/addons")
|
|
body = await resp.json()
|
|
assert body["data"]["addons"][0]["system_managed"] is True
|
|
|
|
resp = await api_client.get("/addons/local_example/info")
|
|
body = await resp.json()
|
|
assert body["data"]["system_managed"] is True
|
|
assert body["data"]["system_managed_config_entry"] == "abc123"
|
|
|
|
# Revert. Log that cannot have a config entry if not system managed
|
|
coresys.addons.data.save_data.reset_mock()
|
|
resp = await api_client.post(
|
|
"/addons/local_example/sys_options",
|
|
json={"system_managed": False, "system_managed_config_entry": "abc123"},
|
|
)
|
|
assert resp.status == 200
|
|
coresys.addons.data.save_data.assert_called_once()
|
|
assert "Ignoring system managed config entry" in caplog.text
|
|
|
|
resp = await api_client.get("/addons")
|
|
body = await resp.json()
|
|
assert body["data"]["addons"][0]["system_managed"] is False
|
|
|
|
resp = await api_client.get("/addons/local_example/info")
|
|
body = await resp.json()
|
|
assert body["data"]["system_managed"] is False
|
|
assert body["data"]["system_managed_config_entry"] is None
|
|
|
|
|
|
async def test_addon_options_boot_mode_manual_only_invalid(
|
|
api_client: TestClient, install_addon_example: Addon
|
|
):
|
|
"""Test changing boot mode is invalid if set to manual only."""
|
|
install_addon_example.data["ingress"] = False
|
|
resp = await api_client.get("/addons/local_example/info")
|
|
assert resp.status == 200
|
|
body = await resp.json()
|
|
assert body["data"]["boot"] == "manual"
|
|
assert body["data"]["boot_config"] == "manual_only"
|
|
|
|
resp = await api_client.post("/addons/local_example/options", json={"boot": "auto"})
|
|
assert resp.status == 400
|
|
body = await resp.json()
|
|
assert (
|
|
body["message"]
|
|
== "Addon local_example boot option is set to manual_only so it cannot be changed"
|
|
)
|
|
assert body["error_key"] == "addon_boot_config_cannot_change_error"
|
|
assert body["extra_fields"] == {
|
|
"addon": "local_example",
|
|
"boot_config": "manual_only",
|
|
}
|
|
|
|
|
|
async def get_message(resp: ClientResponse, json_expected: bool) -> str:
|
|
"""Get message from response based on response type."""
|
|
if json_expected:
|
|
body = await resp.json()
|
|
return body["message"]
|
|
return await resp.text()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "url", "json_expected"),
|
|
[
|
|
("get", "/addons/bad/info", True),
|
|
("post", "/addons/bad/uninstall", True),
|
|
("post", "/addons/bad/start", True),
|
|
("post", "/addons/bad/stop", True),
|
|
("post", "/addons/bad/restart", True),
|
|
("post", "/addons/bad/options", True),
|
|
("post", "/addons/bad/sys_options", True),
|
|
("post", "/addons/bad/options/validate", True),
|
|
("post", "/addons/bad/rebuild", True),
|
|
("post", "/addons/bad/stdin", True),
|
|
("post", "/addons/bad/security", True),
|
|
("get", "/addons/bad/stats", True),
|
|
("get", "/addons/bad/logs", False),
|
|
("get", "/addons/bad/logs/follow", False),
|
|
("get", "/addons/bad/logs/boots/1", False),
|
|
("get", "/addons/bad/logs/boots/1/follow", False),
|
|
],
|
|
)
|
|
async def test_addon_not_found(
|
|
api_client: TestClient, method: str, url: str, json_expected: bool
|
|
):
|
|
"""Test addon not found error."""
|
|
resp = await api_client.request(method, url)
|
|
assert resp.status == 404
|
|
assert await get_message(resp, json_expected) == "Addon bad does not exist"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "url", "json_expected"),
|
|
[
|
|
("post", "/addons/local_ssh/uninstall", True),
|
|
("post", "/addons/local_ssh/start", True),
|
|
("post", "/addons/local_ssh/stop", True),
|
|
("post", "/addons/local_ssh/restart", True),
|
|
("post", "/addons/local_ssh/options", True),
|
|
("post", "/addons/local_ssh/sys_options", True),
|
|
("post", "/addons/local_ssh/options/validate", True),
|
|
("post", "/addons/local_ssh/rebuild", True),
|
|
("post", "/addons/local_ssh/stdin", True),
|
|
("post", "/addons/local_ssh/security", True),
|
|
("get", "/addons/local_ssh/stats", True),
|
|
("get", "/addons/local_ssh/logs", False),
|
|
("get", "/addons/local_ssh/logs/follow", False),
|
|
("get", "/addons/local_ssh/logs/boots/1", False),
|
|
("get", "/addons/local_ssh/logs/boots/1/follow", False),
|
|
],
|
|
)
|
|
@pytest.mark.usefixtures("test_repository")
|
|
async def test_addon_not_installed(
|
|
api_client: TestClient, method: str, url: str, json_expected: bool
|
|
):
|
|
"""Test addon not installed error."""
|
|
resp = await api_client.request(method, url)
|
|
assert resp.status == 400
|
|
assert await get_message(resp, json_expected) == "Addon is not installed"
|
|
|
|
|
|
async def test_addon_set_options(api_client: TestClient, install_addon_example: Addon):
|
|
"""Test setting options for an addon."""
|
|
resp = await api_client.post(
|
|
"/addons/local_example/options", json={"options": {"message": "test"}}
|
|
)
|
|
assert resp.status == 200
|
|
assert install_addon_example.options == {"message": "test"}
|
|
|
|
|
|
async def test_addon_set_options_error(
|
|
api_client: TestClient, install_addon_example: Addon
|
|
):
|
|
"""Test setting options for an addon."""
|
|
resp = await api_client.post(
|
|
"/addons/local_example/options", json={"options": {"message": True}}
|
|
)
|
|
assert resp.status == 400
|
|
body = await resp.json()
|
|
assert (
|
|
body["message"]
|
|
== "Add-on local_example has invalid options: not a valid value. Got {'message': True}"
|
|
)
|
|
assert body["error_key"] == "addon_configuration_invalid_error"
|
|
assert body["extra_fields"] == {
|
|
"addon": "local_example",
|
|
"validation_error": "not a valid value. Got {'message': True}",
|
|
}
|
|
|
|
|
|
async def test_addon_start_options_error(
|
|
api_client: TestClient,
|
|
install_addon_example: Addon,
|
|
caplog: pytest.LogCaptureFixture,
|
|
):
|
|
"""Test error writing options when trying to start addon."""
|
|
install_addon_example.options = {"message": "hello"}
|
|
|
|
# Simulate OS error trying to write the file
|
|
with patch("supervisor.utils.json.atomic_write", side_effect=OSError("fail")):
|
|
resp = await api_client.post("/addons/local_example/start")
|
|
assert resp.status == 500
|
|
body = await resp.json()
|
|
assert (
|
|
body["message"]
|
|
== "An unknown error occurred with addon local_example. Check supervisor logs for details (check with 'ha supervisor logs')"
|
|
)
|
|
assert body["error_key"] == "addon_unknown_error"
|
|
assert body["extra_fields"] == {
|
|
"addon": "local_example",
|
|
"logs_command": "ha supervisor logs",
|
|
}
|
|
assert "Add-on local_example can't write options" in caplog.text
|
|
|
|
# Simulate an update with a breaking change for options schema creating failure on start
|
|
caplog.clear()
|
|
install_addon_example.data["schema"] = {"message": "bool"}
|
|
resp = await api_client.post("/addons/local_example/start")
|
|
assert resp.status == 400
|
|
body = await resp.json()
|
|
assert (
|
|
body["message"]
|
|
== "Add-on local_example has invalid options: expected boolean. Got {'message': 'hello'}"
|
|
)
|
|
assert body["error_key"] == "addon_configuration_invalid_error"
|
|
assert body["extra_fields"] == {
|
|
"addon": "local_example",
|
|
"validation_error": "expected boolean. Got {'message': 'hello'}",
|
|
}
|
|
assert (
|
|
"Add-on local_example has invalid options: expected boolean. Got {'message': 'hello'}"
|
|
in caplog.text
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(("method", "action"), [("get", "stats"), ("post", "stdin")])
|
|
@pytest.mark.usefixtures("install_addon_example")
|
|
async def test_addon_not_running_error(
|
|
api_client: TestClient, method: str, action: str
|
|
):
|
|
"""Test addon not running error for endpoints that require that."""
|
|
with patch.object(Addon, "with_stdin", new=PropertyMock(return_value=True)):
|
|
resp = await api_client.request(method, f"/addons/local_example/{action}")
|
|
|
|
assert resp.status == 400
|
|
body = await resp.json()
|
|
assert body["message"] == "Add-on local_example is not running"
|
|
assert body["error_key"] == "addon_not_running_error"
|
|
assert body["extra_fields"] == {"addon": "local_example"}
|
|
|
|
|
|
@pytest.mark.usefixtures("install_addon_example")
|
|
async def test_addon_write_stdin_not_supported_error(api_client: TestClient):
|
|
"""Test error when trying to write stdin to addon that does not support it."""
|
|
resp = await api_client.post("/addons/local_example/stdin")
|
|
assert resp.status == 400
|
|
body = await resp.json()
|
|
assert body["message"] == "Add-on local_example does not support writing to stdin"
|
|
assert body["error_key"] == "addon_not_supported_write_stdin_error"
|
|
assert body["extra_fields"] == {"addon": "local_example"}
|
|
|
|
|
|
@pytest.mark.usefixtures("install_addon_ssh")
|
|
async def test_addon_rebuild_fails_error(api_client: TestClient, coresys: CoreSys):
|
|
"""Test error when build fails during rebuild for addon."""
|
|
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
|
coresys.docker.containers.run.side_effect = DockerException("fail")
|
|
|
|
with (
|
|
patch.object(
|
|
CpuArchManager, "supported", new=PropertyMock(return_value=["aarch64"])
|
|
),
|
|
patch.object(
|
|
CpuArchManager, "default", new=PropertyMock(return_value="aarch64")
|
|
),
|
|
patch.object(AddonBuild, "get_docker_args", return_value={}),
|
|
):
|
|
resp = await api_client.post("/addons/local_ssh/rebuild")
|
|
assert resp.status == 500
|
|
body = await resp.json()
|
|
assert (
|
|
body["message"]
|
|
== "An unknown error occurred while trying to build the image for addon local_ssh. Check supervisor logs for details (check with 'ha supervisor logs')"
|
|
)
|
|
assert body["error_key"] == "addon_build_failed_unknown_error"
|
|
assert body["extra_fields"] == {
|
|
"addon": "local_ssh",
|
|
"logs_command": "ha supervisor logs",
|
|
}
|