Merge pull request #8394 from francoisfreitag/real-inv

Rewrite test_inspect_main_url to avoid mocking

tests/test_build_linkcheck.py

@@ -11,10 +11,11 @@
 import http.server
 import json
 import re
-import threading
 from unittest import mock
 import pytest
 
+from utils import http_server
+
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck', freshenv=True)
 def test_defaults(app, status, warning):
@@ -110,12 +111,12 @@ def test_anchors_ignored(app, status, warning):
 
 @pytest.mark.sphinx('linkcheck', testroot='linkcheck-localserver', freshenv=True)
 def test_raises_for_invalid_status(app, status, warning):
-    server_thread = HttpServerThread(InternalServerErrorHandler, daemon=True)
-    server_thread.start()
-    try:
+    class InternalServerErrorHandler(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):
+            self.send_error(500, "Internal Server Error")
+
+    with http_server(InternalServerErrorHandler):
         app.builder.build_all()
-    finally:
-        server_thread.terminate()
     content = (app.outdir / 'output.txt').read_text()
     assert content == (
         "index.rst:1: [broken] http://localhost:7777/#anchor: "
@@ -177,22 +178,3 @@ def test_linkcheck_request_headers(app, status, warning):
                 assert headers["X-Secret"] == "open sesami"
             else:
                 assert headers["Accept"] == "text/html,application/xhtml+xml;q=0.9,*/*;q=0.8"
-
-
-class HttpServerThread(threading.Thread):
-    def __init__(self, handler, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.server = http.server.HTTPServer(("localhost", 7777), handler)
-
-    def run(self):
-        self.server.serve_forever(poll_interval=0.01)
-
-    def terminate(self):
-        self.server.shutdown()
-        self.server.server_close()
-        self.join()
-
-
-class InternalServerErrorHandler(http.server.BaseHTTPRequestHandler):
-    def do_GET(self):
-        self.send_error(500, "Internal Server Error")

tests/test_ext_intersphinx.py

@@ -8,13 +8,13 @@
     :license: BSD, see LICENSE for details.
 """
 
+import http.server
 import os
 import unittest
 from io import BytesIO
 from unittest import mock
 
 import pytest
-import requests
 from docutils import nodes
 from test_util_inventory import inventory_v2, inventory_v2_not_having_version
 
@@ -25,6 +25,8 @@ from sphinx.ext.intersphinx import (
 )
 from sphinx.ext.intersphinx import setup as intersphinx_setup
 
+from utils import http_server
+
 
 def fake_node(domain, type, target, content, **attrs):
     contnode = nodes.emphasis(content, content)
@@ -433,24 +435,22 @@ def test_inspect_main_file(capsys, tempdir):
     assert stderr == ""
 
 
-@mock.patch('requests.get')
-def test_inspect_main_url(fake_get, capsys):
+def test_inspect_main_url(capsys):
     """inspect_main interface, with url argument"""
-    raw = BytesIO(inventory_v2)
-    real_read = raw.read
+    class InventoryHandler(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):
+            self.send_response(200, "OK")
+            self.end_headers()
+            self.wfile.write(inventory_v2)
 
-    def fake_read(*args, **kwargs):
-        return real_read()
+        def log_message(*args, **kwargs):
+            # Silenced.
+            pass
 
-    raw.read = fake_read
-    url = 'http://hostname/' + INVENTORY_FILENAME
-    resp = requests.Response()
-    resp.status_code = 200
-    resp.url = url
-    resp.raw = raw
-    fake_get.return_value = resp
+    url = 'http://localhost:7777/' + INVENTORY_FILENAME
 
-    inspect_main([url])
+    with http_server(InventoryHandler):
+        inspect_main([url])
 
     stdout, stderr = capsys.readouterr()
     assert stdout.startswith("c:function\n")

tests/utils.py

@@ -0,0 +1,27 @@
+import contextlib
+import http.server
+import threading
+
+
+class HttpServerThread(threading.Thread):
+    def __init__(self, handler, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.server = http.server.HTTPServer(("localhost", 7777), handler)
+
+    def run(self):
+        self.server.serve_forever(poll_interval=0.01)
+
+    def terminate(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.join()
+
+
+@contextlib.contextmanager
+def http_server(handler):
+    server_thread = HttpServerThread(handler, daemon=True)
+    server_thread.start()
+    try:
+        yield server_thread
+    finally:
+        server_thread.terminate()