Config coverage increase

.travis.yml

@@ -27,10 +27,10 @@ install:
         choco install python3;
         export PATH=/c/Python37:/c/Python37/Scripts:/c/Python38:/c/Python38/Scripts:$PATH;
         python -m pip install -U click h11 wsproto==0.13.* websockets==8.*;
-        python -m pip install -U autoflake black codecov flake8 isort pytest pytest-cov requests watchgod;
+        python -m pip install -U autoflake black codecov flake8 isort pytest pytest-cov requests watchgod python-dotenv;
       elif [ "$TRAVIS_PYTHON_VERSION" = "pypy3" ]; then
         pip install -U click h11 wsproto==0.13.*;
-        pip install -U autoflake codecov flake8 isort pytest pytest-cov requests watchgod;
+        pip install -U autoflake codecov flake8 isort pytest pytest-cov requests watchgod python-dotenv;
       else
         pip install -U -r requirements.txt;
       fi;

requirements.txt

@@ -6,6 +6,7 @@ httptools
 uvloop>=0.14.0
 websockets==8.*
 wsproto==0.13.*
+python-dotenv
 
 # Testing
 autoflake

tests/conftest.py

@@ -70,3 +70,44 @@ def certfile_and_keyfile(tmp_path):
         fout.write(PRIVATE_KEY)
 
     return certfile, keyfile
+
+
+ENV_FILE = """KEY_TRUE="1"
+KEY_FALSE=""
+WEB_CONCURRENCY=2048
+"""
+
+
+@pytest.fixture(scope="function")
+def env_file(tmp_path):
+    envfile = str(tmp_path / ".env")
+    with open(envfile, "w") as fout:
+        fout.write(ENV_FILE)
+    return envfile
+
+
+INI_LOG_CONFIG = """[loggers]
+keys=root
+[handlers]
+keys=h
+[formatters]
+keys=f
+[logger_root]
+level=INFO
+handlers=h
+[handler_h]
+class=StreamHandler
+level=INFO
+formatter=f
+args=(sys.stderr,)
+[formatter_f]
+format=%(asctime)s %(name)s %(levelname)-4s %(message)s
+"""
+
+
+@pytest.fixture(scope="function")
+def ini_log_config(tmp_path):
+    inifile = str(tmp_path / "log_config.ini")
+    with open(inifile, "w") as fout:
+        fout.write(INI_LOG_CONFIG)
+    return inifile

tests/test_config.py

@@ -1,9 +1,12 @@
+import os
+import platform
 import socket
+import sys
 
 import pytest
 
 from uvicorn import protocols
-from uvicorn.config import Config
+from uvicorn.config import LOG_LEVELS, Config
 from uvicorn.middleware.debug import DebugMiddleware
 from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
 from uvicorn.middleware.wsgi import WSGIMiddleware
@@ -66,3 +69,115 @@ def test_ssl_config(certfile_and_keyfile):
     config.load()
 
     assert config.is_ssl is True
+
+
+def test_env_file(env_file):
+    config = Config(app=asgi_app, env_file=env_file)
+    config.load()
+    assert bool(os.environ.get("KEY_TRUE"))
+    assert not bool(os.environ.get("KEY_FALSE"))
+    assert os.environ.get("KEY_NOT_EXISTS") is None
+    # you'd love that a beefy desktop !
+    assert int(os.environ.get("WEB_CONCURRENCY")) == 2048
+    assert config.workers == 2048
+
+
+def test_reload_dir(tmp_path):
+    config = Config(app=asgi_app, reload_dirs=tmp_path)
+    config.load()
+    assert config.reload_dirs == tmp_path
+
+
+def test_forwarded_allow_ips():
+    config = Config(app=asgi_app, forwarded_allow_ips="192.168.0.1")
+    config.load()
+    assert config.forwarded_allow_ips == "192.168.0.1"
+
+
+@pytest.mark.parametrize("use_colors", [(True), (False)])
+def test_log_config_use_colors(use_colors):
+    log_config = {
+        "version": 1,
+        "disable_existing_loggers": False,
+        "formatters": {"default": {}, "access": {}},
+    }
+    config = Config(app=asgi_app, log_config=log_config, use_colors=use_colors)
+    config.load()
+    assert config.use_colors == use_colors
+
+
+def test_log_config_inifile(ini_log_config):
+    config = Config(app=asgi_app, log_config=ini_log_config)
+    config.load()
+    assert config
+
+
+log_lvl_passed = [(k) for k, v in LOG_LEVELS.items()] + [
+    (v) for k, v in LOG_LEVELS.items()
+]
+
+
+@pytest.mark.parametrize("log_lvl_passed", log_lvl_passed)
+def test_log_level_set_as_str_or_int(log_lvl_passed,):
+    config = Config(app=asgi_app, log_level=log_lvl_passed)
+    config.load()
+    assert config.log_level == log_lvl_passed
+
+
+def test_log_access():
+    config = Config(app=asgi_app, access_log=False)
+    config.load()
+    assert not config.access_log
+
+
+def test_fail_asgi_app_import_and_exit():
+    asgi_app_wrong = ""
+    config = Config(app=asgi_app_wrong)
+    with pytest.raises(SystemExit) as pytest_wrapped_e:
+        config.load()
+    assert pytest_wrapped_e.type == SystemExit
+    assert pytest_wrapped_e.value.code == 1
+
+
+def test_should_reload_property():
+    config = Config(app="tests.test_config:asgi_app", reload=True)
+    config.load()
+    assert config.should_reload
+
+
+@pytest.mark.skipif(
+    sys.platform.startswith("win") or platform.python_implementation() == "PyPy",
+    reason="Skipping unix domain tests on Windows and PyPy",
+)
+def test_config_unix_domain_socket(tmp_path):
+    uds = tmp_path / "socket"
+    config = Config(app=asgi_app, uds=uds)
+    config.load()
+    assert config.uds == uds
+
+
+@pytest.mark.skipif(
+    sys.platform.startswith("win") or platform.python_implementation() == "PyPy",
+    reason="Skipping file descriptor tests on Windows and PyPy",
+)
+def test_config_file_descriptor():
+    config = Config(app=asgi_app, fd=1)
+    config.load()
+    assert config.fd == 1
+
+
+@pytest.mark.skipif(
+    sys.platform.startswith("win") or platform.python_implementation() == "PyPy",
+    reason="Skipping unix domain tests on Windows and PyPy",
+)
+def test_config_rebind_socket():
+    sock = socket.socket()
+    config = Config(asgi_app)
+    try:
+        sock.bind((config.host, config.port))
+        with pytest.raises(SystemExit) as pytest_wrapped_e:
+            config.bind_socket()
+        assert pytest_wrapped_e.type == SystemExit
+        assert pytest_wrapped_e.value.code == 1
+    finally:
+        sock.close()

tests/test_main.py

@@ -1,7 +1,11 @@
 import asyncio
+import platform
+import socket
+import sys
 import threading
 import time
 
+import pytest
 import requests
 
 from uvicorn.config import Config
@@ -121,3 +125,83 @@ def safe_run():
     server.should_exit = True
     thread.join()
     assert exc is None
+
+
+@pytest.mark.skipif(
+    sys.platform.startswith("win") or platform.python_implementation() == "PyPy",
+    reason="Skipping uds test on Windows and pypy",
+)
+def test_run_uds(tmp_path):
+    class App:
+        def __init__(self, scope):
+            if scope["type"] != "http":
+                raise Exception()
+
+        async def __call__(self, receive, send):
+            await send({"type": "http.response.start", "status": 204, "headers": []})
+            await send({"type": "http.response.body", "body": b"", "more_body": False})
+
+    class CustomServer(Server):
+        def install_signal_handlers(self):
+            pass
+
+    uds = str(tmp_path / "socket")
+    config = Config(app=App, loop="asyncio", limit_max_requests=1, uds=uds)
+    server = CustomServer(config=config)
+    thread = threading.Thread(target=server.run)
+    thread.start()
+    while not server.started:
+        time.sleep(0.01)
+    data = b"GET / HTTP/1.1\r\n\r\n"
+    sock_client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    try:
+        sock_client.connect(uds)
+        r = sock_client.sendall(data)
+        assert r is None
+    except Exception as e:
+        print(e)
+    finally:
+        sock_client.close()
+    thread.join()
+
+
+@pytest.mark.skipif(
+    sys.platform.startswith("win") or platform.python_implementation() == "PyPy",
+    reason="Skipping fd test on Windows and pypy",
+)
+def test_run_fd(tmp_path):
+    class App:
+        def __init__(self, scope):
+            if scope["type"] != "http":
+                raise Exception()
+
+        async def __call__(self, receive, send):
+            await send({"type": "http.response.start", "status": 204, "headers": []})
+            await send({"type": "http.response.body", "body": b"", "more_body": False})
+
+    class CustomServer(Server):
+        def install_signal_handlers(self):
+            pass
+
+    uds = str(tmp_path / "socket")
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    fd = sock.fileno()
+    sock.bind(uds)
+    config = Config(app=App, loop="asyncio", limit_max_requests=1, fd=fd)
+    server = CustomServer(config=config)
+    thread = threading.Thread(target=server.run)
+    thread.start()
+    while not server.started:
+        time.sleep(0.01)
+    data = b"GET / HTTP/1.1\r\n\r\n"
+    sock_client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    try:
+        sock_client.connect(uds)
+        r = sock_client.sendall(data)
+        assert r is None
+    except Exception as e:
+        print(e)
+    finally:
+        sock_client.close()
+        sock.close()
+    thread.join()

uvicorn/config.py

@@ -207,20 +207,6 @@ def is_ssl(self) -> bool:
     def configure_logging(self):
         logging.addLevelName(TRACE_LOG_LEVEL, "TRACE")
 
-        if sys.version_info < (3, 7):
-            # https://bugs.python.org/issue30520
-            import pickle
-
-            def __reduce__(self):
-                if isinstance(self, logging.RootLogger):
-                    return logging.getLogger, ()
-
-                if logging.getLogger(self.name) is not self:
-                    raise pickle.PicklingError("logger cannot be pickled")
-                return logging.getLogger, (self.name,)
-
-            logging.Logger.__reduce__ = __reduce__
-
         if self.log_config is not None:
             if isinstance(self.log_config, dict):
                 if self.use_colors in (True, False):
@@ -232,7 +218,9 @@ def __reduce__(self):
                     ] = self.use_colors
                 logging.config.dictConfig(self.log_config)
             else:
-                logging.config.fileConfig(self.log_config)
+                logging.config.fileConfig(
+                    self.log_config, disable_existing_loggers=False
+                )
 
         if self.log_level is not None:
             if isinstance(self.log_level, str):