mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-09-26 11:20:31 +00:00
Fold the outbound guard's public-address patch into a named fixture
Six tests in test_network_integration.py and one in test_workflows.py each patched paperless.network.is_public_ip inline to make every resolved address count as public, so the guard's own address policy would not get in the way of whatever the test was actually checking. The line was repeated verbatim at every call site and understated what it did: read literally it makes every address public, while each docstring around it already explained the intent as loopback being treated as public. Add an every_address_is_public fixture to conftest.py, following the existing pattern for the other outbound test fixtures: a thin wrapper that imports a new allow_all_addresses helper from paperless_testing.outbound and calls it with mocker. Apply it at each call site with pytest.mark.usefixtures, dropping the mocker parameter from tests that no longer need it directly. The test that patches is_public_ip with a selective side effect to allow only one address is left untouched, since that selectivity is load-bearing for what it verifies. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
46c150eb67
commit
de8edc31c5
@@ -178,3 +178,16 @@ def dial_recorder(mocker: MockerFixture) -> DialRecorder:
|
||||
from paperless_testing.outbound import install_dial_recorder
|
||||
|
||||
return install_dial_recorder(mocker)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def every_address_is_public(mocker: MockerFixture) -> None:
|
||||
"""Disable the outbound guard's address policy: every address passes.
|
||||
|
||||
For tests that are not themselves exercising which addresses the guard
|
||||
accepts, so loopback and other private addresses dial just like a
|
||||
public one.
|
||||
"""
|
||||
from paperless_testing.outbound import allow_all_addresses
|
||||
|
||||
allow_all_addresses(mocker)
|
||||
|
||||
@@ -5162,9 +5162,9 @@ class TestWebhookSecurity:
|
||||
assert dial_recorder.hosts() == []
|
||||
|
||||
@override_settings(WEBHOOKS_ALLOW_INTERNAL_REQUESTS=False)
|
||||
@pytest.mark.usefixtures("every_address_is_public")
|
||||
def test_sends_to_validated_address(
|
||||
self,
|
||||
mocker: MockerFixture,
|
||||
local_http_server: LocalHTTPServer,
|
||||
fake_dns: FakeDNS,
|
||||
) -> None:
|
||||
@@ -5178,7 +5178,6 @@ class TestWebhookSecurity:
|
||||
- The payload arrives with the webhook hostname in the Host header
|
||||
"""
|
||||
fake_dns.add("webhook.test", "127.0.0.1")
|
||||
mocker.patch("paperless.network.is_public_ip", return_value=True)
|
||||
|
||||
send_webhook(
|
||||
url=f"http://webhook.test:{local_http_server.port}",
|
||||
|
||||
@@ -17,9 +17,9 @@ from paperless_testing.outbound import running_http_server
|
||||
|
||||
|
||||
class TestGuardedTransportSync:
|
||||
@pytest.mark.usefixtures("every_address_is_public")
|
||||
def test_pinned_connection_falls_back_to_next_address(
|
||||
self,
|
||||
mocker: MockerFixture,
|
||||
local_http_server: LocalHTTPServer,
|
||||
fake_dns: FakeDNS,
|
||||
dial_recorder: DialRecorder,
|
||||
@@ -35,7 +35,6 @@ class TestGuardedTransportSync:
|
||||
- ::1 fails, 127.0.0.1 is dialled next and the request succeeds
|
||||
"""
|
||||
fake_dns.add("dual-stack.test", "::1", "127.0.0.1")
|
||||
mocker.patch("paperless.network.is_public_ip", return_value=True)
|
||||
|
||||
with httpx.Client(
|
||||
transport=GuardedHTTPTransport(allow_internal=False),
|
||||
@@ -93,9 +92,9 @@ class TestGuardedTransportSync:
|
||||
assert local_http_server.connections == 0
|
||||
assert dial_recorder.hosts() == []
|
||||
|
||||
@pytest.mark.usefixtures("every_address_is_public")
|
||||
def test_host_header_is_the_hostname(
|
||||
self,
|
||||
mocker: MockerFixture,
|
||||
local_http_server: LocalHTTPServer,
|
||||
fake_dns: FakeDNS,
|
||||
) -> None:
|
||||
@@ -108,7 +107,6 @@ class TestGuardedTransportSync:
|
||||
- The server receives the hostname in Host, not the dialled IP
|
||||
"""
|
||||
fake_dns.add("pinned.test", "127.0.0.1")
|
||||
mocker.patch("paperless.network.is_public_ip", return_value=True)
|
||||
|
||||
with httpx.Client(
|
||||
transport=GuardedHTTPTransport(allow_internal=False),
|
||||
@@ -161,9 +159,9 @@ class TestGuardedTransportSync:
|
||||
assert dial_recorder.hosts() == ["127.0.0.1"]
|
||||
assert len(local_http_server.requests) == 1
|
||||
|
||||
@pytest.mark.usefixtures("every_address_is_public")
|
||||
def test_connections_are_not_shared_between_hosts_on_one_address(
|
||||
self,
|
||||
mocker: MockerFixture,
|
||||
local_http_server: LocalHTTPServer,
|
||||
fake_dns: FakeDNS,
|
||||
dial_recorder: DialRecorder,
|
||||
@@ -181,7 +179,6 @@ class TestGuardedTransportSync:
|
||||
"""
|
||||
fake_dns.add("first.test", "127.0.0.1")
|
||||
fake_dns.add("second.test", "127.0.0.1")
|
||||
mocker.patch("paperless.network.is_public_ip", return_value=True)
|
||||
|
||||
with httpx.Client(
|
||||
transport=GuardedHTTPTransport(allow_internal=False),
|
||||
@@ -199,6 +196,7 @@ class TestGuardedTransportSync:
|
||||
f"second.test:{local_http_server.port}",
|
||||
]
|
||||
|
||||
@pytest.mark.usefixtures("every_address_is_public")
|
||||
def test_tls_uses_the_hostname_not_the_dialled_address(
|
||||
self,
|
||||
mocker: MockerFixture,
|
||||
@@ -217,7 +215,6 @@ class TestGuardedTransportSync:
|
||||
- TLS is started with the hostname for SNI and certificate checks
|
||||
"""
|
||||
fake_dns.add("pinned.test", "127.0.0.1")
|
||||
mocker.patch("paperless.network.is_public_ip", return_value=True)
|
||||
start_tls = mocker.spy(httpcore._backends.sync.SyncStream, "start_tls")
|
||||
|
||||
with (
|
||||
@@ -269,6 +266,7 @@ class TestGuardedTransportSync:
|
||||
assert local_http_server.connections == 0
|
||||
assert dial_recorder.hosts() == []
|
||||
|
||||
@pytest.mark.usefixtures("every_address_is_public")
|
||||
def test_environment_proxy_is_not_used(
|
||||
self,
|
||||
mocker: MockerFixture,
|
||||
@@ -297,7 +295,6 @@ class TestGuardedTransportSync:
|
||||
},
|
||||
)
|
||||
fake_dns.add("origin.test", "127.0.0.1")
|
||||
mocker.patch("paperless.network.is_public_ip", return_value=True)
|
||||
|
||||
url = f"http://origin.test:{local_http_server.port}/"
|
||||
with create_guarded_httpx_client(
|
||||
@@ -323,9 +320,9 @@ class TestGuardedTransportAsync:
|
||||
return "asyncio"
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.usefixtures("every_address_is_public")
|
||||
async def test_pinned_connection_falls_back_to_next_address(
|
||||
self,
|
||||
mocker: MockerFixture,
|
||||
local_http_server: LocalHTTPServer,
|
||||
fake_dns: FakeDNS,
|
||||
dial_recorder: DialRecorder,
|
||||
@@ -341,7 +338,6 @@ class TestGuardedTransportAsync:
|
||||
- ::1 fails, 127.0.0.1 is dialled next and the request succeeds
|
||||
"""
|
||||
fake_dns.add("dual-stack.test", "::1", "127.0.0.1")
|
||||
mocker.patch("paperless.network.is_public_ip", return_value=True)
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=GuardedAsyncHTTPTransport(allow_internal=False),
|
||||
|
||||
@@ -198,6 +198,15 @@ def install_dial_recorder(mocker: MockerFixture) -> DialRecorder:
|
||||
)
|
||||
|
||||
|
||||
def allow_all_addresses(mocker: MockerFixture) -> None:
|
||||
"""Patch the guard's public-address check to accept every address.
|
||||
|
||||
Loopback and other private addresses pass just like a public one, for
|
||||
tests that exercise something other than the address policy itself.
|
||||
"""
|
||||
mocker.patch("paperless.network.is_public_ip", return_value=True)
|
||||
|
||||
|
||||
def guard_of(
|
||||
client: httpx.Client | httpx.AsyncClient,
|
||||
) -> _GuardedSyncBackend | _GuardedAsyncBackend:
|
||||
|
||||
Reference in New Issue
Block a user