1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127
|
"""Test resolution center supervisor client."""
from uuid import uuid4
from aioresponses import aioresponses
from yarl import URL
from aiohasupervisor import SupervisorClient
from aiohasupervisor.models import CheckOptions
from . import load_fixture
from .const import SUPERVISOR_URL
async def test_resolution_info(
responses: aioresponses, supervisor_client: SupervisorClient
) -> None:
"""Test resolution center info API."""
responses.get(
f"{SUPERVISOR_URL}/resolution/info",
status=200,
body=load_fixture("resolution_info.json"),
)
info = await supervisor_client.resolution.info()
assert info.checks[4].enabled is True
assert info.checks[4].slug == "backups"
assert info.issues[0].context == "system"
assert info.issues[0].type == "no_current_backup"
assert info.issues[0].reference is None
assert info.issues[0].uuid.hex == "7f0eac2b61c9456dab6970507a276c36"
assert info.suggestions[0].auto is False
assert info.suggestions[0].context == "system"
assert info.suggestions[0].type == "create_full_backup"
assert info.suggestions[0].reference is None
assert info.suggestions[0].uuid.hex == "f87d3556f02c4004a47111c072c76fac"
assert info.unhealthy == ["supervisor"]
assert info.unsupported == []
async def test_resolution_check_options(
responses: aioresponses, supervisor_client: SupervisorClient
) -> None:
"""Test resolution center check options API."""
responses.post(f"{SUPERVISOR_URL}/resolution/check/backups/options", status=200)
assert (
await supervisor_client.resolution.check_options(
"backups", CheckOptions(enabled=False)
)
is None
)
assert responses.requests.keys() == {
("POST", URL(f"{SUPERVISOR_URL}/resolution/check/backups/options"))
}
async def test_resolution_run_check(
responses: aioresponses, supervisor_client: SupervisorClient
) -> None:
"""Test resolution center run check API."""
responses.post(f"{SUPERVISOR_URL}/resolution/check/backups/run", status=200)
assert await supervisor_client.resolution.run_check("backups") is None
assert responses.requests.keys() == {
("POST", URL(f"{SUPERVISOR_URL}/resolution/check/backups/run"))
}
async def test_resolution_apply_suggestion(
responses: aioresponses, supervisor_client: SupervisorClient
) -> None:
"""Test resolution center apply suggestion API."""
uuid = uuid4()
responses.post(f"{SUPERVISOR_URL}/resolution/suggestion/{uuid.hex}", status=200)
assert await supervisor_client.resolution.apply_suggestion(uuid) is None
assert responses.requests.keys() == {
("POST", URL(f"{SUPERVISOR_URL}/resolution/suggestion/{uuid.hex}"))
}
async def test_resolution_dismiss_suggestion(
responses: aioresponses, supervisor_client: SupervisorClient
) -> None:
"""Test resolution center dismiss suggestion API."""
uuid = uuid4()
responses.delete(f"{SUPERVISOR_URL}/resolution/suggestion/{uuid.hex}", status=200)
assert await supervisor_client.resolution.dismiss_suggestion(uuid) is None
assert responses.requests.keys() == {
("DELETE", URL(f"{SUPERVISOR_URL}/resolution/suggestion/{uuid.hex}"))
}
async def test_resolution_dismiss_issue(
responses: aioresponses, supervisor_client: SupervisorClient
) -> None:
"""Test resolution center dismiss issue API."""
uuid = uuid4()
responses.delete(f"{SUPERVISOR_URL}/resolution/issue/{uuid.hex}", status=200)
assert await supervisor_client.resolution.dismiss_issue(uuid) is None
assert responses.requests.keys() == {
("DELETE", URL(f"{SUPERVISOR_URL}/resolution/issue/{uuid.hex}"))
}
async def test_resolution_suggestions_for_issue(
responses: aioresponses, supervisor_client: SupervisorClient
) -> None:
"""Test resolution center suggestions for issue API."""
uuid = uuid4()
responses.get(
f"{SUPERVISOR_URL}/resolution/issue/{uuid.hex}/suggestions",
status=200,
body=load_fixture("resolution_suggestions_for_issue.json"),
)
result = await supervisor_client.resolution.suggestions_for_issue(uuid)
assert result[0].auto is False
assert result[0].context == "system"
assert result[0].type == "create_full_backup"
async def test_resolution_healthcheck(
responses: aioresponses, supervisor_client: SupervisorClient
) -> None:
"""Test resolution center healthcheck API."""
responses.post(f"{SUPERVISOR_URL}/resolution/healthcheck", status=200)
assert await supervisor_client.resolution.healthcheck() is None
assert responses.requests.keys() == {
("POST", URL(f"{SUPERVISOR_URL}/resolution/healthcheck"))
}
|