Home-Assistant/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-12-26 13:37:06 +00:00
Code Activity
Files
e2ca357774b33dbb0ed93ec45fa53b148d67e24c
supervisor/tests/api/test_addons.py
Jan Čermák a894c4589e Use Systemd Journal API for all logs endpoints in API (#4972) 
* Use Systemd Journal API for all logs endpoints in API

Replace all logs endpoints using container logs with wrapped
advanced_logs function, adding possibility to get logs from previous
boots and following the logs. Supervisor logs are an excetion where
Docker logs are still used - in case an exception is raised while
accessing the Systemd logs, they're used as fallback - otherwise we
wouldn't have an easy way to see what went wrong.

* Refactor testing of advanced logs endpoints to a common method

* Send error while fetching Supervisor logs to Sentry; minor cleanup

* Properly handle errors and use consistent content type in logs endpoints

* Replace api_process_custom with reworked api_process_raw per @mdegat01 suggestion
2024-04-04 12:09:08 +02:00

288 lines
9.6 KiB
Python
Raw Blame History

"""Test addons api."""
import asyncio
from unittest.mock import MagicMock, PropertyMock, patch
from aiohttp.test_utils import TestClient
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 == 400
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", new=PropertyMock(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()
View Git Blame Copy Permalink