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
|
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Test component to manage a container using podman."""
from unittest.mock import call, patch
import pytest
from plinth.app import App, Info
from plinth.container import Container
from plinth.diagnostic_check import DiagnosticCheck, Result
pytestmark = pytest.mark.usefixtures('mock_privileged')
privileged_modules_to_mock = [
'plinth.privileged', 'plinth.privileged.container',
'plinth.privileged.service'
]
class AppTest(App):
"""Test application that contains a daemon."""
app_id = 'test-app'
@pytest.fixture(name='container')
def fixture_container():
app1 = AppTest()
app1.add(Info('test-app', 1))
container = Container('test-container', 'name1', 'image:stable', 'volume1',
'/volume', {'/host1': '/cont1'}, {'KEY1': 'VAL1'},
['service1.service'], {'/dev/host1': '/dev/cont1'},
[(1234, 'tcp4')])
app1.add(container)
with patch('plinth.app.App.list') as app_list:
app_list.return_value = [app1]
yield container
def test_container_init(container):
"""Test initializing the container component."""
component = Container('test-container', 'name1', 'image:stable', 'volume1',
'/volume')
assert component.component_id == 'test-container'
assert component.name == 'name1'
assert component.image_name == 'image:stable'
assert component.volume_name == 'volume1'
assert component.volume_path == '/volume'
assert component.volumes is None
assert component.env is None
assert component.binds_to is None
assert component.devices is None
assert component.listen_ports == []
assert container.component_id == 'test-container'
assert container.name == 'name1'
assert container.image_name == 'image:stable'
assert container.volume_name == 'volume1'
assert container.volume_path == '/volume'
assert container.volumes == {'/host1': '/cont1'}
assert container.env == {'KEY1': 'VAL1'}
assert container.binds_to == ['service1.service']
assert container.devices == {'/dev/host1': '/dev/cont1'}
assert container.listen_ports == [(1234, 'tcp4')]
@patch('plinth.action_utils.podman_is_enabled')
def test_container_is_enabled(podman_is_enabled, container):
"""Test checking if container is enabled."""
podman_is_enabled.return_value = False
assert not container.is_enabled()
podman_is_enabled.return_value = True
assert container.is_enabled()
@patch('plinth.action_utils.service_enable')
@patch('plinth.action_utils.podman_enable')
def test_container_enable(podman_enable, enable, container):
"""Test enabling a container component."""
container.enable()
assert podman_enable.mock_calls == [call('name1')]
assert enable.mock_calls == [call('name1')]
@patch('plinth.action_utils.service_disable')
@patch('plinth.action_utils.podman_disable')
def test_container_disable(podman_disable, disable, container):
"""Test disabling a container component."""
container.disable()
assert podman_disable.mock_calls == [call('name1')]
assert disable.mock_calls == [call('name1')]
@patch('plinth.action_utils.service_is_running')
def test_container_is_running(service_is_running, container):
"""Test checking of container component is running."""
service_is_running.return_value = False
assert not container.is_running()
assert service_is_running.mock_calls == [call('name1')]
service_is_running.reset_mock()
service_is_running.return_value = True
assert container.is_running()
@patch('plinth.action_utils.service_disable')
@patch('plinth.action_utils.service_enable')
@patch('plinth.action_utils.service_is_running')
def test_container_ensure_running(service_is_running, enable, disable,
container):
"""Test checking of container component can be ensured to be running."""
service_is_running.return_value = True
with container.ensure_running() as state:
assert state
assert enable.mock_calls == []
assert disable.mock_calls == []
service_is_running.return_value = False
with container.ensure_running() as state:
assert not state
assert enable.mock_calls == [call('name1')]
assert disable.mock_calls == [call('name1')]
@patch('plinth.action_utils.service_disable')
@patch('plinth.action_utils.service_start')
@patch('plinth.action_utils.podman_disable')
@patch('plinth.action_utils.podman_is_enabled')
@patch('plinth.action_utils.podman_create')
def test_container_setup(podman_create, is_enabled, disable, service_start,
service_disable, container):
"""Test setting up the container."""
is_enabled.return_value = True
container.setup(0)
assert podman_create.mock_calls == [
call('name1', 'image:stable', 'volume1', '/volume',
{'/host1': '/cont1'}, {'KEY1': 'VAL1'}, ['service1.service'],
{'/dev/host1': '/dev/cont1'})
]
assert service_start.mock_calls == [call('name1', check=True)]
assert disable.mock_calls == []
is_enabled.return_value = False
container.setup(0)
assert disable.mock_calls == []
is_enabled.return_value = False
container.setup(1)
assert disable.mock_calls == [call('name1')]
assert service_disable.mock_calls == [call('name1')]
@patch('plinth.action_utils.podman_uninstall')
def test_container_uninstall(podman_uninstall, container):
"""Test uninstalling the container."""
container.uninstall()
assert podman_uninstall.mock_calls == [
call(container_name='name1', image_name='image:stable',
volume_name='volume1', volume_path='/volume')
]
@patch('plinth.action_utils.service_is_running')
@patch('plinth.container.diagnose_port_listening')
def test_container_diagnose(diagnose_port_listening, service_is_running,
container):
"""Test diagnosing the container."""
expected_results = [
DiagnosticCheck('container-running-name1',
'Container {container_name} is running', Result.PASSED,
{'container_name': 'name1'}, 'test-container'),
DiagnosticCheck('daemon-listening-tcp4-1234',
'Listening on tcp4 port 1234', Result.PASSED, {
'kind': 'tcp4',
'port': 1234
}, 'test-container'),
]
diagnose_port_listening.return_value = expected_results[1]
service_is_running.return_value = True
results = container.diagnose()
assert results == expected_results
service_is_running.return_value = False
expected_results[0].result = Result.FAILED
results = container.diagnose()
assert results == expected_results
|
