-
Notifications
You must be signed in to change notification settings - Fork 3.3k
/
test_cmd_apps.py
143 lines (119 loc) · 6.46 KB
/
test_cmd_apps.py
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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
from unittest import mock
from unittest.mock import MagicMock
import pytest as pytest
from lightning_cloud.openapi import (
Externalv1LightningappInstance,
V1LightningappInstanceSpec,
V1LightningappInstanceState,
V1LightningappInstanceStatus,
V1LightningworkState,
V1ListLightningappInstancesResponse,
V1ListLightningworkResponse,
V1ListMembershipsResponse,
V1Membership,
)
from rich.text import Text
from lightning_app.cli.cmd_apps import _AppList, _AppManager
@pytest.mark.parametrize(
"current_state,desired_state,expected",
[
(
V1LightningappInstanceStatus(phase=V1LightningappInstanceState.RUNNING),
V1LightningappInstanceState.DELETED,
Text("terminating"),
),
(
V1LightningappInstanceStatus(phase=V1LightningappInstanceState.STOPPED),
V1LightningappInstanceState.RUNNING,
Text("restarting"),
),
(
V1LightningappInstanceStatus(phase=V1LightningappInstanceState.PENDING),
V1LightningappInstanceState.RUNNING,
Text("restarting"),
),
(
V1LightningappInstanceStatus(phase=V1LightningappInstanceState.UNSPECIFIED, start_timestamp=None),
V1LightningappInstanceState.RUNNING,
Text("not yet started"),
),
],
)
def test_state_transitions(current_state, desired_state, expected):
actual = _AppList._textualize_state_transitions(current_state=current_state, desired_state=desired_state)
assert actual == expected
@mock.patch("lightning_cloud.login.Auth.authenticate", MagicMock())
@mock.patch("lightning_app.utilities.network.LightningClient.lightningapp_instance_service_list_lightningapp_instances")
@mock.patch("lightning_app.utilities.network.LightningClient.projects_service_list_memberships")
def test_list_all_apps_paginated(list_memberships: mock.MagicMock, list_instances: mock.MagicMock):
list_memberships.return_value = V1ListMembershipsResponse(memberships=[V1Membership(project_id="default-project")])
list_instances.side_effect = [
V1ListLightningappInstancesResponse(
lightningapps=[
Externalv1LightningappInstance(
name="test1",
spec=V1LightningappInstanceSpec(desired_state=V1LightningappInstanceState.RUNNING),
status=V1LightningappInstanceStatus(phase=V1LightningappInstanceState.RUNNING),
)
],
next_page_token="page-2",
),
V1ListLightningappInstancesResponse(
lightningapps=[
Externalv1LightningappInstance(
name="test2",
spec=V1LightningappInstanceSpec(desired_state=V1LightningappInstanceState.STOPPED),
status=V1LightningappInstanceStatus(phase=V1LightningappInstanceState.RUNNING),
)
],
),
]
cluster_manager = _AppManager()
cluster_manager.list()
list_memberships.assert_called_once()
assert list_instances.mock_calls == [
mock.call(project_id="default-project", limit=100, phase_in=[]),
mock.call(project_id="default-project", page_token="page-2", limit=100, phase_in=[]),
]
@mock.patch("lightning_cloud.login.Auth.authenticate", MagicMock())
@mock.patch("lightning_app.utilities.network.LightningClient.lightningapp_instance_service_list_lightningapp_instances")
@mock.patch("lightning_app.utilities.network.LightningClient.projects_service_list_memberships")
def test_list_all_apps(list_memberships: mock.MagicMock, list_instances: mock.MagicMock):
list_memberships.return_value = V1ListMembershipsResponse(memberships=[V1Membership(project_id="default-project")])
list_instances.return_value = V1ListLightningappInstancesResponse(lightningapps=[])
cluster_manager = _AppManager()
cluster_manager.list()
list_memberships.assert_called_once()
list_instances.assert_called_once_with(project_id="default-project", limit=100, phase_in=[])
@mock.patch("lightning_cloud.login.Auth.authenticate", MagicMock())
@mock.patch("lightning_app.utilities.network.LightningClient.lightningwork_service_list_lightningwork")
@mock.patch("lightning_app.utilities.network.LightningClient.projects_service_list_memberships")
def test_list_components(list_memberships: mock.MagicMock, list_components: mock.MagicMock):
list_memberships.return_value = V1ListMembershipsResponse(memberships=[V1Membership(project_id="default-project")])
list_components.return_value = V1ListLightningworkResponse(lightningworks=[])
cluster_manager = _AppManager()
cluster_manager.list_components(app_id="cheese")
list_memberships.assert_called_once()
list_components.assert_called_once_with(project_id="default-project", app_id="cheese", phase_in=[])
@mock.patch("lightning_cloud.login.Auth.authenticate", MagicMock())
@mock.patch("lightning_app.utilities.network.LightningClient.lightningwork_service_list_lightningwork")
@mock.patch("lightning_app.utilities.network.LightningClient.projects_service_list_memberships")
def test_list_components_with_phase(list_memberships: mock.MagicMock, list_components: mock.MagicMock):
list_memberships.return_value = V1ListMembershipsResponse(memberships=[V1Membership(project_id="default-project")])
list_components.return_value = V1ListLightningworkResponse(lightningworks=[])
cluster_manager = _AppManager()
cluster_manager.list_components(app_id="cheese", phase_in=[V1LightningworkState.RUNNING])
list_memberships.assert_called_once()
list_components.assert_called_once_with(
project_id="default-project", app_id="cheese", phase_in=[V1LightningworkState.RUNNING]
)
@mock.patch("lightning_cloud.login.Auth.authenticate", MagicMock())
@mock.patch("lightning_app.utilities.network.LightningClient.lightningapp_instance_service_list_lightningapp_instances")
@mock.patch("lightning_app.utilities.network.LightningClient.projects_service_list_memberships")
def test_list_apps_on_cluster(list_memberships: mock.MagicMock, list_instances: mock.MagicMock):
list_memberships.return_value = V1ListMembershipsResponse(memberships=[V1Membership(project_id="default-project")])
list_instances.return_value = V1ListLightningappInstancesResponse(lightningapps=[])
cluster_manager = _AppManager()
cluster_manager.list(cluster_id="12345")
list_memberships.assert_called_once()
list_instances.assert_called_once_with(project_id="default-project", cluster_id="12345", limit=100, phase_in=[])