mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-04-19 10:47:15 +00:00
6ef4f3cc67
* Add blockbuster library and find I/O from unit tests * Fix lint and test issue * Fixes from feedback * Avoid modifying webapp object in executor * Split su options validation and only validate timezone on change
438 lines
15 KiB
Python
438 lines
15 KiB
Python
"""Test addons api."""
|
|
|
|
import asyncio
|
|
from unittest.mock import MagicMock, PropertyMock, patch
|
|
|
|
from aiohttp import ClientResponse
|
|
from aiohttp.test_utils import TestClient
|
|
import pytest
|
|
|
|
from supervisor.addons.addon import Addon
|
|
from supervisor.addons.build import AddonBuild
|
|
from supervisor.arch import CpuArch
|
|
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.monitor import DockerContainerStateEvent
|
|
from supervisor.exceptions import HassioError
|
|
from supervisor.store.repository import Repository
|
|
|
|
from ..const import TEST_ADDON_SLUG
|
|
from . import common_test_api_advanced_logs
|
|
|
|
|
|
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, 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(
|
|
api_client: TestClient, journald_logs: MagicMock, install_addon_ssh: Addon
|
|
):
|
|
"""Test addon logs."""
|
|
await common_test_api_advanced_logs(
|
|
"/addons/local_ssh", "addon_local_ssh", api_client, journald_logs
|
|
)
|
|
|
|
|
|
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(CpuArch, "supported", new=PropertyMock(return_value=["amd64"])),
|
|
patch.object(DockerAddon, "run", new=container_events_task),
|
|
):
|
|
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_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"
|
|
)
|
|
|
|
|
|
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("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"
|