Unit test Python library for testing cache systems like varnish
Based on the work of @onyxfish (https://gist.github.com/onyxfish)
- cachetest.py : productive
- wordpresscachetest.py : not usable yet
user@localhost> python3.3 exampletest 
-
-
Save paje/69ccd26c38e2b0210de8 to your computer and use it in GitHub Desktop.
Varnish Unit Test
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import random | |
| import unittest | |
| import sys, getopt | |
| import http.cookiejar | |
| import json | |
| import re | |
| import time | |
| import requests | |
| import http.client | |
| __unittest = True | |
| class CacheTest(unittest.TestCase): | |
| """ | |
| Basis class common to all CMS and tests | |
| """ | |
| def setUp(self, proxy, port, website): | |
| self.website = "" | |
| self.headers = None | |
| self.proxies = None | |
| if proxy != "": | |
| # print("proxying website ", website, " with proxy ", proxy) | |
| self.proxies = { "http": proxy, "https": proxy} | |
| self.proxy = proxy | |
| self.port = 80 | |
| if port != "": | |
| self.port = port | |
| if website != "": | |
| self.website = website | |
| self.headers = {'Host' : self.website} | |
| # Add specific header | |
| def set_header(self, header_value): | |
| self.headers.update(header_value) | |
| def get_request(self, url): | |
| # Recoding get request to allow proxy | |
| conn = http.client.HTTPConnection(self.proxy, self.port) | |
| conn.putrequest("GET", url, skip_host=True) | |
| conn.putheader(self.headers) | |
| conn.endheaders() | |
| response = conn.getresponse() | |
| conn.close() | |
| return response | |
| def build_url(self, path): | |
| """ | |
| Construct an absolute url by appending a path to a domain. | |
| """ | |
| return 'http://%s%s' % (self.website, path) | |
| def get_once(self, url, **kwargs): | |
| response = self.get_request(url) | |
| return response | |
| def get_twice(self, url, **kwargs): | |
| """ | |
| Fetch a url twice and return the second response (for testing cache hits). | |
| """ | |
| self.get_request(url) | |
| time.sleep(2) | |
| response = self.get_request(url) | |
| return response | |
| def get_twice_tokenized(self, url, tokenname=None, **kwargs): | |
| """ | |
| Fetch a url twice with two different tokens and return the 2nd response | |
| """ | |
| if tokenname != None: | |
| token = tokenname + "=" + str(random.randint(10000,999999)) | |
| else: | |
| token = str(random.randint(10000,999999)) | |
| #print("url1: " + url + "?" + token) | |
| self.get_request(url + "?" + token) | |
| time.sleep(2) | |
| if tokenname != None : | |
| token = tokenname + "=" + str(random.randint(10000,999999)) | |
| else: | |
| token = str(random.randint(10000,999999)) | |
| #print("url2: " + url + "?" + token) | |
| response = self.get_request(url + "?" + token) | |
| return response | |
| """ | |
| Assertions | |
| """ | |
| def assertHit(self, response): | |
| """ | |
| Assert that a given response contains the header indicating a cache hit. | |
| """ | |
| self.assertEqual(response.headers['X-Cache'].lower(), 'HIT'.lower(), msg='Uncached while cache was expected') | |
| def assertMiss(self, response): | |
| """ | |
| Assert that a given response contains the header indicating a cache miss. | |
| """ | |
| self.assertEqual(response.headers['X-Cache'].lower(), 'miss'.lower()) | |
| def assertMaxAge(self, response, value): | |
| """ | |
| Assert that a given response contains the header indicating specific "max-age" value. | |
| """ | |
| MAX_AGE_REGEX = re.compile('max-age\s*=\s*(\d+)') | |
| try: | |
| cache_control = response.headers['cache-control'] | |
| except KeyError: | |
| try: | |
| cache_control = response.headers['Cache-Control'] | |
| except: | |
| raise AssertionError('No cache-control header.') | |
| max_age = MAX_AGE_REGEX.match(cache_control) | |
| if not max_age: | |
| raise AssertionError('No max-age specified in cache-control header.') | |
| self.assertEqual(int(max_age.group(1)), value) | |
| def assert200(self, response): | |
| # Ok | |
| self.assertEqual(response.status, 200) | |
| def assert30X(self, response): | |
| self.assertRegex(str(response.status), '30?') | |
| def assert301(self, response): | |
| # Permanent redirect | |
| self.assertEqual(response.status, 301) | |
| def assert302(self, response): | |
| # Temporary redirect | |
| self.assertEqual(response.status, 302) | |
| def assert304(self, response): | |
| # Not modified | |
| self.assertEqual(response.status, 304) | |
| def assert40X(self, response): | |
| self.assertRegex(str(response.status), '40?') | |
| def assert400(self, response): | |
| # Bad Request | |
| self.assertEqual(response.status, 400) | |
| def assert401(self, response): | |
| # Unauthorized | |
| self.assertEqual(response.status, 401) | |
| def assert403(self, response): | |
| # Forbidden | |
| self.assertEqual(response.status, 403) | |
| def assert404(self, response): | |
| # Not found | |
| self.assertEqual(response.status, 404) | |
| def assert405(self, response): | |
| # Method Not allowed | |
| self.assertEqual(response.status, 405) | |
| def assert50X(self, response): | |
| # Method Not allowed | |
| self.assertRegex(str(response.status), '50?') | |
| def assertBackend(self, response, backend): | |
| self.assertEqual(response.headers['X-Back'].lower(), backend.lower()) | |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import sys, getopt | |
| import json | |
| import re | |
| import random | |
| import unittest | |
| from cachetest import CacheTest | |
| PROXYHOST='127.0.0.1' | |
| PROXYPORT='80' | |
| """ | |
| ARTE Homepage | |
| """ | |
| class TestHome(CacheTest): | |
| def setUp(self): | |
| CacheTest.setUp(self, PROXYHOST, PROXYPORT, "www.arte.tv") | |
| def test_homeLangRedirect(self): | |
| url = '/' | |
| response = self.get_once(url) | |
| # Should be a redirect (lang test) | |
| self.assert301(response) | |
| def test_homeCache(self): | |
| url = '/fr' | |
| response = self.get_twice(url) | |
| # Content should be cached | |
| self.assertHit(response) | |
| # Cache delay should be 60s | |
| self.assertMaxAge(response, 60) | |
| # Backend server should be HPV2 | |
| self.assertBackend(response, "HPV2") | |
| if __name__ == "__main__": | |
| unittest.main(verbosity=2) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python | |
| import cookielib | |
| import json | |
| import re | |
| import random | |
| import unittest | |
| import requests | |
| from wordpress_xmlrpc import Client, WordPressComment, WordPressPost | |
| from wordpress_xmlrpc.methods.posts import EditPost, DeletePost, GetPost, NewPost | |
| from wordpress_xmlrpc.methods.comments import EditComment, DeleteComment, NewComment | |
| DOMAIN = 'chicagonow.dev' | |
| USERNAME = 'admin' | |
| PASSWORD = 'password' | |
| TEST_BLOG = 'cta-tattler' | |
| MAX_AGE_REGEX = re.compile('max-age\s*=\s*(\d+)') | |
| def build_url(path): | |
| """ | |
| Construct an absolute url by appending a path to a domain. | |
| """ | |
| return 'http://%s%s' % (DOMAIN, path) | |
| class TestCachingBase(unittest.TestCase): | |
| """ | |
| Base class Wordpress + Varnish TestCases. | |
| Provides utils for login/logout and asserting hits/misses. | |
| """ | |
| def login(self): | |
| """ | |
| Login and return a cookie jar holding the login cookie. | |
| """ | |
| cookies = cookielib.CookieJar() | |
| response = requests.post( | |
| build_url('/wp-admin/admin-ajax.php'), | |
| cookies=cookies, | |
| data = { | |
| 'action': 'chicagonow', | |
| 'url': 'users/login', | |
| 'data': '{"user_email":"%s","user_pass":"%s"}' % (USERNAME, PASSWORD) | |
| } | |
| ) | |
| self.assertEqual(response.status_code, 200) | |
| self.assertMiss(response) | |
| return cookies | |
| def logout(self, cookies): | |
| """ | |
| Logout and return a now-empty cookie jar. | |
| """ | |
| response = requests.post( | |
| build_url('/wp-admin/admin-ajax.php'), | |
| cookies=cookies, | |
| data = { | |
| 'action': 'chicagonow', | |
| 'url': 'users/logout' | |
| } | |
| ) | |
| self.assertEqual(response.status_code, 200) | |
| self.assertMiss(response) | |
| response_json = json.loads(response.read()) | |
| self.assertEqual(response_json['logged_out'], True) | |
| return cookies | |
| def get_xmlrpc(self, blog=None): | |
| """ | |
| Fetch an XML-RPC client for a given blog (or the root blog). | |
| """ | |
| if blog: | |
| path = '/%s/xmlrpc.php' % blog | |
| else: | |
| path = '/xmlrpc.php' | |
| return Client(build_url(path), USERNAME, PASSWORD) | |
| def new_post(self, blog): | |
| """ | |
| Create a new post. | |
| """ | |
| xmlrpc = self.get_xmlrpc(blog) | |
| post = WordPressPost() | |
| post.title = 'New post from api' | |
| post.description = 'New description' | |
| return xmlrpc.call(NewPost(post, True)) | |
| def edit_post(self, blog, post_id): | |
| """ | |
| Edit a post. | |
| """ | |
| xmlrpc = self.get_xmlrpc(blog) | |
| post = WordPressPost() | |
| post.title = 'Edited post from api' | |
| post.description = 'Edited description' | |
| return xmlrpc.call(EditPost(post_id, post, True)) | |
| def delete_post(self, blog, post_id): | |
| """ | |
| Delete a post. | |
| """ | |
| xmlrpc = self.get_xmlrpc(blog) | |
| return xmlrpc.call(DeletePost(post_id)) | |
| def get_post(self, blog, post_id): | |
| """ | |
| Fetch a post object. | |
| """ | |
| xmlrpc = self.get_xmlrpc(blog) | |
| return xmlrpc.call(GetPost(post_id)) | |
| def new_comment(self, blog, post_id): | |
| """ | |
| Create a new comment. | |
| """ | |
| xmlrpc = self.get_xmlrpc(blog) | |
| comment = WordPressComment() | |
| comment.content = 'Test comment from api. (%s)' % str(random.random() * 99999999) | |
| return xmlrpc.call(NewComment(post_id, comment)) | |
| def edit_comment(self, blog, comment_id): | |
| """ | |
| Edit a comment. | |
| """ | |
| xmlrpc = self.get_xmlrpc(blog) | |
| comment = WordPressComment() | |
| comment.content = 'Edited comment from api. (%s)' % str(random.random() * 99999999) | |
| return xmlrpc.call(EditComment(comment_id, comment)) | |
| def delete_comment(self, blog, comment_id): | |
| """ | |
| Delete a comment. | |
| """ | |
| xmlrpc = self.get_xmlrpc(blog) | |
| return xmlrpc.call(DeleteComment(comment_id)) | |
| def get_twice(self, url, **kwargs): | |
| """ | |
| Fetch a url twice and return the second response (for testing cache hits). | |
| """ | |
| requests.get(url, **kwargs) | |
| response = requests.get(url, **kwargs) | |
| return response | |
| def assertHit(self, response): | |
| """ | |
| Assert that a given response contains the header indicating a cache hit. | |
| """ | |
| self.assertEqual(response.headers['X-Cache'], 'Hit') | |
| def assertMiss(self, response): | |
| """ | |
| Assert that a given response contains the header indicating a cache miss. | |
| """ | |
| self.assertEqual(response.headers['X-Cache'], 'Miss') | |
| def assertMaxAge(self, response, value): | |
| """ | |
| Assert that a given response contains the header indicating specific "max-age" value. | |
| """ | |
| try: | |
| cache_control = response.headers['cache-control'] | |
| except KeyError: | |
| try: | |
| cache_control = response.headers['Cache-Control'] | |
| except: | |
| raise AssertionError('No cache-control header.') | |
| max_age = MAX_AGE_REGEX.match(cache_control) | |
| if not max_age: | |
| raise AssertionError('No max-age specified in cache-control header.') | |
| self.assertEqual(int(max_age.group(1)), value) | |
| class TestLoggedIn(TestCachingBase): | |
| """ | |
| Tests for logged-in users. | |
| """ | |
| def setUp(self): | |
| self.cookies = self.login() | |
| def test_homepage(self): | |
| url = build_url('/') | |
| response = self.get_twice(url, cookies=self.cookies) | |
| self.assertHit(response) | |
| def test_post_preview(self): | |
| url = build_url('/2400-north1200-west/?p=4&preview=true&url=preview/4') | |
| response = requests.get(url, cookies=self.cookies) | |
| self.assertMiss(response) | |
| class TestLoggedOut(TestCachingBase): | |
| """ | |
| Tests for logged-out users. | |
| """ | |
| def test_homepage(self): | |
| url = build_url('/') | |
| response = self.get_twice(url) | |
| self.assertHit(response) | |
| self.assertMaxAge(response, 300) | |
| def test_homepage_login_logout(self): | |
| url = build_url('/') | |
| cookies = self.login() | |
| cookies = self.logout(cookies) | |
| response = self.get_twice(url, cookies=cookies) | |
| self.assertHit(response) | |
| def test_search(self): | |
| url = build_url('/search/') | |
| response = self.get_twice(url, params={ 'blog_s': 'test' }) | |
| self.assertMiss(response) | |
| def test_static_content(self): | |
| url = build_url('/wp-content/themes/chicagonow/images/home-logo.png') | |
| response = self.get_twice(url) | |
| self.assertHit(response) | |
| self.assertMaxAge(response, 604800) | |
| def test_avatar(self): | |
| url = build_url('/avatar/user-1-16.png') | |
| response = self.get_twice(url) | |
| self.assertHit(response) | |
| self.assertMaxAge(response, 604800) | |
| def test_ajax_users(self): | |
| url = build_url('/wp-admin/admin-ajax.php') | |
| data = { | |
| 'action': 'chicagonow', | |
| 'url': 'users', | |
| 'data': 'null' | |
| } | |
| response = self.get_twice(url, data=data) | |
| self.assertMiss(response) | |
| def test_ajax_comment_form(self): | |
| url = build_url('/wp-admin/admin-ajax.php') | |
| data = { | |
| 'action': 'commentform', | |
| 'data': '{ "post_id": 61 }' | |
| } | |
| response = self.get_twice(url, data=data) | |
| self.assertMiss(response) | |
| def test_new_post(self): | |
| url = build_url('/%s/' % TEST_BLOG) | |
| response = self.get_twice(url) | |
| self.assertHit(response) | |
| self.new_post(TEST_BLOG) | |
| response = requests.get(url) | |
| self.assertMiss(response) | |
| def test_edit_post(self): | |
| url = build_url('/%s/' % TEST_BLOG) | |
| post_id = self.new_post(TEST_BLOG) | |
| response = self.get_twice(url) | |
| self.assertHit(response) | |
| self.edit_post(TEST_BLOG, post_id) | |
| response = requests.get(url) | |
| self.assertMiss(response) | |
| def test_delete_post(self): | |
| url = build_url('/%s/' % TEST_BLOG) | |
| post_id = self.new_post(TEST_BLOG) | |
| response = self.get_twice(url) | |
| self.assertHit(response) | |
| self.delete_post(TEST_BLOG, post_id) | |
| response = requests.get(url) | |
| self.assertMiss(response) | |
| def test_preview_post(self): | |
| post_id = self.new_post(TEST_BLOG) | |
| post = self.get_post(TEST_BLOG, post_id) | |
| response = self.get_twice('%s?preview=true' % post.permalink) | |
| self.assertMiss(response) | |
| def test_new_comment(self): | |
| post_id = self.new_post(TEST_BLOG) | |
| post = self.get_post(TEST_BLOG, post_id) | |
| response = self.get_twice(post.permalink) | |
| self.assertHit(response) | |
| self.new_comment(TEST_BLOG, post_id) | |
| response = requests.get(post.permalink) | |
| self.assertMiss(response) | |
| def test_edit_comment(self): | |
| post_id = self.new_post(TEST_BLOG) | |
| post = self.get_post(TEST_BLOG, post_id) | |
| comment_id = self.new_comment(TEST_BLOG, post_id) | |
| response = self.get_twice(post.permalink) | |
| self.assertHit(response) | |
| self.edit_comment(TEST_BLOG, comment_id) | |
| response = requests.get(post.permalink) | |
| self.assertMiss(response) | |
| def test_delete_comment(self): | |
| post_id = self.new_post(TEST_BLOG) | |
| post = self.get_post(TEST_BLOG, post_id) | |
| comment_id = self.new_comment(TEST_BLOG, post_id) | |
| response = self.get_twice(post.permalink) | |
| self.assertHit(response) | |
| self.delete_comment(TEST_BLOG, comment_id) | |
| response = requests.get(post.permalink) | |
| self.assertMiss(response) | |
| def test_new_comments_purge_too_much(self): | |
| # Comments were incorrectly busting site root | |
| post_id = self.new_post(TEST_BLOG) | |
| url = build_url('/') | |
| response = self.get_twice(url) | |
| self.assertHit(response) | |
| self.new_comment(TEST_BLOG, post_id) | |
| response = requests.get(url) | |
| self.assertHit(response) | |
| if __name__ == '__main__': | |
| unittest.main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment