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