1 import email.utils 2 import httplib2 3 import pytest 4 import re 5 import tests 6 import time 7 8 dummy_url = "http://127.0.0.1:1" 9 10 11 def test_get_only_if_cached_cache_hit(): 12 # Test that can do a GET with cache and 'only-if-cached' 13 http = httplib2.Http(cache=tests.get_cache_path()) 14 with tests.server_const_http(add_etag=True) as uri: 15 http.request(uri, "GET") 16 response, content = http.request( 17 uri, "GET", headers={"cache-control": "only-if-cached"} 18 ) 19 assert response.fromcache 20 assert response.status == 200 21 22 23 def test_get_only_if_cached_cache_miss(): 24 # Test that can do a GET with no cache with 'only-if-cached' 25 http = httplib2.Http(cache=tests.get_cache_path()) 26 with tests.server_const_http(request_count=0) as uri: 27 response, content = http.request( 28 uri, "GET", headers={"cache-control": "only-if-cached"} 29 ) 30 assert not response.fromcache 31 assert response.status == 504 32 33 34 def test_get_only_if_cached_no_cache_at_all(): 35 # Test that can do a GET with no cache with 'only-if-cached' 36 # Of course, there might be an intermediary beyond us 37 # that responds to the 'only-if-cached', so this 38 # test can't really be guaranteed to pass. 39 http = httplib2.Http() 40 with tests.server_const_http(request_count=0) as uri: 41 response, content = http.request( 42 uri, "GET", headers={"cache-control": "only-if-cached"} 43 ) 44 assert not response.fromcache 45 assert response.status == 504 46 47 48 @pytest.mark.skip(reason="was commented in legacy code") 49 def test_TODO_vary_no(): 50 pass 51 # when there is no vary, a different Accept header (e.g.) should not 52 # impact if the cache is used 53 # test that the vary header is not sent 54 # uri = urllib.parse.urljoin(base, "vary/no-vary.asis") 55 # response, content = http.request(uri, 'GET', headers={'Accept': 'text/plain'}) 56 # assert response.status == 200 57 # assert 'vary' not in response 58 # 59 # response, content = http.request(uri, 'GET', headers={'Accept': 'text/plain'}) 60 # assert response.status == 200 61 # assert response.fromcache, "Should be from cache" 62 # 63 # response, content = http.request(uri, 'GET', headers={'Accept': 'text/html'}) 64 # assert response.status == 200 65 # assert response.fromcache, "Should be from cache" 66 67 68 def test_vary_header_is_sent(): 69 # Verifies RFC 2616 13.6. 70 # See https://www.w3.org/Protocols/rfc2616/rfc2616-sec13.html. 71 http = httplib2.Http(cache=tests.get_cache_path()) 72 response = tests.http_response_bytes( 73 headers={"vary": "Accept", "cache-control": "max-age=300"}, add_date=True 74 ) 75 with tests.server_const_bytes(response, request_count=3) as uri: 76 response, content = http.request(uri, "GET", headers={"accept": "text/plain"}) 77 assert response.status == 200 78 assert "vary" in response 79 80 # get the resource again, from the cache since accept header in this 81 # request is the same as the request 82 response, content = http.request(uri, "GET", headers={"Accept": "text/plain"}) 83 assert response.status == 200 84 assert response.fromcache, "Should be from cache" 85 86 # get the resource again, not from cache since Accept headers does not match 87 response, content = http.request(uri, "GET", headers={"Accept": "text/html"}) 88 assert response.status == 200 89 assert not response.fromcache, "Should not be from cache" 90 91 # get the resource again, without any Accept header, so again no match 92 response, content = http.request(uri, "GET") 93 assert response.status == 200 94 assert not response.fromcache, "Should not be from cache" 95 96 97 def test_vary_header_double(): 98 http = httplib2.Http(cache=tests.get_cache_path()) 99 response = tests.http_response_bytes( 100 headers={"vary": "Accept, Accept-Language", "cache-control": "max-age=300"}, 101 add_date=True, 102 ) 103 with tests.server_const_bytes(response, request_count=3) as uri: 104 response, content = http.request( 105 uri, 106 "GET", 107 headers={ 108 "Accept": "text/plain", 109 "Accept-Language": "da, en-gb;q=0.8, en;q=0.7", 110 }, 111 ) 112 assert response.status == 200 113 assert "vary" in response 114 115 # we are from cache 116 response, content = http.request( 117 uri, 118 "GET", 119 headers={ 120 "Accept": "text/plain", 121 "Accept-Language": "da, en-gb;q=0.8, en;q=0.7", 122 }, 123 ) 124 assert response.fromcache, "Should be from cache" 125 126 response, content = http.request(uri, "GET", headers={"Accept": "text/plain"}) 127 assert response.status == 200 128 assert not response.fromcache 129 130 # get the resource again, not from cache, varied headers don't match exact 131 response, content = http.request(uri, "GET", headers={"Accept-Language": "da"}) 132 assert response.status == 200 133 assert not response.fromcache, "Should not be from cache" 134 135 136 def test_vary_unused_header(): 137 http = httplib2.Http(cache=tests.get_cache_path()) 138 response = tests.http_response_bytes( 139 headers={"vary": "X-No-Such-Header", "cache-control": "max-age=300"}, 140 add_date=True, 141 ) 142 with tests.server_const_bytes(response, request_count=1) as uri: 143 # A header's value is not considered to vary if it's not used at all. 144 response, content = http.request(uri, "GET", headers={"Accept": "text/plain"}) 145 assert response.status == 200 146 assert "vary" in response 147 148 # we are from cache 149 response, content = http.request(uri, "GET", headers={"Accept": "text/plain"}) 150 assert response.fromcache, "Should be from cache" 151 152 153 def test_get_cache_control_no_cache(): 154 # Test Cache-Control: no-cache on requests 155 http = httplib2.Http(cache=tests.get_cache_path()) 156 with tests.server_const_http( 157 add_date=True, 158 add_etag=True, 159 headers={"cache-control": "max-age=300"}, 160 request_count=2, 161 ) as uri: 162 response, _ = http.request(uri, "GET", headers={"accept-encoding": "identity"}) 163 assert response.status == 200 164 assert response["etag"] != "" 165 assert not response.fromcache 166 response, _ = http.request(uri, "GET", headers={"accept-encoding": "identity"}) 167 assert response.status == 200 168 assert response.fromcache 169 response, _ = http.request( 170 uri, 171 "GET", 172 headers={"accept-encoding": "identity", "Cache-Control": "no-cache"}, 173 ) 174 assert response.status == 200 175 assert not response.fromcache 176 177 178 def test_get_cache_control_pragma_no_cache(): 179 # Test Pragma: no-cache on requests 180 http = httplib2.Http(cache=tests.get_cache_path()) 181 with tests.server_const_http( 182 add_date=True, 183 add_etag=True, 184 headers={"cache-control": "max-age=300"}, 185 request_count=2, 186 ) as uri: 187 response, _ = http.request(uri, "GET", headers={"accept-encoding": "identity"}) 188 assert response["etag"] != "" 189 response, _ = http.request(uri, "GET", headers={"accept-encoding": "identity"}) 190 assert response.status == 200 191 assert response.fromcache 192 response, _ = http.request( 193 uri, "GET", headers={"accept-encoding": "identity", "Pragma": "no-cache"} 194 ) 195 assert response.status == 200 196 assert not response.fromcache 197 198 199 def test_get_cache_control_no_store_request(): 200 # A no-store request means that the response should not be stored. 201 http = httplib2.Http(cache=tests.get_cache_path()) 202 with tests.server_const_http( 203 add_date=True, 204 add_etag=True, 205 headers={"cache-control": "max-age=300"}, 206 request_count=2, 207 ) as uri: 208 response, _ = http.request(uri, "GET", headers={"Cache-Control": "no-store"}) 209 assert response.status == 200 210 assert not response.fromcache 211 response, _ = http.request(uri, "GET", headers={"Cache-Control": "no-store"}) 212 assert response.status == 200 213 assert not response.fromcache 214 215 216 def test_get_cache_control_no_store_response(): 217 # A no-store response means that the response should not be stored. 218 http = httplib2.Http(cache=tests.get_cache_path()) 219 with tests.server_const_http( 220 add_date=True, 221 add_etag=True, 222 headers={"cache-control": "max-age=300, no-store"}, 223 request_count=2, 224 ) as uri: 225 response, _ = http.request(uri, "GET") 226 assert response.status == 200 227 assert not response.fromcache 228 response, _ = http.request(uri, "GET") 229 assert response.status == 200 230 assert not response.fromcache 231 232 233 def test_get_cache_control_no_cache_no_store_request(): 234 # Test that a no-store, no-cache clears the entry from the cache 235 # even if it was cached previously. 236 http = httplib2.Http(cache=tests.get_cache_path()) 237 with tests.server_const_http( 238 add_date=True, 239 add_etag=True, 240 headers={"cache-control": "max-age=300"}, 241 request_count=3, 242 ) as uri: 243 response, _ = http.request(uri, "GET") 244 response, _ = http.request(uri, "GET") 245 assert response.fromcache 246 response, _ = http.request( 247 uri, "GET", headers={"Cache-Control": "no-store, no-cache"} 248 ) 249 assert response.status == 200 250 assert not response.fromcache 251 response, _ = http.request( 252 uri, "GET", headers={"Cache-Control": "no-store, no-cache"} 253 ) 254 assert response.status == 200 255 assert not response.fromcache 256 257 258 def test_update_invalidates_cache(): 259 # Test that calling PUT or DELETE on a 260 # URI that is cache invalidates that cache. 261 http = httplib2.Http(cache=tests.get_cache_path()) 262 263 def handler(request): 264 if request.method in ("PUT", "PATCH", "DELETE"): 265 return tests.http_response_bytes(status=405) 266 return tests.http_response_bytes( 267 add_date=True, add_etag=True, headers={"cache-control": "max-age=300"} 268 ) 269 270 with tests.server_request(handler, request_count=3) as uri: 271 response, _ = http.request(uri, "GET") 272 response, _ = http.request(uri, "GET") 273 assert response.fromcache 274 response, _ = http.request(uri, "DELETE") 275 assert response.status == 405 276 assert not response.fromcache 277 response, _ = http.request(uri, "GET") 278 assert not response.fromcache 279 280 281 def handler_conditional_update(request): 282 respond = tests.http_response_bytes 283 if request.method == "GET": 284 if request.headers.get("if-none-match", "") == "12345": 285 return respond(status=304) 286 return respond( 287 add_date=True, headers={"etag": "12345", "cache-control": "max-age=300"} 288 ) 289 elif request.method in ("PUT", "PATCH", "DELETE"): 290 if request.headers.get("if-match", "") == "12345": 291 return respond(status=200) 292 return respond(status=412) 293 return respond(status=405) 294 295 296 @pytest.mark.parametrize("method", ("PUT", "PATCH")) 297 def test_update_uses_cached_etag(method): 298 # Test that we natively support http://www.w3.org/1999/04/Editing/ 299 http = httplib2.Http(cache=tests.get_cache_path()) 300 with tests.server_request(handler_conditional_update, request_count=3) as uri: 301 response, _ = http.request(uri, "GET") 302 assert response.status == 200 303 assert not response.fromcache 304 response, _ = http.request(uri, "GET") 305 assert response.status == 200 306 assert response.fromcache 307 response, _ = http.request(uri, method, body=b"foo") 308 assert response.status == 200 309 response, _ = http.request(uri, method, body=b"foo") 310 assert response.status == 412 311 312 313 def test_update_uses_cached_etag_and_oc_method(): 314 # Test that we natively support http://www.w3.org/1999/04/Editing/ 315 http = httplib2.Http(cache=tests.get_cache_path()) 316 with tests.server_request(handler_conditional_update, request_count=2) as uri: 317 response, _ = http.request(uri, "GET") 318 assert response.status == 200 319 assert not response.fromcache 320 response, _ = http.request(uri, "GET") 321 assert response.status == 200 322 assert response.fromcache 323 http.optimistic_concurrency_methods.append("DELETE") 324 response, _ = http.request(uri, "DELETE") 325 assert response.status == 200 326 327 328 def test_update_uses_cached_etag_overridden(): 329 # Test that we natively support http://www.w3.org/1999/04/Editing/ 330 http = httplib2.Http(cache=tests.get_cache_path()) 331 with tests.server_request(handler_conditional_update, request_count=2) as uri: 332 response, content = http.request(uri, "GET") 333 assert response.status == 200 334 assert not response.fromcache 335 response, content = http.request(uri, "GET") 336 assert response.status == 200 337 assert response.fromcache 338 response, content = http.request( 339 uri, "PUT", body=b"foo", headers={"if-match": "fred"} 340 ) 341 assert response.status == 412 342 343 344 @pytest.mark.parametrize( 345 "data", 346 ( 347 ({}, {}), 348 ({"cache-control": " no-cache"}, {"no-cache": 1}), 349 ( 350 {"cache-control": " no-store, max-age = 7200"}, 351 {"no-store": 1, "max-age": "7200"}, 352 ), 353 ({"cache-control": " , "}, {"": 1}), # FIXME 354 ( 355 {"cache-control": "Max-age=3600;post-check=1800,pre-check=3600"}, 356 {"max-age": "3600;post-check=1800", "pre-check": "3600"}, 357 ), 358 ), 359 ids=lambda data: str(data[0]), 360 ) 361 def test_parse_cache_control(data): 362 header, expected = data 363 assert httplib2._parse_cache_control(header) == expected 364 365 366 def test_normalize_headers(): 367 # Test that we normalize headers to lowercase 368 h = httplib2._normalize_headers({"Cache-Control": "no-cache", "Other": "Stuff"}) 369 assert "cache-control" in h 370 assert "other" in h 371 assert h["other"] == "Stuff" 372 373 374 @pytest.mark.parametrize( 375 "data", 376 ( 377 ( 378 {"cache-control": "no-cache"}, 379 {"cache-control": "max-age=7200"}, 380 "TRANSPARENT", 381 ), 382 ({}, {"cache-control": "max-age=fred, min-fresh=barney"}, "STALE"), 383 ({}, {"date": "{now}", "expires": "{now+3}"}, "FRESH"), 384 ( 385 {}, 386 {"date": "{now}", "expires": "{now+3}", "cache-control": "no-cache"}, 387 "STALE", 388 ), 389 ({"cache-control": "must-revalidate"}, {}, "STALE"), 390 ({}, {"cache-control": "must-revalidate"}, "STALE"), 391 ({}, {"date": "{now}", "cache-control": "max-age=0"}, "STALE"), 392 ({"cache-control": "only-if-cached"}, {}, "FRESH"), 393 ({}, {"date": "{now}", "expires": "0"}, "STALE"), 394 ({}, {"data": "{now+3}"}, "STALE"), 395 ( 396 {"cache-control": "max-age=0"}, 397 {"date": "{now}", "cache-control": "max-age=2"}, 398 "STALE", 399 ), 400 ( 401 {"cache-control": "min-fresh=2"}, 402 {"date": "{now}", "expires": "{now+2}"}, 403 "STALE", 404 ), 405 ( 406 {"cache-control": "min-fresh=2"}, 407 {"date": "{now}", "expires": "{now+4}"}, 408 "FRESH", 409 ), 410 ), 411 ids=lambda data: str(data), 412 ) 413 def test_entry_disposition(data): 414 now = time.time() 415 nowre = re.compile(r"{now([\+\-]\d+)?}") 416 417 def render(s): 418 m = nowre.match(s) 419 if m: 420 offset = int(m.expand(r"\1")) if m.group(1) else 0 421 s = email.utils.formatdate(now + offset, usegmt=True) 422 return s 423 424 request, response, expected = data 425 request = {k: render(v) for k, v in request.items()} 426 response = {k: render(v) for k, v in response.items()} 427 assert httplib2._entry_disposition(response, request) == expected 428 429 430 def test_expiration_model_fresh(): 431 response_headers = { 432 "date": email.utils.formatdate(usegmt=True), 433 "cache-control": "max-age=2", 434 } 435 assert httplib2._entry_disposition(response_headers, {}) == "FRESH" 436 # TODO: add current time as _entry_disposition argument to avoid sleep in tests 437 time.sleep(3) 438 assert httplib2._entry_disposition(response_headers, {}) == "STALE" 439 440 441 def test_expiration_model_date_and_expires(): 442 now = time.time() 443 response_headers = { 444 "date": email.utils.formatdate(now, usegmt=True), 445 "expires": email.utils.formatdate(now + 2, usegmt=True), 446 } 447 assert httplib2._entry_disposition(response_headers, {}) == "FRESH" 448 time.sleep(3) 449 assert httplib2._entry_disposition(response_headers, {}) == "STALE" 450 451 452 # TODO: Repeat all cache tests with memcache. pytest.mark.parametrize 453 # cache = memcache.Client(['127.0.0.1:11211'], debug=0) 454 # #cache = memcache.Client(['10.0.0.4:11211'], debug=1) 455 # http = httplib2.Http(cache) 456
