1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
# stdlib
import json
import logging
import re
import sys
import time
from datetime import datetime
# 3rd party
import pytest
import requests
from pytest_httpserver import HTTPServer
from werkzeug import Request, Response
# this package
from apeye.rate_limiter import HTTPCache, rate_limit
if sys.version_info < (3, 7):
# 3rd party
from backports.datetime_fromisoformat import datetime_fromisoformat
else:
datetime_fromisoformat = datetime.fromisoformat
logging.basicConfig()
@rate_limit(3)
def rate_limited_function():
print("Inside function")
return 42
def test_rate_limit(capsys, caplog):
assert rate_limited_function() == 42
assert rate_limited_function() == 42
assert rate_limited_function() == 42
assert capsys.readouterr().out.splitlines() == ["Inside function"] * 3
caplog.set_level(logging.DEBUG)
assert rate_limited_function() == 42
assert rate_limited_function() == 42
time.sleep(2)
assert rate_limited_function() == 42
assert capsys.readouterr().out.splitlines() == ["Inside function"] * 3
last_ran_re = re.compile(r"rate_limited_function: Last ran (\d+(\.\d+)?|\d+e-\d+) seconds ago\.")
print(caplog.record_tuples)
assert last_ran_re.match(caplog.record_tuples[0][2])
assert re.match(r"rate_limited_function: Waiting (\d+(\.\d+)?|\d+e-\d+) seconds\.", caplog.record_tuples[1][2])
assert last_ran_re.match(caplog.record_tuples[2][2])
assert last_ran_re.match(caplog.record_tuples[3][2])
@pytest.fixture(scope="session")
def testing_http_cache():
cache = HTTPCache("testing_apeye_http")
assert cache.clear()
yield cache
cache.clear()
@pytest.fixture()
def timeserver(httpserver: HTTPServer):
def time_handler(request: Request):
time.sleep(1)
now = datetime.utcnow()
headers = {
"Cache-Control": "max-age=0, private, must-revalidate",
"Date": now.strftime("%a, %d %B %Y %H:%M:%S GMT"),
}
response_json = {
"abbreviation": "GMT",
"client_ip": "127.0.0.1",
"datetime": now.strftime("%Y-%m-%dT%H:%M:%S%z"),
"day_of_week": now.strftime("%w"),
"day_of_year": now.strftime("%j"),
}
return Response(json.dumps(response_json, indent=4), 200, headers, None, None)
httpserver.expect_request("/time", method="GET").respond_with_handler(time_handler)
return httpserver
def test_cache_canary(timeserver: HTTPServer):
# Proves that worldtimeapi.org returns a different time for two sequential requests.
session = requests.session()
# target_url = "http://worldtimeapi.org/api/ip"
target_url = timeserver.url_for("/time")
response = session.get(target_url)
assert response.status_code == 200
original_time = datetime_fromisoformat(response.json()["datetime"])
response = session.get(target_url)
assert response.status_code == 200
current_time = datetime_fromisoformat(response.json()["datetime"])
assert current_time > original_time
@pytest.mark.parametrize("run_number", [1, 2])
def test_http_cache(testing_http_cache, capsys, run_number: int, timeserver: HTTPServer):
session = testing_http_cache.session
# target_url = "http://worldtimeapi.org/api/ip"
target_url = timeserver.url_for("/time")
response = session.get(target_url)
assert response.status_code == 200
assert not response.from_cache
original_time = datetime_fromisoformat(response.json()["datetime"])
response = session.get(target_url)
assert response.status_code == 200
assert response.from_cache
current_time = datetime_fromisoformat(response.json()["datetime"])
# If the times have changed the cache has failed.
assert current_time == original_time
assert testing_http_cache.cache_dir.is_dir()
assert testing_http_cache.clear()
assert not testing_http_cache.cache_dir.is_dir()
# make a new request
response = session.get(target_url)
assert response.status_code == 200
assert not response.from_cache
current_time = datetime_fromisoformat(response.json()["datetime"])
assert current_time > original_time
assert testing_http_cache.clear()
|
