mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-07-16 05:36:29 +00:00
Use CPE name for boards (#3990)
* Use CPE name for boards * Simplify test
This commit is contained in:
parent
89de909020
commit
a0b28ebb97
@ -8,7 +8,6 @@ from aiohttp import web
|
|||||||
|
|
||||||
from ..const import AddonState
|
from ..const import AddonState
|
||||||
from ..coresys import CoreSys, CoreSysAttributes
|
from ..coresys import CoreSys, CoreSysAttributes
|
||||||
from ..dbus.agent.boards.const import BOARD_NAME_SUPERVISED, BOARD_NAME_YELLOW
|
|
||||||
from ..exceptions import APIAddonNotInstalled
|
from ..exceptions import APIAddonNotInstalled
|
||||||
from .addons import APIAddons
|
from .addons import APIAddons
|
||||||
from .audio import APIAudio
|
from .audio import APIAudio
|
||||||
@ -181,29 +180,10 @@ class RestAPI(CoreSysAttributes):
|
|||||||
)
|
)
|
||||||
|
|
||||||
# Boards endpoints
|
# Boards endpoints
|
||||||
def get_board_routes(
|
|
||||||
board: str,
|
|
||||||
info_handler,
|
|
||||||
options_handler=None,
|
|
||||||
) -> list[web.RouteDef]:
|
|
||||||
routes = [
|
|
||||||
web.get(f"/os/boards/{board}", info_handler),
|
|
||||||
web.get(f"/os/boards/{board.lower()}", info_handler),
|
|
||||||
]
|
|
||||||
if options_handler:
|
|
||||||
routes.insert(1, web.post(f"/os/boards/{board}", options_handler))
|
|
||||||
routes.append(web.post(f"/os/boards/{board.lower()}", options_handler))
|
|
||||||
|
|
||||||
return routes
|
|
||||||
|
|
||||||
self.webapp.add_routes(
|
self.webapp.add_routes(
|
||||||
[
|
[
|
||||||
*get_board_routes(
|
web.get("/os/boards/yellow", api_os.boards_yellow_info),
|
||||||
BOARD_NAME_YELLOW,
|
web.post("/os/boards/yellow", api_os.boards_yellow_options),
|
||||||
api_os.boards_yellow_info,
|
|
||||||
api_os.boards_yellow_options,
|
|
||||||
),
|
|
||||||
*get_board_routes(BOARD_NAME_SUPERVISED, api_os.boards_supervised_info),
|
|
||||||
web.get("/os/boards/{board}", api_os.boards_other_info),
|
web.get("/os/boards/{board}", api_os.boards_other_info),
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
@ -10,7 +10,6 @@ import voluptuous as vol
|
|||||||
from ..const import (
|
from ..const import (
|
||||||
ATTR_BOARD,
|
ATTR_BOARD,
|
||||||
ATTR_BOOT,
|
ATTR_BOOT,
|
||||||
ATTR_CPE_BOARD,
|
|
||||||
ATTR_DEVICES,
|
ATTR_DEVICES,
|
||||||
ATTR_UPDATE_AVAILABLE,
|
ATTR_UPDATE_AVAILABLE,
|
||||||
ATTR_VERSION,
|
ATTR_VERSION,
|
||||||
@ -54,10 +53,9 @@ class APIOS(CoreSysAttributes):
|
|||||||
ATTR_VERSION: self.sys_os.version,
|
ATTR_VERSION: self.sys_os.version,
|
||||||
ATTR_VERSION_LATEST: self.sys_os.latest_version,
|
ATTR_VERSION_LATEST: self.sys_os.latest_version,
|
||||||
ATTR_UPDATE_AVAILABLE: self.sys_os.need_update,
|
ATTR_UPDATE_AVAILABLE: self.sys_os.need_update,
|
||||||
ATTR_BOARD: self.sys_dbus.agent.board.board,
|
ATTR_BOARD: self.sys_os.board,
|
||||||
ATTR_BOOT: self.sys_dbus.rauc.boot_slot,
|
ATTR_BOOT: self.sys_dbus.rauc.boot_slot,
|
||||||
ATTR_DATA_DISK: self.sys_os.datadisk.disk_used,
|
ATTR_DATA_DISK: self.sys_os.datadisk.disk_used,
|
||||||
ATTR_CPE_BOARD: self.sys_os.board,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@api_process
|
@api_process
|
||||||
@ -116,17 +114,10 @@ class APIOS(CoreSysAttributes):
|
|||||||
suggestions=[SuggestionType.EXECUTE_REBOOT],
|
suggestions=[SuggestionType.EXECUTE_REBOOT],
|
||||||
)
|
)
|
||||||
|
|
||||||
@api_process
|
|
||||||
async def boards_supervised_info(self, request: web.Request) -> dict[str, Any]:
|
|
||||||
"""Get supervised board settings."""
|
|
||||||
# There are none currently, this rasises an error if a different board is in use
|
|
||||||
if self.sys_dbus.agent.board.supervised:
|
|
||||||
return {}
|
|
||||||
|
|
||||||
@api_process
|
@api_process
|
||||||
async def boards_other_info(self, request: web.Request) -> dict[str, Any]:
|
async def boards_other_info(self, request: web.Request) -> dict[str, Any]:
|
||||||
"""Empty success return if board is in use, error otherwise."""
|
"""Empty success return if board is in use, error otherwise."""
|
||||||
if request.match_info["board"] != self.sys_dbus.agent.board.board:
|
if request.match_info["board"] != self.sys_os.board:
|
||||||
raise BoardInvalidError(
|
raise BoardInvalidError(
|
||||||
f"{request.match_info['board']} board is not in use", _LOGGER.error
|
f"{request.match_info['board']} board is not in use", _LOGGER.error
|
||||||
)
|
)
|
||||||
|
@ -128,7 +128,6 @@ ATTR_CONTAINERS = "containers"
|
|||||||
ATTR_CONTENT = "content"
|
ATTR_CONTENT = "content"
|
||||||
ATTR_CONTENT_TRUST = "content_trust"
|
ATTR_CONTENT_TRUST = "content_trust"
|
||||||
ATTR_CPE = "cpe"
|
ATTR_CPE = "cpe"
|
||||||
ATTR_CPE_BOARD = "cpe_board"
|
|
||||||
ATTR_CPU_PERCENT = "cpu_percent"
|
ATTR_CPU_PERCENT = "cpu_percent"
|
||||||
ATTR_CRYPTO = "crypto"
|
ATTR_CRYPTO = "crypto"
|
||||||
ATTR_DATA = "data"
|
ATTR_DATA = "data"
|
||||||
|
@ -9,6 +9,7 @@ from awesomeversion import AwesomeVersion, AwesomeVersionException
|
|||||||
from cpe import CPE
|
from cpe import CPE
|
||||||
|
|
||||||
from ..coresys import CoreSys, CoreSysAttributes
|
from ..coresys import CoreSys, CoreSysAttributes
|
||||||
|
from ..dbus.agent.boards.const import BOARD_NAME_SUPERVISED
|
||||||
from ..dbus.rauc import RaucState
|
from ..dbus.rauc import RaucState
|
||||||
from ..exceptions import DBusError, HassOSJobError, HassOSUpdateError
|
from ..exceptions import DBusError, HassOSJobError, HassOSUpdateError
|
||||||
from ..jobs.const import JobCondition, JobExecutionLimit
|
from ..jobs.const import JobCondition, JobExecutionLimit
|
||||||
@ -133,6 +134,7 @@ class OSManager(CoreSysAttributes):
|
|||||||
cpe = CPE(self.sys_host.info.cpe)
|
cpe = CPE(self.sys_host.info.cpe)
|
||||||
os_name = cpe.get_product()[0]
|
os_name = cpe.get_product()[0]
|
||||||
if os_name not in ("hassos", "haos"):
|
if os_name not in ("hassos", "haos"):
|
||||||
|
self._board = BOARD_NAME_SUPERVISED.lower()
|
||||||
raise NotImplementedError()
|
raise NotImplementedError()
|
||||||
except NotImplementedError:
|
except NotImplementedError:
|
||||||
_LOGGER.info("No Home Assistant Operating System found")
|
_LOGGER.info("No Home Assistant Operating System found")
|
||||||
|
@ -10,6 +10,7 @@ import pytest
|
|||||||
from supervisor.coresys import CoreSys
|
from supervisor.coresys import CoreSys
|
||||||
from supervisor.dbus.agent.boards import BoardManager
|
from supervisor.dbus.agent.boards import BoardManager
|
||||||
from supervisor.hardware.data import Device
|
from supervisor.hardware.data import Device
|
||||||
|
from supervisor.os.manager import OSManager
|
||||||
from supervisor.resolution.const import ContextType, IssueType, SuggestionType
|
from supervisor.resolution.const import ContextType, IssueType, SuggestionType
|
||||||
from supervisor.resolution.data import Issue, Suggestion
|
from supervisor.resolution.data import Issue, Suggestion
|
||||||
|
|
||||||
@ -29,7 +30,6 @@ async def test_api_os_info(api_client: TestClient):
|
|||||||
"board",
|
"board",
|
||||||
"boot",
|
"boot",
|
||||||
"data_disk",
|
"data_disk",
|
||||||
"cpe_board",
|
|
||||||
):
|
):
|
||||||
assert attr in result["data"]
|
assert attr in result["data"]
|
||||||
|
|
||||||
@ -96,14 +96,11 @@ async def test_api_os_datadisk_list(api_client: TestClient, coresys: CoreSys):
|
|||||||
assert result["data"]["devices"] == ["/dev/sda"]
|
assert result["data"]["devices"] == ["/dev/sda"]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("name", ["Yellow", "yellow"])
|
async def test_api_board_yellow_info(api_client: TestClient, coresys: CoreSys):
|
||||||
async def test_api_board_yellow_info(
|
|
||||||
api_client: TestClient, coresys: CoreSys, name: str
|
|
||||||
):
|
|
||||||
"""Test yellow board info."""
|
"""Test yellow board info."""
|
||||||
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
|
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
|
||||||
|
|
||||||
resp = await api_client.get(f"/os/boards/{name}")
|
resp = await api_client.get("/os/boards/yellow")
|
||||||
assert resp.status == 200
|
assert resp.status == 200
|
||||||
|
|
||||||
result = await resp.json()
|
result = await resp.json()
|
||||||
@ -112,12 +109,11 @@ async def test_api_board_yellow_info(
|
|||||||
assert result["data"]["power_led"] is True
|
assert result["data"]["power_led"] is True
|
||||||
|
|
||||||
assert (await api_client.get("/os/boards/supervised")).status == 400
|
assert (await api_client.get("/os/boards/supervised")).status == 400
|
||||||
assert (await api_client.get("/os/boards/NotReal")).status == 400
|
assert (await api_client.get("/os/boards/not-real")).status == 400
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("name", ["Yellow", "yellow"])
|
|
||||||
async def test_api_board_yellow_options(
|
async def test_api_board_yellow_options(
|
||||||
api_client: TestClient, coresys: CoreSys, dbus: list[str], name: str
|
api_client: TestClient, coresys: CoreSys, dbus: list[str]
|
||||||
):
|
):
|
||||||
"""Test yellow board options."""
|
"""Test yellow board options."""
|
||||||
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
|
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
|
||||||
@ -125,7 +121,7 @@ async def test_api_board_yellow_options(
|
|||||||
assert len(coresys.resolution.issues) == 0
|
assert len(coresys.resolution.issues) == 0
|
||||||
dbus.clear()
|
dbus.clear()
|
||||||
resp = await api_client.post(
|
resp = await api_client.post(
|
||||||
f"/os/boards/{name}",
|
"/os/boards/yellow",
|
||||||
json={"disk_led": False, "heartbeat_led": False, "power_led": False},
|
json={"disk_led": False, "heartbeat_led": False, "power_led": False},
|
||||||
)
|
)
|
||||||
assert resp.status == 200
|
assert resp.status == 200
|
||||||
@ -147,28 +143,29 @@ async def test_api_board_yellow_options(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("name", ["Supervised", "supervised"])
|
async def test_api_board_supervised_info(api_client: TestClient, coresys: CoreSys):
|
||||||
async def test_api_board_supervised_info(
|
|
||||||
api_client: TestClient, coresys: CoreSys, name: str
|
|
||||||
):
|
|
||||||
"""Test supervised board info."""
|
"""Test supervised board info."""
|
||||||
with patch.object(
|
with patch(
|
||||||
BoardManager, "board", new=PropertyMock(return_value="Supervised")
|
"supervisor.os.manager.CPE.get_product", return_value=["not-hassos"]
|
||||||
):
|
), patch.object(BoardManager, "board", new=PropertyMock(return_value="Supervised")):
|
||||||
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
|
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
|
||||||
|
await coresys.dbus.hostname.connect(coresys.dbus.bus)
|
||||||
|
await coresys.os.load()
|
||||||
|
|
||||||
assert (await api_client.get(f"/os/boards/{name}")).status == 200
|
assert (await api_client.get("/os/boards/supervised")).status == 200
|
||||||
assert (await api_client.post(f"/os/boards/{name}", json={})).status == 405
|
assert (await api_client.post("/os/boards/supervised", json={})).status == 405
|
||||||
assert (await api_client.get("/os/boards/yellow")).status == 400
|
assert (await api_client.get("/os/boards/yellow")).status == 400
|
||||||
assert (await api_client.get("/os/boards/NotReal")).status == 400
|
assert (await api_client.get("/os/boards/not-real")).status == 400
|
||||||
|
|
||||||
|
|
||||||
async def test_api_board_other_info(api_client: TestClient, coresys: CoreSys):
|
async def test_api_board_other_info(api_client: TestClient, coresys: CoreSys):
|
||||||
"""Test info for other board without dbus object."""
|
"""Test info for other board without dbus object."""
|
||||||
with patch.object(BoardManager, "board", new=PropertyMock(return_value="NotReal")):
|
with patch.object(
|
||||||
|
BoardManager, "board", new=PropertyMock(return_value="not-real")
|
||||||
|
), patch.object(OSManager, "board", new=PropertyMock(return_value="not-real")):
|
||||||
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
|
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
|
||||||
|
|
||||||
assert (await api_client.get("/os/boards/NotReal")).status == 200
|
assert (await api_client.get("/os/boards/not-real")).status == 200
|
||||||
assert (await api_client.post("/os/boards/NotReal", json={})).status == 405
|
assert (await api_client.post("/os/boards/not-real", json={})).status == 405
|
||||||
assert (await api_client.get("/os/boards/yellow")).status == 400
|
assert (await api_client.get("/os/boards/yellow")).status == 400
|
||||||
assert (await api_client.get("/os/boards/supervised")).status == 400
|
assert (await api_client.get("/os/boards/supervised")).status == 400
|
||||||
|
@ -29,7 +29,6 @@ async def test_ota_url_generic_x86_64_rename(coresys: CoreSys) -> None:
|
|||||||
|
|
||||||
def test_ota_url_os_name(coresys: CoreSys) -> None:
|
def test_ota_url_os_name(coresys: CoreSys) -> None:
|
||||||
"""Test download URL generated with os_name."""
|
"""Test download URL generated with os_name."""
|
||||||
|
|
||||||
board = "generic-x86-64"
|
board = "generic-x86-64"
|
||||||
os_name = "haos"
|
os_name = "haos"
|
||||||
versionstr = "6.0"
|
versionstr = "6.0"
|
||||||
@ -47,7 +46,6 @@ def test_ota_url_os_name(coresys: CoreSys) -> None:
|
|||||||
|
|
||||||
def test_ota_url_os_name_rel_5_downgrade(coresys: CoreSys) -> None:
|
def test_ota_url_os_name_rel_5_downgrade(coresys: CoreSys) -> None:
|
||||||
"""Test download URL generated with os_name."""
|
"""Test download URL generated with os_name."""
|
||||||
|
|
||||||
board = "generic-x86-64"
|
board = "generic-x86-64"
|
||||||
versionstr = "5.9"
|
versionstr = "5.9"
|
||||||
|
|
||||||
@ -74,3 +72,11 @@ async def test_update_fails_if_out_of_date(coresys: CoreSys) -> None:
|
|||||||
HassOSJobError
|
HassOSJobError
|
||||||
):
|
):
|
||||||
await coresys.os.update()
|
await coresys.os.update()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_board_name_supervised(coresys: CoreSys) -> None:
|
||||||
|
"""Test board name is supervised when not on haos."""
|
||||||
|
with patch("supervisor.os.manager.CPE.get_product", return_value=["not-hassos"]):
|
||||||
|
await coresys.dbus.hostname.connect(coresys.dbus.bus)
|
||||||
|
await coresys.os.load()
|
||||||
|
assert coresys.os.board == "supervised"
|
||||||
|
Loading…
x
Reference in New Issue
Block a user