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
|
"""Test for prawcore.requestor.Requestor class."""
import pickle
import unittest
from inspect import signature
from mock import Mock, patch
import prawcore
from prawcore import RequestException
class RequestorTest(unittest.TestCase):
def test_initialize(self):
requestor = prawcore.Requestor("prawcore:test (by /u/bboe)")
self.assertEqual(
f"prawcore:test (by /u/bboe) prawcore/{prawcore.__version__}",
requestor._http.headers["User-Agent"],
)
def test_initialize__failures(self):
for agent in [None, "shorty"]:
self.assertRaises(
prawcore.InvalidInvocation, prawcore.Requestor, agent
)
@patch("requests.Session")
def test_request__wrap_request_exceptions(self, mock_session):
exception = Exception("prawcore wrap_request_exceptions")
session_instance = mock_session.return_value
session_instance.request.side_effect = exception
requestor = prawcore.Requestor("prawcore:test (by /u/bboe)")
with self.assertRaises(prawcore.RequestException) as context_manager:
requestor.request("get", "http://a.b", data="bar")
self.assertIsInstance(context_manager.exception, RequestException)
self.assertIs(exception, context_manager.exception.original_exception)
self.assertEqual(
("get", "http://a.b"), context_manager.exception.request_args
)
self.assertEqual(
{"data": "bar"}, context_manager.exception.request_kwargs
)
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(
f"prawcore:test (by /u/bboe) prawcore/{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 test_request__session_timeout_default(self):
requestor = prawcore.Requestor("prawcore:test (by /u/bboe)")
requestor_signature = signature(requestor._http.request)
self.assertEqual(
str(requestor_signature.parameters["timeout"]), "timeout=None"
)
def test_pickle(self):
requestor = prawcore.Requestor("prawcore:test (by /u/bboe)")
for protocol in range(pickle.HIGHEST_PROTOCOL + 1):
pickle.loads(pickle.dumps(requestor, protocol=protocol))
|