Skip to content

Commit 56242fe

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

File tree

1 file changed

+122
-216
lines changed

1 file changed

+122
-216
lines changed

‎test/test_base_service.py

+122-216
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,135 @@ 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-
792+
class TestRedirect:
862793
safe_headers = {
863794
'Authorization': 'foo',
864795
'WWW-Authenticate': 'bar',
865796
'Cookie': 'baz',
866797
'Cookie2': 'baz2',
867798
}
868799

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-
961-
safe_headers = {
962-
'Authorization': 'foo',
963-
'WWW-Authenticate': 'bar',
964-
'Cookie': 'baz',
965-
'Cookie2': 'baz2',
966-
}
967-
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',
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+
# pylint: disable=too-many-locals
806+
def run_test(self, url_from_base: str, url_to_base: str, safe_headers_included: bool):
807+
paths = [
808+
# 1. port, 1. path, 2. port with path
809+
['', '/', '/'],
810+
[':3000', '/', '/'],
811+
[':3000', '/', ':3333/'],
812+
['', '/a/very/long/path/with?some=query&params=the_end', '/'],
813+
[':3000', '/a/very/long/path/with?some=query&params=the_end', '/'],
814+
[':3000', '/a/very/long/path/with?some=query&params=the_end', '/api/v1'],
815+
[':3000', '/a/very/long/path/with?some=query&params=the_end', ':3000/api/v1'],
816+
]
817+
818+
# Different test cases to make sure different status codes handled correctly.
819+
TestCase = namedtuple(
820+
'TestCase', ['status_1', 'status_2', 'method_1', 'method_2', 'method_expected', 'body_returned']
1006821
)
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'
822+
test_matrix = [
823+
TestCase(301, 200, responses.GET, responses.GET, responses.GET, False),
824+
TestCase(301, 200, responses.POST, responses.GET, responses.GET, False),
825+
TestCase(302, 200, responses.GET, responses.GET, responses.GET, False),
826+
TestCase(302, 200, responses.POST, responses.GET, responses.GET, False),
827+
TestCase(303, 200, responses.GET, responses.GET, responses.GET, False),
828+
TestCase(303, 200, responses.POST, responses.GET, responses.GET, False),
829+
TestCase(307, 200, responses.GET, responses.GET, responses.GET, True),
830+
TestCase(307, 200, responses.POST, responses.POST, responses.POST, True),
831+
TestCase(308, 200, responses.GET, responses.GET, responses.GET, True),
832+
TestCase(308, 200, responses.POST, responses.POST, responses.POST, True),
833+
]
834+
835+
for path in paths:
836+
url_from = url_from_base + path[0] + path[1]
837+
url_to = url_to_base + path[2]
838+
839+
for test_case in test_matrix:
840+
# Make sure we start with a clean "env".
841+
responses.reset()
842+
843+
# Add our mock responses.
844+
responses.add(
845+
test_case.method_1,
846+
url_from,
847+
status=test_case.status_1,
848+
adding_headers={'Location': url_to},
849+
body='just about to redirect',
850+
)
851+
responses.add(test_case.method_2, url_to, status=test_case.status_2, body='successfully redirected')
852+
853+
# Create the service, prepare the request and call it.
854+
service = BaseService(service_url=url_from_base + path[0], authenticator=NoAuthAuthenticator())
855+
prepped = service.prepare_request(test_case.method_1, path[1], headers=self.safe_headers)
856+
response = service.send(prepped)
857+
result = response.get_result()
858+
859+
# Check the status code, URL, body and the method of the last request (redirected).
860+
assert result.status_code == test_case.status_2
861+
assert result.url == url_to
862+
assert result.text == 'successfully redirected'
863+
assert result.request.method == test_case.method_expected
864+
865+
# Check each headers based on the kind of the current test.
866+
redirected_request = responses.calls[1].request
867+
for key in self.safe_headers:
868+
if safe_headers_included:
869+
assert key in redirected_request.headers
870+
else:
871+
assert key not in redirected_request.headers
872+
873+
# We don't always want to see a body in the last response.
874+
if not test_case.body_returned:
875+
assert redirected_request.body is None
876+
877+
@responses.activate
878+
def test_redirect_ibm_to_ibm(self):
879+
self.run_test(self.url_cloud_1, self.url_cloud_2, True)
880+
881+
@responses.activate
882+
def test_redirect_not_ibm_to_ibm(self):
883+
self.run_test(self.url_notcloud_1, self.url_cloud_2, False)
884+
885+
@responses.activate
886+
def test_redirect_ibm_to_not_ibm(self):
887+
self.run_test(self.url_cloud_1, self.url_notcloud_2, False)
888+
889+
@responses.activate
890+
def test_redirect_not_ibm_to_not_ibm(self):
891+
self.run_test(self.url_notcloud_1, self.url_notcloud_2, False)
892+
893+
@responses.activate
894+
def test_redirect_ibm_same_host(self):
895+
self.run_test(self.url_cloud_1, self.url_cloud_1, True)
896+
897+
@responses.activate
898+
def test_redirect_not_ibm_same_host(self):
899+
self.run_test(self.url_notcloud_1, self.url_notcloud_1, True)
900+
901+
@responses.activate
902+
def test_redirect_ibm_to_ibm_exhausted(self):
903+
redirects = 11
904+
905+
for i in range(redirects):
906+
responses.add(
907+
responses.GET,
908+
f'https://region{i+1}.cloud.ibm.com/',
909+
status=302,
910+
adding_headers={'Location': f'https://region{i+2}.cloud.ibm.com/'},
911+
body='just about to redirect',
912+
)
913+
914+
service = BaseService(service_url='https://region1.cloud.ibm.com/', authenticator=NoAuthAuthenticator())
915+
916+
with pytest.raises(MaxRetryError) as ex:
917+
prepped = service.prepare_request('GET', '', headers=self.safe_headers)
918+
service.send(prepped)
919+
920+
assert ex.value.reason == 'reached the maximum number of redirects: 10'
1015921

1016922

1017923
@responses.activate

0 commit comments

Comments
 (0)