From 574018590e4be647e983beb81ce2a980aef0db7f Mon Sep 17 00:00:00 2001 From: Amuthan Mannar Date: Wed, 12 Aug 2026 18:22:00 +0530 Subject: [PATCH 1/2] gh-67765: Add tests for wsgiref.validate (GH-112398) 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 (cherry picked from commit e96cf738e810df38531de0973f561ef9fc868e77) --- Lib/test/test_wsgiref.py | 109 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 109 insertions(+) diff --git a/Lib/test/test_wsgiref.py b/Lib/test/test_wsgiref.py index 34e52fe40f500ec..978ae0021906276 100644 --- a/Lib/test/test_wsgiref.py +++ b/Lib/test/test_wsgiref.py @@ -66,6 +66,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)) @@ -192,6 +212,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.assertTrue(out.endswith( + 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.assertTrue(out.endswith(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.assertTrue(out.endswith( + 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.assertTrue(out.endswith(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.assertTrue(out.endswith( + 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.assertTrue(out.endswith(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.assertTrue(out.endswith(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.assertTrue(out.endswith(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.assertTrue(err.startswith("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.assertTrue(err.startswith("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", [ From c08cbc88a1d5426f68d4ca82f63c6be70ee2b693 Mon Sep 17 00:00:00 2001 From: Serhiy Storchaka Date: Wed, 12 Aug 2026 16:03:48 +0300 Subject: [PATCH 2/2] 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) --- Lib/test/test_wsgiref.py | 29 +++++++++++++++-------------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/Lib/test/test_wsgiref.py b/Lib/test/test_wsgiref.py index 978ae0021906276..7772e2a6399e5e4 100644 --- a/Lib/test/test_wsgiref.py +++ b/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 @@ -122,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() + "/" + @@ -217,51 +218,51 @@ def test_wsgi_input_read(self): good_app = input_app("read", 5) out, err = run_amock(validator(bad_app)) - self.assertTrue(out.endswith( + 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.assertTrue(out.endswith(b"Test ")) + 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.assertTrue(out.endswith( + 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.assertTrue(out.endswith(b"Test Line 1\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.assertTrue(out.endswith( + 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.assertTrue(out.endswith(b"Te")) + 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.assertTrue(out.endswith(b"A server error occurred. Please contact the administrator.")) + self.assertEndsWith(out, b"A server error occurred. Please contact the administrator.") def test_wsgi_input_iter(self): def app(e,s): @@ -272,7 +273,7 @@ def app(e,s): return [b';'.join(req)] out, err = run_amock(validator(app), b"GET / HTTP/1.0\n\nTest 1\nTest 2\n") - self.assertTrue(out.endswith(b"Test 1\n;Test 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") @@ -282,7 +283,7 @@ def test_wsgi_errors_write(self): self.assertEqual(err.splitlines()[-2], 'AssertionError') out, err = run_amock(validator(good_app), b"GET / HTTP/1.0\n\n") - self.assertTrue(err.startswith("Test")) + self.assertStartsWith(err, "Test") def test_wsgi_errors_writelines(self): bad_app = errors_app("writelines", [1, "Test"]) @@ -292,7 +293,7 @@ def test_wsgi_errors_writelines(self): self.assertEqual(err.splitlines()[-2], 'AssertionError') out, err = run_amock(validator(good_app), b"GET / HTTP/1.0\n\n") - self.assertTrue(err.startswith("TestTest")) + self.assertStartsWith(err, "TestTest") def test_wsgi_errors_close(self): app = errors_app("close")