|
12 | 12 | import unittest
|
13 | 13 | import textwrap
|
14 | 14 | from click.testing import CliRunner
|
| 15 | +from unittest.mock import Mock, patch |
15 | 16 |
|
16 | 17 | from safety import safety
|
17 | 18 | from safety import cli
|
@@ -256,6 +257,136 @@ def test_get_packages_licenses(self):
|
256 | 257 | "unexpected package '" + pkg_license['package'] + "' was found"
|
257 | 258 | )
|
258 | 259 |
|
| 260 | + def test_get_packages_licenses_without_api_key(self): |
| 261 | + from safety.errors import InvalidKeyError |
| 262 | + |
| 263 | + # without providing an API-KEY |
| 264 | + with self.assertRaises(InvalidKeyError) as error: |
| 265 | + safety.get_licenses( |
| 266 | + db_mirror=False, |
| 267 | + cached=False, |
| 268 | + proxy={}, |
| 269 | + key=None |
| 270 | + ) |
| 271 | + db_generic_exception = error.exception |
| 272 | + self.assertEqual(str(db_generic_exception), 'API-KEY not provided.') |
| 273 | + |
| 274 | + @patch("safety.safety.requests") |
| 275 | + def test_get_packages_licenses_with_invalid_api_key(self, requests): |
| 276 | + from safety.errors import InvalidKeyError |
| 277 | + |
| 278 | + mock = Mock() |
| 279 | + mock.status_code = 403 |
| 280 | + requests.get.return_value = mock |
| 281 | + |
| 282 | + # proving an invalid API-KEY |
| 283 | + with self.assertRaises(InvalidKeyError): |
| 284 | + safety.get_licenses( |
| 285 | + db_mirror=False, |
| 286 | + cached=False, |
| 287 | + proxy={}, |
| 288 | + key="INVALID" |
| 289 | + ) |
| 290 | + |
| 291 | + @patch("safety.safety.requests") |
| 292 | + def test_get_packages_licenses_db_fetch_error(self, requests): |
| 293 | + from safety.errors import DatabaseFetchError |
| 294 | + |
| 295 | + mock = Mock() |
| 296 | + mock.status_code = 500 |
| 297 | + requests.get.return_value = mock |
| 298 | + |
| 299 | + with self.assertRaises(DatabaseFetchError): |
| 300 | + safety.get_licenses( |
| 301 | + db_mirror=False, |
| 302 | + cached=False, |
| 303 | + proxy={}, |
| 304 | + key="MY-VALID-KEY" |
| 305 | + ) |
| 306 | + |
| 307 | + def test_get_packages_licenses_with_invalid_db_file(self): |
| 308 | + from safety.errors import DatabaseFileNotFoundError |
| 309 | + |
| 310 | + with self.assertRaises(DatabaseFileNotFoundError): |
| 311 | + safety.get_licenses( |
| 312 | + db_mirror='/my/invalid/path', |
| 313 | + cached=False, |
| 314 | + proxy={}, |
| 315 | + key=None |
| 316 | + ) |
| 317 | + |
| 318 | + @patch("safety.safety.requests") |
| 319 | + def test_get_packages_licenses_very_often(self, requests): |
| 320 | + from safety.errors import TooManyRequestsError |
| 321 | + |
| 322 | + # if the request is made too often, an 429 error is raise by PyUp.io |
| 323 | + mock = Mock() |
| 324 | + mock.status_code = 429 |
| 325 | + requests.get.return_value = mock |
| 326 | + |
| 327 | + with self.assertRaises(TooManyRequestsError): |
| 328 | + safety.get_licenses( |
| 329 | + db_mirror=False, |
| 330 | + cached=False, |
| 331 | + proxy={}, |
| 332 | + key="MY-VALID-KEY" |
| 333 | + ) |
| 334 | + |
| 335 | + @patch("safety.safety.requests") |
| 336 | + def test_get_cached_packages_licenses(self, requests): |
| 337 | + import copy |
| 338 | + from safety.constants import CACHE_FILE |
| 339 | + |
| 340 | + licenses_db = { |
| 341 | + "licenses": { |
| 342 | + "BSD-3-Clause": 2 |
| 343 | + }, |
| 344 | + "packages": { |
| 345 | + "django": [ |
| 346 | + { |
| 347 | + "start_version": "0.0", |
| 348 | + "license_id": 2 |
| 349 | + } |
| 350 | + ] |
| 351 | + } |
| 352 | + } |
| 353 | + original_db = copy.deepcopy(licenses_db) |
| 354 | + |
| 355 | + mock = Mock() |
| 356 | + mock.json.return_value = licenses_db |
| 357 | + mock.status_code = 200 |
| 358 | + requests.get.return_value = mock |
| 359 | + |
| 360 | + # lets clear the cache first |
| 361 | + try: |
| 362 | + with open(CACHE_FILE, 'w') as f: |
| 363 | + f.write(json.dumps({})) |
| 364 | + except Exception: |
| 365 | + pass |
| 366 | + |
| 367 | + # In order to cache the db (and get), we must set cached as True |
| 368 | + response = safety.get_licenses( |
| 369 | + db_mirror=False, |
| 370 | + cached=True, |
| 371 | + proxy={}, |
| 372 | + key="MY-VALID-KEY" |
| 373 | + ) |
| 374 | + self.assertEqual(response, licenses_db) |
| 375 | + |
| 376 | + # now we should have the db in cache |
| 377 | + # changing the "live" db to test if we are getting the cached db |
| 378 | + licenses_db['licenses']['BSD-3-Clause'] = 123 |
| 379 | + |
| 380 | + resp = safety.get_licenses( |
| 381 | + db_mirror=False, |
| 382 | + cached=True, |
| 383 | + proxy={}, |
| 384 | + key="MY-VALID-KEY" |
| 385 | + ) |
| 386 | + |
| 387 | + self.assertNotEqual(resp, licenses_db) |
| 388 | + self.assertEqual(resp, original_db) |
| 389 | + |
259 | 390 |
|
260 | 391 | class ReadRequirementsTestCase(unittest.TestCase):
|
261 | 392 |
|
|
0 commit comments