1
|
|
|
"""kytos.core.napps tests.""" |
2
|
1 |
|
from unittest.mock import MagicMock, patch |
3
|
|
|
|
4
|
1 |
|
from kytos.core.napps import KytosNApp, NApp |
5
|
|
|
|
6
|
|
|
|
7
|
1 |
|
class TestNapp: |
8
|
|
|
"""NApp tests.""" |
9
|
|
|
|
10
|
1 |
|
def setup_method(self): |
11
|
|
|
"""Execute steps before each tests.""" |
12
|
1 |
|
self.napp = NApp(username='kytos', name='napp', version='1.0', |
13
|
|
|
repository='any') |
14
|
1 |
|
self.napp.description = 'desc' |
15
|
1 |
|
self.napp.tags = ['tag1', 'tag2'] |
16
|
|
|
|
17
|
1 |
|
def test__str__(self): |
18
|
|
|
"""Test __str__ method.""" |
19
|
1 |
|
assert str(self.napp) == 'kytos/napp' |
20
|
|
|
|
21
|
1 |
|
def test__repr__(self): |
22
|
|
|
"""Test __repr__ method.""" |
23
|
1 |
|
assert repr(self.napp) == 'NApp(kytos/napp)' |
24
|
|
|
|
25
|
1 |
|
def test_id(self): |
26
|
|
|
"""Test id property.""" |
27
|
1 |
|
assert self.napp.id == 'kytos/napp' |
28
|
|
|
|
29
|
1 |
|
@patch('kytos.core.napps.NApp._has_valid_repository', return_value=True) |
30
|
1 |
|
def test_uri(self, _): |
31
|
|
|
"""Test uri property.""" |
32
|
1 |
|
assert self.napp.uri == 'any/kytos/napp-1.0' |
33
|
|
|
|
34
|
1 |
|
@patch('kytos.core.napps.NApp._has_valid_repository', return_value=False) |
35
|
1 |
|
def test_uri__false(self, _): |
36
|
|
|
"""Test uri property when repository is invalid.""" |
37
|
1 |
|
assert not self.napp.uri |
38
|
|
|
|
39
|
1 |
|
def test_package_url(self): |
40
|
|
|
"""Test package_url property.""" |
41
|
1 |
|
assert self.napp.package_url == 'any/kytos/napp-1.0.napp' |
42
|
|
|
|
43
|
1 |
|
@patch('kytos.core.napps.NApp._has_valid_repository', return_value=False) |
44
|
1 |
|
def test_package_url__none(self, _): |
45
|
|
|
"""Test package_url property when uri does not exist.""" |
46
|
1 |
|
assert not self.napp.package_url |
47
|
|
|
|
48
|
1 |
|
def test_create_from_uri(self): |
49
|
|
|
"""Test create_from_uri method.""" |
50
|
1 |
|
napp = NApp.create_from_uri('file://any/kytos/napp:1.0') |
51
|
|
|
|
52
|
1 |
|
assert napp.username == 'kytos' |
53
|
1 |
|
assert napp.name == 'napp' |
54
|
1 |
|
assert napp.version == '1.0' |
55
|
1 |
|
assert napp.repository == 'file://any' |
56
|
|
|
|
57
|
1 |
|
def test_create_from_uri__not(self): |
58
|
|
|
"""Test create_from_uri method when uri does not match.""" |
59
|
1 |
|
napp = NApp.create_from_uri('') |
60
|
|
|
|
61
|
1 |
|
assert napp is None |
62
|
|
|
|
63
|
1 |
|
@patch('builtins.open') |
64
|
1 |
|
def test_create_from_json(self, mock_open): |
65
|
|
|
"""Test create_from_json method.""" |
66
|
1 |
|
data_file = MagicMock() |
67
|
1 |
|
data_file.read.return_value = '{"username": "kytos", \ |
68
|
|
|
"name": "napp", \ |
69
|
|
|
"version": "1.0", \ |
70
|
|
|
"repository": "any"}' |
71
|
|
|
|
72
|
1 |
|
mock_open.return_value.__enter__.return_value = data_file |
73
|
1 |
|
napp = NApp.create_from_json('filename') |
74
|
|
|
|
75
|
1 |
|
assert napp.username == 'kytos' |
76
|
1 |
|
assert napp.name == 'napp' |
77
|
1 |
|
assert napp.version == '1.0' |
78
|
1 |
|
assert napp.repository == 'any' |
79
|
|
|
|
80
|
1 |
|
def test_create_from_dict(self): |
81
|
|
|
"""Test create_from_dict method.""" |
82
|
1 |
|
data = {'username': 'kytos', 'name': 'napp', 'version': '1.0', |
83
|
|
|
'repository': 'any'} |
84
|
1 |
|
napp = NApp.create_from_dict(data) |
85
|
|
|
|
86
|
1 |
|
assert napp.username == 'kytos' |
87
|
1 |
|
assert napp.name == 'napp' |
88
|
1 |
|
assert napp.version == '1.0' |
89
|
1 |
|
assert napp.repository == 'any' |
90
|
|
|
|
91
|
1 |
|
def test_match(self): |
92
|
|
|
"""Test match method.""" |
93
|
1 |
|
for pattern in ['kytos/napp', 'desc', 'tag1', 'tag2']: |
94
|
1 |
|
assert self.napp.match(pattern) |
95
|
|
|
|
96
|
1 |
|
@patch('os.mkdir') |
97
|
1 |
|
@patch('tarfile.open') |
98
|
1 |
|
@patch('builtins.open') |
99
|
1 |
|
@patch('urllib.request.urlretrieve', return_value=['filename']) |
100
|
1 |
|
@patch('kytos.core.napps.base.randint', return_value=123) |
101
|
1 |
|
@patch('kytos.core.napps.base.Path.unlink') |
102
|
1 |
|
@patch('kytos.core.napps.base.Path.stem', 'stem') |
103
|
1 |
|
def test_download(self, *args): |
104
|
|
|
"""Test download method.""" |
105
|
1 |
|
(_, _, _, mock_open, mock_tarfile_open, mock_mkdir) = args |
106
|
1 |
|
tar = MagicMock() |
107
|
1 |
|
repo_file = MagicMock() |
108
|
|
|
|
109
|
1 |
|
mock_open.return_value.__enter__.return_value = repo_file |
110
|
1 |
|
mock_tarfile_open.return_value.__enter__.return_value = tar |
111
|
|
|
|
112
|
1 |
|
extracted = self.napp.download() |
113
|
|
|
|
114
|
1 |
|
mock_mkdir.assert_called_with('/tmp/kytos-napp-stem-123') |
115
|
1 |
|
tar.extractall.assert_called_with('/tmp/kytos-napp-stem-123') |
116
|
1 |
|
repo_file.write.assert_called_with('any\n') |
117
|
1 |
|
assert str(extracted) == '/tmp/kytos-napp-stem-123' |
118
|
|
|
|
119
|
1 |
|
@patch('kytos.core.napps.NApp._has_valid_repository', return_value=False) |
120
|
1 |
|
def test_download__none(self, _): |
121
|
|
|
"""Test download method when package_url does not exist.""" |
122
|
1 |
|
extracted = self.napp.download() |
123
|
|
|
|
124
|
1 |
|
assert extracted is None |
125
|
|
|
|
126
|
|
|
|
127
|
|
|
# pylint: disable=no-member |
128
|
1 |
|
class KytosNAppChild(KytosNApp): |
129
|
|
|
"""KytosNApp generic class.""" |
130
|
|
|
|
131
|
1 |
|
def setup(self): |
132
|
|
|
"""Setup NApp.""" |
133
|
|
|
|
134
|
1 |
|
def execute(self): |
135
|
|
|
"""Execute NApp.""" |
136
|
|
|
|
137
|
1 |
|
def shutdown(self): |
138
|
|
|
"""End of the NApp.""" |
139
|
|
|
|
140
|
|
|
|
141
|
|
|
# pylint: disable=protected-access |
142
|
1 |
|
class TestKytosNApp: |
143
|
|
|
"""KytosNApp tests.""" |
144
|
|
|
|
145
|
|
|
# pylint: disable=arguments-differ |
146
|
1 |
|
@patch('kytos.core.napps.base.Event') |
147
|
1 |
|
@patch('builtins.open') |
148
|
1 |
|
def setup_method(self, *args): |
149
|
|
|
"""Execute steps before each tests.""" |
150
|
1 |
|
(_, mock_open, mock_event) = args |
151
|
1 |
|
self.event = MagicMock() |
152
|
1 |
|
mock_event.return_value = self.event |
153
|
|
|
|
154
|
1 |
|
data_file = MagicMock() |
155
|
1 |
|
data_file.read.return_value = '{"username": "kytos", \ |
156
|
|
|
"name": "napp", \ |
157
|
|
|
"version": "1.0", \ |
158
|
|
|
"repository": "any"}' |
159
|
|
|
|
160
|
1 |
|
mock_open.return_value.__enter__.return_value = data_file |
161
|
|
|
|
162
|
1 |
|
self.kytos_napp = KytosNAppChild(MagicMock()) |
163
|
1 |
|
self.kytos_napp.execute = MagicMock() |
164
|
1 |
|
self.kytos_napp.shutdown = MagicMock() |
165
|
|
|
|
166
|
1 |
|
def test_napp_id(self): |
167
|
|
|
"""Test napp_id property.""" |
168
|
1 |
|
assert self.kytos_napp.napp_id == 'kytos/napp' |
169
|
|
|
|
170
|
1 |
|
@patch('builtins.open') |
171
|
1 |
|
def test_load_json(self, mock_open): |
172
|
|
|
"""Test _load_json method.""" |
173
|
1 |
|
data_file = MagicMock() |
174
|
1 |
|
data_file.read.return_value = '{"username": "kytos", \ |
175
|
|
|
"name": "napp", \ |
176
|
|
|
"version": "1.0", \ |
177
|
|
|
"repository": "any"}' |
178
|
|
|
|
179
|
1 |
|
mock_open.return_value.__enter__.return_value = data_file |
180
|
|
|
|
181
|
1 |
|
self.kytos_napp._load_json() |
182
|
|
|
|
183
|
1 |
|
assert self.kytos_napp.username == 'kytos' |
184
|
1 |
|
assert self.kytos_napp.name == 'napp' |
185
|
1 |
|
assert self.kytos_napp.version == '1.0' |
186
|
1 |
|
assert self.kytos_napp.repository == 'any' |
187
|
|
|
|
188
|
1 |
|
def test_execute_as_loop_and_run(self): |
189
|
|
|
"""Test execute_as_loop and run methods.""" |
190
|
1 |
|
self.event.is_set.side_effect = [False, True] |
191
|
1 |
|
self.kytos_napp.execute_as_loop(1) |
192
|
|
|
|
193
|
1 |
|
self.kytos_napp.run() |
194
|
|
|
|
195
|
1 |
|
assert self.kytos_napp.execute.call_count == 2 |
196
|
|
|
|
197
|
1 |
|
def test_shutdown_handler(self): |
198
|
|
|
"""Test _shutdown_handler method.""" |
199
|
1 |
|
self.event.is_set.return_value = False |
200
|
|
|
|
201
|
1 |
|
self.kytos_napp._shutdown_handler(MagicMock()) |
202
|
|
|
|
203
|
|
|
self.kytos_napp.shutdown.assert_called_once() |
204
|
|
|
|