tests/test_host_network_capability.py

_Response · class · L31-L46 — class _Response(io.BytesIO)
init · method · L34-L37 — def init(self, body: bytes, *, url: str = "https://example.test/x", ctype="")
geturl · method · L39-L40 — def geturl(self) -> str
enter · method · L42-L43 — def enter(self)
exit · method · L45-L46 — def exit(self, *_exc) -> None
_network_on · function · L50-L71 — def _network_on(monkeypatch)
_offline_dns · function · L54-L65 — def _offline_dns(host, port, _args, *_kwargs)
_stub_urlopen · function · L74-L101 — def _stub_urlopen(monkeypatch, response_factory, recorder=None)
_fake · function · L91-L94 — def _fake(request, timeout=None): # noqa: ANN001
_Opener · class · L96-L98 — class _Opener
open · method · L97-L98 — def open(self, request, timeout=None): # noqa: ANN001
test_a_response_is_bounded_while_it_is_read_not_after · function · L109-L129 — def test_a_response_is_bounded_while_it_is_read_not_after(monkeypatch)
_Endless · class · L119-L122 — class _Endless(io.RawIOBase)
read · method · L120-L122 — def read(self, size=-1): # noqa: ANN001
test_the_cap_is_the_callers_and_a_body_under_it_is_returned_whole · function · L132-L139 — def test_the_cap_is_the_callers_and_a_body_under_it_is_returned_whole(monkeypatch)
test_head_asks_for_no_body_and_does_not_pretend_to_have_one · function · L147-L176 — def test_head_asks_for_no_body_and_does_not_pretend_to_have_one(monkeypatch)
_Opener · class · L159-L164 — class _Opener
open · method · L160-L164 — def open(self, request, timeout=None): # noqa: ANN001
test_an_unsupported_method_is_refused_rather_than_coerced · function · L179-L185 — def test_an_unsupported_method_is_refused_rather_than_coerced(monkeypatch)
test_a_caller_supplied_user_agent_actually_reaches_the_request · function · L188-L203 — def test_a_caller_supplied_user_agent_actually_reaches_the_request(monkeypatch)
_fake_ip_dns · function · L211-L224 — def _fake_ip_dns(host, port, _args, *_kwargs)
test_fake_ip_dns_needs_the_narrow_opt_in · function · L227-L232 — def test_fake_ip_dns_needs_the_narrow_opt_in(monkeypatch)
test_fake_ip_dns_allows_only_catalogued_hostnames · function · L235-L252 — def test_fake_ip_dns_allows_only_catalogued_hostnames(monkeypatch)
test_the_ssrf_guard_applies_to_every_method · function · L256-L264 — def test_the_ssrf_guard_applies_to_every_method(monkeypatch, method)
test_the_guard_reads_the_host_the_client_will_actually_dial · function · L267-L285 — def test_the_guard_reads_the_host_the_client_will_actually_dial(monkeypatch)
test_the_download_path_goes_through_the_same_guard · function · L288-L293 — def test_the_download_path_goes_through_the_same_guard(monkeypatch, tmp_path)
_workspace · function · L301-L311 — def _workspace(root: Path) -> WorkspaceFileService
test_a_download_that_escapes_the_workspace_never_makes_the_request · function · L314-L334 — def test_a_download_that_escapes_the_workspace_never_makes_the_request( monkeypatch, tmp_path )
test_a_download_reports_a_workspace_relative_path_and_its_digest · function · L337-L355 — def test_a_download_reports_a_workspace_relative_path_and_its_digest( monkeypatch, tmp_path )
test_download_publishes_through_acquired_parent_after_directory_swap · function · L358-L389 — def test_download_publishes_through_acquired_parent_after_directory_swap( monkeypatch, tmp_path )
acquire_then_swap · function · L373-L377 — def acquire_then_swap(relative, *, create_parents=False)
test_a_download_streams_instead_of_collecting_the_response · function · L392-L412 — def test_a_download_streams_instead_of_collecting_the_response(monkeypatch, tmp_path)
test_the_streaming_copier_applies_backpressure_between_reads · function · L415-L447 — def test_the_streaming_copier_applies_backpressure_between_reads()
_BackpressureReader · class · L418-L431 — class _BackpressureReader
init · method · L419-L421 — def init(self)
read · method · L423-L431 — def read(self, size=-1): # noqa: ANN001
_BackpressureWriter · class · L435-L439 — class _BackpressureWriter(io.BytesIO)
write · method · L436-L439 — def write(self, data): # noqa: ANN001
test_a_download_binds_publication_to_the_streamed_inode · function · L450-L473 — def test_a_download_binds_publication_to_the_streamed_inode(monkeypatch, tmp_path)
swap_before_link · function · L459-L461 — def swap_before_link(source, target, **kwargs)
test_a_download_rehashes_the_staged_inode_before_replacing_destination · function · L476-L499 — def test_a_download_rehashes_the_staged_inode_before_replacing_destination( monkeypatch, tmp_path )
mutate_after_link · function · L485-L487 — def mutate_after_link(source, target, **kwargs)
test_a_download_falls_back_when_the_filesystem_has_no_hardlinks · function · L502-L523 — def test_a_download_falls_back_when_the_filesystem_has_no_hardlinks( monkeypatch, tmp_path )
unsupported_link · function · L510-L511 — def unsupported_link(_args, *_kwargs)
test_a_post_publish_interrupt_does_not_unlink_the_public_path · function · L526-L553 — def test_a_post_publish_interrupt_does_not_unlink_the_public_path( monkeypatch, tmp_path )
publish_then_interrupt · function · L535-L538 — def publish_then_interrupt(source, target)
test_a_download_streams_through_the_optional_requests_transport · function · L556-L599 — def test_a_download_streams_through_the_optional_requests_transport( monkeypatch, tmp_path )
_RequestsResponse · class · L562-L574 — class _RequestsResponse
raise_for_status · method · L570-L571 — def raise_for_status()
close · method · L573-L574 — def close(self)
_Requests · class · L576-L581 — class _Requests
request · method · L578-L581 — def request(_args, *kwargs)
test_a_cancelled_download_preserves_the_previous_file · function · L602-L622 — def test_a_cancelled_download_preserves_the_previous_file(monkeypatch, tmp_path)
_Interrupted · class · L603-L612 — class _Interrupted(_Response)
init · method · L604-L606 — def init(self)
read · method · L608-L612 — def read(self, size=-1): # noqa: ANN001
test_requests_http_error_preserves_the_previous_file · function · L625-L656 — def test_requests_http_error_preserves_the_previous_file(monkeypatch, tmp_path)
_ErrorResponse · class · L628-L640 — class _ErrorResponse
raise_for_status · method · L636-L637 — def raise_for_status()
close · method · L639-L640 — def close(self)
_Requests · class · L642-L645 — class _Requests
request · method · L644-L645 — def request(_args, *_kwargs)
test_stdlib_http_error_preserves_the_previous_file · function · L659-L673 — def test_stdlib_http_error_preserves_the_previous_file(monkeypatch, tmp_path)
missing · function · L662-L663 — def missing(request)
test_an_oversized_download_preserves_the_previous_file · function · L676-L690 — def test_an_oversized_download_preserves_the_previous_file(monkeypatch, tmp_path)
test_an_oversized_download_is_a_soft_error_not_a_crash · function · L693-L711 — def test_an_oversized_download_is_a_soft_error_not_a_crash(monkeypatch, tmp_path)
_Big · class · L697-L702 — class _Big(_Response)
init · method · L698-L699 — def init(self)
read · method · L701-L702 — def read(self, size=-1): # noqa: ANN001
test_the_download_tool_screens_its_output_like_every_network_tool · function · L719-L730 — def test_the_download_tool_screens_its_output_like_every_network_tool()
test_a_symlink_planted_inside_the_workspace_does_not_widen_it · function · L733-L757 — def test_a_symlink_planted_inside_the_workspace_does_not_widen_it( monkeypatch, tmp_path )
test_a_probe_is_guarded_like_any_other_request · function · L760-L779 — def test_a_probe_is_guarded_like_any_other_request(monkeypatch)
test_an_unreachable_probe_is_an_answer_not_an_exception · function · L782-L796 — def test_an_unreachable_probe_is_an_answer_not_an_exception(monkeypatch)
_Dead · class · L789-L791 — class _Dead
open · method · L790-L791 — def open(self, request, timeout=None): # noqa: ANN001
test_a_404_probe_reports_absence_rather_than_failing · function · L799-L809 — def test_a_404_probe_reports_absence_rather_than_failing(monkeypatch)
_Missing · class · L803-L805 — class _Missing
open · method · L804-L805 — def open(self, request, timeout=None): # noqa: ANN001