refactor: config logic

Author: rexzhang

asgi_webdav/config.py

@@ -27,7 +27,6 @@
     AppEntryParameters,
     DAVCompressLevel,
 )
-from asgi_webdav.exception import DAVExceptionConfigFileNotFound
 
 logger = getLogger(__name__)
 
@@ -300,14 +299,7 @@ def update_from_app_args_and_env_and_default_value(
         self._complete_config()
 
 
-_config: Config = Config()
-
-
-def get_config() -> Config:
-    return _config
-
-
-def get_config_copy_from_dict(
+def generate_config_from_dict(
     data: dict[str, Any], complete_config: bool = False
 ) -> Config:
     config = Config.from_dict(data)
@@ -317,21 +309,13 @@ def get_config_copy_from_dict(
     return config
 
 
-def reinit_config_from_dict(
-    data: dict[str, Any], complete_config: bool = False
-) -> Config:
-    global _config
-
-    logger.debug("Load config value from python object(dict)")
-    _config = get_config_copy_from_dict(data, complete_config)
-
-    return _config
-
-
-def reinit_config_from_file(file_name: str, complete_config: bool = False) -> bool:
+def generate_config_from_file(
+    file: Path | str, complete_config: bool = False
+) -> Config | None:
     load_func: Callable[[Any], Any]
+    if not isinstance(file, Path):
+        file = Path(file)
 
-    file = Path(file_name)
     match file.suffix:
         case ".json":
             load_func = json.load
@@ -340,54 +324,75 @@ def reinit_config_from_file(file_name: str, complete_config: bool = False) -> bo
         case _:
             message = f"Unsupported config file type: {file.suffix}"
             logger.error(message)
-            return False
+            return None
 
     try:
         with open(file, "rb") as f:
             data = load_func(f)
 
     except FileNotFoundError as e:
-        message = f"Can not open config file[{file}]!"
-        logger.error(message)
-        logger.error(e)
-        # return False
-        raise DAVExceptionConfigFileNotFound(e)
+        logger.error(f"Can not open config file[{file}]!, {e}")
+        return None
 
     except (json.JSONDecodeError, tomllib.TOMLDecodeError) as e:
         message = f"Load config from file[{file}] failed!"
         logger.error(message)
         logger.error(e)
-        return False
+        return None
 
-    reinit_config_from_dict(data, complete_config)
+    config = generate_config_from_dict(data, complete_config)
     logger.info(f"Load config from file: [{file}] success!")
-    return True
+    return config
 
 
-def reinit_config_from_file_multi_suffix(
-    file_name: str, complete_config: bool = False
-) -> bool:
+def generate_config_from_file_with_multi_suffix(
+    file: Path | str, complete_config: bool = False
+) -> Config | None:
     """help users in switching from .json to .toml configuration files."""
 
-    try:
-        return reinit_config_from_file(file_name, complete_config)
-    except DAVExceptionConfigFileNotFound:
-        logger.warning(f"Can not found config file: {file_name}!")
+    if not isinstance(file, Path):
+        file = Path(file)
+
+    config = generate_config_from_file(file, complete_config)
+    if config is not None:
+        return config
 
     # try other suffix
-    file = Path(file_name)
     stem = file.stem
     suffix = file.suffix
 
     suffixs = {".json", ".toml"}
-    suffixs.remove(suffix)
+    try:
+        suffixs.remove(suffix)
+    except KeyError as e:
+        logger.warning(f"Wrong file extension: {suffix}, {e}")
 
     for suffix in suffixs:
-        file_name = file.parent.joinpath(f"{stem}{suffix}").as_posix()
-        logger.warning(f"Try load config file: {file_name}!")
-        try:
-            return reinit_config_from_file(file_name, complete_config)
-        except DAVExceptionConfigFileNotFound:
-            logger.warning(f"Can not found config file: {file_name}!")
-
-    return False
+        file = file.parent.joinpath(f"{stem}{suffix}")
+        logger.warning(f"Try load config file: {file}!")
+
+        config = generate_config_from_file(file, complete_config)
+        if config is None:
+            logger.warning(f"Can not found config file: {file}!")
+        else:
+            # loaded
+            return config
+
+    # all failed
+    return None
+
+
+_config: Config = Config()
+
+
+def get_config() -> Config:
+    return _config
+
+
+def reinit_global_config(config: Config | None = None) -> None:
+    global _config
+
+    if config is None:
+        config = Config()
+
+    _config = config

asgi_webdav/server.py

@@ -12,9 +12,10 @@
 from asgi_webdav.auth import DAVAuth
 from asgi_webdav.config import (
     Config,
+    generate_config_from_dict,
+    generate_config_from_file_with_multi_suffix,
     get_config,
-    reinit_config_from_dict,
-    reinit_config_from_file_multi_suffix,
+    reinit_global_config,
 )
 from asgi_webdav.constants import AppEntryParameters, DAVMethod, DevMode
 from asgi_webdav.exception import DAVExceptionProviderInitFailed
@@ -109,14 +110,19 @@ def get_asgi_app(aep: AppEntryParameters, config_obj: dict[str, Any] | None = No
     logging.config.dictConfig(get_dav_logging_config(config=get_config()))
 
     # init config
+    config: Config | None = None
     if aep.config_file is not None:
-        reinit_config_from_file_multi_suffix(aep.config_file)
+        config = generate_config_from_file_with_multi_suffix(aep.config_file)
     elif config_obj is not None:
-        reinit_config_from_dict(config_obj)
+        config = generate_config_from_dict(config_obj)
+
+    if config is None:
+        raise
 
-    config = get_config()
     config.update_from_app_args_and_env_and_default_value(aep=aep)
 
+    reinit_global_config(config)
+
     # init logging
     if config.logging.enable:
         logging.config.dictConfig(get_dav_logging_config(config=config))

examples/config/anonymous-default.toml examples/config/anonymous-enable.tomlexamples/config/anonymous-default.toml renamed to examples/config/anonymous-enable.toml

@@ -0,0 +1 @@
+decode failed

examples/config/decode-failed.json

@@ -6,11 +6,11 @@
 
 from asgi_webdav.auth import DAVAuth, DAVPassword, DAVPasswordType
 from asgi_webdav.cache import DAVCacheType
-from asgi_webdav.config import Config, get_config_copy_from_dict
+from asgi_webdav.config import Config, generate_config_from_dict
 from asgi_webdav.constants import DAVPath, DAVUser
 from asgi_webdav.request import DAVRequest
 
-from .asgi_test_kit import ASGITestClient, create_dav_request_object, get_webdav_app
+from .testkit_asgi import ASGITestClient, create_dav_request_object, get_webdav_app
 
 USERNAME = "username"
 PASSWORD = "password"
@@ -431,7 +431,7 @@ def get_dav_request(extra_headers: dict[str, str]) -> DAVRequest:
 @pytest.mark.asyncio
 async def test_dav_auth_pick_out_user_anonymous_user():
     # anonymous user : ok
-    config = get_config_copy_from_dict(
+    config = generate_config_from_dict(
         BASIC_AUTHORIZATION_CONFIG_DATA_FOR_ANONYMOUS_USER_DEFAULT, complete_config=True
     )
     dav_auth = DAVAuth(config)
@@ -456,7 +456,7 @@ async def test_dav_auth_pick_out_user_anonymous_user():
     assert message is not None
 
     # anonymous user : disable
-    config = get_config_copy_from_dict(
+    config = generate_config_from_dict(
         BASIC_AUTHORIZATION_CONFIG_DATA_FOR_ANONYMOUS_USER_DISABLE, complete_config=True
     )
     dav_auth = DAVAuth(config)
@@ -474,7 +474,7 @@ async def test_dav_auth_pick_out_user_anonymous_user():
     assert message is not None
 
     # anonymous user: allow_missing_auth_header is False
-    config = get_config_copy_from_dict(
+    config = generate_config_from_dict(
         BASIC_AUTHORIZATION_CONFIG_DATA_FOR_ANONYMOUS_USER_ALLOW_MISSING_AUTH_HEADER_FALSE,
         complete_config=True,
     )

examples/config/no-support.ext

@@ -0,0 +1,143 @@
+import pytest
+
+from asgi_webdav.config import (
+    Config,
+    LoggingLevel,
+    generate_config_from_dict,
+    generate_config_from_file,
+    generate_config_from_file_with_multi_suffix,
+    get_config,
+    reinit_global_config,
+)
+from asgi_webdav.constants import (
+    DEFAULT_PASSWORD_ANONYMOUS,
+    DEFAULT_PERMISSIONS,
+    DEFAULT_USERNAME_ANONYMOUS,
+)
+
+from .testkit_common import get_project_root_path
+
+EXAMPLE_CONFIG_ROOT_PATH = get_project_root_path().joinpath("examples/config")
+
+TEST_PERMISSIONS = ["+^/$"]
+
+
+@pytest.fixture
+def default_config():
+    config = Config()
+
+    yield config
+
+    pass
+
+
+def test_defalut_config(default_config):
+    config = default_config
+    assert config.anonymous.enable is False
+    assert config.anonymous.user.username == DEFAULT_USERNAME_ANONYMOUS
+    assert config.anonymous.user.password == DEFAULT_PASSWORD_ANONYMOUS
+    assert config.anonymous.user.permissions == DEFAULT_PERMISSIONS
+
+    assert config.logging.enable is True
+    assert config.logging.level == LoggingLevel.INFO
+
+
+def test_generate_config_from_dict():
+    config = generate_config_from_dict(
+        {
+            "anonymous": {
+                "enable": True,
+                "user": {
+                    "username": "test_user",
+                    "password": "test_password",
+                    "permissions": TEST_PERMISSIONS,
+                },
+            },
+            "logging": {
+                "enable": False,
+                "level": "DEBUG",
+            },
+        }
+    )
+
+    assert config.anonymous.enable is True
+    assert config.anonymous.user.username == "test_user"
+    assert config.anonymous.user.password == "test_password"
+    assert config.anonymous.user.permissions == TEST_PERMISSIONS
+
+
+def test_generate_config_from_file():
+
+    config = generate_config_from_file(
+        EXAMPLE_CONFIG_ROOT_PATH.joinpath("anonymous-limited-permission.toml")
+    )
+
+    assert config.anonymous.enable is True
+    assert config.anonymous.user.permissions == TEST_PERMISSIONS
+
+    # no file
+    assert generate_config_from_file("no-file.toml") is None
+    assert generate_config_from_file("no-file.json") is None
+
+    # not support file
+    assert (
+        generate_config_from_file(EXAMPLE_CONFIG_ROOT_PATH.joinpath("no-support.ext"))
+        is None
+    )
+
+    # decode failed
+    assert (
+        generate_config_from_file(
+            EXAMPLE_CONFIG_ROOT_PATH.joinpath("decode-failed.json")
+        )
+        is None
+    )
+
+
+def test_generate_config_from_file_with_multi_suffix():
+    assert (
+        generate_config_from_file_with_multi_suffix(
+            EXAMPLE_CONFIG_ROOT_PATH.joinpath("anonymous-enable.toml")
+        ).anonymous.enable
+        is True
+    )
+
+    assert (
+        generate_config_from_file_with_multi_suffix(
+            EXAMPLE_CONFIG_ROOT_PATH.joinpath("anonymous-enable.toml").as_posix()
+        ).anonymous.enable
+        is True
+    )
+
+    assert (
+        generate_config_from_file_with_multi_suffix(
+            EXAMPLE_CONFIG_ROOT_PATH.joinpath("anonymous-enable.json")
+        ).anonymous.enable
+        is True
+    )
+
+    assert (
+        generate_config_from_file_with_multi_suffix(
+            EXAMPLE_CONFIG_ROOT_PATH.joinpath("no-support.ext")
+        )
+        is None
+    )
+
+
+def test_global_config():
+    # reset
+    reinit_global_config()
+
+    config = get_config()
+    assert config.anonymous.enable is False
+
+    # set
+    reinit_global_config(
+        generate_config_from_file_with_multi_suffix(
+            EXAMPLE_CONFIG_ROOT_PATH.joinpath("anonymous-enable.toml")
+        )
+    )
+
+    # check
+    config = get_config()
+    assert config.anonymous.enable is True

tests/test_auth.py

@@ -5,7 +5,7 @@
 from asgi_webdav.constants import DAVPath
 from asgi_webdav.lock import DAVLock, DAVLockInfo, DAVLockScope
 
-from .asgi_test_kit import create_dav_request_object
+from .testkit_asgi import create_dav_request_object
 
 
 def create_request(path: str):

tests/test_config.py

@@ -3,7 +3,7 @@
 
 from asgi_webdav.middleware.cors import ASGIMiddlewareCORS
 
-from .asgi_test_kit import ASGIApp, ASGITestClient
+from .testkit_asgi import ASGIApp, ASGITestClient
 
 
 def get_middleware_app(middleware, **kwargs):