mirror of
https://github.com/netbox-community/netbox.git
synced 2024-05-10 07:54:54 +00:00
Add unittests
This commit is contained in:
Sander Steffann
160
netbox/netbox/tests/test_get_releases.py
Normal file
160
netbox/netbox/tests/test_get_releases.py
Normal file
@ -0,0 +1,160 @@
|
||||
from io import BytesIO
|
||||
from logging import ERROR
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import requests
|
||||
from cacheops import CacheMiss, RedisCache
|
||||
from django.test import SimpleTestCase, override_settings
|
||||
from packaging.version import Version
|
||||
from requests import Response
|
||||
|
||||
from utilities.background_tasks import get_releases
|
||||
|
||||
|
||||
def successful_github_response(url, *_args, **_kwargs):
|
||||
r = Response()
|
||||
r.url = url
|
||||
r.status_code = 200
|
||||
r.reason = 'OK'
|
||||
r.headers = {
|
||||
'Content-Type': 'application/json; charset=utf-8',
|
||||
}
|
||||
r.raw = BytesIO(b'''[
|
||||
{
|
||||
"html_url": "https://github.com/netbox-community/netbox/releases/tag/v2.7.8",
|
||||
"tag_name": "v2.7.8",
|
||||
"prerelease": false
|
||||
},
|
||||
{
|
||||
"html_url": "https://github.com/netbox-community/netbox/releases/tag/v2.6-beta1",
|
||||
"tag_name": "v2.6-beta1",
|
||||
"prerelease": true
|
||||
},
|
||||
{
|
||||
"html_url": "https://github.com/netbox-community/netbox/releases/tag/v2.5.9",
|
||||
"tag_name": "v2.5.9",
|
||||
"prerelease": false
|
||||
}
|
||||
]
|
||||
''')
|
||||
return r
|
||||
|
||||
|
||||
def unsuccessful_github_response(url, *_args, **_kwargs):
|
||||
r = Response()
|
||||
r.url = url
|
||||
r.status_code = 404
|
||||
r.reason = 'Not Found'
|
||||
r.headers = {
|
||||
'Content-Type': 'application/json; charset=utf-8',
|
||||
}
|
||||
r.raw = BytesIO(b'''{
|
||||
"message": "Not Found",
|
||||
"documentation_url": "https://developer.github.com/v3/repos/releases/#list-releases-for-a-repository"
|
||||
}
|
||||
''')
|
||||
return r
|
||||
|
||||
|
||||
@override_settings(UPDATE_REPO_URL='https://localhost/unittest', UPDATE_CACHE_TIMEOUT=160876)
|
||||
class GetReleasesTestCase(SimpleTestCase):
|
||||
@patch.object(requests, 'get')
|
||||
@patch.object(RedisCache, 'set')
|
||||
@patch.object(RedisCache, 'get')
|
||||
def test_pre_releases(self, dummy_cache_get: Mock, dummy_cache_set: Mock, dummy_request_get: Mock):
|
||||
dummy_cache_get.side_effect = CacheMiss()
|
||||
dummy_request_get.side_effect = successful_github_response
|
||||
|
||||
releases = get_releases(pre_releases=True)
|
||||
|
||||
# Check result
|
||||
self.assertListEqual(releases, [
|
||||
(Version('2.7.8'), 'https://github.com/netbox-community/netbox/releases/tag/v2.7.8'),
|
||||
(Version('2.6b1'), 'https://github.com/netbox-community/netbox/releases/tag/v2.6-beta1'),
|
||||
(Version('2.5.9'), 'https://github.com/netbox-community/netbox/releases/tag/v2.5.9')
|
||||
])
|
||||
|
||||
# Check if correct request is made
|
||||
dummy_request_get.assert_called_once()
|
||||
dummy_request_get.assert_called_with('https://localhost/unittest/releases',
|
||||
headers={
|
||||
'Accept': 'application/vnd.github.v3+json'
|
||||
})
|
||||
|
||||
# Check if result is put in cache
|
||||
dummy_cache_set.assert_called_once()
|
||||
dummy_cache_set.assert_called_with('netbox_releases', releases, 160876)
|
||||
|
||||
@patch.object(requests, 'get')
|
||||
@patch.object(RedisCache, 'set')
|
||||
@patch.object(RedisCache, 'get')
|
||||
def test_no_pre_releases(self, dummy_cache_get: Mock, dummy_cache_set: Mock, dummy_request_get: Mock):
|
||||
dummy_cache_get.side_effect = CacheMiss()
|
||||
dummy_request_get.side_effect = successful_github_response
|
||||
|
||||
releases = get_releases(pre_releases=False)
|
||||
|
||||
# Check result
|
||||
self.assertListEqual(releases, [
|
||||
(Version('2.7.8'), 'https://github.com/netbox-community/netbox/releases/tag/v2.7.8'),
|
||||
(Version('2.5.9'), 'https://github.com/netbox-community/netbox/releases/tag/v2.5.9')
|
||||
])
|
||||
|
||||
# Check if correct request is made
|
||||
dummy_request_get.assert_called_once()
|
||||
dummy_request_get.assert_called_with('https://localhost/unittest/releases',
|
||||
headers={
|
||||
'Accept': 'application/vnd.github.v3+json'
|
||||
})
|
||||
|
||||
# Check if result is put in cache
|
||||
dummy_cache_set.assert_called_once()
|
||||
dummy_cache_set.assert_called_with('netbox_releases', releases, 160876)
|
||||
|
||||
@patch.object(requests, 'get')
|
||||
@patch.object(RedisCache, 'set')
|
||||
@patch.object(RedisCache, 'get')
|
||||
def test_failed_request(self, dummy_cache_get: Mock, dummy_cache_set: Mock, dummy_request_get: Mock):
|
||||
dummy_cache_get.side_effect = CacheMiss()
|
||||
dummy_request_get.side_effect = unsuccessful_github_response
|
||||
|
||||
with self.assertLogs(level=ERROR) as cm:
|
||||
releases = get_releases()
|
||||
|
||||
# Check log entry
|
||||
self.assertEqual(len(cm.output), 1)
|
||||
log_output = cm.output[0]
|
||||
last_log_line = log_output.split('\n')[-1]
|
||||
self.assertRegex(last_log_line, '404 .* Not Found')
|
||||
|
||||
# Check result
|
||||
self.assertListEqual(releases, [])
|
||||
|
||||
# Check if correct request is made
|
||||
dummy_request_get.assert_called_once()
|
||||
dummy_request_get.assert_called_with('https://localhost/unittest/releases',
|
||||
headers={
|
||||
'Accept': 'application/vnd.github.v3+json'
|
||||
})
|
||||
|
||||
# Check if failure is put in cache
|
||||
dummy_cache_set.assert_called_once()
|
||||
dummy_cache_set.assert_called_with('netbox_releases_no_retry', 'https://localhost/unittest/releases', 900)
|
||||
|
||||
@patch.object(requests, 'get')
|
||||
@patch.object(RedisCache, 'set')
|
||||
@patch.object(RedisCache, 'get')
|
||||
def test_blocked_retry(self, dummy_cache_get: Mock, dummy_cache_set: Mock, dummy_request_get: Mock):
|
||||
dummy_cache_get.return_value = 'https://localhost/unittest/releases'
|
||||
dummy_request_get.side_effect = successful_github_response
|
||||
|
||||
releases = get_releases()
|
||||
|
||||
# Check result
|
||||
self.assertListEqual(releases, [])
|
||||
|
||||
# Check if request is NOT made
|
||||
dummy_request_get.assert_not_called()
|
||||
|
||||
# Check if cache is not updated
|
||||
dummy_cache_set.assert_not_called()
|
||||
@ -1,7 +1,7 @@
|
||||
import logging
|
||||
|
||||
import requests
|
||||
from cacheops import cache
|
||||
from cacheops.simple import cache, CacheMiss
|
||||
from django.conf import settings
|
||||
from django_rq import job
|
||||
from packaging import version
|
||||
@ -17,24 +17,35 @@ def get_releases(pre_releases=False):
|
||||
'Accept': 'application/vnd.github.v3+json',
|
||||
}
|
||||
|
||||
# Check whether this URL has failed and shouldn't be retried yet
|
||||
try:
|
||||
failed_url = cache.get('netbox_releases_no_retry')
|
||||
if url == failed_url:
|
||||
return []
|
||||
except CacheMiss:
|
||||
pass
|
||||
|
||||
releases = []
|
||||
|
||||
# noinspection PyBroadException
|
||||
try:
|
||||
response = requests.get(url, headers=headers)
|
||||
response.raise_for_status()
|
||||
|
||||
for release in response.json():
|
||||
if 'tag_name' not in release:
|
||||
continue
|
||||
|
||||
if not pre_releases and (release.get('is_devrelease') or release.get('is_prerelease')):
|
||||
if not pre_releases and (release.get('devrelease') or release.get('prerelease')):
|
||||
continue
|
||||
|
||||
releases.append((version.parse(release['tag_name']), release.get('html_url')))
|
||||
except Exception:
|
||||
# Don't retry this URL for 15 minutes
|
||||
cache.set('netbox_releases_no_retry', url, 900)
|
||||
|
||||
logger.exception("Error while fetching {}".format(url))
|
||||
return []
|
||||
|
||||
logger.debug("Found NetBox releases {}".format([str(release) for release, url in releases]))
|
||||
|
||||
cache.set('netbox_releases', releases, settings.UPDATE_CACHE_TIMEOUT)
|
||||
return releases
|
||||
|
||||
Reference in New Issue
Block a user