mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-12-20 10:28:45 +00:00
696dcf6149
* Initialize Supervisor Core state in constructor Make sure the Supervisor Core state is set to a value early on. This makes sure that the state is always of type CoreState, and makes sure that any use of the state can rely on it being an actual value from the CoreState enum. This fixes Sentry filter during early startup, where the state previously was None. Because of that, the Sentry filter tried to collect more Context, which lead to an exception and not reporting errors. * Fix pytest It seems that with initializing the state early, the pytest actually runs a system evaluation with: Starting system evaluation with state initialize Before it did that with: Starting system evaluation with state None It detects that the container runs as privileged, and declares the system as unhealthy. It is unclear to me why coresys.core.healthy was checked in this context, it doesn't seem useful. Just remove the check, and validate the state through the getter instead. * Update supervisor/core.py Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> * Make sure Supervisor container is privileged in pytest With the Supervisor Core state being valid now, some evaluations now actually run when loading the resolution center. This leads to Supervisor getting declared unhealthy due to not running in a privileged container under pytest. Fake the host container to be privileged to make evaluations not causing the system to be declared unhealthy under pytest. * Avoid writing actual Supervisor run state file With the Supervisor Core state being valid from the very start, we end up writing a state everytime. Instead of actually writing a state file, simply validate the the necessary calls are being made. This is more conform to typical unit tests and avoids writing a file for every test. * Extend WebSocket client fixture and use it consistently Extend the ha_ws_client WebSocket client fixture to set Supervisor Core into run state and clear all pending messages. Currently only some tests use the ha_ws_client WebSocket client fixture. Use it consistently for all tests. --------- Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
232 lines
6.5 KiB
Python
232 lines
6.5 KiB
Python
"""Test the condition decorators."""
|
|
|
|
import asyncio
|
|
from unittest.mock import ANY, AsyncMock
|
|
|
|
import pytest
|
|
|
|
from supervisor.coresys import CoreSys
|
|
from supervisor.exceptions import JobStartException
|
|
|
|
TEST_JOB = "test"
|
|
|
|
|
|
async def test_add_job(coresys: CoreSys):
|
|
"""Test adding jobs."""
|
|
job = coresys.jobs.new_job(TEST_JOB)
|
|
|
|
assert job in coresys.jobs.jobs
|
|
|
|
|
|
async def test_remove_job_directly(coresys: CoreSys, caplog: pytest.LogCaptureFixture):
|
|
"""Test removing jobs from manager."""
|
|
job = coresys.jobs.new_job(TEST_JOB)
|
|
assert job in coresys.jobs.jobs
|
|
|
|
coresys.jobs.remove_job(job)
|
|
assert job not in coresys.jobs.jobs
|
|
assert f"Removing incomplete job {job.name}" not in caplog.text
|
|
|
|
job = coresys.jobs.new_job(TEST_JOB)
|
|
assert job in coresys.jobs.jobs
|
|
|
|
with job.start():
|
|
coresys.jobs.remove_job(job)
|
|
assert job not in coresys.jobs.jobs
|
|
assert f"Removing incomplete job {job.name}" in caplog.text
|
|
|
|
|
|
async def test_job_done(coresys: CoreSys):
|
|
"""Test done set correctly with jobs."""
|
|
job = coresys.jobs.new_job(TEST_JOB)
|
|
assert not job.done
|
|
assert not coresys.jobs.is_job
|
|
|
|
with job.start():
|
|
assert coresys.jobs.is_job
|
|
assert coresys.jobs.current == job
|
|
assert not job.done
|
|
|
|
assert not coresys.jobs.is_job
|
|
assert job.done
|
|
|
|
with pytest.raises(JobStartException), job.start():
|
|
pass
|
|
|
|
|
|
async def test_job_start_bad_parent(coresys: CoreSys):
|
|
"""Test job cannot be started outside of parent."""
|
|
job = coresys.jobs.new_job(TEST_JOB)
|
|
job2 = coresys.jobs.new_job(f"{TEST_JOB}_2")
|
|
|
|
with job.start(), pytest.raises(JobStartException), job2.start():
|
|
pass
|
|
|
|
with job2.start():
|
|
assert coresys.jobs.current == job2
|
|
|
|
|
|
async def test_update_job(coresys: CoreSys):
|
|
"""Test updating jobs."""
|
|
job = coresys.jobs.new_job(TEST_JOB)
|
|
|
|
job.progress = 50
|
|
assert job.progress == 50
|
|
|
|
job.stage = "stage"
|
|
assert job.stage == "stage"
|
|
|
|
with pytest.raises(ValueError):
|
|
job.progress = 110
|
|
|
|
with pytest.raises(ValueError):
|
|
job.progress = -10
|
|
|
|
|
|
async def test_notify_on_change(coresys: CoreSys, ha_ws_client: AsyncMock):
|
|
"""Test jobs notify Home Assistant on changes."""
|
|
job = coresys.jobs.new_job(TEST_JOB)
|
|
|
|
job.progress = 50
|
|
await asyncio.sleep(0)
|
|
# pylint: disable=protected-access
|
|
ha_ws_client.async_send_command.assert_called_with(
|
|
{
|
|
"type": "supervisor/event",
|
|
"data": {
|
|
"event": "job",
|
|
"data": {
|
|
"name": TEST_JOB,
|
|
"reference": None,
|
|
"uuid": ANY,
|
|
"progress": 50,
|
|
"stage": None,
|
|
"done": None,
|
|
"parent_id": None,
|
|
"errors": [],
|
|
"created": ANY,
|
|
},
|
|
},
|
|
}
|
|
)
|
|
|
|
job.stage = "test"
|
|
await asyncio.sleep(0)
|
|
ha_ws_client.async_send_command.assert_called_with(
|
|
{
|
|
"type": "supervisor/event",
|
|
"data": {
|
|
"event": "job",
|
|
"data": {
|
|
"name": TEST_JOB,
|
|
"reference": None,
|
|
"uuid": ANY,
|
|
"progress": 50,
|
|
"stage": "test",
|
|
"done": None,
|
|
"parent_id": None,
|
|
"errors": [],
|
|
"created": ANY,
|
|
},
|
|
},
|
|
}
|
|
)
|
|
|
|
job.reference = "test"
|
|
await asyncio.sleep(0)
|
|
ha_ws_client.async_send_command.assert_called_with(
|
|
{
|
|
"type": "supervisor/event",
|
|
"data": {
|
|
"event": "job",
|
|
"data": {
|
|
"name": TEST_JOB,
|
|
"reference": "test",
|
|
"uuid": ANY,
|
|
"progress": 50,
|
|
"stage": "test",
|
|
"done": None,
|
|
"parent_id": None,
|
|
"errors": [],
|
|
"created": ANY,
|
|
},
|
|
},
|
|
}
|
|
)
|
|
|
|
with job.start():
|
|
await asyncio.sleep(0)
|
|
ha_ws_client.async_send_command.assert_called_with(
|
|
{
|
|
"type": "supervisor/event",
|
|
"data": {
|
|
"event": "job",
|
|
"data": {
|
|
"name": TEST_JOB,
|
|
"reference": "test",
|
|
"uuid": ANY,
|
|
"progress": 50,
|
|
"stage": "test",
|
|
"done": False,
|
|
"parent_id": None,
|
|
"errors": [],
|
|
"created": ANY,
|
|
},
|
|
},
|
|
}
|
|
)
|
|
|
|
job.capture_error()
|
|
await asyncio.sleep(0)
|
|
ha_ws_client.async_send_command.assert_called_with(
|
|
{
|
|
"type": "supervisor/event",
|
|
"data": {
|
|
"event": "job",
|
|
"data": {
|
|
"name": TEST_JOB,
|
|
"reference": "test",
|
|
"uuid": ANY,
|
|
"progress": 50,
|
|
"stage": "test",
|
|
"done": False,
|
|
"parent_id": None,
|
|
"errors": [
|
|
{
|
|
"type": "HassioError",
|
|
"message": "Unknown error, see supervisor logs",
|
|
}
|
|
],
|
|
"created": ANY,
|
|
},
|
|
},
|
|
}
|
|
)
|
|
|
|
await asyncio.sleep(0)
|
|
ha_ws_client.async_send_command.assert_called_with(
|
|
{
|
|
"type": "supervisor/event",
|
|
"data": {
|
|
"event": "job",
|
|
"data": {
|
|
"name": TEST_JOB,
|
|
"reference": "test",
|
|
"uuid": ANY,
|
|
"progress": 50,
|
|
"stage": "test",
|
|
"done": True,
|
|
"parent_id": None,
|
|
"errors": [
|
|
{
|
|
"type": "HassioError",
|
|
"message": "Unknown error, see supervisor logs",
|
|
}
|
|
],
|
|
"created": ANY,
|
|
},
|
|
},
|
|
}
|
|
)
|
|
# pylint: enable=protected-access
|