Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_requestor_class(self):
class CustomRequestor(Requestor):
pass
reddit = Reddit(
client_id="dummy",
client_secret="dummy",
password="dummy",
user_agent="dummy",
username="dummy",
requestor_class=CustomRequestor,
)
assert isinstance(reddit._core._requestor, CustomRequestor)
assert not isinstance(self.reddit._core._requestor, CustomRequestor)
reddit = Reddit(
client_id="dummy",
client_secret="dummy",
def test_initialize(self):
requestor = prawcore.Requestor("prawcore:test (by /u/bboe)")
self.assertEqual(
"prawcore:test (by /u/bboe) prawcore/{}".format(
prawcore.__version__
),
requestor._http.headers["User-Agent"],
)
def test_request__read_timeout_retry(self, mock_session):
session_instance = mock_session.return_value
# Handle Auth
response_dict = {"access_token": "", "expires_in": 99, "scope": ""}
session_instance.request.return_value = Mock(
headers={}, json=lambda: response_dict, status_code=200
)
requestor = prawcore.Requestor("prawcore:test (by /u/bboe)")
authorizer = readonly_authorizer(requestor=requestor)
session_instance.request.reset_mock()
# Fail on subsequent request
exception = ReadTimeout()
session_instance.request.side_effect = exception
expected = (
"prawcore",
"WARNING",
"Retrying due to ReadTimeout() status: GET "
"https://oauth.reddit.com/",
)
with LogCapture(level=logging.WARNING) as log_capture:
with self.assertRaises(RequestException) as context_manager:
def test_request__chunked_encoding_retry(self, mock_session):
session_instance = mock_session.return_value
# Handle Auth
response_dict = {"access_token": "", "expires_in": 99, "scope": ""}
session_instance.request.return_value = Mock(
headers={}, json=lambda: response_dict, status_code=200
)
requestor = prawcore.Requestor("prawcore:test (by /u/bboe)")
authorizer = readonly_authorizer(requestor=requestor)
session_instance.request.reset_mock()
# Fail on subsequent request
exception = ChunkedEncodingError()
session_instance.request.side_effect = exception
expected = (
"prawcore",
"WARNING",
"Retrying due to ChunkedEncodingError() status: GET "
"https://oauth.reddit.com/",
)
with LogCapture(level=logging.WARNING) as log_capture:
with self.assertRaises(RequestException) as context_manager:
def test_request__use_custom_session(self):
override = "REQUEST OVERRIDDEN"
custom_header = "CUSTOM SESSION HEADER"
headers = {"session_header": custom_header}
attrs = {"request.return_value": override, "headers": headers}
session = Mock(**attrs)
requestor = prawcore.Requestor(
"prawcore:test (by /u/bboe)", session=session
)
self.assertEqual(
"prawcore:test (by /u/bboe) prawcore/{}".format(
prawcore.__version__
),
requestor._http.headers["User-Agent"],
)
self.assertEqual(
requestor._http.headers["session_header"], custom_header
)
self.assertEqual(requestor.request("https://reddit.com"), override)
def main():
"""Provide the program's entry point when directly executed."""
if len(sys.argv) < 2:
print("Usage: {} SCOPE...".format(sys.argv[0]))
return 1
authenticator = prawcore.TrustedAuthenticator(
prawcore.Requestor("prawcore_refresh_token_example"),
os.environ["PRAWCORE_CLIENT_ID"],
os.environ["PRAWCORE_CLIENT_SECRET"],
os.environ["PRAWCORE_REDIRECT_URI"],
)
state = str(random.randint(0, 65000))
url = authenticator.authorize_url("permanent", sys.argv[1:], state)
print(url)
client = receive_connection()
data = client.recv(1024).decode("utf-8")
param_tokens = data.split(" ", 2)[1].split("?", 1)[1].split("&")
params = {
key: value
for (key, value) in [token.split("=") for token in param_tokens]
}
def main():
"""Provide the program's entry point when directly executed."""
authenticator = prawcore.TrustedAuthenticator(
prawcore.Requestor("prawcore_script_auth_example"),
os.environ["PRAWCORE_CLIENT_ID"],
os.environ["PRAWCORE_CLIENT_SECRET"],
)
authorizer = prawcore.ScriptAuthorizer(
authenticator,
os.environ["PRAWCORE_USERNAME"],
os.environ["PRAWCORE_PASSWORD"],
)
authorizer.refresh()
with prawcore.session(authorizer) as session:
data = session.request("GET", "/api/v1/me/friends")
for friend in data["data"]["children"]:
print(friend["name"])
def main():
"""Provide the program's entry point when directly executed."""
if len(sys.argv) != 2:
print("Usage: {} USERNAME".format(sys.argv[0]))
return 1
caching_requestor = prawcore.Requestor(
"prawcore_device_id_auth_example", session=CachingSession()
)
authenticator = prawcore.TrustedAuthenticator(
caching_requestor,
os.environ["PRAWCORE_CLIENT_ID"],
os.environ["PRAWCORE_CLIENT_SECRET"],
)
authorizer = prawcore.ReadOnlyAuthorizer(authenticator)
authorizer.refresh()
user = sys.argv[1]
with prawcore.session(authorizer) as session:
data1 = session.request("GET", "/api/v1/user/{}/trophies".format(user))
with prawcore.session(authorizer) as session:
data2 = session.request("GET", "/api/v1/user/{}/trophies".format(user))