Replace mock-based proxy redirect tests with integration tests

Use mock_http_server instead of patching httpx.AsyncClient for the
three redirect tests. Add redirect_to_self parameter to
MockHttpServer.get_url() to support self-referencing redirect loops.
This commit is contained in:
mvdbeek
2026-03-30 12:02:02 +02:00
parent 8a5a3b1c76
commit fe2a78db5d
2 changed files with 34 additions and 76 deletions
+28 -75
View File
@@ -1,9 +1,3 @@
from unittest.mock import (
AsyncMock,
MagicMock,
patch,
)
import pytest
import requests
@@ -137,82 +131,41 @@ class TestProxyApi(ApiTestCase):
# Verify content-encoding header was properly filtered out (no double decompression)
assert "content-encoding" not in response.headers
@patch("galaxy.webapps.galaxy.api.proxy.httpx.AsyncClient")
def test_proxy_validates_redirects(self, mock_client_class):
"""Test that redirects are validated."""
# Create mock responses - redirect to a local file (invalid scheme)
redirect_response = MagicMock()
redirect_response.status_code = 302
redirect_response.headers = {"location": "file://internal-server/secret-files"}
redirect_response.aclose = AsyncMock()
# Setup mock client
mock_client = MagicMock()
mock_client.build_request = MagicMock(return_value=MagicMock())
mock_client.send = AsyncMock(return_value=redirect_response)
mock_client.aclose = AsyncMock()
mock_client_class.return_value = mock_client
# Attempt to proxy a URL that redirects to file:// (should be blocked)
response = self._get("proxy?url=https://evil.com/redirect")
# Should fail with 400 Bad Request due to invalid redirect URL scheme
def test_proxy_validates_redirects(self, mock_http_server):
"""Test that redirects to invalid schemes are blocked."""
url = mock_http_server.get_url(
remote_url="https://evil.com/redirect",
status=302,
response_headers={"Location": "file://internal-server/secret-files"},
)
response = self._get(f"proxy?url={url}")
self._assert_status_code_is(response, 400)
assert "Invalid URL format" in response.json()["err_msg"]
@patch("galaxy.webapps.galaxy.api.proxy.httpx.AsyncClient")
def test_proxy_follows_valid_redirects(self, mock_client_class):
def test_proxy_follows_valid_redirects(self, mock_http_server):
"""Test that valid redirects are followed after validation."""
# Create mock responses
redirect_response = MagicMock()
redirect_response.status_code = 301
redirect_response.headers = {"location": "https://example.com/final"}
redirect_response.aclose = AsyncMock()
final_response = MagicMock()
final_response.status_code = 200
final_response.headers = {"content-type": "text/plain"}
final_response.aclose = AsyncMock()
# Create async generator for streaming
async def mock_stream():
yield b"test content"
final_response.aiter_bytes = mock_stream
# Setup mock client to return redirect first, then final response
mock_client = MagicMock()
mock_client.build_request = MagicMock(return_value=MagicMock())
mock_client.send = AsyncMock(side_effect=[redirect_response, final_response])
mock_client.aclose = AsyncMock()
mock_client_class.return_value = mock_client
# Proxy a URL that redirects to a valid external URL
response = self._get("proxy?url=https://example.com/redirect")
# Should succeed and follow the redirect
final_url = mock_http_server.get_url(
remote_url="https://example.com/final",
status=200,
body="test content",
content_type="text/plain",
)
redirect_url = mock_http_server.get_url(
remote_url="https://example.com/redirect",
status=301,
response_headers={"Location": final_url},
)
response = self._get(f"proxy?url={redirect_url}")
self._assert_status_code_is_ok(response)
assert b"test content" in response.content
@patch("galaxy.webapps.galaxy.api.proxy.httpx.AsyncClient")
def test_proxy_blocks_too_many_redirects(self, mock_client_class):
def test_proxy_blocks_too_many_redirects(self, mock_http_server):
"""Test that excessive redirects are blocked to prevent redirect loops."""
# Create a mock response that always redirects
redirect_response = MagicMock()
redirect_response.status_code = 302
redirect_response.headers = {"location": "https://example.com/loop"}
redirect_response.aclose = AsyncMock()
# Setup mock client
mock_client = MagicMock()
mock_client.build_request = MagicMock(return_value=MagicMock())
mock_client.send = AsyncMock(return_value=redirect_response)
mock_client.aclose = AsyncMock()
mock_client_class.return_value = mock_client
# Attempt to proxy a URL that loops redirects
response = self._get("proxy?url=https://example.com/loop")
# Should fail with 400 Bad Request due to too many redirects
url = mock_http_server.get_url(
remote_url="https://example.com/loop",
status=302,
redirect_to_self=True,
)
response = self._get(f"proxy?url={url}")
self._assert_status_code_is(response, 400)
assert "Too many redirects" in response.json()["err_msg"]
+6 -1
View File
@@ -141,6 +141,7 @@ class MockHttpServer:
request_method: str = "GET",
support_head: bool = False,
support_ranges: bool = False,
redirect_to_self: bool = False,
) -> str:
"""Register a mock endpoint and return its URL, or return remote_url with skip-if-down.
@@ -155,6 +156,7 @@ class MockHttpServer:
request_method: HTTP method to match (GET, POST, etc.).
support_head: Whether to also respond to HEAD requests.
support_ranges: Whether to support Range request headers (HTTP 206).
redirect_to_self: Whether to add a Location header pointing to this route's own URL.
"""
if self.is_remote:
if not is_site_up(remote_url):
@@ -194,7 +196,10 @@ class MockHttpServer:
support_head=support_head,
support_ranges=support_ranges,
)
return f"{self.base_url}{path}"
url = f"{self.base_url}{path}"
if redirect_to_self:
headers["Location"] = url
return url
def start_mock_http_server(host: str = "127.0.0.1", port: int = 0) -> tuple[HTTPServer, str]: