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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
|
"""
Tests for the Docker API routes
"""
from unittest.mock import PropertyMock, patch
import pytest
from async_asgi_testclient import TestClient
from async_asgi_testclient.response import Response
from docker.client import DockerClient
from docker.errors import APIError, ImageNotFound, NotFound
from fastapi import status
from serverctl_deployd.dependencies import get_docker_client
from serverctl_deployd.main import app
from serverctl_deployd.models.docker import (BuildCachesDeleted,
ContainerDetails,
ContainersDeleted, ImagesDeleted,
LogsResponse, NetworksDeleted,
PruneResponse, VolumesDeleted)
from tests.fakes.fake_docker_api import (FAKE_CONTAINER_ID,
FAKE_CONTAINER_NAME, FAKE_IMAGE_ID,
FAKE_LOG_LINE_CONTENT,
FAKE_LOG_LINE_COUNT,
FAKE_LOGS_MESSAGE, FAKE_LONG_ID,
FAKE_TAG)
from tests.fakes.fake_docker_client import make_fake_client
async def _get_fake_docker_client() -> DockerClient:
return make_fake_client()
app.dependency_overrides[get_docker_client] = _get_fake_docker_client
@pytest.mark.asyncio
async def test_docker_prune() -> None:
"""Test the docker prune endpoint"""
async with TestClient(app) as client:
# Test successful prune all
response: Response = await client.post("/docker/prune", json={"all": True})
assert response.status_code == status.HTTP_200_OK
prune_response = PruneResponse.parse_obj(response.json())
assert isinstance(prune_response.containers, ContainersDeleted)
assert prune_response.containers.containers_deleted[0] == FAKE_LONG_ID
assert isinstance(prune_response.images, ImagesDeleted)
assert prune_response.images.images_deleted[0] == FAKE_LONG_ID
assert isinstance(prune_response.volumes, VolumesDeleted)
assert prune_response.volumes.volumes_deleted[0] == FAKE_LONG_ID
assert isinstance(prune_response.networks, NetworksDeleted)
assert prune_response.networks.networks_deleted[0] == FAKE_LONG_ID
assert isinstance(prune_response.build_cache, BuildCachesDeleted)
assert prune_response.build_cache.caches_deleted[0] == FAKE_LONG_ID
# Test successful prune only containers
response = await client.post("/docker/prune", json={"containers": True})
assert response.status_code == status.HTTP_200_OK
prune_response = PruneResponse.parse_obj(response.json())
assert isinstance(prune_response.containers, ContainersDeleted)
assert prune_response.containers.containers_deleted[0] == FAKE_LONG_ID
assert prune_response.images is None
assert prune_response.volumes is None
assert prune_response.networks is None
assert prune_response.build_cache is None
# Test condition where Docker API fails
with patch.object(DockerClient, "containers") as containers:
containers.prune.side_effect = APIError("API error")
response = await client.post("/docker/prune", json={"containers": True})
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
@pytest.mark.asyncio
async def test_docker_attach() -> None:
"""Test the docker attach endpoint"""
async with TestClient(app) as client:
# Test successful attach
response: Response = await client.get(
f"/docker/containers/{FAKE_CONTAINER_ID}/attach", stream=True)
line_count = 0
async for line in response.iter_content(chunk_size=512):
if line:
assert line == FAKE_LOG_LINE_CONTENT
line_count += 1
assert line_count == FAKE_LOG_LINE_COUNT
# Test condition where container does not exist
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = NotFound("Not found")
response = await client.get(
"/docker/containers/wrong_id/attach")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json()["detail"] == "Container not found"
# Test condition where Docker API fails
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = APIError("API Error")
response = await client.get(
f"/docker/containers/{FAKE_CONTAINER_ID}/attach")
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert response.json()["detail"] == "Internal server error"
@pytest.mark.asyncio
async def test_docker_get_containers() -> None:
"""Test the docker get containers endpoint"""
async with TestClient(app) as client:
# Test successful get containers
containers: Response = await client.get("/docker/containers")
assert containers.status_code == status.HTTP_200_OK
for container in containers.json():
# pylint: disable-next=fixme
# TODO: Fix the test as image name is not being returned currently
container_details = ContainerDetails.parse_obj(container)
assert container_details.id == FAKE_CONTAINER_ID
assert container_details.name == FAKE_CONTAINER_NAME
assert isinstance(container_details, ContainerDetails)
# Test condition where Docker API fails
with patch.object(DockerClient, "containers") as containers:
containers.list.side_effect = APIError("API Error")
response = await client.get("/docker/containers")
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert response.json()["detail"] == "Internal server error"
@pytest.mark.asyncio
async def test_docker_get_container_details() -> None:
"""Test the docker get container details endpoint"""
async with TestClient(app) as client:
# Test successful get container details
container_details: Response = await client.get(
f"/docker/containers/{FAKE_CONTAINER_ID}")
assert container_details.status_code == status.HTTP_200_OK
container_details = ContainerDetails.parse_obj(
container_details.json())
assert container_details.id == FAKE_CONTAINER_ID
assert container_details.name == FAKE_CONTAINER_NAME
assert isinstance(container_details, ContainerDetails)
# Test condition where container does not exist
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = NotFound("Not found")
response = await client.get(
f"/docker/containers/{FAKE_CONTAINER_ID}")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json()["detail"] == "Container not found"
# Test condition where Docker API fails
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = APIError("API Error")
response = await client.get(
f"/docker/containers/{FAKE_CONTAINER_ID}")
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert response.json()["detail"] == "Internal server error"
@pytest.mark.asyncio
async def test_docker_post_container_delete() -> None:
"""Test the docker post container delete endpoint"""
async with TestClient(app) as client:
# Test successful delete container
response: Response = await client.post(
"/docker/containers/delete",
json={
"container_id": FAKE_CONTAINER_ID
}
)
assert response.status_code == status.HTTP_200_OK
assert response.json()[
"message"] == f"Container {FAKE_CONTAINER_ID} deleted"
# Test condition where container does not exist
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = NotFound("Not found")
response = await client.post(
"/docker/containers/delete",
json={
"container_id": FAKE_CONTAINER_ID
}
)
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json()["detail"] == "Container not found"
# Test condition where Docker API fails as container can't be removed
with patch("docker.models.containers.Container.remove") as container_remove,\
patch(
"docker.models.containers.Container.status",
new_callable=PropertyMock
) as container_status:
container_remove.side_effect = APIError("API Error")
container_status.return_value = "running"
response = await client.post(
"/docker/containers/delete",
json={
"container_id": FAKE_CONTAINER_ID
}
)
assert response.status_code == status.HTTP_403_FORBIDDEN
assert response.json()[
"detail"] == "Cannot remove running containers, try forcing"
# Test condition where Docker API fails
with patch("docker.models.containers.Container.remove") as container_remove,\
patch(
"docker.models.containers.Container.status",
new_callable=PropertyMock
) as container_status:
container_remove.side_effect = APIError("API Error")
container_status.return_value = "exited"
response = await client.post(
"/docker/containers/delete",
json={
"container_id": FAKE_CONTAINER_ID
}
)
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert response.json()["detail"] == "Internal server error"
@pytest.mark.asyncio
async def test_docker_post_container_start() -> None:
"""Test the docker post container start endpoint"""
async with TestClient(app) as client:
# Test successful start container
response: Response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/start")
assert response.status_code == status.HTTP_200_OK
assert response.json()[
"message"] == f"Container {FAKE_CONTAINER_ID} started"
# Test condition where container does not exist
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = NotFound("Not found")
response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/start")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json()["detail"] == "Container not found"
# Test condition where Docker API fails
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = APIError("API Error")
response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/start")
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert response.json()["detail"] == "Internal server error"
@pytest.mark.asyncio
async def test_docker_post_container_stop() -> None:
"""Test the docker post container stop endpoint"""
async with TestClient(app) as client:
# Test successful stop container
response: Response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/stop")
assert response.status_code == status.HTTP_200_OK
assert response.json()[
"message"] == f"Container {FAKE_CONTAINER_ID} stopped"
# Test condition where container does not exist
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = NotFound("Not found")
response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/stop")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json()["detail"] == "Container not found"
# Test condition where Docker API fails
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = APIError("API Error")
response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/stop")
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert response.json()["detail"] == "Internal server error"
@pytest.mark.asyncio
async def test_docker_post_container_restart() -> None:
"""Test the docker post container restart endpoint"""
async with TestClient(app) as client:
# Test successful restart container
response: Response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/restart")
assert response.status_code == status.HTTP_200_OK
assert response.json()[
"message"] == f"Container {FAKE_CONTAINER_ID} restarted"
# Test condition where container does not exist
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = NotFound("Not found")
response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/restart")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json()["detail"] == "Container not found"
# Test condition where Docker API fails
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = APIError("API Error")
response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/restart")
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert response.json()["detail"] == "Internal server error"
@pytest.mark.asyncio
async def test_docker_post_container_kill() -> None:
"""Test the docker post kill container endpoint"""
async with TestClient(app) as client:
# Test successful kill container
response: Response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/kill")
assert response.status_code == status.HTTP_200_OK
assert response.json()[
"message"] == f"Container {FAKE_CONTAINER_ID} killed"
# Test condition where container does not exist
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = NotFound("Not found")
response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/kill")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json()["detail"] == "Container not found"
# Test condition where Docker API fails as container is not running
with patch("docker.models.containers.Container.kill") as container_kill,\
patch(
"docker.models.containers.Container.status",
new_callable=PropertyMock
) as container_status:
container_kill.side_effect = APIError("API Error")
container_status.return_value = "exited"
response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/kill")
assert response.status_code == status.HTTP_403_FORBIDDEN
assert response.json()[
"detail"] == "Cannot kill containers that are not running"
# Test condition where Docker API fails
with patch("docker.models.containers.Container.kill") as container_kill,\
patch(
"docker.models.containers.Container.status",
new_callable=PropertyMock
) as container_status:
container_kill.side_effect = APIError("API Error")
container_status.return_value = "running"
response = await client.post(f"/docker/containers/{FAKE_CONTAINER_ID}/kill")
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert response.json()["detail"] == "Internal server error"
@pytest.mark.asyncio
async def test_docker_post_container_logs() -> None:
"""Test get container logs endpoint"""
async with TestClient(app) as client:
# Test successful get container logs
response: Response = await client.get(f"/docker/containers/{FAKE_CONTAINER_ID}/logs")
logs_response = LogsResponse.parse_obj(response.json())
assert logs_response.container_id == FAKE_CONTAINER_ID
assert logs_response.logs == FAKE_LOGS_MESSAGE
# Test condition where container does not exist
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = NotFound("Not found")
response = await client.get(f"/docker/containers/{FAKE_CONTAINER_ID}/logs")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json()["detail"] == "Container not found"
# Test condition where Docker API fails
with patch.object(DockerClient, "containers") as containers:
containers.get.side_effect = APIError("API Error")
response = await client.get(f"/docker/containers/{FAKE_CONTAINER_ID}/logs")
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert response.json()["detail"] == "Internal server error"
@pytest.mark.asyncio
async def test_docker_post_tag_image() -> None:
"""Test post tag image endpoint"""
async with TestClient(app) as client:
# Test successful tag image
response: Response = await client.post("/docker/images/tag", json={
"image_id": FAKE_IMAGE_ID,
"tag": FAKE_TAG
})
assert response.status_code == status.HTTP_200_OK
assert response.json()["message"] == f"Image {FAKE_IMAGE_ID} tagged"
# Test condition where image does not exist
with patch.object(DockerClient, "images") as images:
images.get.side_effect = ImageNotFound("Not found")
response = await client.post("/docker/images/tag", json={
"image_id": FAKE_IMAGE_ID,
"tag": FAKE_TAG
})
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json()["detail"] == "Image not found"
# Test condition where Docker API fails
with patch.object(DockerClient, "images") as images:
images.get.side_effect = APIError("API Error")
response = await client.post("/docker/images/tag", json={
"image_id": FAKE_IMAGE_ID,
"tag": FAKE_TAG
})
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert response.json()["detail"] == "Internal server error"
|