File: test_resolution.py

package info (click to toggle)
python-aiohasupervisor 0.3.3-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 932 kB
  • sloc: python: 4,666; sh: 37; makefile: 3
file content (127 lines) | stat: -rw-r--r-- 4,727 bytes parent folder | download | duplicates (2)
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"))
    }