• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1import email.utils
2import httplib2
3import pytest
4import re
5import tests
6import time
7
8dummy_url = "http://127.0.0.1:1"
9
10
11def 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
23def 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
34def 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")
49def 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
68def 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
97def 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
136def 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
153def 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
178def 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
199def 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
216def 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
233def 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
258def 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
281def 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"))
297def 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
313def 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
328def 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)
361def test_parse_cache_control(data):
362    header, expected = data
363    assert httplib2._parse_cache_control(header) == expected
364
365
366def 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)
413def 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
430def 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
441def 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