<s> from __future__ import unicode_literals <EOL> import unittest <EOL> from datetime import timedelta , datetime <EOL> from six . moves . cPickle import dumps , loads <EOL> from six import iteritems <EOL> from flask import Flask , send_file <EOL> from werkzeug . wrappers import Response <EOL> from werkzeug . datastructures import HeaderSet <EOL> from werkzeug . contrib . cache import SimpleCache <EOL> from flask_webcache . storage import Config , Metadata , Store , Retrieval <EOL> from flask_webcache . storage import ( CacheMiss , NoResourceMetadata , NoMatchingRepresentation , NotFreshEnoughForClient , <EOL> RecacheRequested ) <EOL> from flask_webcache . recache import RECACHE_HEADER <EOL> from flask_webcache . utils import werkzeug_cache_get_or_add <EOL> from testutils import compare_numbers <EOL> a = Flask ( __name__ ) <EOL> class UtilsTestCase ( unittest . TestCase ) : <EOL> def test_config_kwargs ( self ) : <EOL> with self . assertRaises ( TypeError ) : <EOL> Config ( foo = <NUM_LIT:1> ) <EOL> def test_metadata_datastructure ( self ) : <EOL> def check_metadata ( m ) : <EOL> self . assertEquals ( m . salt , '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT:foo>' , m . vary ) <EOL> self . assertIn ( '<STR_LIT:bar>' , m . vary ) <EOL> m = Metadata ( HeaderSet ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) ) , '<STR_LIT>' ) <EOL> check_metadata ( m ) <EOL> check_metadata ( loads ( dumps ( m ) ) ) <EOL> m2 = Metadata ( HeaderSet ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) ) , '<STR_LIT>' ) <EOL> self . assertEquals ( m , m2 ) <EOL> m3 = Metadata ( HeaderSet ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) ) , '<STR_LIT>' ) <EOL> self . assertNotEquals ( m2 , m3 ) <EOL> class StorageTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . c = SimpleCache ( ) <EOL> self . s = Store ( self . c ) <EOL> self . r = Retrieval ( self . c ) <EOL> def test_basic_cachability ( self ) : <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> self . assertFalse ( self . s . should_cache_response ( Response ( x for x in '<STR_LIT:foo>' ) ) ) <EOL> self . assertTrue ( self . s . should_cache_response ( Response ( status = <NUM_LIT> ) ) ) <EOL> self . assertFalse ( self . s . should_cache_response ( Response ( status = <NUM_LIT> ) ) ) <EOL> self . assertTrue ( self . s . should_cache_response ( Response ( '<STR_LIT:foo>' ) ) ) <EOL> self . assertTrue ( self . s . should_cache_response ( Response ( ) ) ) <EOL> r = Response ( ) <EOL> r . vary . add ( '<STR_LIT:*>' ) <EOL> self . assertFalse ( self . s . should_cache_response ( r ) ) <EOL> with a . test_request_context ( '<STR_LIT>' , method = '<STR_LIT>' ) : <EOL> self . assertFalse ( self . s . should_cache_response ( Response ( '<STR_LIT:foo>' ) ) ) <EOL> with a . test_request_context ( '<STR_LIT>' , method = '<STR_LIT:POST>' ) : <EOL> self . assertFalse ( self . s . should_cache_response ( Response ( '<STR_LIT:foo>' ) ) ) <EOL> def test_cache_control_cachability ( self ) : <EOL> def check_response_with_cache_control ( ** cc ) : <EOL> r = Response ( ) <EOL> for k , v in iteritems ( cc ) : <EOL> setattr ( r . cache_control , k , v ) <EOL> return self . s . should_cache_response ( r ) <EOL> with a . test_request_context ( ) : <EOL> self . assertTrue ( check_response_with_cache_control ( max_age = <NUM_LIT:10> ) ) <EOL> self . assertTrue ( check_response_with_cache_control ( must_revalidate = True ) ) <EOL> self . assertFalse ( check_response_with_cache_control ( max_age = <NUM_LIT:0> ) ) <EOL> self . assertFalse ( check_response_with_cache_control ( private = True ) ) <EOL> self . assertFalse ( check_response_with_cache_control ( no_cache = True ) ) <EOL> self . assertFalse ( check_response_with_cache_control ( no_store = True ) ) <EOL> def test_expire_cachability ( self ) : <EOL> def check_response_with_expires ( dt ) : <EOL> r = Response ( ) <EOL> r . expires = dt <EOL> return self . s . should_cache_response ( r ) <EOL> with a . test_request_context ( ) : <EOL> self . assertFalse ( check_response_with_expires ( datetime . utcnow ( ) - timedelta ( seconds = <NUM_LIT:1> ) ) ) <EOL> self . assertTrue ( check_response_with_expires ( datetime . utcnow ( ) + timedelta ( seconds = <NUM_LIT:1> ) ) ) <EOL> def test_default_cachability ( self ) : <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> self . assertTrue ( self . s . should_cache_response ( Response ( ) ) ) <EOL> with a . test_request_context ( '<STR_LIT>' , query_string = '<STR_LIT>' ) : <EOL> self . assertFalse ( self . s . should_cache_response ( Response ( ) ) ) <EOL> def test_x_cache_headers ( self ) : <EOL> r = Response ( ) <EOL> self . s . mark_cache_hit ( r ) <EOL> self . assertEquals ( r . headers [ self . s . X_CACHE_HEADER ] , '<STR_LIT>' ) <EOL> self . s . mark_cache_miss ( r ) <EOL> self . assertEquals ( r . headers [ self . s . X_CACHE_HEADER ] , '<STR_LIT>' ) <EOL> def test_metadata_miss ( self ) : <EOL> with self . assertRaises ( NoResourceMetadata ) : <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> self . r . fetch_metadata ( ) <EOL> def test_response_miss ( self ) : <EOL> with self . assertRaises ( NoResourceMetadata ) : <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> self . r . fetch_response ( ) <EOL> def test_store_retrieve_cycle ( self ) : <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> r = Response ( '<STR_LIT:foo>' ) <EOL> self . s . cache_response ( r ) <EOL> self . assertEquals ( len ( self . c . _cache ) , <NUM_LIT:2> ) <EOL> r2 = self . r . fetch_response ( ) <EOL> self . assertEquals ( r . data , r2 . data ) <EOL> def test_vary_miss ( self ) : <EOL> with a . test_request_context ( '<STR_LIT>' , headers = ( ( '<STR_LIT>' , '<STR_LIT>' ) , ) ) : <EOL> r = Response ( '<STR_LIT:foo>' ) <EOL> r . vary . add ( '<STR_LIT>' ) <EOL> r . content_encoding = '<STR_LIT>' <EOL> self . s . cache_response ( r ) <EOL> with self . assertRaises ( NoMatchingRepresentation ) : <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> self . r . fetch_response ( ) <EOL> def test_invalidation_condition ( self ) : <EOL> with a . test_request_context ( '<STR_LIT>' , method = "<STR_LIT>" ) : <EOL> r = Response ( '<STR_LIT:foo>' ) <EOL> self . assertTrue ( self . s . should_invalidate_resource ( r ) ) <EOL> r = Response ( '<STR_LIT:foo>' , status = <NUM_LIT> ) <EOL> self . assertFalse ( self . s . should_invalidate_resource ( r ) ) <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> r = Response ( '<STR_LIT:foo>' ) <EOL> self . assertFalse ( self . s . should_invalidate_resource ( r ) ) <EOL> def test_invalidation ( self ) : <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> r = Response ( '<STR_LIT:foo>' ) <EOL> self . s . cache_response ( r ) <EOL> self . assertEquals ( len ( self . c . _cache ) , <NUM_LIT:2> ) <EOL> with a . test_request_context ( '<STR_LIT>' , method = "<STR_LIT>" ) : <EOL> r = Response ( '<STR_LIT:foo>' ) <EOL> self . assertTrue ( self . s . should_invalidate_resource ( r ) ) <EOL> self . s . invalidate_resource ( ) <EOL> self . assertEquals ( len ( self . c . _cache ) , <NUM_LIT:1> ) <EOL> with self . assertRaises ( CacheMiss ) : <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> self . r . fetch_response ( ) <EOL> def test_master_salt_invalidation ( self ) : <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> r = Response ( '<STR_LIT:foo>' ) <EOL> self . s . cache_response ( r ) <EOL> self . assertEquals ( self . r . fetch_response ( ) . data , b'<STR_LIT:foo>' ) <EOL> self . r . config . master_salt = '<STR_LIT>' <EOL> with self . assertRaises ( NoMatchingRepresentation ) : <EOL> self . r . fetch_response ( ) <EOL> def test_request_cache_controls ( self ) : <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> self . assertTrue ( self . r . should_fetch_response ( ) ) <EOL> with a . test_request_context ( '<STR_LIT>' , method = '<STR_LIT>' ) : <EOL> self . assertTrue ( self . r . should_fetch_response ( ) ) <EOL> with a . test_request_context ( '<STR_LIT>' , method = '<STR_LIT:POST>' ) : <EOL> self . assertFalse ( self . r . should_fetch_response ( ) ) <EOL> with a . test_request_context ( '<STR_LIT>' , headers = ( ( '<STR_LIT>' , '<STR_LIT>' ) , ) ) : <EOL> self . assertFalse ( self . r . should_fetch_response ( ) ) <EOL> with a . test_request_context ( '<STR_LIT>' , headers = ( ( '<STR_LIT>' , '<STR_LIT>' ) , ) ) : <EOL> self . assertFalse ( self . r . should_fetch_response ( ) ) <EOL> with a . test_request_context ( '<STR_LIT>' , headers = ( ( '<STR_LIT>' , '<STR_LIT>' ) , ) ) : <EOL> self . assertFalse ( self . r . should_fetch_response ( ) ) <EOL> with a . test_request_context ( '<STR_LIT>' , headers = ( ( '<STR_LIT>' , '<STR_LIT>' ) , ) ) : <EOL> self . assertTrue ( self . r . should_fetch_response ( ) ) <EOL> def test_response_freshness_seconds ( self ) : <EOL> r = Response ( ) <EOL> self . assertEquals ( <NUM_LIT:0> , self . r . response_freshness_seconds ( r ) ) <EOL> r . date = datetime . utcnow ( ) <EOL> self . assertTrue ( compare_numbers ( self . s . DEFAULT_EXPIRATION_SECONDS , <EOL> self . r . response_freshness_seconds ( r ) , <EOL> <NUM_LIT:1> ) ) <EOL> r . expires = datetime . utcnow ( ) + timedelta ( seconds = <NUM_LIT> ) <EOL> self . assertTrue ( compare_numbers ( <NUM_LIT> , self . r . response_freshness_seconds ( r ) , <NUM_LIT:1> ) ) <EOL> r . cache_control . max_age = <NUM_LIT> <EOL> self . assertTrue ( compare_numbers ( <NUM_LIT> , self . r . response_freshness_seconds ( r ) , <NUM_LIT:1> ) ) <EOL> def test_min_fresh ( self ) : <EOL> r = Response ( ) <EOL> r . date = datetime . utcnow ( ) - timedelta ( seconds = <NUM_LIT:100> ) <EOL> r . cache_control . max_age = <NUM_LIT:200> <EOL> f = self . r . response_freshness_seconds ( r ) <EOL> with a . test_request_context ( '<STR_LIT>' , headers = ( ( '<STR_LIT>' , '<STR_LIT>' ) , ) ) : <EOL> try : <EOL> self . r . verify_response_freshness_or_miss ( r , f ) <EOL> except CacheMiss : <EOL> self . fail ( '<STR_LIT>' ) <EOL> with a . test_request_context ( '<STR_LIT>' , headers = ( ( '<STR_LIT>' , '<STR_LIT>' ) , ) ) : <EOL> self . assertRaises ( NotFreshEnoughForClient , self . r . verify_response_freshness_or_miss , r , f ) <EOL> def test_request_cache_control_disobedience ( self ) : <EOL> c = SimpleCache ( ) <EOL> cfg = Config ( request_controls_cache = False ) <EOL> s = Store ( c , cfg ) <EOL> r = Retrieval ( c , cfg ) <EOL> with a . test_request_context ( '<STR_LIT>' , headers = ( ( '<STR_LIT>' , '<STR_LIT>' ) , ) ) : <EOL> self . assertTrue ( r . should_fetch_response ( ) ) <EOL> with a . test_request_context ( '<STR_LIT>' , headers = ( ( '<STR_LIT>' , '<STR_LIT>' ) , ) ) : <EOL> self . assertTrue ( s . should_cache_response ( Response ( ) ) ) <EOL> with a . test_request_context ( '<STR_LIT>' , headers = ( ( '<STR_LIT>' , '<STR_LIT>' ) , ) ) : <EOL> self . assertTrue ( s . should_cache_response ( Response ( ) ) ) <EOL> resp = Response ( ) <EOL> resp . date = datetime . utcnow ( ) - timedelta ( seconds = <NUM_LIT:100> ) <EOL> resp . cache_control . max_age = <NUM_LIT:200> <EOL> with a . test_request_context ( '<STR_LIT>' , headers = ( ( '<STR_LIT>' , '<STR_LIT>' ) , ) ) : <EOL> f = self . r . response_freshness_seconds ( resp ) <EOL> try : <EOL> r . verify_response_freshness_or_miss ( resp , f ) <EOL> except CacheMiss : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_sequence_converted_responses ( self ) : <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> r = Response ( f for f in '<STR_LIT:foo>' ) <EOL> r . make_sequence ( ) <EOL> self . assertFalse ( self . s . should_cache_response ( r ) ) <EOL> r = send_file ( __file__ ) <EOL> r . make_sequence ( ) <EOL> self . assertFalse ( self . s . should_cache_response ( r ) ) <EOL> class RecacheTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . recached = False <EOL> def dispatcher ( salt ) : <EOL> self . recached = True <EOL> self . c = SimpleCache ( ) <EOL> cfg = Config ( preemptive_recache_seconds = <NUM_LIT:10> , preemptive_recache_callback = dispatcher ) <EOL> self . s = Store ( self . c , cfg ) <EOL> self . r = Retrieval ( self . c , cfg ) <EOL> def test_preemptive_recaching_predicate ( self ) : <EOL> m = Metadata ( HeaderSet ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) ) , '<STR_LIT>' ) <EOL> def mkretr ( ** kwargs ) : <EOL> return Retrieval ( self . c , Config ( ** kwargs ) ) <EOL> with a . test_request_context ( '<STR_LIT:/>' ) : <EOL> self . assertFalse ( mkretr ( preemptive_recache_seconds = <NUM_LIT:10> ) . should_recache_preemptively ( <NUM_LIT:10> , m ) ) <EOL> self . assertFalse ( mkretr ( preemptive_recache_callback = lambda x : <NUM_LIT:0> ) . should_recache_preemptively ( <NUM_LIT:10> , m ) ) <EOL> self . assertFalse ( self . r . should_recache_preemptively ( <NUM_LIT:11> , m ) ) <EOL> self . assertTrue ( self . r . should_recache_preemptively ( <NUM_LIT:10> , m ) ) <EOL> self . assertFalse ( self . r . should_recache_preemptively ( <NUM_LIT:10> , m ) ) <EOL> self . c . clear ( ) <EOL> self . assertTrue ( self . r . should_recache_preemptively ( <NUM_LIT:10> , m ) ) <EOL> def test_preemptive_recaching_cache_bypass ( self ) : <EOL> fresh = Response ( '<STR_LIT:foo>' ) <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> self . s . cache_response ( fresh ) <EOL> metadata = self . r . fetch_metadata ( ) <EOL> with a . test_request_context ( '<STR_LIT>' ) : <EOL> cached = self . r . fetch_response ( ) <EOL> self . assertEquals ( cached . headers [ self . r . X_CACHE_HEADER ] , '<STR_LIT>' ) <EOL> with a . test_request_context ( '<STR_LIT>' , headers = { RECACHE_HEADER : metadata . salt } ) : <EOL> self . assertRaises ( RecacheRequested , self . r . fetch_response ) <EOL> with a . test_request_context ( '<STR_LIT>' , headers = { RECACHE_HEADER : '<STR_LIT>' } ) : <EOL> try : <EOL> self . r . fetch_response ( ) <EOL> except RecacheRequested : <EOL> self . fail ( '<STR_LIT>' ) <EOL> class UtilityTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . c = SimpleCache ( ) <EOL> def test_werkzeug_cache_get_or_add_missing_key ( self ) : <EOL> self . assertEquals ( '<STR_LIT:bar>' , werkzeug_cache_get_or_add ( self . c , '<STR_LIT:foo>' , '<STR_LIT:bar>' , <NUM_LIT:10> ) ) <EOL> def test_werkzeug_cache_get_or_add_existing_key ( self ) : <EOL> self . c . set ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> self . assertEquals ( '<STR_LIT:bar>' , werkzeug_cache_get_or_add ( self . c , '<STR_LIT:foo>' , '<STR_LIT>' , <NUM_LIT:10> ) ) </s>
