[3.13] gh-67765: Add tests for wsgiref.validate (GH-112398) (GH-155625)

Cover the InputWrapper and ErrorWrapper methods of wsgiref.validate:
read, readline, readlines, __iter__, write, writelines and flush.
Each is tested both for the AssertionError raised on an invalid call
and for the data passed through on a valid one.

Co-authored-by: Alex Shkop <[email protected]>
(cherry picked from commit e96cf73)

* Use ExtraAssertions instead of rewriting the assertions

assertStartsWith() and assertEndsWith() are provided by
test.support.testcase.ExtraAssertions in 3.13, so the new tests can be
kept identical to the 3.14+ version.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>

---------

Co-authored-by: Amuthan Mannar <[email protected]>
Co-authored-by: Alex Shkop <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>

Author: 4 people

Lib/test/test_wsgiref.py

@@ -1,6 +1,7 @@
 from unittest import mock
 from test import support
 from test.support import socket_helper, control_characters_c0
+from test.support.testcase import ExtraAssertions
 from test.test_httpservers import NoLogRequestHandler
 from unittest import TestCase
 from wsgiref.util import setup_testing_defaults
@@ -66,6 +67,26 @@ def header_app(environ, start_response):
     ]).encode('iso-8859-1')]
 
 
+def input_app(func_name, *args):
+    def app(e,s):
+        req = getattr(e['wsgi.input'], func_name)(*args)
+        s("200 OK", [("Content-Type", "text/plain; charset=utf-8")])
+        if type(req) is list:
+            resp = b";".join(req)
+        else:
+            resp = req
+        return [resp]
+    return app
+
+
+def errors_app(func_name, *args):
+    def app(e,s):
+        getattr(e['wsgi.errors'], func_name)(*args)
+        s("200 OK", [("Content-Type", "text/plain; charset=utf-8")])
+        return [b"data"]
+    return app
+
+
 def run_amock(app=hello_app, data=b"GET / HTTP/1.0\n\n"):
     server = make_server("", 80, app, MockServer, MockHandler)
     inp = BufferedReader(BytesIO(data))
@@ -102,7 +123,7 @@ def compare_generic_iter(make_it, match):
         raise AssertionError("Too many items from .__next__()", it)
 
 
-class IntegrationTests(TestCase):
+class IntegrationTests(TestCase, ExtraAssertions):
 
     def check_hello(self, out, has_length=True):
         pyver = (python_implementation() + "/" +
@@ -192,6 +213,95 @@ def bad_app(e,s):
             err.splitlines()[-2], "AssertionError"
         )
 
+    def test_wsgi_input_read(self):
+        bad_app = input_app("read")
+        good_app = input_app("read", 5)
+
+        out, err = run_amock(validator(bad_app))
+        self.assertEndsWith(out,
+             b"A server error occurred.  Please contact the administrator."
+        )
+
+        self.assertEqual(
+            err.splitlines()[-2], "AssertionError"
+        )
+
+        out, err = run_amock(validator(good_app), b"GET / HTTP/1.0\n\nTest 1\nTest 2\n")
+        self.assertEndsWith(out, b"Test ")
+
+    def test_wsgi_input_readlines(self):
+        bad_app = input_app("readlines", 3, 5)
+        good_app = input_app("readlines", 1)
+
+        out, err = run_amock(validator(bad_app))
+        self.assertEndsWith(out,
+            b"A server error occurred.  Please contact the administrator."
+        )
+        self.assertEqual(
+            err.splitlines()[-2], "AssertionError"
+        )
+        out, err = run_amock(validator(good_app), b"GET / HTTP/1.0\n\nTest Line 1\nTest Line 2\n")
+        self.assertEndsWith(out, b"Test Line 1\n")
+
+    def test_wsgi_input_readline(self):
+        bad_app = input_app("readline", 3, 4)
+        good_app = input_app("readline", 2)
+
+        out, err = run_amock(validator(bad_app))
+        self.assertEndsWith(out,
+            b"A server error occurred.  Please contact the administrator."
+        )
+        self.assertEqual(
+            err.splitlines()[-2], "AssertionError"
+        )
+
+        out, err = run_amock(validator(good_app), b"GET / HTTP/1.0\n\nTest 1\nTest 2\n")
+        self.assertEndsWith(out, b"Te")
+
+    def test_wsgi_input_close(self):
+        app = input_app("close")
+        out, err = run_amock(validator(app), b"GET / HTTP/1.0\n\nTest 1\nTest 2\n")
+        self.assertEqual(err.splitlines()[-2], 'AssertionError: input.close() must not be called')
+        self.assertEndsWith(out, b"A server error occurred.  Please contact the administrator.")
+
+    def test_wsgi_input_iter(self):
+        def app(e,s):
+            req = []
+            for line in e['wsgi.input']:
+                req.append(line)
+            s("200 OK", [('Content-Type', 'text/plain; charser=utf-8')])
+            return [b';'.join(req)]
+
+        out, err = run_amock(validator(app), b"GET / HTTP/1.0\n\nTest 1\nTest 2\n")
+        self.assertEndsWith(out, b"Test 1\n;Test 2\n")
+
+    def test_wsgi_errors_write(self):
+        bad_app = errors_app("write", b"Test")
+        good_app = errors_app("write", "Test")
+
+        out, err = run_amock(validator(bad_app), b"GET / HTTP/1.0\n\n")
+        self.assertEqual(err.splitlines()[-2], 'AssertionError')
+
+        out, err = run_amock(validator(good_app), b"GET / HTTP/1.0\n\n")
+        self.assertStartsWith(err, "Test")
+
+    def test_wsgi_errors_writelines(self):
+        bad_app = errors_app("writelines", [1, "Test"])
+        good_app = errors_app("writelines", ["Test", "Test"])
+
+        out, err = run_amock(validator(bad_app), b"GET / HTTP/1.0\n\n")
+        self.assertEqual(err.splitlines()[-2], 'AssertionError')
+
+        out, err = run_amock(validator(good_app), b"GET / HTTP/1.0\n\n")
+        self.assertStartsWith(err, "TestTest")
+
+    def test_wsgi_errors_close(self):
+        app = errors_app("close")
+
+        out, err = run_amock(validator(app), b"GET / HTTP/1.0\n\n")
+        self.assertEqual(err.splitlines()[-2],
+                         'AssertionError: errors.close() must not be called')
+
     def test_bytes_validation(self):
         def app(e, s):
             s("200 OK", [