Skip to content

Commit a3536b2

Browse files
pyrookapyrooka
committed
chore: revamp redirection tests
Signed-off-by: Norbert Biczo <[email protected]>
1 parent 9bee88c commit a3536b2

File tree

1 file changed

+117
-217
lines changed

1 file changed

+117
-217
lines changed

‎test/test_base_service.py

+117-217
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import ssl
77
import tempfile
88
import time
9+
from collections import namedtuple
910
from shutil import copyfile
1011
from typing import Optional
1112
from urllib3.exceptions import ConnectTimeoutError, MaxRetryError
@@ -788,230 +789,129 @@ def test_retry_config_external():
788789
assert retry_err.value.reason == error
789790

790791

791-
@responses.activate
792-
def test_redirect_ibm_to_ibm():
793-
url_from = 'https://region1.cloud.ibm.com/'
794-
url_to = 'https://region2.cloud.ibm.com/'
795-
796-
safe_headers = {
797-
'Authorization': 'foo',
798-
'WWW-Authenticate': 'bar',
799-
'Cookie': 'baz',
800-
'Cookie2': 'baz2',
801-
}
802-
803-
responses.add(
804-
responses.GET, url_from, status=302, adding_headers={'Location': url_to}, body='just about to redirect'
805-
)
806-
responses.add(responses.GET, url_to, status=200, body='successfully redirected')
807-
808-
service = BaseService(service_url=url_from, authenticator=NoAuthAuthenticator())
809-
810-
prepped = service.prepare_request('GET', '', headers=safe_headers)
811-
response = service.send(prepped)
812-
result = response.get_result()
813-
814-
assert result.status_code == 200
815-
assert result.url == url_to
816-
assert result.text == 'successfully redirected'
817-
818-
# Make sure the headers are included in the 2nd, redirected request.
819-
redirected_headers = responses.calls[1].request.headers
820-
for key in safe_headers:
821-
assert key in redirected_headers
822-
823-
824-
@responses.activate
825-
def test_redirect_not_ibm_to_ibm():
826-
url_from = 'https://region1.notcloud.ibm.com/'
827-
url_to = 'https://region2.cloud.ibm.com/'
828-
829-
safe_headers = {
830-
'Authorization': 'foo',
831-
'WWW-Authenticate': 'bar',
832-
'Cookie': 'baz',
833-
'Cookie2': 'baz2',
834-
}
835-
836-
responses.add(
837-
responses.GET, url_from, status=302, adding_headers={'Location': url_to}, body='just about to redirect'
838-
)
839-
responses.add(responses.GET, url_to, status=200, body='successfully redirected')
840-
841-
service = BaseService(service_url=url_from, authenticator=NoAuthAuthenticator())
842-
843-
prepped = service.prepare_request('GET', '', headers=safe_headers)
844-
response = service.send(prepped)
845-
result = response.get_result()
846-
847-
assert result.status_code == 200
848-
assert result.url == url_to
849-
assert result.text == 'successfully redirected'
850-
851-
# Make sure the headers have been excluded from the 2nd, redirected request.
852-
redirected_headers = responses.calls[1].request.headers
853-
for key in safe_headers:
854-
assert key not in redirected_headers
855-
856-
857-
@responses.activate
858-
def test_redirect_ibm_to_not_ibm():
859-
url_from = 'https://region1.cloud.ibm.com/'
860-
url_to = 'https://region2.notcloud.ibm.com/'
861-
862-
safe_headers = {
863-
'Authorization': 'foo',
864-
'WWW-Authenticate': 'bar',
865-
'Cookie': 'baz',
866-
'Cookie2': 'baz2',
867-
}
868-
869-
responses.add(
870-
responses.GET, url_from, status=302, adding_headers={'Location': url_to}, body='just about to redirect'
871-
)
872-
responses.add(responses.GET, url_to, status=200, body='successfully redirected')
873-
874-
service = BaseService(service_url=url_from, authenticator=NoAuthAuthenticator())
875-
876-
prepped = service.prepare_request('GET', '', headers=safe_headers)
877-
response = service.send(prepped)
878-
result = response.get_result()
879-
880-
assert result.status_code == 200
881-
assert result.url == url_to
882-
assert result.text == 'successfully redirected'
883-
884-
# Make sure the headers have been excluded from the 2nd, redirected request.
885-
redirected_headers = responses.calls[1].request.headers
886-
for key in safe_headers:
887-
assert key not in redirected_headers
888-
889-
890-
@responses.activate
891-
def test_redirect_not_ibm_to_not_ibm():
892-
url_from = 'https://region1.notcloud.ibm.com/'
893-
url_to = 'https://region2.notcloud.ibm.com/'
894-
895-
safe_headers = {
896-
'Authorization': 'foo',
897-
'WWW-Authenticate': 'bar',
898-
'Cookie': 'baz',
899-
'Cookie2': 'baz2',
900-
}
901-
902-
responses.add(
903-
responses.GET, url_from, status=302, adding_headers={'Location': url_to}, body='just about to redirect'
904-
)
905-
responses.add(responses.GET, url_to, status=200, body='successfully redirected')
906-
907-
service = BaseService(service_url=url_from, authenticator=NoAuthAuthenticator())
908-
909-
prepped = service.prepare_request('GET', '', headers=safe_headers)
910-
response = service.send(prepped)
911-
result = response.get_result()
912-
913-
assert result.status_code == 200
914-
assert result.url == url_to
915-
assert result.text == 'successfully redirected'
916-
917-
# Make sure the headers have been excluded from the 2nd, redirected request.
918-
redirected_headers = responses.calls[1].request.headers
919-
for key in safe_headers:
920-
assert key not in redirected_headers
921-
922-
923-
@responses.activate
924-
def test_redirect_ibm_same_host():
925-
url_from = 'https://region1.cloud.ibm.com/'
926-
url_to = 'https://region1.cloud.ibm.com/'
927-
928-
safe_headers = {
929-
'Authorization': 'foo',
930-
'WWW-Authenticate': 'bar',
931-
'Cookie': 'baz',
932-
'Cookie2': 'baz2',
933-
}
934-
935-
responses.add(
936-
responses.GET, url_from, status=302, adding_headers={'Location': url_to}, body='just about to redirect'
937-
)
938-
responses.add(responses.GET, url_to, status=200, body='successfully redirected')
939-
940-
service = BaseService(service_url=url_from, authenticator=NoAuthAuthenticator())
941-
942-
prepped = service.prepare_request('GET', '', headers=safe_headers)
943-
response = service.send(prepped)
944-
result = response.get_result()
945-
946-
assert result.status_code == 200
947-
assert result.url == url_to
948-
assert result.text == 'successfully redirected'
949-
950-
# Make sure the headers have been excluded from the 2nd, redirected request.
951-
redirected_headers = responses.calls[1].request.headers
952-
for key in safe_headers:
953-
assert key in redirected_headers
954-
955-
956-
@responses.activate
957-
def test_redirect_not_ibm_same_host():
958-
url_from = 'https://region1.notcloud.ibm.com/'
959-
url_to = 'https://region1.notcloud.ibm.com/'
960-
792+
class TestRedirect:
961793
safe_headers = {
962794
'Authorization': 'foo',
963795
'WWW-Authenticate': 'bar',
964796
'Cookie': 'baz',
965797
'Cookie2': 'baz2',
966798
}
967799

968-
responses.add(
969-
responses.GET, url_from, status=302, adding_headers={'Location': url_to}, body='just about to redirect'
970-
)
971-
responses.add(responses.GET, url_to, status=200, body='successfully redirected')
972-
973-
service = BaseService(service_url=url_from, authenticator=NoAuthAuthenticator())
974-
975-
prepped = service.prepare_request('GET', '', headers=safe_headers)
976-
response = service.send(prepped)
977-
result = response.get_result()
978-
979-
assert result.status_code == 200
980-
assert result.url == url_to
981-
assert result.text == 'successfully redirected'
982-
983-
# Make sure the headers have been excluded from the 2nd, redirected request.
984-
redirected_headers = responses.calls[1].request.headers
985-
for key in safe_headers:
986-
assert key in redirected_headers
987-
988-
989-
@responses.activate
990-
def test_redirect_ibm_to_ibm_exhausted():
991-
redirects = 11
992-
safe_headers = {
993-
'Authorization': 'foo',
994-
'WWW-Authenticate': 'bar',
995-
'Cookie': 'baz',
996-
'Cookie2': 'baz2',
997-
}
998-
999-
for i in range(redirects):
1000-
responses.add(
1001-
responses.GET,
1002-
f'https://region{i+1}.cloud.ibm.com/',
1003-
status=302,
1004-
adding_headers={'Location': f'https://region{i+2}.cloud.ibm.com/'},
1005-
body='just about to redirect',
1006-
)
1007-
1008-
service = BaseService(service_url='https://region1.cloud.ibm.com/', authenticator=NoAuthAuthenticator())
1009-
1010-
with pytest.raises(MaxRetryError) as ex:
1011-
prepped = service.prepare_request('GET', '', headers=safe_headers)
1012-
service.send(prepped)
1013-
1014-
assert ex.value.reason == 'reached the maximum number of redirects: 10'
800+
url_cloud_1 = 'https://region1.cloud.ibm.com'
801+
url_cloud_2 = 'https://region2.cloud.ibm.com'
802+
url_notcloud_1 = 'https://region1.notcloud.ibm.com'
803+
url_notcloud_2 = 'https://region2.notcloud.ibm.com'
804+
805+
def run_test(self, url_from_base: str, url_to_base: str, safe_headers_included: bool):
806+
paths = [
807+
# 1. port, 1. path, 2. port with path
808+
['', '/', '/'],
809+
[':3000', '/', '/'],
810+
[':3000', '/', ':3333/'],
811+
['', '/a/very/long/path/with?some=query&params=the_end', '/'],
812+
[':3000', '/a/very/long/path/with?some=query&params=the_end', '/'],
813+
[':3000', '/a/very/long/path/with?some=query&params=the_end', '/api/v1'],
814+
[':3000', '/a/very/long/path/with?some=query&params=the_end', ':3000/api/v1'],
815+
]
816+
817+
# Different test cases to make sure different status codes handled correctly.
818+
TestCase = namedtuple('TestCase', ['status_1', 'status_2', 'method_1', 'method_2', 'method_expected', 'body_returned'])
819+
test_matrix = [
820+
TestCase(301, 200, responses.GET, responses.GET, responses.GET, False),
821+
TestCase(301, 200, responses.POST, responses.GET, responses.GET, False),
822+
TestCase(302, 200, responses.GET, responses.GET, responses.GET, False),
823+
TestCase(302, 200, responses.POST, responses.GET, responses.GET, False),
824+
TestCase(303, 200, responses.GET, responses.GET, responses.GET, False),
825+
TestCase(303, 200, responses.POST, responses.GET, responses.GET, False),
826+
TestCase(307, 200, responses.GET, responses.GET, responses.GET, True),
827+
TestCase(307, 200, responses.POST, responses.POST, responses.POST, True),
828+
TestCase(308, 200, responses.GET, responses.GET, responses.GET, True),
829+
TestCase(308, 200, responses.POST, responses.POST, responses.POST, True),
830+
]
831+
832+
for path in paths:
833+
url_from = url_from_base + path[0] + path[1]
834+
url_to = url_to_base + path[2]
835+
836+
for test_case in test_matrix:
837+
# Make sure we start with a clean "env".
838+
responses.reset()
839+
840+
# Add our mock responses.
841+
responses.add(
842+
test_case.method_1, url_from, status=test_case.status_1, adding_headers={'Location': url_to}, body='just about to redirect'
843+
)
844+
responses.add(test_case.method_2, url_to, status=test_case.status_2, body='successfully redirected')
845+
846+
# Create the service, prepare the request and call it.
847+
service = BaseService(service_url=url_from_base + path[0], authenticator=NoAuthAuthenticator())
848+
prepped = service.prepare_request(test_case.method_1, path[1], headers=self.safe_headers)
849+
response = service.send(prepped)
850+
result = response.get_result()
851+
852+
853+
# Check the status code, URL, body and the method of the last request (redirected).
854+
assert result.status_code == test_case.status_2
855+
assert result.url == url_to
856+
assert result.text == 'successfully redirected'
857+
assert result.request.method == test_case.method_expected
858+
859+
# Check each headers based on the kind of the current test.
860+
redirected_request = responses.calls[1].request
861+
for key in self.safe_headers:
862+
if safe_headers_included:
863+
assert key in redirected_request.headers
864+
else:
865+
assert key not in redirected_request.headers
866+
867+
# We don't always want to see a body in the last response.
868+
if not test_case.body_returned:
869+
assert redirected_request.body is None
870+
871+
@responses.activate
872+
def test_redirect_ibm_to_ibm(self):
873+
self.run_test(self.url_cloud_1, self.url_cloud_2, True)
874+
875+
@responses.activate
876+
def test_redirect_not_ibm_to_ibm(self):
877+
self.run_test(self.url_notcloud_1, self.url_cloud_2, False)
878+
879+
@responses.activate
880+
def test_redirect_ibm_to_not_ibm(self):
881+
self.run_test(self.url_cloud_1, self.url_notcloud_2, False)
882+
883+
@responses.activate
884+
def test_redirect_not_ibm_to_not_ibm(self):
885+
self.run_test(self.url_notcloud_1, self.url_notcloud_2, False)
886+
887+
@responses.activate
888+
def test_redirect_ibm_same_host(self):
889+
self.run_test(self.url_cloud_1, self.url_cloud_1, True)
890+
891+
@responses.activate
892+
def test_redirect_not_ibm_same_host(self):
893+
self.run_test(self.url_notcloud_1, self.url_notcloud_1, True)
894+
895+
@responses.activate
896+
def test_redirect_ibm_to_ibm_exhausted(self):
897+
redirects = 11
898+
899+
for i in range(redirects):
900+
responses.add(
901+
responses.GET,
902+
f'https://region{i+1}.cloud.ibm.com/',
903+
status=302,
904+
adding_headers={'Location': f'https://region{i+2}.cloud.ibm.com/'},
905+
body='just about to redirect',
906+
)
907+
908+
service = BaseService(service_url='https://region1.cloud.ibm.com/', authenticator=NoAuthAuthenticator())
909+
910+
with pytest.raises(MaxRetryError) as ex:
911+
prepped = service.prepare_request('GET', '', headers=self.safe_headers)
912+
service.send(prepped)
913+
914+
assert ex.value.reason == 'reached the maximum number of redirects: 10'
1015915

1016916

1017917
@responses.activate

0 commit comments

Comments
 (0)