| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| # |
| |
| import os, sys |
| from time import sleep |
| import system_test |
| from system_test import TestCase, Qdrouterd, Process, SkipIfNeeded |
| from subprocess import PIPE |
| |
| |
| def python_37_available(): |
| if sys.version_info >= (3, 7): |
| return True |
| |
| def curl_available(): |
| popen_args = ['curl', '--version'] |
| try: |
| process = Process(popen_args, |
| name='curl_check', |
| stdout=PIPE, |
| expect=None, |
| universal_newlines=True) |
| out = process.communicate()[0] |
| return True |
| except: |
| return False |
| |
| def quart_available(): |
| popen_args = ['quart', '--version'] |
| try: |
| process = Process(popen_args, |
| name='curl_check', |
| stdout=PIPE, |
| expect=None, |
| universal_newlines=True) |
| out = process.communicate()[0] |
| parts = out.split(".") |
| if int(parts[1]) >= 13: |
| return True |
| return False |
| except Exception as e: |
| print (e) |
| print("quart_not_available") |
| return False |
| |
| def skip_test(): |
| if python_37_available() and quart_available() and curl_available(): |
| return False |
| return True |
| |
| class Http2TestBase(TestCase): |
| def run_curl(self, args=None, regexp=None, address=None): |
| # Tell with -m / --max-time the maximum time, in seconds, that you |
| # allow the command line to spend before curl exits with a |
| # timeout error code (28). |
| local_args = ["--http2-prior-knowledge"] |
| if args: |
| local_args = args + ["--http2-prior-knowledge"] |
| |
| popen_args = ['curl', |
| str(address), |
| '--max-time', str(system_test.TIMEOUT)] + local_args |
| p = self.popen(popen_args, |
| name='curl-' + self.id(), stdout=PIPE, expect=None, |
| universal_newlines=True) |
| |
| out = p.communicate()[0] |
| assert (p.returncode == 0) |
| return out |
| |
| |
| class CommonHttp2Tests(): |
| """ |
| The tests in this class are run by both Http2TestOneRouter and |
| Http2TestTwoRouter |
| """ |
| @SkipIfNeeded(skip_test(), "Python 3.7 or greater, Quart 0.13.0 or greater and curl needed to run http2 tests") |
| # Tests the HTTP2 head request |
| def test_head_request(self): |
| # Run curl 127.0.0.1:port --http2-prior-knowledge --head |
| address = self.router_qdra.http_addresses[0] |
| out = self.run_curl(args=["--head"], address=address) |
| self.assertIn('HTTP/2 200', out) |
| self.assertIn('server: hypercorn-h2', out) |
| self.assertIn('content-type: text/html; charset=utf-8', out) |
| |
| @SkipIfNeeded(skip_test(), "Python 3.7 or greater, Quart 0.13.0 or greater and curl needed to run http2 tests") |
| def test_get_request(self): |
| # Run curl 127.0.0.1:port --http2-prior-knowledge |
| address = self.router_qdra.http_addresses[0] |
| out = self.run_curl(address=address) |
| i = 0 |
| ret_string = "" |
| while (i < 1000): |
| ret_string += str(i) + "," |
| i += 1 |
| self.assertIn(ret_string, out) |
| |
| @SkipIfNeeded(skip_test(), "Python 3.7 or greater, Quart 0.13.0 or greater and curl needed to run http2 tests") |
| def test_large_get_request(self): |
| # Tests a large get request. Response is more than 50k which means it |
| # will span many qd_http2_buffer_t objects. |
| # Run curl 127.0.0.1:port/largeget --http2-prior-knowledge |
| address = self.router_qdra.http_addresses[0] + "/largeget" |
| out = self.run_curl(address=address) |
| self.assertIn("49996,49997,49998,49999", out) |
| |
| @SkipIfNeeded(skip_test(), "Python 3.7 or greater, Quart 0.13.0 or greater and curl needed to run http2 tests") |
| def test_post_request(self): |
| # curl -d "fname=John&lname=Doe" -X POST 127.0.0.1:9000/myinfo --http2-prior-knowledge |
| address = self.router_qdra.http_addresses[0] + "/myinfo" |
| out = self.run_curl(args=['-d', 'fname=John&lname=Doe', '-X', 'POST'], address=address) |
| self.assertIn('Success! Your first name is John, last name is Doe', out) |
| |
| @SkipIfNeeded(skip_test(), "Python 3.7 or greater, Quart 0.13.0 or greater and curl needed to run http2 tests") |
| def test_delete_request(self): |
| # curl -X DELETE "http://127.0.0.1:9000/myinfo/delete/22122" -H "accept: application/json" --http2-prior-knowledge |
| address = self.router_qdra.http_addresses[0] + "/myinfo/delete/22122" |
| out = self.run_curl(args=['-X', 'DELETE'], address=address) |
| self.assertIn('{"fname": "John", "lname": "Doe", "id": "22122"}', out) |
| |
| |
| @SkipIfNeeded(skip_test(), "Python 3.7 or greater, Quart 0.13.0 or greater and curl needed to run http2 tests") |
| def test_put_request(self): |
| # curl -d "fname=John&lname=Doe" -X PUT 127.0.0.1:9000/myinfo --http2-prior-knowledge |
| address = self.router_qdra.http_addresses[0] + "/myinfo" |
| out = self.run_curl(args=['-d', 'fname=John&lname=Doe', '-X', 'PUT'], address=address) |
| self.assertIn('Success! Your first name is John, last name is Doe', out) |
| |
| @SkipIfNeeded(skip_test(), "Python 3.7 or greater, Quart 0.13.0 or greater and curl needed to run http2 tests") |
| def test_patch_request(self): |
| # curl -d "fname=John&lname=Doe" -X PATCH 127.0.0.1:9000/myinfo --http2-prior-knowledge |
| address = self.router_qdra.http_addresses[0] + "/patch" |
| out = self.run_curl(args=['--data', '{\"op\":\"add\",\"path\":\"/user\",\"value\":\"jane\"}', '-X', 'PATCH'], address=address) |
| self.assertIn('"op":"add"', out) |
| |
| @SkipIfNeeded(skip_test(), "Python 3.7 or greater, Quart 0.13.0 or greater and curl needed to run http2 tests") |
| def test_404(self): |
| # Run curl 127.0.0.1:port/unavilable --http2-prior-knowledge |
| address = self.router_qdra.http_addresses[0] + "/unavilable" |
| out = self.run_curl(address=address) |
| self.assertIn('404 Not Found', out) |
| |
| @SkipIfNeeded(skip_test(), "Python 3.7 or greater, Quart 0.13.0 or greater and curl needed to run http2 tests") |
| def test_500(self): |
| # Run curl 127.0.0.1:port/unavilable --http2-prior-knowledge |
| address = self.router_qdra.http_addresses[0] + "/test/500" |
| out = self.run_curl(address=address) |
| self.assertIn('500 Internal Server Error', out) |
| |
| @SkipIfNeeded(skip_test(), "Python 3.7 or greater, Quart 0.13.0 or greater and curl needed to run http2 tests") |
| def test_get_image_png(self): |
| # Run curl 127.0.0.1:port --http2-prior-knowledge |
| passed = False |
| try: |
| address = self.router_qdra.http_addresses[0] + "/images/balanced-routing.png" |
| self.run_curl(address=address) |
| except UnicodeDecodeError as u: |
| if "codec can't decode byte 0x89" in str(u): |
| passed = True |
| self.assertTrue(passed) |
| |
| @SkipIfNeeded(skip_test(), "Python 3.7 or greater, Quart 0.13.0 or greater and curl needed to run http2 tests") |
| def test_get_image_jpg(self): |
| # Run curl 127.0.0.1:port --http2-prior-knowledge |
| passed = False |
| try: |
| address = self.router_qdra.http_addresses[0] + "/images/apache.jpg" |
| self.run_curl(address=address) |
| except UnicodeDecodeError as u: |
| print (u) |
| if "codec can't decode byte 0xff" in str(u): |
| passed = True |
| self.assertTrue(passed) |
| |
| |
| class Http2TestOneRouter(Http2TestBase, CommonHttp2Tests): |
| @classmethod |
| def setUpClass(cls): |
| super(Http2TestOneRouter, cls).setUpClass() |
| if skip_test(): |
| return |
| cls.http2_server_name = "http2_server" |
| os.environ["QUART_APP"] = "http2server:app" |
| os.environ['SERVER_LISTEN_PORT'] = str(cls.tester.get_port()) |
| cls.http2_server = cls.tester.http2server(name=cls.http2_server_name, |
| listen_port=int(os.getenv('SERVER_LISTEN_PORT')), |
| py_string='python3', |
| server_file="http2_server.py") |
| name = "http2-test-router" |
| config = Qdrouterd.Config([ |
| ('router', {'mode': 'standalone', 'id': 'QDR'}), |
| ('listener', {'port': cls.tester.get_port(), 'role': 'normal', 'host': '0.0.0.0'}), |
| |
| ('httpListener', {'port': cls.tester.get_port(), 'address': 'examples', |
| 'host': '127.0.0.1', 'protocolVersion': 'HTTP2'}), |
| ('httpConnector', |
| {'port': os.getenv('SERVER_LISTEN_PORT'), 'address': 'examples', |
| 'host': '127.0.0.1', 'protocolVersion': 'HTTP2'}) |
| ]) |
| cls.router_qdra = cls.tester.qdrouterd(name, config, wait=True) |
| |
| |
| |
| class Http2TestTwoRouter(Http2TestBase, CommonHttp2Tests): |
| @classmethod |
| def setUpClass(cls): |
| super(Http2TestTwoRouter, cls).setUpClass() |
| if skip_test(): |
| return |
| cls.http2_server_name = "http2_server" |
| os.environ["QUART_APP"] = "http2server:app" |
| os.environ['SERVER_LISTEN_PORT'] = str(cls.tester.get_port()) |
| cls.http2_server = cls.tester.http2server(name=cls.http2_server_name, |
| listen_port=int(os.getenv('SERVER_LISTEN_PORT')), |
| py_string='python3', |
| server_file="http2_server.py") |
| name = "http2-test-router" |
| inter_router_port = cls.tester.get_port() |
| |
| config_qdra = Qdrouterd.Config([ |
| ('router', {'mode': 'interior', 'id': 'QDR.A'}), |
| ('listener', {'port': cls.tester.get_port(), 'role': 'normal', 'host': '0.0.0.0'}), |
| ('httpListener', {'port': cls.tester.get_port(), 'address': 'examples', |
| 'host': '127.0.0.1', 'protocolVersion': 'HTTP2'}), |
| ('listener', {'role': 'inter-router', 'port': inter_router_port}) |
| ]) |
| |
| config_qdrb = Qdrouterd.Config([ |
| ('router', {'mode': 'interior', 'id': 'QDR.B'}), |
| ('listener', {'port': cls.tester.get_port(), 'role': 'normal', 'host': '0.0.0.0'}), |
| ('httpConnector', |
| {'port': os.getenv('SERVER_LISTEN_PORT'), 'address': 'examples', |
| 'host': '127.0.0.1', 'protocolVersion': 'HTTP2'}), |
| ('connector', {'name': 'connectorToA', 'role': 'inter-router', |
| 'port': inter_router_port, |
| 'verifyHostname': 'no'}) |
| |
| ]) |
| |
| cls.router_qdra = cls.tester.qdrouterd(name, config_qdra, wait=True) |
| cls.router_qdrb = cls.tester.qdrouterd(name, config_qdrb, wait=True) |
| |
| cls.router_qdra.wait_router_connected('QDR.B') |
| cls.router_qdrb.wait_router_connected('QDR.A') |
| |
| sleep(2) |
