Firstly, let's write test functions
Test functions
Web-tier
Checking web-tier doesn't require logic, if we get response from app, then web tier works.def check_web():
"Stub function to check web tier"
return True
Checking django cache
Test django cache and Redis connection. Just set and get some value.TEST_CACHE_KEY = 'key'
TEST_CACHE_VAL = 'value'
from django.core.cache import cache
def check_cache_set():
try:
return cache.set(TEST_CACHE_KEY, TEST_CACHE_VAL)
except:
return False
def check_cache_get():
try:
data = cache.get(TEST_CACHE_KEY)
if data != TEST_CACHE_VAL:
raise ValueError("invalid value")
except:
return False
return True
def check_cache():
"Checks is cache works ok"
return check_cache_set() and check_cache_get()
Database connection
To check DB we can simply do some SQL request. Simpliest thing we can do is to retrieve list of tables:def check_db():
from django.db import connection
try:
connection.introspection.table_names()
except:
return False
return True
Worker tier
Checks RQ and Worker nodeimport time
TIMEOUT = 5 # seconds
def check_worker():
from globalapp.tasks import ping_task
try:
start = time.time()
job = ping_task.delay("test")
while time.time() - start < TIMEOUT:
if job.is_finished:
return job.result == 'tset' # reversed
return False
except:
return False
Where ping_task is something like
@job
def ping_task(say):
return say[::-1]
Running tests
Approach is to add special view that will run server-side checks. So, firstly add url:
urlpatterns += patterns(
'globalapp.views',
url(r'^status/(?P<check>[A-z0-9]+)/$', 'status_view', name='status'),
)
View is simply need to choose check-function in respect of 'check' argument and return response with True or False. Here Django REST Framework was used, but it's pretty simple to rewrite to general django app.
from rest_framework.generics import GenericAPIView
import globalapp.checks as checks # here your module that contains check functions
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from rest_framework.authentication import TokenAuthentication
from rest_framework import status
from rest_framework.generics import GenericAPIView
class StatusView(GenericAPIView):
permission_classes = (IsAuthenticated, )
authentication_classes = (TokenAuthentication, )
def get(self, request, check):
resp = {'name': check}
stat = status.HTTP_200_OK
check_func = None
try:
check_func = getattr(checks, 'check_{}'.format(check))
except AttributeError:
stat = status.HTTP_400_BAD_REQUEST
resp['error'] = 'Unknown check'
else:
resp['result'] = check_func()
return Response(resp, status=stat)
Testing framework
To execute tests I use nosetests, it's pretty simple to use it andclass HerokuTest(TestCase):
def setUp(self):
super().setUp()
self.base_url = "http://example.com"
self.headers = {}
def check(self, name):
conn = requests.get('%s/status/%s/' % (self.base_url, name), headers=self.headers)
assert (conn.status_code == 200), "Failed %s tier test" % name
r = json.loads(conn.content.decode())
assert 'result' in r, "External error"
assert r['result'], "%s not works" % name
def test_web(self):
self.check('web')
def test_db(self):
self.check('db')
def test_cache(self):
self.check('web')
def test_worker(self):
self.check('worker')
Here it's very simple to add custom test.
Integration to Jenkins
It's pretty simple because nosetests can produce results in JUnit formatAdd shell command task
$ nosetests -v --with-xunit || true
and add post-build task "Publish JUnit" and choose path to nosetests.xml file
Комментариев нет
Отправить комментарий