/
conftest.py
103 lines (75 loc) · 2.71 KB
/
conftest.py
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
import itertools
import os
import unittest
import pytest
import vcr
from google.oauth2.credentials import Credentials as UserCredentials
from google.oauth2.service_account import Credentials as ServiceAccountCredentials
import gspread
from gspread.exceptions import APIError
CREDS_FILENAME = os.getenv("GS_CREDS_FILENAME")
RECORD_MODE = os.getenv("GS_RECORD_MODE", "none")
SCOPE = [
"https://spreadsheets.google.com/feeds",
"https://www.googleapis.com/auth/drive.file",
]
DUMMY_ACCESS_TOKEN = "<ACCESS_TOKEN>"
I18N_STR = "Iñtërnâtiônàlizætiøn" # .encode('utf8')
def read_credentials(filename):
return ServiceAccountCredentials.from_service_account_file(filename, scopes=SCOPE)
def prefixed_counter(prefix, start=1):
c = itertools.count(start)
for value in c:
yield "{} {}".format(prefix, value)
def get_method_name(self_id):
return self_id.split(".")[-1]
@pytest.fixture(scope="module")
def vcr_config():
return {
"match_on": ["uri", "method"], # match each query using the uri and the method
"decode_compressed_response": True, # decode requests to save clear content
"record_mode": RECORD_MODE,
"serializer": "json",
"path_transformer": vcr.VCR.ensure_suffix(".json"),
"ignore_hosts": [
"oauth2.googleapis.com", # skip oauth requests, in replay mode we don't use them
],
"filter_headers": [
("authorization", DUMMY_ACCESS_TOKEN)
], # hide token from the recording
}
class SleepyClient(gspread.Client):
HTTP_TOO_MANY_REQUESTS = 429
DEFAULT_SLEEP_SECONDS = 1
def request(self, *args, **kwargs):
try:
return super().request(*args, **kwargs)
except APIError as err:
data = err.response.json()
if data["error"]["code"] == self.HTTP_TOO_MANY_REQUESTS:
import time
time.sleep(self.DEFAULT_SLEEP_SECONDS)
return self.request(*args, **kwargs)
else:
raise err
class DummyCredentials(UserCredentials):
pass
class GspreadTest(unittest.TestCase):
@classmethod
def get_temporary_spreadsheet_title(cls):
return "Test %s" % cls.__name__
@classmethod
def get_cassette_name(cls):
return cls.__name__
def _sequence_generator(self):
return prefixed_counter(get_method_name(self.id()))
@pytest.fixture(scope="module")
def client():
if CREDS_FILENAME:
auth_credentials = read_credentials(CREDS_FILENAME)
else:
auth_credentials = DummyCredentials(DUMMY_ACCESS_TOKEN)
gc = SleepyClient(auth_credentials)
if not isinstance(gc, gspread.client.Client) is True:
raise AssertionError
return gc