kytos-ng /
kytos-utils
| Total Complexity | 57 |
| Total Lines | 645 |
| Duplicated Lines | 7.44 % |
| Coverage | 100% |
| Changes | 0 | ||
Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.
Common duplication problems, and corresponding solutions are:
Complex classes like tests.unit.test_napps often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
| 1 | """kytos.utils.napps tests.""" |
||
| 2 | 1 | import json |
|
| 3 | 1 | import re |
|
| 4 | 1 | import tempfile |
|
| 5 | 1 | import unittest |
|
| 6 | 1 | import urllib |
|
| 7 | 1 | from pathlib import Path, PurePosixPath |
|
| 8 | 1 | from unittest.mock import MagicMock, Mock, PropertyMock, call, patch |
|
| 9 | 1 | from urllib.error import HTTPError |
|
| 10 | |||
| 11 | 1 | from kytos.utils.exceptions import KytosException |
|
| 12 | 1 | from kytos.utils.napps import NAppsManager |
|
| 13 | 1 | from kytos.utils.settings import SKEL_PATH |
|
| 14 | |||
| 15 | |||
| 16 | # pylint: disable=protected-access, too-many-public-methods |
||
| 17 | 1 | class TestNapps(unittest.TestCase): |
|
| 18 | """Test the class NAppsManager.""" |
||
| 19 | |||
| 20 | 1 | def setUp(self): |
|
| 21 | """Execute steps before each tests.""" |
||
| 22 | 1 | self.napps_manager = NAppsManager() |
|
| 23 | |||
| 24 | 1 | @staticmethod |
|
| 25 | 1 | def get_napps_response_mock(napps=None): |
|
| 26 | """Get mock to napps response.""" |
||
| 27 | 1 | if napps is None: |
|
| 28 | 1 | napps = [["kytos", "mef_eline"], ["kytos", "of_lldp"]] |
|
| 29 | 1 | response = json.dumps({"napps": napps}) |
|
| 30 | |||
| 31 | 1 | mock_response = MagicMock() |
|
| 32 | 1 | mock_response.getcode.return_value = 200 |
|
| 33 | 1 | mock_response.read.return_value = response |
|
| 34 | 1 | mock_response.__enter__.return_value = mock_response |
|
| 35 | 1 | return mock_response |
|
| 36 | |||
| 37 | 1 | @patch('urllib.request.urlopen') |
|
| 38 | 1 | def test_enabled_property(self, mock_urlopen): |
|
| 39 | """Test enabled property.""" |
||
| 40 | 1 | data = MagicMock() |
|
| 41 | 1 | data.read.return_value = '{"napps": "ABC", "installed_napps": "DEF"}' |
|
| 42 | 1 | mock_urlopen.return_value = data |
|
| 43 | |||
| 44 | 1 | self.assertEqual(str(self.napps_manager._enabled), 'ABC') |
|
| 45 | |||
| 46 | 1 | @patch('urllib.request.urlopen') |
|
| 47 | 1 | def test_enabled_property__error(self, mock_urlopen): |
|
| 48 | """Test enabled property to error case.""" |
||
| 49 | |||
| 50 | 1 | mock_urlopen.side_effect = urllib.error.URLError("some_error") |
|
| 51 | 1 | with self.assertRaises(SystemExit): |
|
| 52 | # pylint: disable=pointless-statement |
||
| 53 | 1 | self.napps_manager._enabled |
|
| 54 | # pylint: enable=pointless-statement |
||
| 55 | |||
| 56 | 1 | self.assertIsNone(self.napps_manager._NAppsManager__local_enabled) |
|
| 57 | |||
| 58 | 1 | @patch('urllib.request.urlopen') |
|
| 59 | 1 | def test_installed_property(self, mock_urlopen): |
|
| 60 | """Test installed property.""" |
||
| 61 | 1 | data = MagicMock() |
|
| 62 | 1 | data.read.return_value = '{"napps": "ABC", "installed_napps": "DEF"}' |
|
| 63 | 1 | mock_urlopen.return_value = data |
|
| 64 | |||
| 65 | 1 | self.assertEqual(str(self.napps_manager._installed), 'DEF') |
|
| 66 | |||
| 67 | 1 | @patch('urllib.request.urlopen') |
|
| 68 | 1 | def test_installed_property__error(self, mock_urlopen): |
|
| 69 | """Test installed property to error case.""" |
||
| 70 | |||
| 71 | 1 | mock_urlopen.side_effect = urllib.error.URLError("some_error") |
|
| 72 | 1 | with self.assertRaises(SystemExit): |
|
| 73 | # pylint: disable=pointless-statement |
||
| 74 | 1 | self.napps_manager._installed |
|
| 75 | # pylint: enable=pointless-statement |
||
| 76 | |||
| 77 | 1 | self.assertIsNone(self.napps_manager._NAppsManager__local_installed) |
|
| 78 | |||
| 79 | 1 | def test_napp_id_property(self): |
|
| 80 | """Test napp_id property.""" |
||
| 81 | 1 | self.napps_manager.user = 'user' |
|
| 82 | 1 | self.napps_manager.napp = 'napp' |
|
| 83 | |||
| 84 | 1 | self.assertEqual(self.napps_manager.napp_id, 'user/napp') |
|
| 85 | |||
| 86 | 1 | def test_set_napp(self): |
|
| 87 | """Test set_napp method.""" |
||
| 88 | 1 | self.napps_manager.set_napp('user', 'napp', 'version') |
|
| 89 | |||
| 90 | 1 | self.assertEqual(self.napps_manager.user, 'user') |
|
| 91 | 1 | self.assertEqual(self.napps_manager.napp, 'napp') |
|
| 92 | 1 | self.assertEqual(self.napps_manager.version, 'version') |
|
| 93 | |||
| 94 | 1 | def test_get_napps(self): |
|
| 95 | """Test method get_napps used to find |
||
| 96 | enabled and installed napps. |
||
| 97 | """ |
||
| 98 | 1 | mock_path = Mock() |
|
| 99 | |||
| 100 | 1 | def glob_side_effect(args): |
|
| 101 | """Path.glob to mock finding paths with kytos.json file.""" |
||
| 102 | 1 | self.assertEqual(args, "*/*/kytos.json") |
|
| 103 | |||
| 104 | 1 | mock_path1 = Mock() |
|
| 105 | 1 | mock_path1.parts = ['kytos', 'of_core', 'kytos.json'] |
|
| 106 | |||
| 107 | 1 | mock_path2 = Mock() |
|
| 108 | 1 | mock_path2.parts = ['kytos', 'of_lldp', 'kytos.json'] |
|
| 109 | |||
| 110 | 1 | return [mock_path1, mock_path2] |
|
| 111 | |||
| 112 | 1 | mock_path.glob = glob_side_effect |
|
| 113 | |||
| 114 | # pylint: disable=protected-access |
||
| 115 | 1 | get_return = self.napps_manager._get_napps(mock_path) |
|
| 116 | 1 | self.assertEqual(get_return[0][0], 'kytos') |
|
| 117 | 1 | self.assertEqual(get_return[0][1], 'of_core') |
|
| 118 | 1 | self.assertEqual(get_return[1][0], 'kytos') |
|
| 119 | 1 | self.assertEqual(get_return[1][1], 'of_lldp') |
|
| 120 | |||
| 121 | 1 | View Code Duplication | def test_get_enabled_local(self): |
|
Duplication
introduced
by
|
|||
| 122 | """Test get_enabled_local used to find |
||
| 123 | enabled napps in local machine""" |
||
| 124 | # Mock kytos.json path |
||
| 125 | 1 | mock_path = Mock() |
|
| 126 | |||
| 127 | 1 | def glob_side_effect(args): |
|
| 128 | """Path.glob to mock finding paths with kytos.json file.""" |
||
| 129 | 1 | self.assertEqual(args, "*/*/kytos.json") |
|
| 130 | |||
| 131 | 1 | mock_path1 = Mock() |
|
| 132 | 1 | mock_path1.parts = ['kytos', 'of_core', 'kytos.json'] |
|
| 133 | 1 | return [mock_path1] |
|
| 134 | 1 | mock_path.glob = glob_side_effect |
|
| 135 | |||
| 136 | 1 | mock_prop_enabled = PropertyMock() |
|
| 137 | 1 | with patch.object(NAppsManager, '_enabled', mock_prop_enabled): |
|
| 138 | 1 | mock_prop_enabled.return_value = mock_path |
|
| 139 | |||
| 140 | 1 | get_return = self.napps_manager.get_enabled_local() |
|
| 141 | |||
| 142 | 1 | self.assertEqual(get_return[0][0], 'kytos') |
|
| 143 | 1 | self.assertEqual(get_return[0][1], 'of_core') |
|
| 144 | 1 | self.assertEqual(mock_prop_enabled.call_count, 1) |
|
| 145 | |||
| 146 | 1 | View Code Duplication | def test_get_installed_local(self): |
| 147 | """Test get_installed_local used to find |
||
| 148 | installed napps in local machine""" |
||
| 149 | # Mock kytos.json path |
||
| 150 | 1 | mock_path = Mock() |
|
| 151 | |||
| 152 | 1 | def glob_side_effect(args): |
|
| 153 | """Path.glob to mock finding paths with kytos.json file.""" |
||
| 154 | 1 | self.assertEqual(args, "*/*/kytos.json") |
|
| 155 | |||
| 156 | 1 | mock_path1 = Mock() |
|
| 157 | 1 | mock_path1.parts = ['kytos', 'of_core', 'kytos.json'] |
|
| 158 | 1 | return [mock_path1] |
|
| 159 | 1 | mock_path.glob = glob_side_effect |
|
| 160 | |||
| 161 | 1 | mock_prop_installed = PropertyMock() |
|
| 162 | 1 | with patch.object(NAppsManager, '_installed', mock_prop_installed): |
|
| 163 | 1 | mock_prop_installed.return_value = mock_path |
|
| 164 | |||
| 165 | 1 | get_return = self.napps_manager.get_installed_local() |
|
| 166 | |||
| 167 | 1 | self.assertEqual(get_return[0][0], 'kytos') |
|
| 168 | 1 | self.assertEqual(get_return[0][1], 'of_core') |
|
| 169 | 1 | self.assertEqual(mock_prop_installed.call_count, 1) |
|
| 170 | |||
| 171 | 1 | @patch('urllib.request.urlopen') |
|
| 172 | 1 | def test_get_installed(self, mock_urlopen): |
|
| 173 | """Test method get_installed to find all installed napps.""" |
||
| 174 | 1 | mock_urlopen.return_value = self.get_napps_response_mock() |
|
| 175 | |||
| 176 | 1 | installed_napps = self.napps_manager.get_installed() |
|
| 177 | |||
| 178 | 1 | self.assertEqual(len(installed_napps), 2) |
|
| 179 | 1 | self.assertEqual(installed_napps[0], ("kytos", "mef_eline")) |
|
| 180 | 1 | self.assertEqual(installed_napps[1], ("kytos", "of_lldp")) |
|
| 181 | |||
| 182 | 1 | @patch('urllib.request.urlopen') |
|
| 183 | 1 | def test_get_installed__connection_error(self, mock_urlopen): |
|
| 184 | """Test method get_installed to connection error case.""" |
||
| 185 | 1 | err = '[Errno 111] Connection refused' |
|
| 186 | 1 | mock_urlopen.side_effect = urllib.error.URLError(err) |
|
| 187 | 1 | with self.assertRaises(KytosException) as context: |
|
| 188 | 1 | self.napps_manager.get_installed() |
|
| 189 | |||
| 190 | 1 | self.assertEqual('<urlopen error [Errno 111] Connection refused>', |
|
| 191 | str(context.exception)) |
||
| 192 | |||
| 193 | 1 | @patch('urllib.request.urlopen') |
|
| 194 | 1 | def test_get_installed__error(self, mock_urlopen): |
|
| 195 | """Test method get_installed with API error.""" |
||
| 196 | 1 | mock_response = MagicMock() |
|
| 197 | 1 | mock_response.getcode.return_value = 500 |
|
| 198 | 1 | mock_urlopen.return_value = mock_response |
|
| 199 | |||
| 200 | 1 | with self.assertRaises(KytosException) as context: |
|
| 201 | 1 | self.napps_manager.get_installed() |
|
| 202 | |||
| 203 | 1 | self.assertEqual('Error calling Kytos to check installed NApps.', |
|
| 204 | str(context.exception)) |
||
| 205 | |||
| 206 | 1 | @patch('urllib.request.urlopen') |
|
| 207 | 1 | def test_get_enabled(self, mock_urlopen): |
|
| 208 | """Test method get_enabled to find all enabled napps.""" |
||
| 209 | 1 | mock_urlopen.return_value = self.get_napps_response_mock() |
|
| 210 | |||
| 211 | 1 | installed_napps = self.napps_manager.get_enabled() |
|
| 212 | |||
| 213 | 1 | self.assertEqual(len(installed_napps), 2) |
|
| 214 | 1 | self.assertEqual(installed_napps[0], ("kytos", "mef_eline")) |
|
| 215 | 1 | self.assertEqual(installed_napps[1], ("kytos", "of_lldp")) |
|
| 216 | |||
| 217 | 1 | @patch('urllib.request.urlopen') |
|
| 218 | 1 | def test_get_enabled__connection_error(self, mock_urlopen): |
|
| 219 | """Test method get_enabled to connection error case.""" |
||
| 220 | 1 | err = '[Errno 111] Connection refused' |
|
| 221 | 1 | mock_urlopen.side_effect = urllib.error.URLError(err) |
|
| 222 | 1 | with self.assertRaises(KytosException) as context: |
|
| 223 | 1 | self.napps_manager.get_enabled() |
|
| 224 | |||
| 225 | 1 | self.assertEqual('<urlopen error [Errno 111] Connection refused>', |
|
| 226 | str(context.exception)) |
||
| 227 | |||
| 228 | 1 | @patch('urllib.request.urlopen') |
|
| 229 | 1 | def test_get_enabled__error(self, mock_urlopen): |
|
| 230 | """Test method get_enabled with API error.""" |
||
| 231 | 1 | mock_response = MagicMock() |
|
| 232 | 1 | mock_response.getcode.return_value = 500 |
|
| 233 | 1 | mock_urlopen.return_value = mock_response |
|
| 234 | |||
| 235 | 1 | with self.assertRaises(KytosException) as context: |
|
| 236 | 1 | self.napps_manager.get_enabled() |
|
| 237 | |||
| 238 | 1 | self.assertEqual('Error calling Kytos to check enabled NApps.', |
|
| 239 | str(context.exception)) |
||
| 240 | |||
| 241 | 1 | @patch('urllib.request.urlopen') |
|
| 242 | 1 | def test_is_enabled(self, mock_urlopen): |
|
| 243 | """Test is_enabled method.""" |
||
| 244 | 1 | mock_urlopen.return_value = self.get_napps_response_mock() |
|
| 245 | |||
| 246 | 1 | self.napps_manager.user = 'kytos' |
|
| 247 | 1 | self.napps_manager.napp = 'mef_eline' |
|
| 248 | |||
| 249 | 1 | self.assertTrue(self.napps_manager.is_enabled()) |
|
| 250 | |||
| 251 | 1 | @patch('urllib.request.urlopen') |
|
| 252 | 1 | def test_is_installed(self, mock_urlopen): |
|
| 253 | """Test is_installed method.""" |
||
| 254 | 1 | mock_urlopen.return_value = self.get_napps_response_mock() |
|
| 255 | |||
| 256 | 1 | self.napps_manager.user = 'kytos' |
|
| 257 | 1 | self.napps_manager.napp = 'mef_eline' |
|
| 258 | |||
| 259 | 1 | self.assertTrue(self.napps_manager.is_installed()) |
|
| 260 | |||
| 261 | 1 | @patch('urllib.request.urlopen') |
|
| 262 | 1 | def test_get_disabled(self, mock_urlopen): |
|
| 263 | """Test get_disabled method.""" |
||
| 264 | 1 | enabled = [["kytos", "mef_eline"]] |
|
| 265 | 1 | mock_urlopen.side_effect = [self.get_napps_response_mock(), |
|
| 266 | self.get_napps_response_mock(enabled)] |
||
| 267 | |||
| 268 | 1 | disabled = self.napps_manager.get_disabled() |
|
| 269 | |||
| 270 | 1 | self.assertEqual(disabled, [('kytos', 'of_lldp')]) |
|
| 271 | |||
| 272 | 1 | @patch('urllib.request.urlopen') |
|
| 273 | 1 | def test_dependencies(self, mock_urlopen): |
|
| 274 | """Test dependencies method.""" |
||
| 275 | 1 | napps = {"napp_dependencies": ["kytos/mef_eline", "kytos/of_lldp"]} |
|
| 276 | 1 | data = MagicMock() |
|
| 277 | 1 | data.read.return_value = json.dumps(napps) |
|
| 278 | 1 | mock_urlopen.return_value = data |
|
| 279 | |||
| 280 | 1 | dependencies = self.napps_manager.dependencies() |
|
| 281 | |||
| 282 | 1 | expected_dependencies = [('kytos', 'mef_eline'), ('kytos', 'of_lldp')] |
|
| 283 | 1 | self.assertEqual(dependencies, expected_dependencies) |
|
| 284 | |||
| 285 | 1 | @patch('urllib.request.urlopen') |
|
| 286 | 1 | def test_get_description(self, mock_urlopen): |
|
| 287 | """Test get_description method.""" |
||
| 288 | 1 | data = MagicMock() |
|
| 289 | 1 | data.read.return_value = '{"description": "ABC"}' |
|
| 290 | 1 | mock_urlopen.return_value = data |
|
| 291 | |||
| 292 | 1 | description = self.napps_manager.get_description() |
|
| 293 | |||
| 294 | 1 | self.assertEqual(description, 'ABC') |
|
| 295 | |||
| 296 | 1 | @patch('urllib.request.urlopen') |
|
| 297 | 1 | def test_get_version(self, mock_urlopen): |
|
| 298 | """Test get_version method.""" |
||
| 299 | 1 | data = MagicMock() |
|
| 300 | 1 | data.read.return_value = '{"version": "123"}' |
|
| 301 | 1 | mock_urlopen.return_value = data |
|
| 302 | |||
| 303 | 1 | version = self.napps_manager.get_version() |
|
| 304 | |||
| 305 | 1 | self.assertEqual(version, '123') |
|
| 306 | |||
| 307 | 1 | @patch('urllib.request.urlopen') |
|
| 308 | 1 | def test_get_napp_key(self, mock_urlopen): |
|
| 309 | """Test _get_napp_key method.""" |
||
| 310 | 1 | data = MagicMock() |
|
| 311 | 1 | data.read.return_value = '{"key": "ABC"}' |
|
| 312 | 1 | mock_urlopen.return_value = data |
|
| 313 | |||
| 314 | 1 | self.napps_manager.user = 'kytos' |
|
| 315 | 1 | self.napps_manager.napp = 'mef_eline' |
|
| 316 | |||
| 317 | 1 | meta_key = self.napps_manager._get_napp_key('key') |
|
| 318 | |||
| 319 | 1 | self.assertEqual(meta_key, 'ABC') |
|
| 320 | |||
| 321 | 1 | @patch('urllib.request.urlopen') |
|
| 322 | 1 | def test_disable(self, mock_urlopen): |
|
| 323 | """Test disable method.""" |
||
| 324 | 1 | data = MagicMock() |
|
| 325 | 1 | data.read.return_value = '{}' |
|
| 326 | 1 | mock_urlopen.return_value = data |
|
| 327 | |||
| 328 | 1 | self.napps_manager.user = 'kytos' |
|
| 329 | 1 | self.napps_manager.napp = 'mef_eline' |
|
| 330 | |||
| 331 | 1 | self.napps_manager.disable() |
|
| 332 | |||
| 333 | 1 | uri = self.napps_manager._kytos_api + self.napps_manager._NAPP_DISABLE |
|
| 334 | 1 | uri = uri.format('kytos', 'mef_eline') |
|
| 335 | |||
| 336 | 1 | mock_urlopen.assert_called_with(uri) |
|
| 337 | |||
| 338 | 1 | @patch('kytos.utils.napps.LOG') |
|
| 339 | 1 | @patch('urllib.request.urlopen') |
|
| 340 | 1 | def test_disable__error(self, *args): |
|
| 341 | """Test disable method to error case.""" |
||
| 342 | 1 | (mock_urlopen, mock_logger) = args |
|
| 343 | 1 | http_errors = [HTTPError('url', 400, 'msg', 'hdrs', MagicMock()), |
|
| 344 | HTTPError('url', 500, 'msg', 'hdrs', MagicMock())] |
||
| 345 | 1 | mock_urlopen.side_effect = http_errors |
|
| 346 | |||
| 347 | 1 | self.napps_manager.disable() |
|
| 348 | 1 | self.napps_manager.disable() |
|
| 349 | |||
| 350 | 1 | self.assertEqual(mock_logger.error.call_count, 2) |
|
| 351 | |||
| 352 | 1 | @patch('urllib.request.urlopen') |
|
| 353 | 1 | def test_enable(self, mock_urlopen): |
|
| 354 | """Test enable method.""" |
||
| 355 | 1 | data = MagicMock() |
|
| 356 | 1 | data.read.return_value = '{}' |
|
| 357 | 1 | mock_urlopen.return_value = data |
|
| 358 | |||
| 359 | 1 | self.napps_manager.user = 'kytos' |
|
| 360 | 1 | self.napps_manager.napp = 'mef_eline' |
|
| 361 | |||
| 362 | 1 | self.napps_manager.enable() |
|
| 363 | |||
| 364 | 1 | uri = self.napps_manager._kytos_api + self.napps_manager._NAPP_ENABLE |
|
| 365 | 1 | uri = uri.format('kytos', 'mef_eline') |
|
| 366 | |||
| 367 | 1 | mock_urlopen.assert_called_with(uri) |
|
| 368 | |||
| 369 | 1 | @patch('kytos.utils.napps.LOG') |
|
| 370 | 1 | @patch('urllib.request.urlopen') |
|
| 371 | 1 | def test_enable__error(self, *args): |
|
| 372 | """Test enable method to error case.""" |
||
| 373 | 1 | (mock_urlopen, mock_logger) = args |
|
| 374 | 1 | http_errors = [HTTPError('url', 400, 'msg', 'hdrs', MagicMock()), |
|
| 375 | HTTPError('url', 500, 'msg', 'hdrs', MagicMock())] |
||
| 376 | 1 | mock_urlopen.side_effect = http_errors |
|
| 377 | |||
| 378 | 1 | self.napps_manager.enable() |
|
| 379 | 1 | self.napps_manager.enable() |
|
| 380 | |||
| 381 | 1 | self.assertEqual(mock_logger.error.call_count, 2) |
|
| 382 | |||
| 383 | 1 | @patch('urllib.request.urlopen') |
|
| 384 | 1 | def test_enabled_dir(self, mock_urlopen): |
|
| 385 | """Test enabled_dir method.""" |
||
| 386 | 1 | data = MagicMock() |
|
| 387 | 1 | data.read.return_value = '{"napps": "ABC", "installed_napps": "DEF"}' |
|
| 388 | 1 | mock_urlopen.return_value = data |
|
| 389 | |||
| 390 | 1 | self.napps_manager.user = 'kytos' |
|
| 391 | 1 | self.napps_manager.napp = 'mef_eline' |
|
| 392 | |||
| 393 | 1 | enabled_dir = self.napps_manager.enabled_dir() |
|
| 394 | 1 | self.assertEqual(str(enabled_dir), 'ABC/kytos/mef_eline') |
|
| 395 | |||
| 396 | 1 | @patch('urllib.request.urlopen') |
|
| 397 | 1 | def test_installed_dir(self, mock_urlopen): |
|
| 398 | """Test installed_dir method.""" |
||
| 399 | 1 | data = MagicMock() |
|
| 400 | 1 | data.read.return_value = '{"napps": "ABC", "installed_napps": "DEF"}' |
|
| 401 | 1 | mock_urlopen.return_value = data |
|
| 402 | |||
| 403 | 1 | self.napps_manager.user = 'kytos' |
|
| 404 | 1 | self.napps_manager.napp = 'mef_eline' |
|
| 405 | |||
| 406 | 1 | installed_dir = self.napps_manager.installed_dir() |
|
| 407 | 1 | self.assertEqual(str(installed_dir), 'DEF/kytos/mef_eline') |
|
| 408 | |||
| 409 | 1 | @patch('urllib.request.urlopen') |
|
| 410 | 1 | def test_remote_uninstall(self, mock_urlopen): |
|
| 411 | """Test remote_uninstall method.""" |
||
| 412 | 1 | data = MagicMock() |
|
| 413 | 1 | data.read.return_value = '{}' |
|
| 414 | 1 | mock_urlopen.return_value = data |
|
| 415 | |||
| 416 | 1 | self.napps_manager.user = 'kytos' |
|
| 417 | 1 | self.napps_manager.napp = 'mef_eline' |
|
| 418 | |||
| 419 | 1 | self.napps_manager.remote_uninstall() |
|
| 420 | |||
| 421 | 1 | uninstall_uri = self.napps_manager._NAPP_UNINSTALL |
|
| 422 | 1 | uri = self.napps_manager._kytos_api + uninstall_uri |
|
| 423 | 1 | uri = uri.format('kytos', 'mef_eline') |
|
| 424 | |||
| 425 | 1 | mock_urlopen.assert_called_with(uri) |
|
| 426 | |||
| 427 | 1 | @patch('kytos.utils.napps.LOG') |
|
| 428 | 1 | @patch('urllib.request.urlopen') |
|
| 429 | 1 | def test_remote_uninstall__error(self, *args): |
|
| 430 | """Test remote_uninstall method to error case.""" |
||
| 431 | 1 | (mock_urlopen, mock_logger) = args |
|
| 432 | 1 | http_errors = [HTTPError('url', 400, 'msg', 'hdrs', MagicMock()), |
|
| 433 | HTTPError('url', 500, 'msg', 'hdrs', MagicMock())] |
||
| 434 | 1 | mock_urlopen.side_effect = http_errors |
|
| 435 | |||
| 436 | 1 | self.napps_manager.remote_uninstall() |
|
| 437 | 1 | self.napps_manager.remote_uninstall() |
|
| 438 | |||
| 439 | 1 | self.assertEqual(mock_logger.error.call_count, 2) |
|
| 440 | |||
| 441 | 1 | @patch('urllib.request.urlopen') |
|
| 442 | 1 | def test_remote_install(self, mock_urlopen): |
|
| 443 | """Test remote_install method.""" |
||
| 444 | 1 | data = MagicMock() |
|
| 445 | 1 | data.read.return_value = '{}' |
|
| 446 | 1 | mock_urlopen.return_value = data |
|
| 447 | |||
| 448 | 1 | self.napps_manager.user = 'kytos' |
|
| 449 | 1 | self.napps_manager.napp = 'mef_eline' |
|
| 450 | |||
| 451 | 1 | self.napps_manager.remote_install() |
|
| 452 | |||
| 453 | 1 | install_uri = self.napps_manager._NAPP_INSTALL |
|
| 454 | 1 | uri = self.napps_manager._kytos_api + install_uri |
|
| 455 | 1 | uri = uri.format('kytos', 'mef_eline') |
|
| 456 | |||
| 457 | 1 | mock_urlopen.assert_called_with(uri) |
|
| 458 | |||
| 459 | 1 | def test_valid_name(self): |
|
| 460 | """Test valid_name method.""" |
||
| 461 | 1 | valid_name = self.napps_manager.valid_name('username') |
|
| 462 | 1 | invalid_name = self.napps_manager.valid_name('_username') |
|
| 463 | |||
| 464 | 1 | self.assertTrue(valid_name) |
|
| 465 | 1 | self.assertFalse(invalid_name) |
|
| 466 | |||
| 467 | 1 | @patch('jinja2.Environment.get_template') |
|
| 468 | 1 | def test_render_template(self, mock_get_template): |
|
| 469 | """Test render_template method.""" |
||
| 470 | 1 | template = MagicMock() |
|
| 471 | 1 | mock_get_template.return_value = template |
|
| 472 | |||
| 473 | 1 | self.napps_manager.render_template('', 'filename', 'context') |
|
| 474 | |||
| 475 | 1 | mock_get_template.assert_called_with('filename') |
|
| 476 | 1 | template.render.assert_called_with('context') |
|
| 477 | |||
| 478 | 1 | @patch('kytos.utils.napps.NAppsClient') |
|
| 479 | 1 | def test_search(self, mock_napps_client): |
|
| 480 | """Test search method.""" |
||
| 481 | 1 | napp_1 = {'username': 'kytos', 'name': 'mef_eline', 'description': '', |
|
| 482 | 'tags': ['A', 'B']} |
||
| 483 | 1 | napp_2 = {'username': '0_kytos', 'name': 'any', 'description': '', |
|
| 484 | 'tags': ['A', 'B']} |
||
| 485 | 1 | napps_client = MagicMock() |
|
| 486 | 1 | napps_client.get_napps.return_value = [napp_1, napp_2] |
|
| 487 | 1 | mock_napps_client.return_value = napps_client |
|
| 488 | |||
| 489 | # pattern to match strings that start with letters |
||
| 490 | 1 | pattern = re.compile('^[a-z]+') |
|
| 491 | 1 | napps = self.napps_manager.search(pattern) |
|
| 492 | |||
| 493 | 1 | self.assertEqual(napps, [napp_1]) |
|
| 494 | |||
| 495 | 1 | @patch('os.makedirs') |
|
| 496 | 1 | @patch('builtins.open') |
|
| 497 | 1 | @patch('builtins.input') |
|
| 498 | 1 | @patch('kytos.utils.napps.NAppsManager.render_template') |
|
| 499 | 1 | def test_create_napp(self, *args): |
|
| 500 | """Test create_napp method.""" |
||
| 501 | 1 | (mock_render_template, mock_input, _, mock_mkdirs) = args |
|
| 502 | 1 | mock_input.side_effect = ['username', 'napp', None] |
|
| 503 | |||
| 504 | 1 | self.napps_manager.create_napp() |
|
| 505 | |||
| 506 | 1 | tmpl_path = SKEL_PATH / 'napp-structure/username/napp' |
|
| 507 | 1 | description = '# TODO: <<<< Insert your NApp description here >>>>' |
|
| 508 | 1 | context = {'username': 'username', 'napp': 'napp', |
|
| 509 | 'description': description} |
||
| 510 | |||
| 511 | 1 | calls = [] |
|
| 512 | 1 | for tmp in ['__init__.py', 'main.py', '.gitignore', 'kytos.json', |
|
| 513 | 'README.rst', 'settings.py']: |
||
| 514 | 1 | calls.append(call(tmpl_path, '{}.template'.format(tmp), context)) |
|
| 515 | 1 | calls.append(call('{}/ui'.format(tmpl_path), 'README.rst.template', |
|
| 516 | context)) |
||
| 517 | |||
| 518 | 1 | mock_mkdirs.assert_has_calls([call('username', exist_ok=True), |
|
| 519 | call('username/napp'), |
||
| 520 | call('username/napp/ui/k-info-panel'), |
||
| 521 | call('username/napp/ui/k-toolbar'), |
||
| 522 | call('username/napp/ui/k-action-menu')]) |
||
| 523 | 1 | mock_render_template.assert_has_calls(calls, any_order=True) |
|
| 524 | |||
| 525 | 1 | def test_check_module(self): |
|
| 526 | """Test _check_module method.""" |
||
| 527 | 1 | folder = MagicMock() |
|
| 528 | 1 | folder.exists.return_value = False |
|
| 529 | |||
| 530 | 1 | self.napps_manager._check_module(folder) |
|
| 531 | |||
| 532 | 1 | folder.mkdir.assert_called() |
|
| 533 | 1 | (folder / '__init__.py').touch.assert_called() |
|
| 534 | |||
| 535 | 1 | @patch('pathspec.pathspec.PathSpec.match_tree') |
|
| 536 | 1 | @patch('tarfile.TarFile.add') |
|
| 537 | 1 | @patch('os.remove') |
|
| 538 | 1 | @patch('os.walk') |
|
| 539 | 1 | @patch('os.getcwd') |
|
| 540 | 1 | @patch('builtins.open') |
|
| 541 | 1 | def test_build_napp_package(self, *args): |
|
| 542 | """Test build_napp_package method.""" |
||
| 543 | 1 | (_, mock_getcwd, mock_walk, _, mock_add, mock_match_tree) = args |
|
| 544 | 1 | with tempfile.TemporaryDirectory() as tmp_dir: |
|
| 545 | 1 | mock_getcwd.return_value = tmp_dir |
|
| 546 | |||
| 547 | 1 | files = ['username/napp/A', 'username/napp/B', 'username/napp/C'] |
|
| 548 | 1 | mock_walk.return_value = [(tmp_dir, ['username/napp/.git'], files)] |
|
| 549 | |||
| 550 | 1 | mock_match_tree.return_value = ['username/napp/C'] |
|
| 551 | |||
| 552 | 1 | self.napps_manager.build_napp_package('username/napp') |
|
| 553 | |||
| 554 | 1 | calls = [call(PurePosixPath('username/napp/A')), |
|
| 555 | call(PurePosixPath('username/napp/B'))] |
||
| 556 | 1 | mock_add.assert_has_calls(calls) |
|
| 557 | |||
| 558 | 1 | @patch('ruamel.yaml.YAML.load', return_value='openapi') |
|
| 559 | 1 | @patch('pathlib.Path.open') |
|
| 560 | 1 | @patch('builtins.open') |
|
| 561 | 1 | def test_create_metadata(self, *args): |
|
| 562 | """Test create_metadata method.""" |
||
| 563 | 1 | (mock_open, _, _) = args |
|
| 564 | 1 | enter_file_1 = MagicMock() |
|
| 565 | 1 | enter_file_1.read.return_value = '{}' |
|
| 566 | |||
| 567 | 1 | enter_file_2 = MagicMock() |
|
| 568 | 1 | enter_file_2.read.return_value = 'readme' |
|
| 569 | |||
| 570 | 1 | mock_open.return_value.__enter__.side_effect = [enter_file_1, |
|
| 571 | enter_file_2] |
||
| 572 | |||
| 573 | 1 | metadata = self.napps_manager.create_metadata() |
|
| 574 | |||
| 575 | 1 | self.assertEqual(metadata, {'readme': 'readme', |
|
| 576 | 'OpenAPI_Spec': '"openapi"'}) |
||
| 577 | |||
| 578 | 1 | @patch('kytos.utils.napps.NAppsClient') |
|
| 579 | 1 | @patch('kytos.utils.napps.NAppsManager.build_napp_package') |
|
| 580 | 1 | @patch('kytos.utils.napps.NAppsManager.create_metadata') |
|
| 581 | 1 | @patch('kytos.utils.napps.NAppsManager.prepare') |
|
| 582 | 1 | def test_upload(self, *args): |
|
| 583 | """Test upload method.""" |
||
| 584 | 1 | (mock_prepare, mock_create, mock_build, mock_napps_client) = args |
|
| 585 | 1 | mock_create.return_value = {'name': 'ABC'} |
|
| 586 | 1 | mock_build.return_value = 'package' |
|
| 587 | 1 | napps_client = MagicMock() |
|
| 588 | 1 | mock_napps_client.return_value = napps_client |
|
| 589 | |||
| 590 | 1 | self.napps_manager.upload() |
|
| 591 | |||
| 592 | 1 | mock_prepare.assert_called() |
|
| 593 | 1 | mock_create.assert_called() |
|
| 594 | 1 | mock_build.assert_called_with('ABC') |
|
| 595 | 1 | napps_client.upload_napp.assert_called_with({'name': 'ABC'}, 'package') |
|
| 596 | |||
| 597 | 1 | @patch('kytos.utils.napps.NAppsClient') |
|
| 598 | 1 | def test_delete(self, mock_napps_client): |
|
| 599 | """Test delete method.""" |
||
| 600 | 1 | napps_client = MagicMock() |
|
| 601 | 1 | mock_napps_client.return_value = napps_client |
|
| 602 | |||
| 603 | 1 | self.napps_manager.user = 'kytos' |
|
| 604 | 1 | self.napps_manager.napp = 'mef_eline' |
|
| 605 | |||
| 606 | 1 | self.napps_manager.delete() |
|
| 607 | |||
| 608 | 1 | napps_client.delete.assert_called_with('kytos', 'mef_eline') |
|
| 609 | |||
| 610 | 1 | @patch('sys.exit') |
|
| 611 | 1 | @patch('kytos.utils.napps.OpenAPI') |
|
| 612 | 1 | @patch('kytos.utils.napps.NAppsManager._ask_openapi', return_value=True) |
|
| 613 | 1 | def test_prepare(self, *args): |
|
| 614 | """Test prepare method.""" |
||
| 615 | 1 | (_, mock_openapi, _) = args |
|
| 616 | 1 | self.napps_manager.prepare() |
|
| 617 | |||
| 618 | 1 | napp_path = Path() |
|
| 619 | 1 | tpl_path = SKEL_PATH / 'napp-structure/username/napp' |
|
| 620 | 1 | mock_openapi.assert_called_with(napp_path, tpl_path) |
|
| 621 | 1 | mock_openapi.return_value.render_template.assert_called() |
|
| 622 | |||
| 623 | 1 | @patch('pathlib.Path.exists') |
|
| 624 | 1 | @patch('builtins.input') |
|
| 625 | 1 | def test_ask_openapi(self, *args): |
|
| 626 | """Test _ask_openapi method.""" |
||
| 627 | 1 | (mock_input, mock_exists) = args |
|
| 628 | 1 | mock_input.side_effect = ['', '', 'yes', 'no'] |
|
| 629 | 1 | mock_exists.side_effect = [True, False, False, False] |
|
| 630 | |||
| 631 | 1 | for expected in [False, True, True, False]: |
|
| 632 | 1 | response = self.napps_manager._ask_openapi() |
|
| 633 | 1 | self.assertEqual(response, expected) |
|
| 634 | |||
| 635 | 1 | @patch('kytos.utils.napps.NAppsClient') |
|
| 636 | 1 | def test_reload(self, mock_napps_client): |
|
| 637 | """Test reload method.""" |
||
| 638 | 1 | napps_client = MagicMock() |
|
| 639 | 1 | mock_napps_client.return_value = napps_client |
|
| 640 | |||
| 641 | 1 | napps = [] |
|
| 642 | 1 | self.napps_manager.reload(napps) |
|
| 643 | |||
| 644 | napps_client.reload_napps.assert_called_with(napps) |
||
| 645 |
Loading history...