test libbrotli

.coveragerc

@@ -0,0 +1,3 @@
+[run]
+omit =
+    */site-packages/*

lib/brotli.py

@@ -8,15 +8,15 @@ import mimetypes
 
 
 class BrotliCache(object):
-    def __init__(self, redis_uri='redis://', max_wait=0.020, expire=60*60*6):
+    def __init__(self, redis_uri='redis://', timeout=0.020, expire=60*60*6):
         self.redis = redis.StrictRedis.from_url(redis_uri)
-        self.max_wait = max_wait
+        self.timeout = timeout
         self.expire = expire
         self.redis.client_setname('brotlicache')
 
     def compress(self, cache_key, lock_key, body, mode=brotli_.MODE_GENERIC):
         encbody = brotli_.compress(body, mode=mode)
-        self.redis.set(cache_key, encbody, ex=self.expire)
+        self.redis.set(cache_key, encbody, px=int(self.expire*1000))
         self.redis.delete(lock_key)
 
     def wrap_response(self, response):
@@ -41,8 +41,8 @@ class BrotliCache(object):
                     target=self.compress,
                     args=(cache_key, lock_key, body, mode))
                 t.start()
-                if self.max_wait > 0:
-                    t.join(self.max_wait)
+                if self.timeout > 0:
+                    t.join(self.timeout)
                     encbody = self.redis.get(cache_key)
                     if not encbody:
                         response.headers.set('x-brotli-cache', 'TIMEOUT')
@@ -57,7 +57,7 @@ class BrotliCache(object):
         return response
 
 
-def brotli(app, static=True, dynamic=True):
+def brotli(app, static=True, dynamic=True, **kwargs):
     original_static = app.view_functions['static']
 
     def static_maybe_gzip_brotli(filename=None):
@@ -79,5 +79,5 @@ def brotli(app, static=True, dynamic=True):
     if static:
         app.view_functions['static'] = static_maybe_gzip_brotli
     if dynamic:
-        cache = BrotliCache(redis_uri = app.config.get('REDIS_URI'))
+        cache = BrotliCache(redis_uri=app.config.get('REDIS_URI'), **kwargs)
         app.after_request(cache.wrap_response)

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+redis_port = 15487

test/static/bee.txt

@@ -0,0 +1 @@
+According to all known laws of aviation, there is no way a bee should be able to fly. Its wings are too small to get its fat little body off the ground.

test/test_libbrotli.py

@@ -0,0 +1,120 @@
+import pytest
+import lib.brotli
+
+
+BEE_SCRIPT = bytes("According to all known laws of aviation,", 'UTF-8')
+TIMEOUT_TARGET = 0.2
+
+
+@pytest.fixture
+def app(redisdb):
+    from flask import Flask
+    app_ = Flask(__name__)
+    app_.config['REDIS_URI'] = 'redis://localhost:15487'
+
+    @app_.route('/')
+    def hello():
+        return 'Hello, world!'
+    with app_.app_context():
+        yield app_
+
+
+@pytest.fixture
+def br_app(app):
+    lib.brotli.brotli(app, timeout=TIMEOUT_TARGET)
+    return app
+
+
+@pytest.fixture
+def br_client(br_app):
+    return br_app.test_client()
+
+
+def test_brotli_static_passthru(br_client):
+    resp = br_client.get('/static/bee.txt')
+    assert BEE_SCRIPT in resp.data
+
+
+def test_brotli_static_gzip(br_client):
+    from gzip import decompress
+
+    gzip_resp = br_client.get(
+            '/static/bee.txt',
+            headers=[('accept-encoding', 'gzip')])
+    assert gzip_resp.headers.get('content-encoding') == 'gzip'
+
+    assert BEE_SCRIPT in decompress(gzip_resp.data)
+
+
+def test_brotli_static_br(br_client):
+    from brotli import decompress
+
+    br_resp = br_client.get(
+            '/static/bee.txt',
+            headers=[('accept-encoding', 'gzip, br')])
+    assert br_resp.headers.get('content-encoding') == 'br'
+
+    assert BEE_SCRIPT in decompress(br_resp.data)
+
+
+def test_brotli_dynamic(br_client):
+    from brotli import decompress
+
+    resp = br_client.get(
+            '/',
+            headers=[('accept-encoding', 'gzip, br')])
+
+    assert resp.headers.get('x-brotli-cache') == 'MISS'
+    assert resp.headers.get('content-encoding') == 'br'
+    assert b"Hello, world!" in decompress(resp.data)
+
+
+def test_brotli_dynamic_cache(br_client):
+    from brotli import decompress
+    from time import sleep
+
+    br_client.get(
+            '/',
+            headers=[('accept-encoding', 'gzip, br')])
+
+    sleep(0.5)
+
+    resp = br_client.get(
+            '/',
+            headers=[('accept-encoding', 'gzip, br')])
+
+    assert resp.headers.get('x-brotli-cache') == 'HIT'
+    assert resp.headers.get('content-encoding') == 'br'
+    assert b"Hello, world!" in decompress(resp.data)
+
+
+def test_brotli_dynamic_timeout(app):
+    lib.brotli.brotli(app, timeout=0.0001)
+
+    client = app.test_client()
+
+    resp = client.get(
+            '/',
+            headers=[('accept-encoding', 'gzip, br')])
+
+    assert resp.headers.get('x-brotli-cache') == 'TIMEOUT'
+    assert resp.headers.get('content-encoding') != 'br'
+
+
+def test_brotli_dynamic_expire(app):
+    from time import sleep
+
+    lib.brotli.brotli(app, expire=0.1)
+
+    client = app.test_client()
+    client.get(
+            '/',
+            headers=[('accept-encoding', 'gzip, br')])
+
+    sleep(1)
+
+    resp = client.get(
+            '/',
+            headers=[('accept-encoding', 'gzip, br')])
+
+    assert resp.headers.get('x-brotli-cache') != 'HIT'