Coverage for src/lexigram/admin/middleware/setup.py: 0%

1"""Middleware for initial setup redirection.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from starlette.responses import RedirectResponse 

8 

9from lexigram.admin.auth.store.protocols import AdminUserStoreProtocol 

10from lexigram.logging import get_logger 

11 

12if TYPE_CHECKING: 

13 from collections.abc import Callable 

14 

15logger = get_logger(__name__) 

16 

17 

18class SetupMiddleware: 

19 """Redirect to the first-run setup wizard when no admin accounts exist. 

20 

21 This middleware intercepts every HTTP request to the admin sub-app and 

22 checks whether at least one admin-panel account has been created. If the 

23 ``admin_users`` table is empty it issues a 302 to ``{prefix}/setup`` so 

24 the operator is guided through initial configuration. 

25 

26 The middleware is mounted by 

27 :class:`~lexigram.admin.di.bundle_provider.AdminProvider` during 

28 :meth:`~lexigram.admin.di.bundle_provider.AdminProvider.mount_to_app` 

29 and receives a fully-resolved :class:`AdminUserStoreProtocol` instance — 

30 no duck-typing, no ``hasattr`` guards. 

31 

32 Paths that are always allowed through (bypass the redirect): 

33 - ``/setup`` and any path ending in ``/setup`` 

34 - ``/static`` paths (asset serving must never redirect) 

35 """ 

36 

37 def __init__( 

38 self, 

39 app: Callable, 

40 admin_user_store: AdminUserStoreProtocol, 

41 ) -> None: 

42 self.app = app 

43 self._store = admin_user_store 

44 

45 async def __call__( 

46 self, 

47 scope: dict[str, Any], 

48 receive: Callable, 

49 send: Callable, 

50 ) -> None: 

51 if scope["type"] != "http": 

52 await self.app(scope, receive, send) 

53 return 

54 

55 path = scope.get("path", "") 

56 if ( 

57 path == "/setup" 

58 or path.endswith("/setup") 

59 or path in ("/login", "/login/") 

60 or path in ("/logout", "/logout/") 

61 or path.startswith("/static") 

62 or "/static/" in path 

63 ): 

64 logger.debug("setup_middleware.skip path=%s", path) 

65 await self.app(scope, receive, send) 

66 return 

67 

68 prefix = scope.get("root_path", "") 

69 try: 

70 admin_count = await self._store.get_admin_count() 

71 logger.debug( 

72 "setup_middleware.check admin_count=%s path=%s", admin_count, path 

73 ) 

74 if admin_count == 0: 

75 logger.debug("setup_middleware.redirect_to_setup path=%s", path) 

76 response = RedirectResponse(url=f"{prefix}/setup") 

77 await response(scope, receive, send) 

78 return 

79 except ( 

80 RuntimeError, 

81 ValueError, 

82 AttributeError, 

83 ConnectionError, 

84 OSError, 

85 ) as e: 

86 logger.warning("setup_middleware.count_failed error=%s degraded", e) 

87 # Degrade gracefully: let the request through. A transient DB 

88 # failure does not mean "no users exist" — it means "can't check." 

89 # AuthMiddleware and SetupController will handle their own errors. 

90 await self.app(scope, receive, send) 

91 return 

92 

93 await self.app(scope, receive, send)