{}
+'''.format(val) + return bytes(response, 'utf-8') + + +class Data: + SMALL_SIZE = 1024 * 10 # 10 KB + LARGE_SIZE = (1024 ** 2) * 10 # 10 MB + + STR_SMALL = generate_random_string(SMALL_SIZE) + STR_LARGE = generate_random_string(LARGE_SIZE) + + EXTRA_SMALL = generate_random_string(1024 * 15) + EXTRA_LARGE = generate_random_string((1024 ** 2) * 15) + + HTML_SMALL = make_html_body(STR_SMALL) + HTML_LARGE = make_html_body(STR_LARGE) + + JSON_SMALL = {'data': STR_SMALL} + JSON_LARGE = {'data': STR_LARGE} + + +class LeafResource(Resource): + isLeaf = True + + +class GetDataHtmlSmall(LeafResource): + def render_GET(self, request: TxRequest): + request.setHeader('Content-Type', 'text/html; charset=UTF-8') + return Data.HTML_SMALL + + +class GetDataHtmlLarge(LeafResource): + def render_GET(self, request: TxRequest): + request.setHeader('Content-Type', 'text/html; charset=UTF-8') + return Data.HTML_LARGE + + +class PostDataJsonMixin: + @staticmethod + def make_response(request: TxRequest, extra_data: str): + response = { + 'request-headers': {}, + 'request-body': json.loads(request.content.read()), + 'extra-data': extra_data + } + for k, v in request.requestHeaders.getAllRawHeaders(): + response['request-headers'][k.decode('utf-8')] = v[0].decode('utf-8') + + response_bytes = bytes(json.dumps(response), 'utf-8') + request.setHeader('Content-Type', 'application/json') + return response_bytes + + +class PostDataJsonSmall(LeafResource, PostDataJsonMixin): + def render_POST(self, request: TxRequest): + return self.make_response(request, Data.EXTRA_SMALL) + + +class PostDataJsonLarge(LeafResource, PostDataJsonMixin): + def render_POST(self, request: TxRequest): + return self.make_response(request, Data.EXTRA_LARGE) + + +def get_client_certificate(key_file, certificate_file): + with open(key_file, 'r') as key, open(certificate_file, 'r') as certificate: + pem = ''.join(key.readlines()) + ''.join(certificate.readlines()) + + return PrivateCertificate.loadPEM(pem) + + +class Https2ClientProtocolTestCase(TestCase): scheme = 'https' + key_file = os.path.join(os.path.dirname(__file__), 'keys', 'localhost.key') + certificate_file = os.path.join(os.path.dirname(__file__), 'keys', 'localhost.crt') + def _init_resource(self): + self.temp_directory = self.mktemp() + os.mkdir(self.temp_directory) + r = File(self.temp_directory) + r.putChild(b'get-data-html-small', GetDataHtmlSmall()) + r.putChild(b'get-data-html-large', GetDataHtmlLarge()) + + r.putChild(b'post-data-json-small', PostDataJsonSmall()) + r.putChild(b'post-data-json-large', PostDataJsonLarge()) + return r + + @inlineCallbacks def setUp(self): + # Initialize resource tree + root = self._init_resource() + self.site = Site(root, timeout=None) + # Start server for testing - self.mockserver = MockServer() - self.mockserver.__enter__() - + self.hostname = u'localhost' if self.scheme == 'https': - self.url = urlparse(self.mockserver.https_address) + context_factory = ssl_context_factory(self.key_file, self.certificate_file) + server_endpoint = SSL4ServerEndpoint(reactor, 0, context_factory, interface=self.hostname) else: - self.url = urlparse(self.mockserver.http_address) + server_endpoint = TCP4ServerEndpoint(reactor, 0, interface=self.hostname) + self.server = yield server_endpoint.listen(self.site) + self.port_number = self.server.getHost().port - self.protocol = H2ClientProtocol() - - # Connect to the server using the custom HTTP2ClientProtocol - options = CertificateOptions(acceptableProtocols=[b'h2']) - endpoint = SSL4ClientEndpoint(reactor, self.url.hostname, self.url.port, options) - connectProtocol(endpoint, self.protocol) - - def getURL(self, path): - return "{}://{}:{}/{}".format(self.url.scheme, self.url.hostname, self.url.port, path) + # Connect H2 client with server + client_certificate = get_client_certificate(self.key_file, self.certificate_file) + client_options = optionsForClientTLS( + hostname=self.hostname, + trustRoot=client_certificate, + acceptableProtocols=[b'h2'] + ) + h2_client_factory = Factory.forProtocol(H2ClientProtocol) + client_endpoint = SSL4ClientEndpoint(reactor, self.hostname, self.port_number, client_options) + self.client = yield client_endpoint.connect(h2_client_factory) + @inlineCallbacks def tearDown(self): - self.mockserver.__exit__(None, None, None) + yield self.client.transport.loseConnection() + yield self.client.transport.abortConnection() + yield self.server.stopListening() + shutil.rmtree(self.temp_directory) - def test_download(self): - request = Request(self.getURL('')) + def get_url(self, path): + """ + :param path: Should have / at the starting compulsorily if not empty + :return: Complete url + """ + assert len(path) > 0 and (path[0] == '/' or path[0] == '&') + return "{}://{}:{}{}".format(self.scheme, self.hostname, self.port_number, path) - def assert_response(response: Response): - self.assertEqual(response.body, b'Scrapy mock HTTP server\n') - self.assertEqual(response.status, 200) + @staticmethod + def _check_repeat(get_deferred, count): + d_list = [] + for _ in range(count): + d = get_deferred() + d_list.append(d) + + return DeferredList(d_list, fireOnOneErrback=True) + + def _check_GET( + self, + request: Request, + expected_body, + expected_status + ): + def check_response(response: Response): + self.assertEqual(response.status, expected_status) + self.assertEqual(response.body, expected_body) self.assertEqual(response.request, request) self.assertEqual(response.url, request.url) - d = self.protocol.request(request) + content_length = int(response.headers.get('Content-Length')) + self.assertEqual(len(response.body), content_length) + + d = self.client.request(request) + d.addCallback(check_response) + d.addErrback(self.fail) + return d + + def test_GET_small_body(self): + request = Request(self.get_url('/get-data-html-small')) + return self._check_GET(request, Data.HTML_SMALL, 200) + + def test_GET_large_body(self): + request = Request(self.get_url('/get-data-html-large')) + return self._check_GET(request, Data.HTML_LARGE, 200) + + def _check_GET_x20(self, *args, **kwargs): + def get_deferred(): + return self._check_GET(*args, **kwargs) + + return self._check_repeat(get_deferred, 20) + + def test_GET_small_body_x20(self): + return self._check_GET_x20( + Request(self.get_url('/get-data-html-small')), + Data.HTML_SMALL, + 200 + ) + + def test_GET_large_body_x20(self): + return self._check_GET_x20( + Request(self.get_url('/get-data-html-large')), + Data.HTML_LARGE, + 200 + ) + + def _check_POST_json( + self, + request: Request, + expected_request_body, + expected_extra_data, + expected_status: int + ): + d = self.client.request(request) + + def assert_response(response: Response): + self.assertEqual(response.status, expected_status) + self.assertEqual(response.request, request) + self.assertEqual(response.url, request.url) + + content_length = int(response.headers.get('Content-Length')) + self.assertEqual(len(response.body), content_length) + + # Parse the body + body = json.loads(response.body.decode('utf-8')) + self.assertIn('request-body', body) + self.assertIn('extra-data', body) + self.assertIn('request-headers', body) + + request_body = body['request-body'] + self.assertEqual(request_body, expected_request_body) + + extra_data = body['extra-data'] + self.assertEqual(extra_data, expected_extra_data) + + # Check if headers were sent successfully + request_headers = body['request-headers'] + for k, v in request.headers.items(): + k_str = k.decode('utf-8') + self.assertIn(k_str, request_headers) + self.assertEqual(request_headers[k_str], v[0].decode('utf-8')) + d.addCallback(assert_response) return d + + def test_POST_small_json(self): + request = JsonRequest(url=self.get_url('/post-data-json-small'), method='POST', data=Data.JSON_SMALL) + return self._check_POST_json( + request, + Data.JSON_SMALL, + Data.EXTRA_SMALL, + 200 + ) + + def test_POST_large_json(self): + request = JsonRequest(url=self.get_url('/post-data-json-large'), method='POST', data=Data.JSON_LARGE) + return self._check_POST_json( + request, + Data.JSON_LARGE, + Data.EXTRA_LARGE, + 200 + ) + + def _check_POST_json_x20(self, *args, **kwargs): + def get_deferred(): + return self._check_POST_json(*args, **kwargs) + + return self._check_repeat(get_deferred, 20) + + def test_POST_small_json_x20(self): + request = JsonRequest(url=self.get_url('/post-data-json-small'), method='POST', data=Data.JSON_SMALL) + return self._check_POST_json_x20( + request, + Data.JSON_SMALL, + Data.EXTRA_SMALL, + 200 + ) + + def test_POST_large_json_x20(self): + request = JsonRequest(url=self.get_url('/post-data-json-large'), method='POST', data=Data.JSON_LARGE) + return self._check_POST_json_x20( + request, + Data.JSON_LARGE, + Data.EXTRA_LARGE, + 200 + ) + + def test_cancel_request(self): + request = Request(url=self.get_url('/get-data-html-large')) + + def assert_response(response: Response): + self.assertEqual(response.status, 499) + self.assertEqual(response.request, request) + self.assertEqual(response.url, request.url) + + d = self.client.request(request) + d.addCallback(assert_response) + d.cancel() + + return d