Home-Assistant/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2026-07-02 19:35:42 +01:00
Code Activity
Files
app-network-isolation
supervisor/tests/api/test_apps.py
T
Stefan Agner a973d22e35 Derive App state from container state (#6890) 
* Derive App state from container state

The App.state setter mixed two responsibilities: it both mutated a
private `_state` field and dispatched side effects (WebSocket events,
issue dismissal, startup_event signaling). On top of that, an installed
but never-started app stayed in AppState.UNKNOWN forever, because the
attach() image-only fallback never fires a container state-change event
and the AppState therefore kept its constructor default. Conceptually,
ContainerState.UNKNOWN ("container does not exist") and AppState.UNKNOWN
("nothing observed yet") happened to share a name but meant different
things, which made the distinction easy to lose.

Make App.state a pure derived property. The source of truth is the last
observed ContainerState (cached on the App), plus a sticky operation-
error flag for start/stop failures that the docker event stream cannot
reflect. When no container has been observed yet, the derivation falls
back to install signals: an attached instance (image present) is
STOPPED, otherwise UNKNOWN. As a side effect, an installed-but-never-
started app now correctly reports STOPPED instead of UNKNOWN.

container_state_changed updates the cached container state and routes
all side effects through a single _emit_state_change helper that diffs
old vs new derived state. The two start/stop failure paths route
through _set_operation_error. Uninstall resets the cached signals so
the derivation naturally returns UNKNOWN.

Tests use a new tests/common.force_app_state helper that pokes the
underlying signals directly; the production class no longer carries
test-only setters.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* Fix App state drive to AppState.UNKNOWN

* Unify state mutation through _update_state

Previously, state-driving signal changes were spread across two helpers
(_set_operation_error, _emit_state_change(old_state)) and required each
caller to capture self.state before mutating a private field — leaking
implementation details to call sites and raising the "why am I emitting
the old state?" question pointed out in code review.

Replace both helpers with a single _update_state(*, container_state=,
operation_error=) entry point. Callers describe what changed via
keyword arguments (None leaves a signal untouched); the helper captures
the previous state, applies the updates, recomputes the derived state
and emits side effects if anything changed.

Diff against a tracked _last_state instead of a freshly derived
"current" state, so that an out-of-band mutation between updates does
not silently shift the comparison baseline. The concrete case is
App.uninstall: instance.remove() clears the docker meta mid-flow, which
would otherwise reshape the derivation (RUNNING with no healthcheck
becomes STARTED instead of STARTUP) and suppress the STARTUP transition
that resolves the start-wait task. As a side effect, the initial
UNKNOWN -> STOPPED transition on attach is also now reliably emitted.

Switch the uninstall path to ContainerState.UNKNOWN ("we know there is
no container") rather than the constructor sentinel None.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* Cache app state instead of deriving on every read

Building on the previous commit, make App.state a plain read of a
cached _state field rather than re-deriving on every property access.
The derivation moves to _derive_state(), and _update_state() is the
sole place that recomputes and assigns _state, so the value consumers
read always matches what was last emitted to listeners.

This removes the _last_state bookkeeping introduced previously: with a
single cached value there is no longer a separate "derived now" vs
"last emitted" distinction to reconcile, and out-of-band mutations
(e.g. instance.remove() clearing _meta during uninstall) can no longer
silently shift what state returns between updates.

Call _update_state() at the end of load() so the cached state settles
once attach() has run. Image-only attaches do not fire a docker event,
so without this an installed app would stay in the constructor-default
UNKNOWN until first start; this also makes the initial transition on
attach observable to listeners.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Pass operation error to _derive_state instead of storing it

The two state-driving signals were not symmetric. _container_state is
genuinely persisted state ("the last thing docker told us") that
re-derivation legitimately reads across calls. _operation_error, on the
other hand, is a momentary "force ERROR for this transition" signal; the
persistence of an error condition already lives in the cached _state.

Storing it as an instance attribute implied a sticky cross-call behavior
that no call path actually exercised: every caller either set it
explicitly right before deriving (start/stop failures, container events)
or ran argless only at load time, where no failure has occurred.

Drop the _operation_error field and pass operation_error as a parameter
to _derive_state(), defaulting to False in _update_state(). A container
observation now supersedes a prior error implicitly via the default,
which lets the container-event and uninstall call sites drop their
explicit operation_error=False.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Settle load state synchronously from current_state

The argless _update_state() settle at the end of load() raced attach()'s
container-state event. attach() fires DOCKER_CONTAINER_STATE_CHANGE via
the bus, which schedules the container_state_changed listener as a task
rather than running it inline. In the deprecated-arch early-return path
there is no await between attach() and the settle, so the listener had
not run yet: _container_state was still None and the settle derived
STOPPED (instance attached) — emitting a transient UNKNOWN->STOPPED even
for a running container before the listener corrected it. The main path
only avoided this incidentally, by having awaits (check_image,
save_persist) in between for the listener to run.

Derive the load-time state synchronously from instance.current_state()
instead of relying on the asynchronously delivered event. current_state()
returns the real container state, or UNKNOWN when only an image is
present (which derives to STOPPED), so both paths settle correctly
without racing the event.

Add a regression test that loading a running container settles to
STARTED, and mock current_state() in the state-listener test which
relies on a clean UNKNOWN baseline.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-06-01 19:50:06 +02:00

724 lines
26 KiB
Python
Raw Permalink Blame History

"""Test apps api."""
import asyncio
from collections.abc import Awaitable, Callable
from pathlib import PurePath
from unittest.mock import MagicMock, PropertyMock, patch
import aiodocker
from aiodocker.containers import DockerContainer
from aiohttp import ClientResponse
from aiohttp.test_utils import TestClient
import pytest
from supervisor.apps.app import App
from supervisor.apps.build import AppBuild
from supervisor.arch import CpuArchManager
from supervisor.const import AppState, CpuArch
from supervisor.coresys import CoreSys
from supervisor.docker.app import DockerApp
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 ..common import force_app_state
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_apps_info(
app_api_client_with_root: tuple[TestClient, str], install_app_ssh: App
):
"""Test getting app info."""
client, root = app_api_client_with_root
force_app_state(install_app_ssh, AppState.STOPPED)
install_app_ssh.ingress_panel = True
install_app_ssh.protected = True
install_app_ssh.watchdog = False
resp = await client.get(f"{root}/{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_apps_info_not_installed(
api_client: TestClient, coresys: CoreSys, test_repository: Repository
):
"""Test getting app info for not installed app."""
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},
}
@pytest.mark.usefixtures("install_app_ssh")
async def test_api_app_logs(
advanced_logs_tester: Callable[[str, str], Awaitable[None]],
):
"""Test app logs."""
await advanced_logs_tester(
"/addons/local_ssh", "addon_local_ssh", v2_path_prefix="/apps/local_ssh"
)
async def test_api_app_logs_not_installed(api_client: TestClient):
"""Test error is returned for non-existing app."""
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 == "App hic_sunt_leones does not exist"
@pytest.mark.usefixtures("docker_logs", "install_app_ssh")
async def test_api_app_logs_error(api_client: TestClient, journald_logs: MagicMock):
"""Test errors are properly handled for app 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!"
@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_api_app_start_healthcheck(
api_client: TestClient, install_app_ssh: App, container: DockerContainer
):
"""Test starting an app waits for healthy."""
install_app_ssh.path_data.mkdir()
container.show.return_value["Config"] = {"Healthcheck": "exists"}
await install_app_ssh.load()
await asyncio.sleep(0)
assert install_app_ssh.state == AppState.STOPPED
state_changes: list[AppState] = []
_container_events_task: asyncio.Task | None = None
async def container_events():
nonlocal state_changes
await asyncio.sleep(0)
await install_app_ssh.container_state_changed(
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.RUNNING)
)
state_changes.append(install_app_ssh.state)
await install_app_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(DockerApp, "run", new=container_events_task):
resp = await api_client.post("/addons/local_ssh/start")
assert state_changes == [AppState.STARTUP]
assert install_app_ssh.state == AppState.STARTED
assert resp.status == 200
@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_api_app_restart_healthcheck(
api_client: TestClient, install_app_ssh: App, container: DockerContainer
):
"""Test restarting an app waits for healthy."""
install_app_ssh.path_data.mkdir()
container.show.return_value["Config"] = {"Healthcheck": "exists"}
await install_app_ssh.load()
await asyncio.sleep(0)
assert install_app_ssh.state == AppState.STOPPED
state_changes: list[AppState] = []
_container_events_task: asyncio.Task | None = None
async def container_events():
nonlocal state_changes
await asyncio.sleep(0)
await install_app_ssh.container_state_changed(
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.RUNNING)
)
state_changes.append(install_app_ssh.state)
await install_app_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(DockerApp, "run", new=container_events_task):
resp = await api_client.post("/addons/local_ssh/restart")
assert state_changes == [AppState.STARTUP]
assert install_app_ssh.state == AppState.STARTED
assert resp.status == 200
@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_api_app_rebuild_healthcheck(
api_client: TestClient,
coresys: CoreSys,
install_app_ssh: App,
container: DockerContainer,
):
"""Test rebuilding an app waits for healthy."""
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
container.show.return_value["State"]["Status"] = "running"
container.show.return_value["State"]["Running"] = True
install_app_ssh.path_data.mkdir()
container.show.return_value["Config"] = {"Healthcheck": "exists"}
await install_app_ssh.load()
await asyncio.sleep(0)
assert install_app_ssh.state == AppState.STARTUP
state_changes: list[AppState] = []
_container_events_task: asyncio.Task | None = None
async def container_events():
nonlocal state_changes
await install_app_ssh.container_state_changed(
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.STOPPED)
)
state_changes.append(install_app_ssh.state)
await install_app_ssh.container_state_changed(
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.RUNNING)
)
state_changes.append(install_app_ssh.state)
await asyncio.sleep(0)
await install_app_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(AppBuild, "is_valid", return_value=True),
patch.object(DockerApp, "is_running", return_value=False),
patch.object(App, "need_build", new=PropertyMock(return_value=True)),
patch.object(
CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"])
),
patch.object(DockerApp, "run", new=container_events_task),
patch.object(
coresys.docker,
"run_command",
return_value=CommandReturn(0, ["Build successful"]),
),
patch.object(
DockerApp, "healthcheck", new=PropertyMock(return_value={"exists": True})
),
patch.object(
type(coresys.config),
"local_to_extern_path",
return_value=PurePath("/addon/path/on/host"),
),
):
resp = await api_client.post("/addons/local_ssh/rebuild")
assert state_changes == [AppState.STOPPED, AppState.STARTUP]
assert install_app_ssh.state == AppState.STARTED
assert resp.status == 200
@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_api_app_rebuild_force(
api_client: TestClient,
coresys: CoreSys,
install_app_ssh: App,
container: DockerContainer,
):
"""Test rebuilding an image-based app with force parameter."""
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
container.show.return_value["State"]["Status"] = "running"
container.show.return_value["State"]["Running"] = True
install_app_ssh.path_data.mkdir()
container.show.return_value["Config"] = {"Healthcheck": "exists"}
await install_app_ssh.load()
await asyncio.sleep(0)
assert install_app_ssh.state == AppState.STARTUP
state_changes: list[AppState] = []
_container_events_task: asyncio.Task | None = None
async def container_events():
nonlocal state_changes
await install_app_ssh.container_state_changed(
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.STOPPED)
)
state_changes.append(install_app_ssh.state)
await install_app_ssh.container_state_changed(
_create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.RUNNING)
)
state_changes.append(install_app_ssh.state)
await asyncio.sleep(0)
await install_app_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 app should fail
with (
patch.object(AppBuild, "is_valid", return_value=True),
patch.object(DockerApp, "is_running", return_value=False),
patch.object(
App, "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 result["message"] == "Cannot rebuild app Terminal & SSH, it is image-based"
# Reset state for next test
state_changes.clear()
# Test 2: With force=True, image-based app should succeed
with (
patch.object(AppBuild, "is_valid", return_value=True),
patch.object(DockerApp, "is_running", return_value=False),
patch.object(
App, "need_build", new=PropertyMock(return_value=False)
), # Image-based
patch.object(
CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"])
),
patch.object(DockerApp, "run", new=container_events_task),
patch.object(
coresys.docker,
"run_command",
return_value=CommandReturn(0, ["Build successful"]),
),
patch.object(
DockerApp, "healthcheck", new=PropertyMock(return_value={"exists": True})
),
patch.object(
type(coresys.config),
"local_to_extern_path",
return_value=PurePath("/addon/path/on/host"),
),
):
resp = await api_client.post("/addons/local_ssh/rebuild", json={"force": True})
assert state_changes == [AppState.STOPPED, AppState.STARTUP]
assert install_app_ssh.state == AppState.STARTED
assert resp.status == 200
await _container_events_task
@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_api_app_uninstall(
app_api_client_with_root: tuple[TestClient, str],
coresys: CoreSys,
install_app_example: App,
):
"""Test uninstall."""
client, root = app_api_client_with_root
install_app_example.data["map"].append({"type": "addon_config", "read_only": False})
install_app_example.path_config.mkdir()
(test_file := install_app_example.path_config / "test.txt").touch()
resp = await client.post(f"{root}/local_example/uninstall")
assert resp.status == 200
assert not coresys.apps.get("local_example", local_only=True)
assert test_file.exists()
@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_api_app_uninstall_remove_config(
app_api_client_with_root: tuple[TestClient, str],
coresys: CoreSys,
install_app_example: App,
):
"""Test uninstall and remove config."""
client, root = app_api_client_with_root
install_app_example.data["map"].append({"type": "addon_config", "read_only": False})
(test_folder := install_app_example.path_config).mkdir()
(install_app_example.path_config / "test.txt").touch()
resp = await client.post(
f"{root}/local_example/uninstall", json={"remove_config": True}
)
assert resp.status == 200
assert not coresys.apps.get("local_example", local_only=True)
assert not test_folder.exists()
@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_api_app_system_managed(
api_client: TestClient,
coresys: CoreSys,
install_app_example: App,
caplog: pytest.LogCaptureFixture,
):
"""Test setting system managed for an app."""
install_app_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.apps.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.apps.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.apps.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.apps.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_app_options_boot_mode_manual_only_invalid(
api_client: TestClient, install_app_example: App
):
"""Test changing boot mode is invalid if set to manual only."""
install_app_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"]
== "App 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", "action", "json_expected"),
[
("get", "bad/info", True),
("post", "bad/uninstall", True),
("post", "bad/start", True),
("post", "bad/stop", True),
("post", "bad/restart", True),
("post", "bad/options", True),
("post", "bad/sys_options", True),
("post", "bad/options/validate", True),
("post", "bad/rebuild", True),
("post", "bad/stdin", True),
("post", "bad/security", True),
("get", "bad/stats", True),
("get", "bad/logs", False),
("get", "bad/logs/follow", False),
("get", "bad/logs/boots/1", False),
("get", "bad/logs/boots/1/follow", False),
],
)
async def test_app_not_found(
app_api_client_with_root: tuple[TestClient, str],
method: str,
action: str,
json_expected: bool,
):
"""Test app not found error."""
client, root = app_api_client_with_root
resp = await client.request(method, f"{root}/{action}")
assert resp.status == 404
assert await get_message(resp, json_expected) == "App bad does not exist"
@pytest.mark.parametrize(
("method", "action", "json_expected"),
[
("post", "local_ssh/uninstall", True),
("post", "local_ssh/start", True),
("post", "local_ssh/stop", True),
("post", "local_ssh/restart", True),
("post", "local_ssh/options", True),
("post", "local_ssh/sys_options", True),
("post", "local_ssh/options/validate", True),
("post", "local_ssh/rebuild", True),
("post", "local_ssh/stdin", True),
("post", "local_ssh/security", True),
("get", "local_ssh/stats", True),
("get", "local_ssh/logs", False),
("get", "local_ssh/logs/follow", False),
("get", "local_ssh/logs/boots/1", False),
("get", "local_ssh/logs/boots/1/follow", False),
],
)
@pytest.mark.usefixtures("test_repository")
async def test_app_not_installed(
app_api_client_with_root: tuple[TestClient, str],
method: str,
action: str,
json_expected: bool,
):
"""Test app not installed error."""
client, root = app_api_client_with_root
resp = await client.request(method, f"{root}/{action}")
assert resp.status == 400
assert await get_message(resp, json_expected) == "App is not installed"
async def test_app_set_options(
app_api_client_with_root: tuple[TestClient, str], install_app_example: App
):
"""Test setting options for an app."""
client, root = app_api_client_with_root
resp = await client.post(
f"{root}/local_example/options", json={"options": {"message": "test"}}
)
assert resp.status == 200
assert install_app_example.options == {"message": "test"}
async def test_app_reset_options(
app_api_client_with_root: tuple[TestClient, str], install_app_example: App
):
"""Test resetting options for an app to defaults.
Fixes SUPERVISOR-171F.
"""
client, root = app_api_client_with_root
# First set some custom options
install_app_example.options = {"message": "custom"}
assert install_app_example.persist["options"] == {"message": "custom"}
# Reset to defaults by sending null
resp = await client.post(f"{root}/local_example/options", json={"options": None})
assert resp.status == 200
# Persisted options should be empty (meaning defaults will be used)
assert install_app_example.persist["options"] == {}
@pytest.mark.usefixtures("install_app_example")
async def test_app_set_options_error(api_client: TestClient):
"""Test setting options for an app."""
resp = await api_client.post(
"/addons/local_example/options", json={"options": {"message": True}}
)
assert resp.status == 400
body = await resp.json()
assert (
body["message"]
== "App 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_app_start_options_error(
api_client: TestClient,
install_app_example: App,
caplog: pytest.LogCaptureFixture,
):
"""Test error writing options when trying to start app."""
install_app_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 app local_example. Check Supervisor logs for details"
)
assert body["error_key"] == "addon_unknown_error"
assert body["extra_fields"] == {
"addon": "local_example",
}
assert "App 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_app_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"]
== "App 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 (
"App 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_app_example")
async def test_app_not_running_error(
app_api_client_with_root: tuple[TestClient, str], method: str, action: str
):
"""Test app not running error for endpoints that require that."""
client, root = app_api_client_with_root
with patch.object(App, "with_stdin", new=PropertyMock(return_value=True)):
resp = await client.request(method, f"{root}/local_example/{action}")
assert resp.status == 400
body = await resp.json()
assert body["message"] == "App local_example is not running"
assert body["error_key"] == "addon_not_running_error"
assert body["extra_fields"] == {"addon": "local_example"}
@pytest.mark.usefixtures("install_app_example")
async def test_app_write_stdin_not_supported_error(
app_api_client_with_root: tuple[TestClient, str],
):
"""Test error when trying to write stdin to app that does not support it."""
client, root = app_api_client_with_root
resp = await client.post(f"{root}/local_example/stdin")
assert resp.status == 400
body = await resp.json()
assert body["message"] == "App 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_app_ssh")
async def test_app_rebuild_fails_error(api_client: TestClient, coresys: CoreSys):
"""Test error when build fails during rebuild for app."""
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
coresys.docker.containers.create.side_effect = aiodocker.DockerError(
500, {"message": "fail"}
)
with (
patch.object(
CpuArchManager,
"supported",
new=PropertyMock(return_value=[CpuArch.AARCH64]),
),
patch.object(
CpuArchManager, "default", new=PropertyMock(return_value=CpuArch.AARCH64)
),
patch.object(AppBuild, "get_docker_args", return_value={"command": ["build"]}),
):
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 app local_ssh. Check Supervisor logs for details"
)
assert body["error_key"] == "addon_build_failed_unknown_error"
assert body["extra_fields"] == {
"addon": "local_ssh",
}
# ── V2 API tests ──────────────────────────────────────────────────────────────
@pytest.mark.usefixtures("install_app_ssh")
async def test_v2_list_apps_uses_apps_key(api_client_v2: TestClient):
"""V2 GET /v2/apps returns 'apps' key (not 'addons')."""
resp = await api_client_v2.get("/v2/apps")
assert resp.status == 200
body = await resp.json()
assert "apps" in body["data"]
assert "addons" not in body["data"]
assert body["data"]["apps"][0]["slug"] == "local_ssh"
View Git Blame Copy Permalink