Add cheroot-8.2.1

lib/cheroot/test/test_server.py

@@ -0,0 +1,235 @@
+"""Tests for the HTTP server."""
+# -*- coding: utf-8 -*-
+# vim: set fileencoding=utf-8 :
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+import os
+import socket
+import tempfile
+import threading
+import uuid
+
+import pytest
+import requests
+import requests_unixsocket
+import six
+
+from .._compat import bton, ntob
+from .._compat import IS_LINUX, IS_MACOS, SYS_PLATFORM
+from ..server import IS_UID_GID_RESOLVABLE, Gateway, HTTPServer
+from ..testing import (
+    ANY_INTERFACE_IPV4,
+    ANY_INTERFACE_IPV6,
+    EPHEMERAL_PORT,
+    get_server_client,
+)
+
+
+unix_only_sock_test = pytest.mark.skipif(
+    not hasattr(socket, 'AF_UNIX'),
+    reason='UNIX domain sockets are only available under UNIX-based OS',
+)
+
+
+non_macos_sock_test = pytest.mark.skipif(
+    IS_MACOS,
+    reason='Peercreds lookup does not work under macOS/BSD currently.',
+)
+
+
+@pytest.fixture(params=('abstract', 'file'))
+def unix_sock_file(request):
+    """Check that bound UNIX socket address is stored in server."""
+    if request.param == 'abstract':
+        yield request.getfixturevalue('unix_abstract_sock')
+        return
+    tmp_sock_fh, tmp_sock_fname = tempfile.mkstemp()
+
+    yield tmp_sock_fname
+
+    os.close(tmp_sock_fh)
+    os.unlink(tmp_sock_fname)
+
+
+@pytest.fixture
+def unix_abstract_sock():
+    """Return an abstract UNIX socket address."""
+    if not IS_LINUX:
+        pytest.skip(
+            '{os} does not support an abstract '
+            'socket namespace'.format(os=SYS_PLATFORM),
+        )
+    return b''.join((
+        b'\x00cheroot-test-socket',
+        ntob(str(uuid.uuid4())),
+    )).decode()
+
+
+def test_prepare_makes_server_ready():
+    """Check that prepare() makes the server ready, and stop() clears it."""
+    httpserver = HTTPServer(
+        bind_addr=(ANY_INTERFACE_IPV4, EPHEMERAL_PORT),
+        gateway=Gateway,
+    )
+
+    assert not httpserver.ready
+    assert not httpserver.requests._threads
+
+    httpserver.prepare()
+
+    assert httpserver.ready
+    assert httpserver.requests._threads
+    for thr in httpserver.requests._threads:
+        assert thr.ready
+
+    httpserver.stop()
+
+    assert not httpserver.requests._threads
+    assert not httpserver.ready
+
+
+def test_stop_interrupts_serve():
+    """Check that stop() interrupts running of serve()."""
+    httpserver = HTTPServer(
+        bind_addr=(ANY_INTERFACE_IPV4, EPHEMERAL_PORT),
+        gateway=Gateway,
+    )
+
+    httpserver.prepare()
+    serve_thread = threading.Thread(target=httpserver.serve)
+    serve_thread.start()
+
+    serve_thread.join(0.5)
+    assert serve_thread.is_alive()
+
+    httpserver.stop()
+
+    serve_thread.join(0.5)
+    assert not serve_thread.is_alive()
+
+
+@pytest.mark.parametrize(
+    'ip_addr',
+    (
+        ANY_INTERFACE_IPV4,
+        ANY_INTERFACE_IPV6,
+    ),
+)
+def test_bind_addr_inet(http_server, ip_addr):
+    """Check that bound IP address is stored in server."""
+    httpserver = http_server.send((ip_addr, EPHEMERAL_PORT))
+
+    assert httpserver.bind_addr[0] == ip_addr
+    assert httpserver.bind_addr[1] != EPHEMERAL_PORT
+
+
+@unix_only_sock_test
+def test_bind_addr_unix(http_server, unix_sock_file):
+    """Check that bound UNIX socket address is stored in server."""
+    httpserver = http_server.send(unix_sock_file)
+
+    assert httpserver.bind_addr == unix_sock_file
+
+
+@unix_only_sock_test
+def test_bind_addr_unix_abstract(http_server, unix_abstract_sock):
+    """Check that bound UNIX abstract sockaddr is stored in server."""
+    httpserver = http_server.send(unix_abstract_sock)
+
+    assert httpserver.bind_addr == unix_abstract_sock
+
+
+PEERCRED_IDS_URI = '/peer_creds/ids'
+PEERCRED_TEXTS_URI = '/peer_creds/texts'
+
+
+class _TestGateway(Gateway):
+    def respond(self):
+        req = self.req
+        conn = req.conn
+        req_uri = bton(req.uri)
+        if req_uri == PEERCRED_IDS_URI:
+            peer_creds = conn.peer_pid, conn.peer_uid, conn.peer_gid
+            self.send_payload('|'.join(map(str, peer_creds)))
+            return
+        elif req_uri == PEERCRED_TEXTS_URI:
+            self.send_payload('!'.join((conn.peer_user, conn.peer_group)))
+            return
+        return super(_TestGateway, self).respond()
+
+    def send_payload(self, payload):
+        req = self.req
+        req.status = b'200 OK'
+        req.ensure_headers_sent()
+        req.write(ntob(payload))
+
+
+@pytest.fixture
+def peercreds_enabled_server_and_client(http_server, unix_sock_file):
+    """Construct a test server with `peercreds_enabled`."""
+    httpserver = http_server.send(unix_sock_file)
+    httpserver.gateway = _TestGateway
+    httpserver.peercreds_enabled = True
+    return httpserver, get_server_client(httpserver)
+
+
+@unix_only_sock_test
+@non_macos_sock_test
+def test_peercreds_unix_sock(peercreds_enabled_server_and_client):
+    """Check that peercred lookup works when enabled."""
+    httpserver, testclient = peercreds_enabled_server_and_client
+    bind_addr = httpserver.bind_addr
+
+    if isinstance(bind_addr, six.binary_type):
+        bind_addr = bind_addr.decode()
+
+    unix_base_uri = 'http+unix://{}'.format(
+        bind_addr.replace('\0', '%00').replace('/', '%2F'),
+    )
+
+    expected_peercreds = os.getpid(), os.getuid(), os.getgid()
+    expected_peercreds = '|'.join(map(str, expected_peercreds))
+
+    with requests_unixsocket.monkeypatch():
+        peercreds_resp = requests.get(unix_base_uri + PEERCRED_IDS_URI)
+        peercreds_resp.raise_for_status()
+        assert peercreds_resp.text == expected_peercreds
+
+        peercreds_text_resp = requests.get(unix_base_uri + PEERCRED_TEXTS_URI)
+        assert peercreds_text_resp.status_code == 500
+
+
+@pytest.mark.skipif(
+    not IS_UID_GID_RESOLVABLE,
+    reason='Modules `grp` and `pwd` are not available '
+           'under the current platform',
+)
+@unix_only_sock_test
+@non_macos_sock_test
+def test_peercreds_unix_sock_with_lookup(peercreds_enabled_server_and_client):
+    """Check that peercred resolution works when enabled."""
+    httpserver, testclient = peercreds_enabled_server_and_client
+    httpserver.peercreds_resolve_enabled = True
+
+    bind_addr = httpserver.bind_addr
+
+    if isinstance(bind_addr, six.binary_type):
+        bind_addr = bind_addr.decode()
+
+    unix_base_uri = 'http+unix://{}'.format(
+        bind_addr.replace('\0', '%00').replace('/', '%2F'),
+    )
+
+    import grp
+    import pwd
+    expected_textcreds = (
+        pwd.getpwuid(os.getuid()).pw_name,
+        grp.getgrgid(os.getgid()).gr_name,
+    )
+    expected_textcreds = '!'.join(map(str, expected_textcreds))
+    with requests_unixsocket.monkeypatch():
+        peercreds_text_resp = requests.get(unix_base_uri + PEERCRED_TEXTS_URI)
+        peercreds_text_resp.raise_for_status()
+        assert peercreds_text_resp.text == expected_textcreds