publishFullContrib: true for ydb

<HIDDEN_URL>
commit_hash:c82a80ac4594723cebf2c7387dec9c60217f603e

1 files changed, 297 insertions, 0 deletions

diff --git a/contrib/python/pytest-localserver/py3/tests/test_http.py b/contrib/python/pytest-localserver/py3/tests/test_http.py
new file mode 100644
index 0000000000..64859ffd25
--- /dev/null
+++ b/contrib/python/pytest-localserver/py3/tests/test_http.py
@@ -0,0 +1,297 @@
+import itertools
+import subprocess
+import sys
+import textwrap
+
+import pytest
+import requests
+
+from pytest_localserver import http
+from pytest_localserver import plugin
+
+
+# define test fixture here again in order to run tests without having to
+# install the plugin anew every single time
+httpserver = plugin.httpserver
+
+
+transfer_encoded = pytest.mark.parametrize(
+    "transfer_encoding_header", ["Transfer-encoding", "Transfer-Encoding", "transfer-encoding", "TRANSFER-ENCODING"]
+)
+
+
+def test_httpserver_funcarg(httpserver):
+    assert isinstance(httpserver, http.ContentServer)
+    assert httpserver.is_alive()
+    assert httpserver.server_address
+
+
+def test_server_does_not_serve_file_at_startup(httpserver):
+    assert httpserver.code == 204
+    assert httpserver.content == ""
+
+
+def test_some_content_retrieval(httpserver):
+    httpserver.serve_content("TEST!")
+    resp = requests.get(httpserver.url)
+    assert resp.text == "TEST!"
+    assert resp.status_code == 200
+
+
+def test_request_is_stored(httpserver):
+    httpserver.serve_content("TEST!")
+    assert len(httpserver.requests) == 0
+    requests.get(httpserver.url)
+    assert len(httpserver.requests) == 1
+
+
+def test_GET_request(httpserver):
+    httpserver.serve_content("TEST!", headers={"Content-type": "text/plain"})
+    resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
+    assert resp.text == "TEST!"
+    assert resp.status_code == 200
+    assert "text/plain" in resp.headers["Content-type"]
+
+
+# FIXME get compression working!
+# def test_gzipped_GET_request(httpserver):
+#     httpserver.serve_content('TEST!', headers={'Content-type': 'text/plain'})
+#     httpserver.compress = 'gzip'
+#     resp = requests.get(httpserver.url, headers={
+#         'User-Agent': 'Test method',
+#         'Accept-encoding': 'gzip'
+#     })
+#     assert resp.text == 'TEST!'
+#     assert resp.status_code == 200
+#     assert resp.content_encoding == 'gzip'
+#     assert resp.headers['Content-type'] == 'text/plain'
+#     assert resp.headers['content-encoding'] == 'gzip'
+
+
+def test_HEAD_request(httpserver):
+    httpserver.serve_content("TEST!", headers={"Content-type": "text/plain"})
+    resp = requests.head(httpserver.url)
+    assert resp.status_code == 200
+    assert resp.headers["Content-type"] == "text/plain"
+
+
+# def test_POST_request(httpserver):
+#     headers = {'Content-type': 'application/x-www-form-urlencoded',
+#                'set-cookie': 'some _cookie_content'}
+#
+#     httpserver.serve_content('TEST!', headers=headers)
+#     resp = requests.post(httpserver.url, data={'data': 'value'}, headers=headers)
+#     assert resp.text == 'TEST!'
+#     assert resp.status_code == 200
+#
+#     httpserver.serve_content('TEST!', headers=headers, show_post_vars=True)
+#     resp = requests.post(httpserver.url, data={'data': 'value'}, headers=headers)
+#     assert resp.json() == {'data': 'value'}
+#     assert resp.status_code == 200
+
+
+@pytest.mark.parametrize("chunked_flag", [http.Chunked.YES, http.Chunked.AUTO, http.Chunked.NO])
+def test_chunked_attribute_without_header(httpserver, chunked_flag):
+    """
+    Test that passing the chunked attribute to serve_content() properly sets
+    the chunked property of the server.
+    """
+    httpserver.serve_content(("TEST!", "test"), headers={"Content-type": "text/plain"}, chunked=chunked_flag)
+    assert httpserver.chunked == chunked_flag
+
+
+@pytest.mark.parametrize("chunked_flag", [http.Chunked.YES, http.Chunked.AUTO, http.Chunked.NO])
+def test_chunked_attribute_with_header(httpserver, chunked_flag):
+    """
+    Test that passing the chunked attribute to serve_content() properly sets
+    the chunked property of the server even when the transfer-encoding header is
+    also set.
+    """
+    httpserver.serve_content(
+        ("TEST!", "test"), headers={"Content-type": "text/plain", "Transfer-encoding": "chunked"}, chunked=chunked_flag
+    )
+    assert httpserver.chunked == chunked_flag
+
+
+@transfer_encoded
+@pytest.mark.parametrize("chunked_flag", [http.Chunked.YES, http.Chunked.AUTO])
+def test_GET_request_chunked_parameter(httpserver, transfer_encoding_header, chunked_flag):
+    """
+    Test that passing YES or AUTO as the chunked parameter to serve_content()
+    causes the response to be sent using chunking when the Transfer-encoding
+    header is also set.
+    """
+    httpserver.serve_content(
+        ("TEST!", "test"),
+        headers={"Content-type": "text/plain", transfer_encoding_header: "chunked"},
+        chunked=chunked_flag,
+    )
+    resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
+    assert resp.text == "TEST!test"
+    assert resp.status_code == 200
+    assert "text/plain" in resp.headers["Content-type"]
+    assert "chunked" in resp.headers["Transfer-encoding"]
+
+
+@transfer_encoded
+@pytest.mark.parametrize("chunked_flag", [http.Chunked.YES, http.Chunked.AUTO])
+def test_GET_request_chunked_attribute(httpserver, transfer_encoding_header, chunked_flag):
+    """
+    Test that setting the chunked attribute of httpserver to YES or AUTO
+    causes the response to be sent using chunking when the Transfer-encoding
+    header is also set.
+    """
+    httpserver.serve_content(
+        ("TEST!", "test"), headers={"Content-type": "text/plain", transfer_encoding_header: "chunked"}
+    )
+    httpserver.chunked = chunked_flag
+    resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
+    assert resp.text == "TEST!test"
+    assert resp.status_code == 200
+    assert "text/plain" in resp.headers["Content-type"]
+    assert "chunked" in resp.headers["Transfer-encoding"]
+
+
+@transfer_encoded
+def test_GET_request_not_chunked(httpserver, transfer_encoding_header):
+    """
+    Test that setting the chunked attribute of httpserver to NO causes
+    the response not to be sent using chunking even if the Transfer-encoding
+    header is set.
+    """
+    httpserver.serve_content(
+        ("TEST!", "test"),
+        headers={"Content-type": "text/plain", transfer_encoding_header: "chunked"},
+        chunked=http.Chunked.NO,
+    )
+    with pytest.raises(requests.exceptions.ChunkedEncodingError):
+        requests.get(httpserver.url, headers={"User-Agent": "Test method"})
+
+
+@pytest.mark.parametrize("chunked_flag", [http.Chunked.NO, http.Chunked.AUTO])
+def test_GET_request_chunked_parameter_no_header(httpserver, chunked_flag):
+    """
+    Test that passing NO or AUTO as the chunked parameter to serve_content()
+    causes the response not to be sent using chunking when the Transfer-encoding
+    header is not set.
+    """
+    httpserver.serve_content(("TEST!", "test"), headers={"Content-type": "text/plain"}, chunked=chunked_flag)
+    resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
+    assert resp.text == "TEST!test"
+    assert resp.status_code == 200
+    assert "text/plain" in resp.headers["Content-type"]
+    assert "Transfer-encoding" not in resp.headers
+
+
+@pytest.mark.parametrize("chunked_flag", [http.Chunked.NO, http.Chunked.AUTO])
+def test_GET_request_chunked_attribute_no_header(httpserver, chunked_flag):
+    """
+    Test that setting the chunked attribute of httpserver to NO or AUTO
+    causes the response not to be sent using chunking when the Transfer-encoding
+    header is not set.
+    """
+    httpserver.serve_content(("TEST!", "test"), headers={"Content-type": "text/plain"})
+    httpserver.chunked = chunked_flag
+    resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
+    assert resp.text == "TEST!test"
+    assert resp.status_code == 200
+    assert "text/plain" in resp.headers["Content-type"]
+    assert "Transfer-encoding" not in resp.headers
+
+
+def test_GET_request_chunked_no_header(httpserver):
+    """
+    Test that setting the chunked attribute of httpserver to YES causes
+    the response to be sent using chunking even if the Transfer-encoding
+    header is not set.
+    """
+    httpserver.serve_content(("TEST!", "test"), headers={"Content-type": "text/plain"}, chunked=http.Chunked.YES)
+    resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
+    # Without the Transfer-encoding header set, requests does not undo the chunk
+    # encoding so it comes through as "raw" chunks
+    assert resp.text == "5\r\nTEST!\r\n4\r\ntest\r\n0\r\n\r\n"
+
+
+def _format_chunk(chunk):
+    r = repr(chunk)
+    if len(r) <= 40:
+        return r
+    else:
+        return r[:13] + "..." + r[-14:] + " (length {})".format(len(chunk))
+
+
+def _compare_chunks(expected, actual):
+    __tracebackhide__ = True
+    if expected != actual:
+        message = [_format_chunk(expected) + " != " + _format_chunk(actual)]
+        if type(expected) == type(actual):
+            for i, (e, a) in enumerate(itertools.zip_longest(expected, actual, fillvalue="<end>")):
+                if e != a:
+                    message += [
+                        "  Chunks differ at index {}:".format(i),
+                        "    Expected: " + (repr(expected[i : i + 5]) + "..." if e != "<end>" else "<end>"),
+                        "    Found:    " + (repr(actual[i : i + 5]) + "..." if a != "<end>" else "<end>"),
+                    ]
+                    break
+        pytest.fail("\n".join(message))
+
+
+@pytest.mark.parametrize("chunk_size", [400, 499, 500, 512, 750, 1024, 4096, 8192])
+def test_GET_request_large_chunks(httpserver, chunk_size):
+    """
+    Test that a response with large chunks comes through correctly
+    """
+    body = b"0123456789abcdef" * 1024  # 16 kb total
+    # Split body into fixed-size chunks, from https://stackoverflow.com/a/18854817/56541
+    chunks = [body[0 + i : chunk_size + i] for i in range(0, len(body), chunk_size)]
+    httpserver.serve_content(
+        chunks, headers={"Content-type": "text/plain", "Transfer-encoding": "chunked"}, chunked=http.Chunked.YES
+    )
+    resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"}, stream=True)
+    assert resp.status_code == 200
+    text = b""
+    for original_chunk, received_chunk in itertools.zip_longest(chunks, resp.iter_content(chunk_size=None)):
+        _compare_chunks(original_chunk, received_chunk)
+        text += received_chunk
+    assert text == body
+    assert "chunked" in resp.headers["Transfer-encoding"]
+
+
+@pytest.mark.parametrize("chunked_flag", [http.Chunked.YES, http.Chunked.AUTO])
+def test_GET_request_chunked_no_content_length(httpserver, chunked_flag):
+    """
+    Test that a chunked response does not include a Content-length header
+    """
+    httpserver.serve_content(
+        ("TEST!", "test"), headers={"Content-type": "text/plain", "Transfer-encoding": "chunked"}, chunked=chunked_flag
+    )
+    resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
+    assert resp.status_code == 200
+    assert "Transfer-encoding" in resp.headers
+    assert "Content-length" not in resp.headers
+
+
+def test_httpserver_init_failure_no_stderr_during_cleanup(tmp_path):
+    """
+    Test that, when the server encounters an error during __init__, its cleanup
+    does not raise an AttributeError in its __del__ method, which would emit a 
+    warning onto stderr.
+    """
+
+    script_path = tmp_path.joinpath("script.py")
+
+    script_path.write_text(textwrap.dedent("""
+        from pytest_localserver import http
+        from unittest.mock import patch
+
+        with patch("pytest_localserver.http.make_server", side_effect=RuntimeError("init failure")):
+            server = http.ContentServer()
+    """))
+
+    result = subprocess.run([sys.executable, str(script_path)], stderr=subprocess.PIPE)
+
+    # We ensure that no warning about an AttributeError is printed on stderr
+    # due to an error in the server's __del__ method. This AttributeError is
+    # raised during cleanup and doesn't affect the exit code of the script, so
+    # we can't use the exit code to tell whether the script did its job.
+    assert b"AttributeError" not in result.stderr