Adding upstream version 7.4.7.upstream/7.4.7upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 169 insertions, 66 deletions

diff --git a/tests/test_builders/test_build_linkcheck.py b/tests/test_builders/test_build_linkcheck.py
index c8d8515..0787661 100644
--- a/tests/test_builders/test_build_linkcheck.py
+++ b/tests/test_builders/test_build_linkcheck.py
@@ -11,6 +11,7 @@ import wsgiref.handlers
 from base64 import b64encode
 from http.server import BaseHTTPRequestHandler
 from queue import Queue
+from typing import TYPE_CHECKING
 from unittest import mock
 
 import docutils
@@ -20,6 +21,7 @@ from urllib3.poolmanager import PoolManager
 import sphinx.util.http_date
 from sphinx.builders.linkcheck import (
     CheckRequest,
+    CheckResult,
     Hyperlink,
     HyperlinkAvailabilityCheckWorker,
     RateLimit,
@@ -33,6 +35,12 @@ from tests.utils import CERT_FILE, serve_application
 
 ts_re = re.compile(r".*\[(?P<ts>.*)\].*")
 
+if TYPE_CHECKING:
+    from collections.abc import Callable, Iterable
+    from io import StringIO
+
+    from sphinx.application import Sphinx
+
 
 class DefaultsHandler(BaseHTTPRequestHandler):
     protocol_version = "HTTP/1.1"
@@ -101,7 +109,7 @@ class ConnectionMeasurement:
 
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck', freshenv=True)
-def test_defaults(app):
+def test_defaults(app: Sphinx) -> None:
     with serve_application(app, DefaultsHandler) as address:
         with ConnectionMeasurement() as m:
             app.build()
@@ -146,7 +154,7 @@ def test_defaults(app):
         'info': '',
     }
 
-    def _missing_resource(filename: str, lineno: int):
+    def _missing_resource(filename: str, lineno: int) -> dict[str, str | int]:
         return {
             'filename': 'links.rst',
             'lineno': lineno,
@@ -178,7 +186,7 @@ def test_defaults(app):
 @pytest.mark.sphinx(
     'linkcheck', testroot='linkcheck', freshenv=True,
     confoverrides={'linkcheck_anchors': False})
-def test_check_link_response_only(app):
+def test_check_link_response_only(app: Sphinx) -> None:
     with serve_application(app, DefaultsHandler) as address:
         app.build()
 
@@ -192,7 +200,7 @@ def test_check_link_response_only(app):
 
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck-too-many-retries', freshenv=True)
-def test_too_many_retries(app):
+def test_too_many_retries(app: Sphinx) -> None:
     with serve_application(app, DefaultsHandler) as address:
         app.build()
 
@@ -221,7 +229,7 @@ def test_too_many_retries(app):
 
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck-raw-node', freshenv=True)
-def test_raw_node(app):
+def test_raw_node(app: Sphinx) -> None:
     with serve_application(app, OKHandler) as address:
         # write an index file that contains a link back to this webserver's root
         # URL.  docutils will replace the raw node with the contents retrieved..
@@ -254,7 +262,7 @@ def test_raw_node(app):
 @pytest.mark.sphinx(
     'linkcheck', testroot='linkcheck-anchors-ignore', freshenv=True,
     confoverrides={'linkcheck_anchors_ignore': ["^!", "^top$"]})
-def test_anchors_ignored(app):
+def test_anchors_ignored(app: Sphinx) -> None:
     with serve_application(app, OKHandler):
         app.build()
 
@@ -266,6 +274,43 @@ def test_anchors_ignored(app):
 
 
 class AnchorsIgnoreForUrlHandler(BaseHTTPRequestHandler):
+    protocol_version = 'HTTP/1.1'
+
+    def _chunk_content(self, content: str, *, max_chunk_size: int) -> Iterable[bytes]:
+
+        def _encode_chunk(chunk: bytes) -> Iterable[bytes]:
+            """Encode a bytestring into a format suitable for HTTP chunked-transfer.
+
+            https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Transfer-Encoding
+            """
+            yield f'{len(chunk):X}'.encode('ascii')
+            yield b'\r\n'
+            yield chunk
+            yield b'\r\n'
+
+        buffer = b''
+        for char in content:
+            buffer += char.encode('utf-8')
+            if len(buffer) >= max_chunk_size:
+                chunk, buffer = buffer[:max_chunk_size], buffer[max_chunk_size:]
+                yield from _encode_chunk(chunk)
+
+        # Flush remaining bytes, if any
+        if buffer:
+            yield from _encode_chunk(buffer)
+
+        # Emit a final empty chunk to close the stream
+        yield from _encode_chunk(b'')
+
+    def _send_chunked(self, content: str) -> bool:
+        for chunk in self._chunk_content(content, max_chunk_size=20):
+            try:
+                self.wfile.write(chunk)
+            except (BrokenPipeError, ConnectionResetError) as e:
+                self.log_message(str(e))
+                return False
+        return True
+
     def do_HEAD(self):
         if self.path in {'/valid', '/ignored'}:
             self.send_response(200, "OK")
@@ -274,17 +319,24 @@ class AnchorsIgnoreForUrlHandler(BaseHTTPRequestHandler):
         self.end_headers()
 
     def do_GET(self):
-        self.do_HEAD()
         if self.path == '/valid':
-            self.wfile.write(b"<h1 id='valid-anchor'>valid anchor</h1>\n")
+            self.send_response(200, 'OK')
+            content = "<h1 id='valid-anchor'>valid anchor</h1>\n"
         elif self.path == '/ignored':
-            self.wfile.write(b"no anchor but page exists\n")
+            self.send_response(200, 'OK')
+            content = 'no anchor but page exists\n'
+        else:
+            self.send_response(404, 'Not Found')
+            content = 'not found\n'
+        self.send_header('Transfer-Encoding', 'chunked')
+        self.end_headers()
+        self._send_chunked(content)
 
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck-anchors-ignore-for-url', freshenv=True)
-def test_anchors_ignored_for_url(app):
+def test_anchors_ignored_for_url(app: Sphinx) -> None:
     with serve_application(app, AnchorsIgnoreForUrlHandler) as address:
-        app.config.linkcheck_anchors_ignore_for_url = [  # type: ignore[attr-defined]
+        app.config.linkcheck_anchors_ignore_for_url = [
             f'http://{address}/ignored',  # existing page
             f'http://{address}/invalid',  # unknown page
         ]
@@ -295,7 +347,7 @@ def test_anchors_ignored_for_url(app):
 
     attrs = ('filename', 'lineno', 'status', 'code', 'uri', 'info')
     data = [json.loads(x) for x in content.splitlines()]
-    assert len(data) == 7
+    assert len(data) == 8
     assert all(all(attr in row for attr in attrs) for row in data)
 
     # rows may be unsorted due to network latency or
@@ -304,6 +356,7 @@ def test_anchors_ignored_for_url(app):
 
     assert rows[f'http://{address}/valid']['status'] == 'working'
     assert rows[f'http://{address}/valid#valid-anchor']['status'] == 'working'
+    assert rows[f'http://{address}/valid#py:module::urllib.parse']['status'] == 'broken'
     assert rows[f'http://{address}/valid#invalid-anchor'] == {
         'status': 'broken',
         'info': "Anchor 'invalid-anchor' not found",
@@ -323,7 +376,7 @@ def test_anchors_ignored_for_url(app):
 
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver-anchor', freshenv=True)
-def test_raises_for_invalid_status(app):
+def test_raises_for_invalid_status(app: Sphinx) -> None:
     class InternalServerErrorHandler(BaseHTTPRequestHandler):
         protocol_version = "HTTP/1.1"
 
@@ -340,6 +393,50 @@ def test_raises_for_invalid_status(app):
     )
 
 
+@pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver-anchor', freshenv=True)
+def test_incomplete_html_anchor(app):
+    class IncompleteHTMLDocumentHandler(BaseHTTPRequestHandler):
+        protocol_version = 'HTTP/1.1'
+
+        def do_GET(self):
+            content = b'this is <div id="anchor">not</div> a valid HTML document'
+            self.send_response(200, 'OK')
+            self.send_header('Content-Length', str(len(content)))
+            self.end_headers()
+            self.wfile.write(content)
+
+    with serve_application(app, IncompleteHTMLDocumentHandler):
+        app.build()
+
+    content = (app.outdir / 'output.json').read_text(encoding='utf8')
+    assert len(content.splitlines()) == 1
+
+    row = json.loads(content)
+    assert row['status'] == 'working'
+
+
+@pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver-anchor', freshenv=True)
+def test_decoding_error_anchor_ignored(app):
+    class NonASCIIHandler(BaseHTTPRequestHandler):
+        protocol_version = 'HTTP/1.1'
+
+        def do_GET(self):
+            content = b'\x80\x00\x80\x00'  # non-ASCII byte-string
+            self.send_response(200, 'OK')
+            self.send_header('Content-Length', str(len(content)))
+            self.end_headers()
+            self.wfile.write(content)
+
+    with serve_application(app, NonASCIIHandler):
+        app.build()
+
+    content = (app.outdir / 'output.json').read_text(encoding='utf8')
+    assert len(content.splitlines()) == 1
+
+    row = json.loads(content)
+    assert row['status'] == 'ignored'
+
+
 def custom_handler(valid_credentials=(), success_criteria=lambda _: True):
     """
     Returns an HTTP request handler that authenticates the client and then determines
@@ -352,25 +449,27 @@ def custom_handler(valid_credentials=(), success_criteria=lambda _: True):
         expected_token = b64encode(":".join(valid_credentials).encode()).decode("utf-8")
         del valid_credentials
 
+    def authenticated(
+        method: Callable[[CustomHandler], None]
+    ) -> Callable[[CustomHandler], None]:
+        def method_if_authenticated(self):
+            if expected_token is None:
+                return method(self)
+            elif not self.headers["Authorization"]:
+                self.send_response(401, "Unauthorized")
+                self.end_headers()
+            elif self.headers["Authorization"] == f"Basic {expected_token}":
+                return method(self)
+            else:
+                self.send_response(403, "Forbidden")
+                self.send_header("Content-Length", "0")
+                self.end_headers()
+
+        return method_if_authenticated
+
     class CustomHandler(BaseHTTPRequestHandler):
         protocol_version = "HTTP/1.1"
 
-        def authenticated(method):
-            def method_if_authenticated(self):
-                if expected_token is None:
-                    return method(self)
-                elif not self.headers["Authorization"]:
-                    self.send_response(401, "Unauthorized")
-                    self.end_headers()
-                elif self.headers["Authorization"] == f"Basic {expected_token}":
-                    return method(self)
-                else:
-                    self.send_response(403, "Forbidden")
-                    self.send_header("Content-Length", "0")
-                    self.end_headers()
-
-            return method_if_authenticated
-
         @authenticated
         def do_HEAD(self):
             self.do_GET()
@@ -389,9 +488,9 @@ def custom_handler(valid_credentials=(), success_criteria=lambda _: True):
 
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver', freshenv=True)
-def test_auth_header_uses_first_match(app):
+def test_auth_header_uses_first_match(app: Sphinx) -> None:
     with serve_application(app, custom_handler(valid_credentials=("user1", "password"))) as address:
-        app.config.linkcheck_auth = [  # type: ignore[attr-defined]
+        app.config.linkcheck_auth = [
             (r'^$', ('no', 'match')),
             (fr'^http://{re.escape(address)}/$', ('user1', 'password')),
             (r'.*local.*', ('user2', 'hunter2')),
@@ -408,7 +507,7 @@ def test_auth_header_uses_first_match(app):
 @pytest.mark.sphinx(
     'linkcheck', testroot='linkcheck-localserver', freshenv=True,
     confoverrides={'linkcheck_allow_unauthorized': False})
-def test_unauthorized_broken(app):
+def test_unauthorized_broken(app: Sphinx) -> None:
     with serve_application(app, custom_handler(valid_credentials=("user1", "password"))):
         app.build()
 
@@ -422,7 +521,7 @@ def test_unauthorized_broken(app):
 @pytest.mark.sphinx(
     'linkcheck', testroot='linkcheck-localserver', freshenv=True,
     confoverrides={'linkcheck_auth': [(r'^$', ('user1', 'password'))]})
-def test_auth_header_no_match(app):
+def test_auth_header_no_match(app: Sphinx) -> None:
     with (
         serve_application(app, custom_handler(valid_credentials=("user1", "password"))),
         pytest.warns(RemovedInSphinx80Warning, match='linkcheck builder encountered an HTTP 401'),
@@ -438,14 +537,14 @@ def test_auth_header_no_match(app):
 
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver', freshenv=True)
-def test_linkcheck_request_headers(app):
+def test_linkcheck_request_headers(app: Sphinx) -> None:
     def check_headers(self):
         if "X-Secret" in self.headers:
             return False
         return self.headers["Accept"] == "text/html"
 
     with serve_application(app, custom_handler(success_criteria=check_headers)) as address:
-        app.config.linkcheck_request_headers = {  # type: ignore[attr-defined]
+        app.config.linkcheck_request_headers = {
             f"http://{address}/": {"Accept": "text/html"},
             "*": {"X-Secret": "open sesami"},
         }
@@ -458,14 +557,14 @@ def test_linkcheck_request_headers(app):
 
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver', freshenv=True)
-def test_linkcheck_request_headers_no_slash(app):
+def test_linkcheck_request_headers_no_slash(app: Sphinx) -> None:
     def check_headers(self):
         if "X-Secret" in self.headers:
             return False
         return self.headers["Accept"] == "application/json"
 
     with serve_application(app, custom_handler(success_criteria=check_headers)) as address:
-        app.config.linkcheck_request_headers = {  # type: ignore[attr-defined]
+        app.config.linkcheck_request_headers = {
             f"http://{address}": {"Accept": "application/json"},
             "*": {"X-Secret": "open sesami"},
         }
@@ -483,7 +582,7 @@ def test_linkcheck_request_headers_no_slash(app):
         "http://do.not.match.org": {"Accept": "application/json"},
         "*": {"X-Secret": "open sesami"},
     }})
-def test_linkcheck_request_headers_default(app):
+def test_linkcheck_request_headers_default(app: Sphinx) -> None:
     def check_headers(self):
         if self.headers["X-Secret"] != "open sesami":
             return False
@@ -566,9 +665,9 @@ def test_follows_redirects_on_GET(app, capsys, warning):
 
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver-warn-redirects')
-def test_linkcheck_allowed_redirects(app, warning):
+def test_linkcheck_allowed_redirects(app: Sphinx, warning: StringIO) -> None:
     with serve_application(app, make_redirect_handler(support_head=False)) as address:
-        app.config.linkcheck_allowed_redirects = {f'http://{address}/.*1': '.*'}  # type: ignore[attr-defined]
+        app.config.linkcheck_allowed_redirects = {f'http://{address}/.*1': '.*'}
         compile_linkcheck_allowed_redirects(app, app.config)
         app.build()
 
@@ -626,7 +725,7 @@ def test_invalid_ssl(get_request, app):
 
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver-https', freshenv=True)
-def test_connect_to_selfsigned_fails(app):
+def test_connect_to_selfsigned_fails(app: Sphinx) -> None:
     with serve_application(app, OKHandler, tls_enabled=True) as address:
         app.build()
 
@@ -639,9 +738,9 @@ def test_connect_to_selfsigned_fails(app):
     assert "[SSL: CERTIFICATE_VERIFY_FAILED]" in content["info"]
 
 
-@pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver-https', freshenv=True)
-def test_connect_to_selfsigned_with_tls_verify_false(app):
-    app.config.tls_verify = False
+@pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver-https', freshenv=True,
+                    confoverrides={'tls_verify': False})
+def test_connect_to_selfsigned_with_tls_verify_false(app: Sphinx) -> None:
     with serve_application(app, OKHandler, tls_enabled=True) as address:
         app.build()
 
@@ -657,9 +756,9 @@ def test_connect_to_selfsigned_with_tls_verify_false(app):
     }
 
 
-@pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver-https', freshenv=True)
-def test_connect_to_selfsigned_with_tls_cacerts(app):
-    app.config.tls_cacerts = CERT_FILE
+@pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver-https', freshenv=True,
+                    confoverrides={'tls_cacerts': CERT_FILE})
+def test_connect_to_selfsigned_with_tls_cacerts(app: Sphinx) -> None:
     with serve_application(app, OKHandler, tls_enabled=True) as address:
         app.build()
 
@@ -693,9 +792,9 @@ def test_connect_to_selfsigned_with_requests_env_var(monkeypatch, app):
     }
 
 
-@pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver-https', freshenv=True)
-def test_connect_to_selfsigned_nonexistent_cert_file(app):
-    app.config.tls_cacerts = "does/not/exist"
+@pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver-https', freshenv=True,
+                    confoverrides={'tls_cacerts': "does/not/exist"})
+def test_connect_to_selfsigned_nonexistent_cert_file(app: Sphinx) -> None:
     with serve_application(app, OKHandler, tls_enabled=True) as address:
         app.build()
 
@@ -863,7 +962,7 @@ def test_too_many_requests_retry_after_without_header(app, capsys):
         'linkcheck_timeout': 0.01,
     }
 )
-def test_requests_timeout(app):
+def test_requests_timeout(app: Sphinx) -> None:
     class DelayedResponseHandler(BaseHTTPRequestHandler):
         protocol_version = "HTTP/1.1"
 
@@ -882,9 +981,9 @@ def test_requests_timeout(app):
     assert content["status"] == "timeout"
 
 
-@pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver', freshenv=True)
-def test_too_many_requests_user_timeout(app):
-    app.config.linkcheck_rate_limit_timeout = 0.0
+@pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver', freshenv=True,
+                    confoverrides={'linkcheck_rate_limit_timeout': 0.0})
+def test_too_many_requests_user_timeout(app: Sphinx) -> None:
     with serve_application(app, make_retry_after_handler([(429, None)])) as address:
         app.build()
     content = (app.outdir / 'output.json').read_text(encoding='utf8')
@@ -903,21 +1002,21 @@ class FakeResponse:
     url = "http://localhost/"
 
 
-def test_limit_rate_default_sleep(app):
+def test_limit_rate_default_sleep(app: Sphinx) -> None:
     worker = HyperlinkAvailabilityCheckWorker(app.config, Queue(), Queue(), {})
     with mock.patch('time.time', return_value=0.0):
         next_check = worker.limit_rate(FakeResponse.url, FakeResponse.headers.get("Retry-After"))
     assert next_check == 60.0
 
 
-def test_limit_rate_user_max_delay(app):
-    app.config.linkcheck_rate_limit_timeout = 0.0
+@pytest.mark.sphinx(confoverrides={'linkcheck_rate_limit_timeout': 0.0})
+def test_limit_rate_user_max_delay(app: Sphinx) -> None:
     worker = HyperlinkAvailabilityCheckWorker(app.config, Queue(), Queue(), {})
     next_check = worker.limit_rate(FakeResponse.url, FakeResponse.headers.get("Retry-After"))
     assert next_check is None
 
 
-def test_limit_rate_doubles_previous_wait_time(app):
+def test_limit_rate_doubles_previous_wait_time(app: Sphinx) -> None:
     rate_limits = {"localhost": RateLimit(60.0, 0.0)}
     worker = HyperlinkAvailabilityCheckWorker(app.config, Queue(), Queue(), rate_limits)
     with mock.patch('time.time', return_value=0.0):
@@ -925,21 +1024,23 @@ def test_limit_rate_doubles_previous_wait_time(app):
     assert next_check == 120.0
 
 
-def test_limit_rate_clips_wait_time_to_max_time(app):
-    app.config.linkcheck_rate_limit_timeout = 90.0
+@pytest.mark.sphinx(confoverrides={'linkcheck_rate_limit_timeout': 90})
+def test_limit_rate_clips_wait_time_to_max_time(app: Sphinx, warning: StringIO) -> None:
     rate_limits = {"localhost": RateLimit(60.0, 0.0)}
     worker = HyperlinkAvailabilityCheckWorker(app.config, Queue(), Queue(), rate_limits)
     with mock.patch('time.time', return_value=0.0):
         next_check = worker.limit_rate(FakeResponse.url, FakeResponse.headers.get("Retry-After"))
     assert next_check == 90.0
+    assert warning.getvalue() == ''
 
 
-def test_limit_rate_bails_out_after_waiting_max_time(app):
-    app.config.linkcheck_rate_limit_timeout = 90.0
+@pytest.mark.sphinx(confoverrides={'linkcheck_rate_limit_timeout': 90.0})
+def test_limit_rate_bails_out_after_waiting_max_time(app: Sphinx, warning: StringIO) -> None:
     rate_limits = {"localhost": RateLimit(90.0, 0.0)}
     worker = HyperlinkAvailabilityCheckWorker(app.config, Queue(), Queue(), rate_limits)
     next_check = worker.limit_rate(FakeResponse.url, FakeResponse.headers.get("Retry-After"))
     assert next_check is None
+    assert warning.getvalue() == ''
 
 
 @mock.patch('sphinx.util.requests.requests.Session.get_adapter')
@@ -957,11 +1058,13 @@ def test_connection_contention(get_adapter, app, capsys):
 
         # Place a workload into the linkcheck queue
         link_count = 10
-        rqueue, wqueue = Queue(), Queue()
+        wqueue: Queue[CheckRequest] = Queue()
+        rqueue: Queue[CheckResult] = Queue()
         for _ in range(link_count):
             wqueue.put(CheckRequest(0, Hyperlink(f"http://{address}", "test", "test.rst", 1)))
 
-        begin, checked = time.time(), []
+        begin = time.time()
+        checked: list[CheckResult] = []
         threads = [
             HyperlinkAvailabilityCheckWorker(
                 config=app.config,
@@ -997,7 +1100,7 @@ class ConnectionResetHandler(BaseHTTPRequestHandler):
 
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver', freshenv=True)
-def test_get_after_head_raises_connection_error(app):
+def test_get_after_head_raises_connection_error(app: Sphinx) -> None:
     with serve_application(app, ConnectionResetHandler) as address:
         app.build()
     content = (app.outdir / 'output.txt').read_text(encoding='utf8')
@@ -1014,7 +1117,7 @@ def test_get_after_head_raises_connection_error(app):
 
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck-documents_exclude', freshenv=True)
-def test_linkcheck_exclude_documents(app):
+def test_linkcheck_exclude_documents(app: Sphinx) -> None:
     with serve_application(app, DefaultsHandler):
         app.build()