File Storage Test: Added tests for file storage and refactored static file mounting

Author: eadwinCode

ellar/app/factory.py

@@ -167,8 +167,7 @@ def _get_config_kwargs() -> t.Dict:
 
         if module_changed:
             app.router.extend(routes)
-            app.reload_static_app()
-            app.rebuild_middleware_stack()
+            app.rebuild_stack()
 
         return app
 

ellar/app/main.py

@@ -31,11 +31,12 @@
     ModuleSetup,
     ModuleTemplateRef,
 )
-from ellar.core.routing import ApplicationRouter
+from ellar.core.routing import ApplicationRouter, AppStaticFileMount
 from ellar.core.services import Reflector
 from ellar.core.versioning import BaseAPIVersioning, VersioningSchemes
 from ellar.di import EllarInjector
-from starlette.routing import BaseRoute, Mount
+from jinja2 import Environment as JinjaEnvironment
+from starlette.routing import BaseRoute
 
 from .lifespan import EllarApplicationLifespan
 from .mixin import AppMixin
@@ -73,6 +74,7 @@ def __init__(
         self.config.DEFAULT_LIFESPAN_HANDLER = (
             lifespan or self.config.DEFAULT_LIFESPAN_HANDLER
         )
+
         self.router = ApplicationRouter(
             routes=self._get_module_routes(),
             redirect_slashes=self.config.REDIRECT_SLASHES,
@@ -81,10 +83,20 @@ def __init__(
                 self.config.DEFAULT_LIFESPAN_HANDLER  # type: ignore[arg-type]
             ).lifespan,
         )
+        if (
+            self.config.STATIC_MOUNT_PATH
+            and self.config.STATIC_MOUNT_PATH not in self.router.routes
+        ):
+            self.router.append(AppStaticFileMount(self))
+
         self._finalize_app_initialization()
         self.middleware_stack = self.build_middleware_stack()
         self._config_logging()
 
+        self.reload_event_manager += lambda app: self._update_jinja_env_filters(  # type:ignore[misc]
+            self.jinja_environment
+        )
+
     def _config_logging(self) -> None:
         log_level = (
             self.config.LOG_LEVEL.value
@@ -117,15 +129,6 @@ async def _statics_func_wrapper(
 
     def _get_module_routes(self) -> t.List[BaseRoute]:
         _routes: t.List[BaseRoute] = []
-        if self.has_static_files:
-            self._static_app = self.create_static_app()
-            _routes.append(
-                Mount(
-                    str(self.config.STATIC_MOUNT_PATH),
-                    app=self._statics_wrapper(),
-                    name="static",
-                )
-            )
 
         for _, module_ref in self._injector.get_modules().items():
             _routes.extend(module_ref.routes)
@@ -157,9 +160,8 @@ def install_module(
         module_ref.run_module_register_services()
         if isinstance(module_ref, ModuleTemplateRef):
             self.router.extend(module_ref.routes)
-            self.reload_static_app()
 
-        self.rebuild_middleware_stack()
+        self.rebuild_stack()
 
         return t.cast(T, module_ref.get_module_instance())
 
@@ -189,12 +191,6 @@ def injector(self) -> EllarInjector:
     def versioning_scheme(self) -> BaseAPIVersioning:
         return t.cast(BaseAPIVersioning, self._config.VERSIONING_SCHEME)
 
-    @property
-    def has_static_files(self) -> bool:  # type: ignore
-        return (
-            True if self.static_files or self.config.STATIC_FOLDER_PACKAGES else False
-        )
-
     @property
     def config(self) -> "Config":
         return self._config
@@ -298,7 +294,9 @@ def _finalize_app_initialization(self) -> None:
         self.injector.container.register_instance(self)
         self.injector.container.register_instance(self.config, Config)
         self.injector.container.register_instance(self.jinja_environment, Environment)
-        self.injector.container.register_instance(self.jinja_environment, Environment)
+        self.injector.container.register_instance(
+            self.jinja_environment, JinjaEnvironment
+        )
 
     def add_exception_handler(
         self,
@@ -310,10 +308,11 @@ def add_exception_handler(
                 self._exception_handlers.append(exception_handler)
                 _added_any = True
         if _added_any:
-            self.rebuild_middleware_stack()
+            self.rebuild_stack()
 
-    def rebuild_middleware_stack(self) -> None:
+    def rebuild_stack(self) -> None:
         self.middleware_stack = self.build_middleware_stack()
+        self.reload_event_manager.run(self)
 
     @property
     def reflector(self) -> Reflector:

ellar/app/mixin.py

@@ -1,6 +1,5 @@
 import json
 import typing as t
-from abc import abstractmethod
 
 from ellar.common.compatible import cached_property
 from ellar.common.constants import TEMPLATE_FILTER_KEY, TEMPLATE_GLOBAL_KEY
@@ -14,8 +13,8 @@
 from ellar.common.types import ASGIApp
 from ellar.core.conf import Config
 from ellar.core.connection import Request
-from ellar.core.staticfiles import StaticFiles
 from ellar.di import EllarInjector
+from ellar.events import EventManager
 from jinja2 import Environment as BaseEnvironment
 from starlette.templating import pass_context
 
@@ -27,15 +26,8 @@ class AppMixin(JinjaTemplating):
     _static_app: t.Optional[ASGIApp]
     _injector: EllarInjector
     _config: Config
-    has_static_files: bool
-
-    @abstractmethod
-    def build_middleware_stack(self) -> t.Callable:  # pragma: no cover
-        pass
-
-    @abstractmethod
-    def rebuild_middleware_stack(self) -> None:  # pragma: no cover
-        pass
+    rebuild_stack: t.Callable
+    reload_event_manager = EventManager()
 
     def get_module_loaders(self) -> t.Generator[ModuleTemplating, None, None]:
         for loader in self._injector.get_templating_modules().values():
@@ -50,7 +42,7 @@ def debug(self, value: bool) -> None:
         del self.__dict__["jinja_environment"]
         self._config.DEBUG = value
         # TODO: Add warning
-        self.rebuild_middleware_stack()
+        self.rebuild_stack()
 
     @cached_property
     def jinja_environment(self) -> BaseEnvironment:  # type: ignore[override]
@@ -94,26 +86,6 @@ def url_for(context: dict, name: str, **path_params: t.Any) -> URL:
     def create_global_jinja_loader(self) -> JinjaLoader:
         return JinjaLoader(t.cast("App", self))
 
-    def create_static_app(self) -> ASGIApp:
-        return StaticFiles(
-            directories=self.static_files,  # type: ignore[arg-type]
-            packages=self._config.STATIC_FOLDER_PACKAGES,
-        )
-
-    def reload_static_app(self) -> None:
-        del self.__dict__["static_files"]
-        if self.has_static_files:
-            self._static_app = self.create_static_app()
-        self._update_jinja_env_filters(self.jinja_environment)
-
     def _update_jinja_env_filters(self, jinja_environment: BaseEnvironment) -> None:
         jinja_environment.globals.update(self._config.get(TEMPLATE_GLOBAL_KEY, {}))
         jinja_environment.filters.update(self._config.get(TEMPLATE_FILTER_KEY, {}))
-
-    @cached_property
-    def static_files(self) -> t.List[str]:
-        static_directories = t.cast(t.List, self._config.STATIC_DIRECTORIES or [])
-        for module in self.get_module_loaders():
-            if module.static_directory:
-                static_directories.append(module.static_directory)
-        return static_directories

ellar/common/datastructures.py

@@ -1,4 +1,5 @@
 import typing as t
+from io import BytesIO, StringIO
 
 from ellar.pydantic import as_pydantic_validator
 from starlette.datastructures import (
@@ -36,6 +37,7 @@
     "UploadFile",
     "URLPath",
     "State",
+    "ContentFile",
 ]
 
 
@@ -89,3 +91,18 @@ def __validate_input__(cls, __input_value: t.Any, _: t.Any) -> "UploadFile":
             filename=__input_value.filename,
             headers=__input_value.headers,
         )
+
+
+class ContentFile(UploadFile):
+    """
+    A File-like object that takes just raw content, rather than an actual file.
+    """
+
+    def __init__(
+        self, content: t.Union[str, bytes], name: t.Optional[str] = None
+    ) -> None:
+        stream_class = StringIO if isinstance(content, str) else BytesIO
+        headers = Headers({"content-type": "text/plain"})
+        super().__init__(
+            stream_class(content), filename=name, size=len(content), headers=headers
+        )

ellar/common/routing/route_collections.py

@@ -15,6 +15,9 @@ def __init__(self, routes: t.Optional[t.Sequence[BaseRoute]] = None) -> None:
         self._served_routes: t.List[BaseRoute] = []
         self.extend([] if routes is None else list(routes))
 
+    def __contains__(self, item: t.Any) -> bool:
+        return item in self._routes
+
     @t.no_type_check
     def __getitem__(self, i: int) -> BaseRoute:
         return self._served_routes.__getitem__(i)

ellar/core/conf/app_settings_models.py

@@ -75,7 +75,7 @@ class ConfigValidationSchema(Serializer, ConfigDefaultTypesMixin):
 
     STATIC_DIRECTORIES: t.Optional[t.List[t.Union[str, t.Any]]] = []
 
-    STATIC_MOUNT_PATH: str = "/static"
+    STATIC_MOUNT_PATH: t.Optional[str] = "/static"
 
     CORS_ALLOW_ORIGINS: t.List[str] = []
     CORS_ALLOW_METHODS: t.List[str] = ["GET"]
@@ -139,7 +139,8 @@ def serializer_custom_encoder(cls, value: t.Any) -> t.Any:
 
     @field_validator("STATIC_MOUNT_PATH", mode="before")
     def pre_static_mount_path(cls, value: t.Any) -> t.Any:
-        assert value.startswith("/"), "Routed paths must start with '/'"
+        if value:
+            assert value.startswith("/"), "Routed paths must start with '/'"
         return value
 
     @field_validator("CACHES", mode="before")

ellar/core/conf/mixins.py

@@ -50,7 +50,7 @@ class ConfigDefaultTypesMixin:
     EXCEPTION_HANDLERS: t.List[IExceptionHandler]
 
     # static route
-    STATIC_MOUNT_PATH: str
+    STATIC_MOUNT_PATH: t.Optional[str]
 
     # defines other custom json encoders
     SERIALIZER_CUSTOM_ENCODER: t.Dict[t.Any, t.Callable[[t.Any], t.Any]]

ellar/core/files/storages/aws_s3.py

@@ -1,17 +1,18 @@
+import typing as t
+import urllib.parse
+from io import BytesIO
+
 try:
     import boto3
 except ImportError as im_ex:  # pragma: no cover
     raise RuntimeError(
         "boto3 must be installed to use the 'S3AWSFileStorage' class."
     ) from im_ex
-import typing as t
-import urllib.parse
-from io import BytesIO
 
 from .base import BaseStorage
 
 
-class S3AWSFileStorage(BaseStorage):
+class S3AWSFileStorage(BaseStorage):  # pragma: no cover
     def service_name(self) -> str:
         return "s3_bucket"
 
@@ -66,8 +67,9 @@ def get_aws_bucket(
 
     def get_s3_path(self, filename: str) -> str:
         if self._prefix:
-            return "{0}/{1}".format(self._prefix, filename)
-        return filename
+            filename = "{0}/{1}".format(self._prefix, filename)
+        self.validate_file_name(filename)
+        return self.generate_filename(filename)
 
     def _upload_file(
         self, filename: str, data: str, content_type: t.Optional[str], rrs: bool = False

ellar/core/files/storages/base.py

@@ -2,7 +2,7 @@
 import pathlib
 from abc import ABC
 
-from .exceptions import SuspiciousFileOperation
+from .exceptions import UnsafeFileOperation
 from .interface import Storage
 from .utils import get_valid_filename, validate_file_name
 
@@ -26,7 +26,7 @@ def generate_filename(self, filename: str) -> str:
         # `filename` may include a path as returned by FileField.upload_to.
         dirname, filename = os.path.split(filename)
         if ".." in pathlib.PurePath(dirname).parts:
-            raise SuspiciousFileOperation(
+            raise UnsafeFileOperation(
                 "Detected path traversal attempt in '%s'" % dirname
             )
         return os.path.normpath(os.path.join(dirname, get_valid_filename(filename)))

ellar/core/files/storages/exceptions.py

@@ -1,8 +1,8 @@
-class SuspiciousOperation(Exception):
+class UnsafeOperation(Exception):
     """The user did something suspicious"""
 
 
-class SuspiciousFileOperation(SuspiciousOperation):
+class UnsafeFileOperation(UnsafeOperation):
     """A Suspicious filesystem operation was attempted"""
 
     pass