diff --git a/admin_app/api/dependencies.py b/admin_app/api/dependencies.py index d4df90a..efd053c 100644 --- a/admin_app/api/dependencies.py +++ b/admin_app/api/dependencies.py @@ -1,7 +1,8 @@ -from fastapi import Depends, HTTPException, Request, status +from fastapi import Depends, HTTPException, Request, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy.orm import Session +from admin_app.api.panel_session import get_panel_access_cookie from admin_app.core import ( AdminSecurityService, AdminSettings, @@ -85,18 +86,72 @@ def get_current_staff_context( ) from exc +def get_current_panel_staff_context( + request: Request, + auth_service: AuthService = Depends(get_auth_service), +) -> AuthenticatedStaffContext: + access_token = get_panel_access_cookie(request) + if not access_token: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Sessao administrativa web obrigatoria.", + ) + + try: + return auth_service.get_authenticated_context(access_token) + except ValueError as exc: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Sessao administrativa web invalida.", + ) from exc + + +def get_optional_panel_staff_context( + request: Request, + auth_service: AuthService = Depends(get_auth_service), +) -> AuthenticatedStaffContext | None: + access_token = get_panel_access_cookie(request) + if not access_token: + return None + + try: + return auth_service.get_authenticated_context(access_token) + except ValueError: + return None + + def get_current_staff_principal( context: AuthenticatedStaffContext = Depends(get_current_staff_context), ) -> AuthenticatedStaffPrincipal: return context.principal +def get_current_panel_staff_principal( + context: AuthenticatedStaffContext = Depends(get_current_panel_staff_context), +) -> AuthenticatedStaffPrincipal: + return context.principal + + +def get_optional_panel_staff_principal( + context: AuthenticatedStaffContext | None = Depends(get_optional_panel_staff_context), +) -> AuthenticatedStaffPrincipal | None: + if context is None: + return None + return context.principal + + def get_current_staff_session_id( context: AuthenticatedStaffContext = Depends(get_current_staff_context), ) -> int: return context.session_id +def get_current_panel_staff_session_id( + context: AuthenticatedStaffContext = Depends(get_current_panel_staff_context), +) -> int: + return context.session_id + + def require_staff_role(minimum_role: StaffRole): def dependency( current_staff: AuthenticatedStaffPrincipal = Depends(get_current_staff_principal), @@ -125,7 +180,21 @@ def require_admin_permission(permission: AdminPermission): return dependency +def require_panel_admin_permission(permission: AdminPermission): + def dependency( + current_staff: AuthenticatedStaffPrincipal = Depends(get_current_panel_staff_principal), + ) -> AuthenticatedStaffPrincipal: + if not role_has_permission(current_staff.role, permission): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=f"Permissao administrativa insuficiente: '{permission.value}'.", + ) + return current_staff + + return dependency + + def get_current_staff_permissions( current_staff: AuthenticatedStaffPrincipal = Depends(get_current_staff_principal), ) -> tuple[str, ...]: - return tuple(permission.value for permission in permissions_for_role(current_staff.role)) \ No newline at end of file + return tuple(permission.value for permission in permissions_for_role(current_staff.role)) diff --git a/admin_app/api/panel_session.py b/admin_app/api/panel_session.py new file mode 100644 index 0000000..32ab4f1 --- /dev/null +++ b/admin_app/api/panel_session.py @@ -0,0 +1,67 @@ +from fastapi import Request, Response + +from admin_app.core import AdminAuthenticatedSession, AdminSettings + +PANEL_ACCESS_COOKIE_NAME = "orquestrador_admin_panel_access" +PANEL_REFRESH_COOKIE_NAME = "orquestrador_admin_panel_refresh" +PANEL_COOKIE_SAMESITE = "lax" + + +def get_panel_access_cookie(request: Request) -> str | None: + return request.cookies.get(PANEL_ACCESS_COOKIE_NAME) + + +def get_panel_refresh_cookie(request: Request) -> str | None: + return request.cookies.get(PANEL_REFRESH_COOKIE_NAME) + + +def set_panel_auth_cookies( + response: Response, + session: AdminAuthenticatedSession, + settings: AdminSettings, +) -> None: + cookie_path = build_panel_cookie_path(settings) + use_secure = should_use_secure_cookies(settings) + response.set_cookie( + key=PANEL_ACCESS_COOKIE_NAME, + value=session.access_token, + max_age=session.expires_in_seconds, + httponly=True, + secure=use_secure, + samesite=PANEL_COOKIE_SAMESITE, + path=cookie_path, + ) + response.set_cookie( + key=PANEL_REFRESH_COOKIE_NAME, + value=session.refresh_token, + max_age=settings.admin_auth_refresh_token_ttl_days * 24 * 60 * 60, + httponly=True, + secure=use_secure, + samesite=PANEL_COOKIE_SAMESITE, + path=cookie_path, + ) + + +def clear_panel_auth_cookies(response: Response, settings: AdminSettings) -> None: + cookie_path = build_panel_cookie_path(settings) + response.delete_cookie( + key=PANEL_ACCESS_COOKIE_NAME, + path=cookie_path, + httponly=True, + samesite=PANEL_COOKIE_SAMESITE, + ) + response.delete_cookie( + key=PANEL_REFRESH_COOKIE_NAME, + path=cookie_path, + httponly=True, + samesite=PANEL_COOKIE_SAMESITE, + ) + + +def build_panel_cookie_path(settings: AdminSettings) -> str: + normalized_prefix = settings.admin_api_prefix.rstrip("/") + return normalized_prefix or "/" + + +def should_use_secure_cookies(settings: AdminSettings) -> bool: + return settings.admin_environment.lower() == "production" and not settings.admin_debug diff --git a/admin_app/api/router.py b/admin_app/api/router.py index a7a97c6..edae7a0 100644 --- a/admin_app/api/router.py +++ b/admin_app/api/router.py @@ -2,10 +2,16 @@ from fastapi import APIRouter from admin_app.api.routes.audit import router as audit_router from admin_app.api.routes.auth import router as auth_router +from admin_app.api.routes.panel_auth import router as panel_auth_router +from admin_app.api.routes.panel_tools import router as panel_tools_router from admin_app.api.routes.system import router as system_router +from admin_app.api.routes.tools import router as tools_router # Agrega as rotas do servico administrativo. api_router = APIRouter() api_router.include_router(auth_router) +api_router.include_router(panel_auth_router) +api_router.include_router(panel_tools_router) api_router.include_router(system_router) -api_router.include_router(audit_router) \ No newline at end of file +api_router.include_router(tools_router) +api_router.include_router(audit_router) diff --git a/admin_app/api/routes/panel_auth.py b/admin_app/api/routes/panel_auth.py new file mode 100644 index 0000000..e597f87 --- /dev/null +++ b/admin_app/api/routes/panel_auth.py @@ -0,0 +1,178 @@ +from fastapi import APIRouter, Depends, HTTPException, Request, Response, status + +from admin_app.api.dependencies import ( + get_auth_service, + get_current_panel_staff_context, + get_settings, +) +from admin_app.api.panel_session import ( + clear_panel_auth_cookies, + get_panel_access_cookie, + get_panel_refresh_cookie, + set_panel_auth_cookies, +) +from admin_app.api.schemas import ( + AdminAuthenticatedStaffResponse, + AdminLoginRequest, + AdminPanelLogoutResponse, + AdminPanelWebSessionResponse, +) +from admin_app.core import AdminAuthenticatedSession, AdminSettings, AuthenticatedStaffContext +from admin_app.services import AuthService + +router = APIRouter(prefix="/panel/auth", tags=["panel-auth"]) + + +@router.post("/login", response_model=AdminPanelWebSessionResponse) +def panel_login( + payload: AdminLoginRequest, + request: Request, + response: Response, + settings: AdminSettings = Depends(get_settings), + auth_service: AuthService = Depends(get_auth_service), +): + ip_address, user_agent = _extract_request_metadata(request) + session = auth_service.login( + email=payload.email, + password=payload.password, + ip_address=ip_address, + user_agent=user_agent, + ) + if session is None: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Credenciais administrativas invalidas.", + ) + + set_panel_auth_cookies(response, session, settings) + return _build_panel_session_response( + session=session, + message="Sessao administrativa web iniciada.", + redirect_to=_build_prefixed_path(settings.admin_api_prefix, "/panel/admin"), + ) + + +@router.post("/refresh", response_model=AdminPanelWebSessionResponse) +def panel_refresh( + request: Request, + response: Response, + settings: AdminSettings = Depends(get_settings), + auth_service: AuthService = Depends(get_auth_service), +): + refresh_token = get_panel_refresh_cookie(request) + if not refresh_token: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Sessao administrativa web sem refresh token.", + ) + + ip_address, user_agent = _extract_request_metadata(request) + session = auth_service.refresh_session( + refresh_token=refresh_token, + ip_address=ip_address, + user_agent=user_agent, + ) + if session is None: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Sessao administrativa web invalida para refresh.", + ) + + set_panel_auth_cookies(response, session, settings) + return _build_panel_session_response( + session=session, + message="Sessao administrativa web renovada.", + ) + + +@router.get("/session", response_model=AdminPanelWebSessionResponse) +def panel_session( + current_context: AuthenticatedStaffContext = Depends(get_current_panel_staff_context), + settings: AdminSettings = Depends(get_settings), +): + return AdminPanelWebSessionResponse( + service="orquestrador-admin", + status="ok", + message="Sessao administrativa web ativa.", + session_id=current_context.session_id, + expires_in_seconds=settings.admin_auth_access_token_ttl_minutes * 60, + staff_account=AdminAuthenticatedStaffResponse(**current_context.principal.model_dump()), + redirect_to=None, + ) + + +@router.post("/logout", response_model=AdminPanelLogoutResponse) +def panel_logout( + request: Request, + response: Response, + settings: AdminSettings = Depends(get_settings), + auth_service: AuthService = Depends(get_auth_service), +): + ip_address, user_agent = _extract_request_metadata(request) + session_id: int | None = None + + access_token = get_panel_access_cookie(request) + if access_token: + try: + current_context = auth_service.get_authenticated_context(access_token) + except ValueError: + current_context = None + if current_context is not None: + auth_service.logout( + current_context.session_id, + actor_staff_account_id=current_context.principal.id, + ip_address=ip_address, + user_agent=user_agent, + ) + session_id = current_context.session_id + + refresh_token = get_panel_refresh_cookie(request) + if session_id is None and refresh_token: + session_id = auth_service.logout_by_refresh_token( + refresh_token, + ip_address=ip_address, + user_agent=user_agent, + ) + + clear_panel_auth_cookies(response, settings) + return AdminPanelLogoutResponse( + service="orquestrador-admin", + status="ok", + message="Sessao administrativa web encerrada.", + session_id=session_id, + redirect_to=_build_prefixed_path(settings.admin_api_prefix, "/login"), + ) + + +def _build_panel_session_response( + session: AdminAuthenticatedSession, + *, + message: str, + redirect_to: str | None = None, +) -> AdminPanelWebSessionResponse: + return AdminPanelWebSessionResponse( + service="orquestrador-admin", + status="ok", + message=message, + session_id=session.session_id, + expires_in_seconds=session.expires_in_seconds, + staff_account=AdminAuthenticatedStaffResponse(**session.principal.model_dump()), + redirect_to=redirect_to, + ) + + +def _extract_request_metadata(request: Request) -> tuple[str | None, str | None]: + ip_address = request.client.host if request.client else None + user_agent = request.headers.get("user-agent") + return ip_address, user_agent + + +def _build_prefixed_path(api_prefix: str, path: str) -> str: + normalized_prefix = api_prefix.rstrip("/") + normalized_path = path if path.startswith("/") else f"/{path}" + if not normalized_prefix: + return normalized_path + if normalized_path == "/": + return f"{normalized_prefix}/" + return f"{normalized_prefix}{normalized_path}" + diff --git a/admin_app/api/routes/panel_tools.py b/admin_app/api/routes/panel_tools.py new file mode 100644 index 0000000..8a396b7 --- /dev/null +++ b/admin_app/api/routes/panel_tools.py @@ -0,0 +1,170 @@ +from fastapi import APIRouter, Depends + +from admin_app.api.dependencies import get_settings, require_panel_admin_permission +from admin_app.api.schemas import ( + AdminToolContractsResponse, + AdminToolDraftListResponse, + AdminToolManagementActionResponse, + AdminToolOverviewResponse, + AdminToolPublicationListResponse, + AdminToolReviewQueueResponse, +) +from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal +from admin_app.services import ToolManagementService +from shared.contracts import AdminPermission + +router = APIRouter(prefix="/panel/tools", tags=["panel-tools"]) + + +def _build_service(settings: AdminSettings) -> ToolManagementService: + return ToolManagementService(settings) + + +@router.get( + "/overview", + response_model=AdminToolOverviewResponse, +) +def panel_tools_overview( + settings: AdminSettings = Depends(get_settings), + _: AuthenticatedStaffPrincipal = Depends( + require_panel_admin_permission(AdminPermission.MANAGE_TOOL_DRAFTS) + ), +): + service = _build_service(settings) + payload = service.build_overview_payload() + return AdminToolOverviewResponse( + service="orquestrador-admin", + mode=payload["mode"], + metrics=payload["metrics"], + workflow=payload["workflow"], + actions=_build_panel_actions(settings), + next_steps=payload["next_steps"], + ) + + +@router.get( + "/contracts", + response_model=AdminToolContractsResponse, +) +def panel_tool_contracts( + settings: AdminSettings = Depends(get_settings), + _: AuthenticatedStaffPrincipal = Depends( + require_panel_admin_permission(AdminPermission.MANAGE_TOOL_DRAFTS) + ), +): + service = _build_service(settings) + payload = service.build_contracts_payload() + return AdminToolContractsResponse( + service="orquestrador-admin", + publication_source_service=payload["publication_source_service"], + publication_target_service=payload["publication_target_service"], + lifecycle_statuses=payload["lifecycle_statuses"], + parameter_types=payload["parameter_types"], + publication_fields=payload["publication_fields"], + published_tool_fields=payload["published_tool_fields"], + ) + + +@router.get( + "/drafts", + response_model=AdminToolDraftListResponse, +) +def panel_tool_drafts( + settings: AdminSettings = Depends(get_settings), + _: AuthenticatedStaffPrincipal = Depends( + require_panel_admin_permission(AdminPermission.MANAGE_TOOL_DRAFTS) + ), +): + service = _build_service(settings) + payload = service.build_drafts_payload() + return AdminToolDraftListResponse( + service="orquestrador-admin", + storage_status=payload["storage_status"], + message=payload["message"], + drafts=payload["drafts"], + supported_statuses=payload["supported_statuses"], + ) + + +@router.get( + "/review-queue", + response_model=AdminToolReviewQueueResponse, +) +def panel_tool_review_queue( + settings: AdminSettings = Depends(get_settings), + _: AuthenticatedStaffPrincipal = Depends( + require_panel_admin_permission(AdminPermission.REVIEW_TOOL_GENERATIONS) + ), +): + service = _build_service(settings) + payload = service.build_review_queue_payload() + return AdminToolReviewQueueResponse( + service="orquestrador-admin", + queue_mode=payload["queue_mode"], + message=payload["message"], + items=payload["items"], + supported_statuses=payload["supported_statuses"], + ) + + +@router.get( + "/publications", + response_model=AdminToolPublicationListResponse, +) +def panel_tool_publications( + settings: AdminSettings = Depends(get_settings), + _: AuthenticatedStaffPrincipal = Depends( + require_panel_admin_permission(AdminPermission.PUBLISH_TOOLS) + ), +): + service = _build_service(settings) + payload = service.build_publications_payload() + return AdminToolPublicationListResponse( + service="orquestrador-admin", + source=payload["source"], + target_service=payload["target_service"], + publications=payload["publications"], + ) + + +def _build_panel_actions(settings: AdminSettings) -> list[AdminToolManagementActionResponse]: + return [ + AdminToolManagementActionResponse( + key="overview", + label="Overview web de tools", + href=_build_prefixed_path(settings.admin_api_prefix, "/panel/tools/overview"), + required_permission=AdminPermission.MANAGE_TOOL_DRAFTS, + description="Snapshot do dominio de tools pronto para leitura no painel.", + ), + AdminToolManagementActionResponse( + key="contracts", + label="Contratos web de tools", + href=_build_prefixed_path(settings.admin_api_prefix, "/panel/tools/contracts"), + required_permission=AdminPermission.MANAGE_TOOL_DRAFTS, + description="Base contratual para a tela de revisao e aprovacao.", + ), + AdminToolManagementActionResponse( + key="review_queue", + label="Fila web de revisao", + href=_build_prefixed_path(settings.admin_api_prefix, "/panel/tools/review-queue"), + required_permission=AdminPermission.REVIEW_TOOL_GENERATIONS, + description="Leitura da fila de revisao sob a sessao web do painel.", + ), + AdminToolManagementActionResponse( + key="publications", + label="Publicacoes web", + href=_build_prefixed_path(settings.admin_api_prefix, "/panel/tools/publications"), + required_permission=AdminPermission.PUBLISH_TOOLS, + description="Catalogo de tools ativas e prontas para ativacao no produto.", + ), + ] + + +def _build_prefixed_path(api_prefix: str, path: str) -> str: + normalized_prefix = api_prefix.rstrip("/") + normalized_path = path if path.startswith("/") else f"/{path}" + if not normalized_prefix: + return normalized_path + if normalized_path == "/": + return f"{normalized_prefix}/" + return f"{normalized_prefix}{normalized_path}" diff --git a/admin_app/api/routes/system.py b/admin_app/api/routes/system.py index c694267..d4da99d 100644 --- a/admin_app/api/routes/system.py +++ b/admin_app/api/routes/system.py @@ -1,18 +1,29 @@ -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, Request +from fastapi.responses import RedirectResponse, Response from admin_app.api.dependencies import ( get_current_staff_permissions, + get_security_service, get_settings, require_admin_permission, ) +from admin_app.api.panel_session import ( + PANEL_ACCESS_COOKIE_NAME, + PANEL_COOKIE_SAMESITE, + PANEL_REFRESH_COOKIE_NAME, + build_panel_cookie_path, + should_use_secure_cookies, +) from admin_app.api.schemas import ( AdminCapabilityResponse, AdminCurrentAccessResponse, AdminHealthResponse, - AdminRootResponse, + AdminSystemConfigurationResponse, AdminSystemInfoResponse, + AdminSystemRuntimeConfigurationResponse, + AdminSystemSecurityConfigurationResponse, ) -from admin_app.core import AuthenticatedStaffPrincipal +from admin_app.core import AdminSecurityService, AuthenticatedStaffPrincipal from admin_app.core.settings import AdminSettings from admin_app.services.system_service import SystemService from shared.contracts import AdminPermission @@ -20,18 +31,29 @@ from shared.contracts import AdminPermission router = APIRouter(tags=["system"]) -def _build_service(settings: AdminSettings) -> SystemService: - return SystemService(settings=settings) +def _build_service( + settings: AdminSettings, + security_service: AdminSecurityService, +) -> SystemService: + return SystemService(settings=settings, security_service=security_service) -@router.get("/", response_model=AdminRootResponse) -def root(settings: AdminSettings = Depends(get_settings)): - return _build_service(settings).build_root_payload() +@router.get("/", response_model=None) +def root( + request: Request, + settings: AdminSettings = Depends(get_settings), +) -> Response | dict: + if "text/html" in request.headers.get("accept", ""): + return RedirectResponse( + url=_build_prefixed_path(settings.admin_api_prefix, "/login"), + status_code=302, + ) + return SystemService(settings=settings).build_root_payload() @router.get("/health", response_model=AdminHealthResponse) def health_check(settings: AdminSettings = Depends(get_settings)): - return _build_service(settings).build_health_payload() + return SystemService(settings=settings).build_health_payload() @router.get( @@ -40,11 +62,12 @@ def health_check(settings: AdminSettings = Depends(get_settings)): ) def system_info( settings: AdminSettings = Depends(get_settings), + security_service: AdminSecurityService = Depends(get_security_service), _: AuthenticatedStaffPrincipal = Depends( require_admin_permission(AdminPermission.VIEW_SYSTEM) ), ): - return _build_service(settings).build_system_info_payload() + return _build_service(settings, security_service).build_system_info_payload() @router.get( @@ -85,3 +108,85 @@ def admin_capabilities( allowed=True, role=current_staff.role, ) + + +@router.get( + "/system/configuration", + response_model=AdminSystemConfigurationResponse, +) +def system_configuration( + settings: AdminSettings = Depends(get_settings), + security_service: AdminSecurityService = Depends(get_security_service), + _: AuthenticatedStaffPrincipal = Depends( + require_admin_permission(AdminPermission.MANAGE_SETTINGS) + ), +): + service = _build_service(settings, security_service) + runtime_payload = _build_runtime_configuration_payload(service, settings) + return AdminSystemConfigurationResponse( + service="orquestrador-admin", + runtime=runtime_payload, + security=service.build_security_configuration_payload(), + sources=service.build_configuration_sources_payload(), + ) + + +@router.get( + "/system/configuration/runtime", + response_model=AdminSystemRuntimeConfigurationResponse, +) +def system_runtime_configuration( + settings: AdminSettings = Depends(get_settings), + security_service: AdminSecurityService = Depends(get_security_service), + _: AuthenticatedStaffPrincipal = Depends( + require_admin_permission(AdminPermission.MANAGE_SETTINGS) + ), +): + service = _build_service(settings, security_service) + return AdminSystemRuntimeConfigurationResponse( + service="orquestrador-admin", + runtime=_build_runtime_configuration_payload(service, settings), + ) + + +@router.get( + "/system/configuration/security", + response_model=AdminSystemSecurityConfigurationResponse, +) +def system_security_configuration( + settings: AdminSettings = Depends(get_settings), + security_service: AdminSecurityService = Depends(get_security_service), + _: AuthenticatedStaffPrincipal = Depends( + require_admin_permission(AdminPermission.MANAGE_SETTINGS) + ), +): + service = _build_service(settings, security_service) + return AdminSystemSecurityConfigurationResponse( + service="orquestrador-admin", + security=service.build_security_configuration_payload(), + ) + + +def _build_runtime_configuration_payload( + service: SystemService, + settings: AdminSettings, +) -> dict: + runtime_payload = service.build_runtime_configuration_payload() + runtime_payload["panel_session"] = { + "access_cookie_name": PANEL_ACCESS_COOKIE_NAME, + "refresh_cookie_name": PANEL_REFRESH_COOKIE_NAME, + "cookie_path": build_panel_cookie_path(settings), + "same_site": PANEL_COOKIE_SAMESITE, + "secure_cookies": should_use_secure_cookies(settings), + } + return runtime_payload + + +def _build_prefixed_path(api_prefix: str, path: str) -> str: + normalized_prefix = api_prefix.rstrip("/") + normalized_path = path if path.startswith("/") else f"/{path}" + if not normalized_prefix: + return normalized_path + if normalized_path == "/": + return f"{normalized_prefix}/" + return f"{normalized_prefix}{normalized_path}" diff --git a/admin_app/api/routes/tools.py b/admin_app/api/routes/tools.py new file mode 100644 index 0000000..1efd938 --- /dev/null +++ b/admin_app/api/routes/tools.py @@ -0,0 +1,177 @@ +from fastapi import APIRouter, Depends + +from admin_app.api.dependencies import get_settings, require_admin_permission +from admin_app.api.schemas import ( + AdminToolContractsResponse, + AdminToolDraftListResponse, + AdminToolManagementActionResponse, + AdminToolOverviewResponse, + AdminToolPublicationListResponse, + AdminToolReviewQueueResponse, +) +from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal +from admin_app.services import ToolManagementService +from shared.contracts import AdminPermission + +router = APIRouter(prefix="/tools", tags=["tools"]) + + +def _build_service(settings: AdminSettings) -> ToolManagementService: + return ToolManagementService(settings) + + +@router.get( + "/overview", + response_model=AdminToolOverviewResponse, +) +def tools_overview( + settings: AdminSettings = Depends(get_settings), + _: AuthenticatedStaffPrincipal = Depends( + require_admin_permission(AdminPermission.MANAGE_TOOL_DRAFTS) + ), +): + service = _build_service(settings) + payload = service.build_overview_payload() + return AdminToolOverviewResponse( + service="orquestrador-admin", + mode=payload["mode"], + metrics=payload["metrics"], + workflow=payload["workflow"], + actions=_build_actions(settings), + next_steps=payload["next_steps"], + ) + + +@router.get( + "/contracts", + response_model=AdminToolContractsResponse, +) +def tool_contracts( + settings: AdminSettings = Depends(get_settings), + _: AuthenticatedStaffPrincipal = Depends( + require_admin_permission(AdminPermission.MANAGE_TOOL_DRAFTS) + ), +): + service = _build_service(settings) + payload = service.build_contracts_payload() + return AdminToolContractsResponse( + service="orquestrador-admin", + publication_source_service=payload["publication_source_service"], + publication_target_service=payload["publication_target_service"], + lifecycle_statuses=payload["lifecycle_statuses"], + parameter_types=payload["parameter_types"], + publication_fields=payload["publication_fields"], + published_tool_fields=payload["published_tool_fields"], + ) + + +@router.get( + "/drafts", + response_model=AdminToolDraftListResponse, +) +def tool_drafts( + settings: AdminSettings = Depends(get_settings), + _: AuthenticatedStaffPrincipal = Depends( + require_admin_permission(AdminPermission.MANAGE_TOOL_DRAFTS) + ), +): + service = _build_service(settings) + payload = service.build_drafts_payload() + return AdminToolDraftListResponse( + service="orquestrador-admin", + storage_status=payload["storage_status"], + message=payload["message"], + drafts=payload["drafts"], + supported_statuses=payload["supported_statuses"], + ) + + +@router.get( + "/review-queue", + response_model=AdminToolReviewQueueResponse, +) +def tool_review_queue( + settings: AdminSettings = Depends(get_settings), + _: AuthenticatedStaffPrincipal = Depends( + require_admin_permission(AdminPermission.REVIEW_TOOL_GENERATIONS) + ), +): + service = _build_service(settings) + payload = service.build_review_queue_payload() + return AdminToolReviewQueueResponse( + service="orquestrador-admin", + queue_mode=payload["queue_mode"], + message=payload["message"], + items=payload["items"], + supported_statuses=payload["supported_statuses"], + ) + + +@router.get( + "/publications", + response_model=AdminToolPublicationListResponse, +) +def tool_publications( + settings: AdminSettings = Depends(get_settings), + _: AuthenticatedStaffPrincipal = Depends( + require_admin_permission(AdminPermission.PUBLISH_TOOLS) + ), +): + service = _build_service(settings) + payload = service.build_publications_payload() + return AdminToolPublicationListResponse( + service="orquestrador-admin", + source=payload["source"], + target_service=payload["target_service"], + publications=payload["publications"], + ) + + +def _build_actions(settings: AdminSettings) -> list[AdminToolManagementActionResponse]: + return [ + AdminToolManagementActionResponse( + key="overview", + label="Overview de tools", + href=_build_prefixed_path(settings.admin_api_prefix, "/tools/overview"), + required_permission=AdminPermission.MANAGE_TOOL_DRAFTS, + description="Snapshot inicial da governanca de tools no admin.", + ), + AdminToolManagementActionResponse( + key="contracts", + label="Contratos compartilhados", + href=_build_prefixed_path(settings.admin_api_prefix, "/tools/contracts"), + required_permission=AdminPermission.MANAGE_TOOL_DRAFTS, + description="Enumera lifecycle, tipos de parametro e campos de publicacao.", + ), + AdminToolManagementActionResponse( + key="drafts", + label="Fila de drafts", + href=_build_prefixed_path(settings.admin_api_prefix, "/tools/drafts"), + required_permission=AdminPermission.MANAGE_TOOL_DRAFTS, + description="Base do cadastro de novas tools e estados vazios da fase atual.", + ), + AdminToolManagementActionResponse( + key="review_queue", + label="Fila de revisao", + href=_build_prefixed_path(settings.admin_api_prefix, "/tools/review-queue"), + required_permission=AdminPermission.REVIEW_TOOL_GENERATIONS, + description="Superficie para validacao, revisao tecnica e aprovacao humana.", + ), + AdminToolManagementActionResponse( + key="publications", + label="Catalogo de publicacoes", + href=_build_prefixed_path(settings.admin_api_prefix, "/tools/publications"), + required_permission=AdminPermission.PUBLISH_TOOLS, + description="Catalogo bootstrap de tools ativas voltadas ao runtime de produto.", + ), + ] + + +def _build_prefixed_path(api_prefix: str, path: str) -> str: + normalized_prefix = api_prefix.rstrip("/") + normalized_path = path if path.startswith("/") else f"/{path}" + if not normalized_prefix: + return normalized_path + if normalized_path == "/": + return f"{normalized_prefix}/" + return f"{normalized_prefix}{normalized_path}" diff --git a/admin_app/api/schemas.py b/admin_app/api/schemas.py index 26c90f8..bf23ccf 100644 --- a/admin_app/api/schemas.py +++ b/admin_app/api/schemas.py @@ -2,7 +2,8 @@ from datetime import datetime from pydantic import BaseModel, Field, field_validator -from shared.contracts import StaffRole +from admin_app.core import AdminCredentialStrategy +from shared.contracts import AdminPermission, ServiceName, StaffRole, ToolLifecycleStatus, ToolParameterType class AdminRootResponse(BaseModel): @@ -67,6 +68,59 @@ class AdminAuditListResponse(BaseModel): events: list[AdminAuditEntryResponse] +class AdminRuntimeApplicationConfigurationResponse(BaseModel): + app_name: str + environment: str + version: str + api_prefix: str + debug: bool + + +class AdminRuntimeDatabaseConfigurationResponse(BaseModel): + host: str + port: int + name: str + cloud_sql_configured: bool + + +class AdminPanelSessionConfigurationResponse(BaseModel): + access_cookie_name: str + refresh_cookie_name: str + cookie_path: str + same_site: str + secure_cookies: bool + + +class AdminSystemRuntimeConfigurationPayload(BaseModel): + application: AdminRuntimeApplicationConfigurationResponse + database: AdminRuntimeDatabaseConfigurationResponse + panel_session: AdminPanelSessionConfigurationResponse + + +class AdminConfigurationSourceResponse(BaseModel): + key: str + source: str + mutable: bool + description: str + + +class AdminSystemRuntimeConfigurationResponse(BaseModel): + service: str + runtime: AdminSystemRuntimeConfigurationPayload + + +class AdminSystemSecurityConfigurationResponse(BaseModel): + service: str + security: AdminCredentialStrategy + + +class AdminSystemConfigurationResponse(BaseModel): + service: str + runtime: AdminSystemRuntimeConfigurationPayload + security: AdminCredentialStrategy + sources: list[AdminConfigurationSourceResponse] + + class AdminLoginRequest(BaseModel): email: str password: str = Field(min_length=1) @@ -97,4 +151,127 @@ class AdminLogoutResponse(BaseModel): service: str status: str message: str - session_id: int \ No newline at end of file + session_id: int + + +class AdminPanelWebSessionResponse(BaseModel): + service: str + status: str + message: str + session_id: int + expires_in_seconds: int + staff_account: AdminAuthenticatedStaffResponse + redirect_to: str | None = None + + +class AdminPanelLogoutResponse(BaseModel): + service: str + status: str + message: str + session_id: int | None + redirect_to: str + + +class AdminToolManagementMetricResponse(BaseModel): + key: str + label: str + value: str + description: str + + +class AdminToolLifecycleStageResponse(BaseModel): + code: ToolLifecycleStatus + label: str + description: str + + +class AdminToolParameterTypeResponse(BaseModel): + code: ToolParameterType + label: str + description: str + + +class AdminToolManagementActionResponse(BaseModel): + key: str + label: str + href: str + required_permission: AdminPermission + description: str + + +class AdminToolOverviewResponse(BaseModel): + service: str + mode: str + metrics: list[AdminToolManagementMetricResponse] + workflow: list[AdminToolLifecycleStageResponse] + actions: list[AdminToolManagementActionResponse] + next_steps: list[str] + + +class AdminToolContractsResponse(BaseModel): + service: str + publication_source_service: ServiceName + publication_target_service: ServiceName + lifecycle_statuses: list[AdminToolLifecycleStageResponse] + parameter_types: list[AdminToolParameterTypeResponse] + publication_fields: list[str] + published_tool_fields: list[str] + + +class AdminToolDraftSummaryResponse(BaseModel): + draft_id: str + tool_name: str + display_name: str + status: ToolLifecycleStatus + summary: str + owner_name: str | None = None + updated_at: datetime | None = None + + +class AdminToolDraftListResponse(BaseModel): + service: str + storage_status: str + message: str + drafts: list[AdminToolDraftSummaryResponse] + supported_statuses: list[ToolLifecycleStatus] + + +class AdminToolReviewQueueEntryResponse(BaseModel): + entry_id: str + tool_name: str + display_name: str + status: ToolLifecycleStatus + gate: str + summary: str + owner_name: str | None = None + queued_at: datetime | None = None + + +class AdminToolReviewQueueResponse(BaseModel): + service: str + queue_mode: str + message: str + items: list[AdminToolReviewQueueEntryResponse] + supported_statuses: list[ToolLifecycleStatus] + + +class AdminToolPublicationSummaryResponse(BaseModel): + publication_id: str + tool_name: str + display_name: str + description: str + domain: str + version: int + status: ToolLifecycleStatus + parameter_count: int + implementation_module: str + implementation_callable: str + published_by: str | None = None + published_at: datetime | None = None + + +class AdminToolPublicationListResponse(BaseModel): + service: str + source: str + target_service: ServiceName + publications: list[AdminToolPublicationSummaryResponse] diff --git a/admin_app/app_factory.py b/admin_app/app_factory.py index 06b2b12..9ca06e3 100644 --- a/admin_app/app_factory.py +++ b/admin_app/app_factory.py @@ -1,7 +1,9 @@ -from fastapi import FastAPI +from fastapi import FastAPI +from fastapi.staticfiles import StaticFiles from admin_app.api.router import api_router from admin_app.core.settings import AdminSettings, get_admin_settings +from admin_app.view import PANEL_STATIC_DIRECTORY, PANEL_STATIC_MOUNT_NAME, panel_router # Fabrica explicita do runtime administrativo para facilitar testes e futura configuracao. @@ -13,5 +15,19 @@ def create_app(settings: AdminSettings | None = None) -> FastAPI: debug=resolved_settings.admin_debug, ) app.state.admin_settings = resolved_settings + app.mount( + _build_panel_static_path(resolved_settings.admin_api_prefix), + StaticFiles(directory=str(PANEL_STATIC_DIRECTORY)), + name=PANEL_STATIC_MOUNT_NAME, + ) app.include_router(api_router, prefix=resolved_settings.admin_api_prefix) + app.include_router(panel_router, prefix=resolved_settings.admin_api_prefix) return app + + +def _build_panel_static_path(api_prefix: str) -> str: + normalized_prefix = api_prefix.rstrip("/") + if normalized_prefix: + return f"{normalized_prefix}/panel/assets" + return "/panel/assets" + diff --git a/admin_app/services/__init__.py b/admin_app/services/__init__.py index 37b3cd8..ea7b64e 100644 --- a/admin_app/services/__init__.py +++ b/admin_app/services/__init__.py @@ -5,6 +5,7 @@ from admin_app.services.audit_service import ( ) from admin_app.services.auth_service import AuthService from admin_app.services.system_service import SystemService +from admin_app.services.tool_management_service import ToolManagementService __all__ = [ "AdminAuditEventType", @@ -12,4 +13,5 @@ __all__ = [ "AuditService", "AuthService", "SystemService", + "ToolManagementService", ] diff --git a/admin_app/services/auth_service.py b/admin_app/services/auth_service.py index 9451df9..27f09c5 100644 --- a/admin_app/services/auth_service.py +++ b/admin_app/services/auth_service.py @@ -1,4 +1,4 @@ -from datetime import datetime, timezone +from datetime import datetime, timezone from admin_app.core import ( AdminAuthenticatedSession, @@ -104,6 +104,28 @@ class AuthService: ) return True + def logout_by_refresh_token( + self, + refresh_token: str, + *, + ip_address: str | None, + user_agent: str | None, + ) -> int | None: + token_hash = self.security_service.hash_refresh_token(refresh_token) + staff_session = self.session_repository.get_by_refresh_token_hash(token_hash) + if staff_session is None: + return None + + account = self.account_repository.get_by_id(staff_session.staff_account_id) + actor_staff_account_id = account.id if account is not None and account.is_active else None + self.logout( + staff_session.id, + actor_staff_account_id=actor_staff_account_id, + ip_address=ip_address, + user_agent=user_agent, + ) + return staff_session.id + def get_authenticated_context(self, access_token: str) -> AuthenticatedStaffContext: claims = self.security_service.decode_access_token(access_token) staff_session = self.session_repository.get_by_id(claims.sid) @@ -202,9 +224,17 @@ class AuthService: ) @staticmethod - def _is_session_active(staff_session: StaffSession | None) -> bool: + def _normalize_datetime(value: datetime) -> datetime: + if value.tzinfo is None or value.tzinfo.utcoffset(value) is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + @classmethod + def _is_session_active(cls, staff_session: StaffSession | None) -> bool: if staff_session is None: return False if staff_session.revoked_at is not None: return False - return staff_session.expires_at >= datetime.now(timezone.utc) \ No newline at end of file + + expires_at = cls._normalize_datetime(staff_session.expires_at) + return expires_at >= datetime.now(timezone.utc) diff --git a/admin_app/services/system_service.py b/admin_app/services/system_service.py index 858f11c..7703fa1 100644 --- a/admin_app/services/system_service.py +++ b/admin_app/services/system_service.py @@ -1,9 +1,15 @@ +from admin_app.core import AdminCredentialStrategy, AdminSecurityService from admin_app.core.settings import AdminSettings class SystemService: - def __init__(self, settings: AdminSettings): + def __init__( + self, + settings: AdminSettings, + security_service: AdminSecurityService | None = None, + ): self.settings = settings + self.security_service = security_service or AdminSecurityService(settings) def build_root_payload(self) -> dict: return { @@ -29,3 +35,51 @@ class SystemService: "api_prefix": self.settings.admin_api_prefix, "debug": self.settings.admin_debug, } + + def build_runtime_configuration_payload(self) -> dict: + return { + "application": { + "app_name": self.settings.admin_app_name, + "environment": self.settings.admin_environment, + "version": self.settings.admin_version, + "api_prefix": self.settings.admin_api_prefix, + "debug": self.settings.admin_debug, + }, + "database": { + "host": self.settings.admin_db_host, + "port": self.settings.admin_db_port, + "name": self.settings.admin_db_name, + "cloud_sql_configured": bool(self.settings.admin_db_cloud_sql_connection_name), + }, + } + + def build_security_configuration_payload(self) -> AdminCredentialStrategy: + return self.security_service.build_credential_strategy() + + def build_configuration_sources_payload(self) -> list[dict]: + return [ + { + "key": "application", + "source": "env", + "mutable": False, + "description": "Metadados principais do admin runtime vindos de AdminSettings.", + }, + { + "key": "database", + "source": "env", + "mutable": False, + "description": "Conexao administrativa derivada das variaveis de ambiente do servico.", + }, + { + "key": "security", + "source": "env", + "mutable": False, + "description": "Politicas de senha, token e bootstrap lidas do runtime administrativo.", + }, + { + "key": "panel_session", + "source": "runtime", + "mutable": False, + "description": "Cookies e sessao web do painel derivam da configuracao ativa do admin.", + }, + ] diff --git a/admin_app/services/tool_management_service.py b/admin_app/services/tool_management_service.py new file mode 100644 index 0000000..99e4e82 --- /dev/null +++ b/admin_app/services/tool_management_service.py @@ -0,0 +1,305 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime + +from admin_app.core.settings import AdminSettings +from shared.contracts import ServiceName, ToolLifecycleStatus, ToolParameterType + + +@dataclass(frozen=True) +class BootstrapToolCatalogEntry: + tool_name: str + display_name: str + description: str + domain: str + parameter_count: int + + +_BOOTSTRAP_TOOL_CATALOG: tuple[BootstrapToolCatalogEntry, ...] = ( + BootstrapToolCatalogEntry( + tool_name="consultar_estoque", + display_name="Consultar estoque", + description="Consulta veiculos disponiveis no estoque comercial.", + domain="vendas", + parameter_count=4, + ), + BootstrapToolCatalogEntry( + tool_name="validar_cliente_venda", + display_name="Validar cliente para venda", + description="Avalia elegibilidade de credito para operacoes de venda.", + domain="vendas", + parameter_count=2, + ), + BootstrapToolCatalogEntry( + tool_name="avaliar_veiculo_troca", + display_name="Avaliar veiculo de troca", + description="Estima o valor de entrada de um veiculo usado.", + domain="vendas", + parameter_count=3, + ), + BootstrapToolCatalogEntry( + tool_name="agendar_revisao", + display_name="Agendar revisao", + description="Abre um agendamento de revisao ou manutencao.", + domain="revisao", + parameter_count=6, + ), + BootstrapToolCatalogEntry( + tool_name="listar_agendamentos_revisao", + display_name="Listar agendamentos de revisao", + description="Consulta a fila de agendamentos de revisao do cliente.", + domain="revisao", + parameter_count=3, + ), + BootstrapToolCatalogEntry( + tool_name="cancelar_agendamento_revisao", + display_name="Cancelar agendamento de revisao", + description="Cancela um agendamento existente por protocolo.", + domain="revisao", + parameter_count=2, + ), + BootstrapToolCatalogEntry( + tool_name="editar_data_revisao", + display_name="Editar data de revisao", + description="Remarca uma revisao para um novo horario.", + domain="revisao", + parameter_count=2, + ), + BootstrapToolCatalogEntry( + tool_name="realizar_pedido", + display_name="Realizar pedido", + description="Efetiva um pedido de compra com o veiculo escolhido.", + domain="vendas", + parameter_count=2, + ), + BootstrapToolCatalogEntry( + tool_name="listar_pedidos", + display_name="Listar pedidos", + description="Consulta pedidos ja abertos pelo cliente.", + domain="vendas", + parameter_count=3, + ), + BootstrapToolCatalogEntry( + tool_name="cancelar_pedido", + display_name="Cancelar pedido", + description="Cancela um pedido existente com motivo registrado.", + domain="vendas", + parameter_count=2, + ), + BootstrapToolCatalogEntry( + tool_name="consultar_frota_aluguel", + display_name="Consultar frota de aluguel", + description="Lista veiculos disponiveis para locacao.", + domain="locacao", + parameter_count=6, + ), + BootstrapToolCatalogEntry( + tool_name="abrir_locacao_aluguel", + display_name="Abrir locacao de aluguel", + description="Inicia um contrato de locacao de veiculo.", + domain="locacao", + parameter_count=7, + ), + BootstrapToolCatalogEntry( + tool_name="registrar_devolucao_aluguel", + display_name="Registrar devolucao de aluguel", + description="Fecha uma locacao e devolve o veiculo para a frota.", + domain="locacao", + parameter_count=4, + ), + BootstrapToolCatalogEntry( + tool_name="registrar_pagamento_aluguel", + display_name="Registrar pagamento de aluguel", + description="Registra comprovantes e pagamentos de contratos de locacao.", + domain="locacao", + parameter_count=7, + ), + BootstrapToolCatalogEntry( + tool_name="limpar_contexto_conversa", + display_name="Limpar contexto de conversa", + description="Reinicia o contexto operacional atual do atendimento.", + domain="orquestracao", + parameter_count=1, + ), + BootstrapToolCatalogEntry( + tool_name="continuar_proximo_pedido", + display_name="Continuar proximo pedido", + description="Retoma o proximo pedido pendente do fluxo atual.", + domain="orquestracao", + parameter_count=0, + ), + BootstrapToolCatalogEntry( + tool_name="descartar_pedidos_pendentes", + display_name="Descartar pedidos pendentes", + description="Descarta apenas a fila pendente de pedidos do contexto.", + domain="orquestracao", + parameter_count=1, + ), + BootstrapToolCatalogEntry( + tool_name="cancelar_fluxo_atual", + display_name="Cancelar fluxo atual", + description="Interrompe o fluxo corrente sem apagar todo o contexto.", + domain="orquestracao", + parameter_count=1, + ), +) + +_LIFECYCLE_DESCRIPTIONS = { + ToolLifecycleStatus.DRAFT: "Estado inicial de uma tool ainda em definicao.", + ToolLifecycleStatus.GENERATED: "Implementacao gerada e pronta para analise tecnica.", + ToolLifecycleStatus.VALIDATED: "Tool validada automaticamente com verificacoes basicas.", + ToolLifecycleStatus.APPROVED: "Versao revisada e aprovada para publicacao controlada.", + ToolLifecycleStatus.ACTIVE: "Tool publicada e apta a abastecer o runtime de produto.", + ToolLifecycleStatus.FAILED: "Falha registrada na geracao, validacao ou ativacao.", + ToolLifecycleStatus.ARCHIVED: "Versao retirada de circulacao e mantida apenas para historico.", +} + +_PARAMETER_TYPE_DESCRIPTIONS = { + ToolParameterType.STRING: "Texto livre, codigos e identificadores.", + ToolParameterType.INTEGER: "Valores inteiros para limites, anos e contagens.", + ToolParameterType.NUMBER: "Valores numericos decimais, como preco e diaria.", + ToolParameterType.BOOLEAN: "Marcadores verdadeiro ou falso para decisoes operacionais.", + ToolParameterType.OBJECT: "Estruturas compostas para payloads complexos.", + ToolParameterType.ARRAY: "Colecoes ordenadas de valores.", +} + + +class ToolManagementService: + def __init__(self, settings: AdminSettings): + self.settings = settings + + def build_overview_payload(self) -> dict: + catalog = self.list_publication_catalog() + return { + "mode": "bootstrap_catalog", + "metrics": [ + { + "key": "active_catalog", + "label": "Tools mapeadas", + "value": str(len(catalog)), + "description": "Catalogo bootstrap refletindo a base de tools conhecida no monorepo.", + }, + { + "key": "lifecycle_stages", + "label": "Etapas de lifecycle", + "value": str(len(ToolLifecycleStatus)), + "description": "Estados compartilhados entre governanca administrativa e publicacao.", + }, + { + "key": "parameter_types", + "label": "Tipos de parametro", + "value": str(len(ToolParameterType)), + "description": "Tipos aceitos pelo contrato inicial de publicacao de tools.", + }, + { + "key": "draft_persistence", + "label": "Persistencia de drafts", + "value": "pendente", + "description": "A fase atual entrega as superficies e o contrato; entidades de draft ainda nao existem.", + }, + ], + "workflow": self.build_lifecycle_payload(), + "next_steps": [ + "Criar entidades administrativas para ToolDraft, ToolValidationRun e ToolPublication.", + "Ligar o formulario de cadastro de novas tools a uma persistencia propria do admin.", + "Abrir filas de revisao, aprovacao e ativacao com auditoria ponta a ponta.", + ], + } + + def build_contracts_payload(self) -> dict: + return { + "publication_source_service": ServiceName.ADMIN, + "publication_target_service": ServiceName.PRODUCT, + "lifecycle_statuses": self.build_lifecycle_payload(), + "parameter_types": [ + { + "code": parameter_type, + "label": parameter_type.value.upper(), + "description": _PARAMETER_TYPE_DESCRIPTIONS[parameter_type], + } + for parameter_type in ToolParameterType + ], + "publication_fields": [ + "source_service", + "target_service", + "publication_id", + "published_tool", + "emitted_at", + ], + "published_tool_fields": [ + "tool_name", + "display_name", + "description", + "version", + "status", + "parameters", + "implementation_module", + "implementation_callable", + "checksum", + "published_at", + "published_by", + ], + } + + def build_drafts_payload(self) -> dict: + return { + "storage_status": "pending_persistence", + "message": ( + "As rotas de gestao de tools ja existem, mas a persistencia de ToolDraft ainda sera criada nas proximas etapas." + ), + "drafts": [], + "supported_statuses": [ToolLifecycleStatus.DRAFT], + } + + def build_review_queue_payload(self) -> dict: + return { + "queue_mode": "bootstrap_empty_state", + "message": ( + "A fila de revisao ainda opera em estado vazio ate a criacao das entidades de geracao e validacao." + ), + "items": [], + "supported_statuses": [ + ToolLifecycleStatus.GENERATED, + ToolLifecycleStatus.VALIDATED, + ToolLifecycleStatus.APPROVED, + ToolLifecycleStatus.FAILED, + ], + } + + def build_publications_payload(self) -> dict: + return { + "source": "bootstrap_catalog", + "target_service": ServiceName.PRODUCT, + "publications": self.list_publication_catalog(), + } + + def build_lifecycle_payload(self) -> list[dict]: + return [ + { + "code": status, + "label": status.value.replace("_", " ").title(), + "description": _LIFECYCLE_DESCRIPTIONS[status], + } + for status in ToolLifecycleStatus + ] + + def list_publication_catalog(self) -> list[dict]: + published_at = datetime.now(UTC) + return [ + { + "publication_id": f"bootstrap::{entry.tool_name}::v1", + "tool_name": entry.tool_name, + "display_name": entry.display_name, + "description": entry.description, + "domain": entry.domain, + "version": 1, + "status": ToolLifecycleStatus.ACTIVE, + "parameter_count": entry.parameter_count, + "implementation_module": "app.services.tools.handlers", + "implementation_callable": entry.tool_name, + "published_by": "bootstrap_catalog", + "published_at": published_at, + } + for entry in _BOOTSTRAP_TOOL_CATALOG + ] diff --git a/admin_app/view/__init__.py b/admin_app/view/__init__.py new file mode 100644 index 0000000..13659de --- /dev/null +++ b/admin_app/view/__init__.py @@ -0,0 +1,9 @@ +from admin_app.view.assets import PANEL_STATIC_DIRECTORY, PANEL_STATIC_MOUNT_NAME +from admin_app.view.router import panel_router + +__all__ = [ + "PANEL_STATIC_DIRECTORY", + "PANEL_STATIC_MOUNT_NAME", + "panel_router", +] + diff --git a/admin_app/view/assets.py b/admin_app/view/assets.py new file mode 100644 index 0000000..80c3185 --- /dev/null +++ b/admin_app/view/assets.py @@ -0,0 +1,4 @@ +from pathlib import Path + +PANEL_STATIC_MOUNT_NAME = "admin_panel_assets" +PANEL_STATIC_DIRECTORY = Path(__file__).resolve().parent / "static" diff --git a/admin_app/view/rendering.py b/admin_app/view/rendering.py new file mode 100644 index 0000000..2b82fa6 --- /dev/null +++ b/admin_app/view/rendering.py @@ -0,0 +1,725 @@ +from html import escape + +from admin_app.view.view_models import ( + AdminLoginPageView, + AdminPanelHomeView, + AdminPanelMetric, + AdminPanelModuleCard, + AdminPanelNavigationItem, + AdminPanelQuickAction, + AdminPanelRoadmapItem, + AdminPanelSurfaceLink, + AdminToolReviewPageView, + AdminToolReviewWorkflowStep, +) + +BOOTSTRAP_CSS_HREF = "https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/css/bootstrap.min.css" +BOOTSTRAP_JS_HREF = "https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/js/bootstrap.bundle.min.js" +_BADGE_CLASS_MAP = { + "success": "bg-success-subtle text-success-emphasis border border-success-subtle", + "warning": "bg-warning-subtle text-warning-emphasis border border-warning-subtle", + "info": "bg-info-subtle text-info-emphasis border border-info-subtle", + "primary": "bg-primary-subtle text-primary-emphasis border border-primary-subtle", + "secondary": "bg-secondary-subtle text-secondary-emphasis border border-secondary-subtle", + "dark": "bg-dark-subtle text-dark-emphasis border border-dark-subtle", +} + + +def render_panel_home( + view: AdminPanelHomeView, + *, + css_href: str, + js_href: str, +) -> str: + navigation_markup = _render_navigation(view.navigation) + quick_actions_markup = _render_quick_actions(view.quick_actions) + metrics_markup = _render_metrics(view.metrics) + modules_markup = _render_modules(view.modules) + surface_links_markup = _render_surface_links(view.surface_links) + roadmap_markup = _render_roadmap(view.roadmap) + + panel_title = escape(view.panel_title) + app_name = escape(view.app_name) + panel_subtitle = escape(view.panel_subtitle) + environment = escape(view.environment) + version = escape(view.version) + api_prefix = escape(view.api_prefix) + service = escape(view.service) + release_label = escape(view.release_label) + + return f""" + + + + + {panel_title} + + + + + +
+
+ + +
+
+
+
+ Area interna protegida + Minimal Bootstrap +
+
+
+

Dashboard do administrador

+

+ A home protegida organiza o trabalho do time interno por fluxo, com foco no que realmente importa depois do login. +

+
+
+
+ {quick_actions_markup} +
+
+
+
+
+ +
+ {metrics_markup} +
+ +
+
+
+
+
+
+

Areas do sistema

+

Onde o time interno opera

+

+ A dashboard agora funciona como ponto de orientacao para entrar nas areas certas sem expor atalhos desnecessarios. +

+
+
+
+ {modules_markup} +
+
+
+
+ +
+
+
+

Acessos disponiveis

+

Entradas claras para as areas protegidas

+
+ {surface_links_markup} +
+
+
+ +
+
+

Fluxo recomendado

+

Como navegar no painel

+
+ {roadmap_markup} +
+
+
+
+
+
+
+
+ + + + + +""" + + +def render_login_page( + view: AdminLoginPageView, + *, + css_href: str, + js_href: str, +) -> str: + security_markup = _render_text_list(view.security_highlights) + notes_markup = _render_text_list(view.integration_notes) + + return f""" + + + + + {escape(view.title)} + + + + + +
+
+
+
+
+ +
+ +
+ +
+
+
+
+
+ + + + + +""" + + +def _render_navigation(items: tuple[AdminPanelNavigationItem, ...]) -> str: + links: list[str] = [] + for item in items: + badge_markup = "" + if item.badge: + badge_markup = ( + f'{escape(item.badge)}' + ) + + active_class = " active shadow-sm" if item.is_active else "" + links.append( + f""" + +
+
+
{escape(item.label)}
+
{escape(item.description)}
+
+ {badge_markup} +
+ +""".strip() + ) + return "\n".join(links) + + +def _render_quick_actions(items: tuple[AdminPanelQuickAction, ...]) -> str: + return "\n".join( + f'{escape(item.label)}' + for item in items + ) + + +def _render_metrics(items: tuple[AdminPanelMetric, ...]) -> str: + cards: list[str] = [] + for item in items: + cards.append( + f""" +
+
+
+

{escape(item.label)}

+
{escape(item.value)}
+

{escape(item.description)}

+
+
+
+""".strip() + ) + return "\n".join(cards) + + +def _render_modules(items: tuple[AdminPanelModuleCard, ...]) -> str: + cards: list[str] = [] + for item in items: + highlights = "".join( + f'
  • {escape(highlight)}
    • ' + for highlight in item.highlights + ) + cta_markup = 'Em preparacao' + if item.href and item.cta_label and item.is_available: + cta_markup = ( + f'{escape(item.cta_label)}' + ) + + badge_classes = _BADGE_CLASS_MAP.get(item.status_variant, _BADGE_CLASS_MAP["secondary"]) + cards.append( + f""" +
    +
    +
    +
    +
    +

    {escape(item.eyebrow)}

    +

    {escape(item.title)}

    +
    + {escape(item.status_label)} +
    +

    {escape(item.description)}

    +
      + {highlights} +
    +
    + {cta_markup} +
    +
    +
    +
    +""".strip() + ) + return "\n".join(cards) + + +def _render_surface_links(items: tuple[AdminPanelSurfaceLink, ...]) -> str: + cards: list[str] = [] + for item in items: + cards.append( + f""" + +
    +
    +
    {escape(item.method)}
    +
    {escape(item.label)}
    +
    {escape(item.description)}
    +
    + Abrir +
    +     +""".strip() + ) + return "\n".join(cards) + + +def _render_roadmap(items: tuple[AdminPanelRoadmapItem, ...]) -> str: + cards: list[str] = [] + for item in items: + cards.append( + f""" +
    +
    +
    +
    Etapa {escape(item.step)}
    +
    {escape(item.title)}
    +
    {escape(item.description)}
    +
    + {escape(item.status_label)} +
    +
    +""".strip() + ) + return "\n".join(cards) + + +def _render_text_list(items: tuple[str, ...]) -> str: + return "\n".join( + f"
  • {escape(item)}
    • " + for item in items + ) + + + +def render_tool_review_page( + view: AdminToolReviewPageView, + *, + css_href: str, + js_href: str, +) -> str: + workflow_markup = _render_tool_review_workflow(view.workflow) + review_notes_markup = _render_text_list(view.review_notes) + approval_notes_markup = _render_text_list(view.approval_notes) + activation_notes_markup = _render_text_list(view.activation_notes) + + return f""" + + + + + {escape(view.title)} + + + + + +
    +
    + + +
    +
    +
    +
    +
    +
    + Governanca de tools + Revisao e ativacao +
    +

    Fluxo visual de aprovacao no painel

    +

    + Esta tela conecta a sessao web do painel aos snapshots administrativos de tools para que o time consiga revisar a fila, conferir contratos e acompanhar o catalogo ativo. +

    +
    +
    + + Voltar ao overview +
    +
    +
    +
    + +
    + +
    +
    +
    +
    +

    Fila de revisao

    +
    0
    +

    Items aguardando leitura tecnica ou aprovacao humana.

    +
    +
    +
    +
    +
    +
    +

    Publicacoes ativas

    +
    0
    +

    Catalogo publicado e pronto para abastecer o runtime de produto.

    +
    +
    +
    +
    +
    +
    +

    Etapas do contrato

    +
    {len(view.workflow)}
    +

    Workflow compartilhado entre revisao, aprovacao e ativacao.

    +
    +
    +
    +
    +
    +
    +
    +
    +

    Pipeline visual

    +

    Etapas que a tela acompanha

    +

    Os cards abaixo resumem o trajeto de uma tool desde a analise ate a ativacao no produto.

    +
    +
    +
    + {workflow_markup} +
    +
    +
    + +
    +
    +
    +
    +
    +
    +

    Fila atual

    +

    Revisao tecnica e aprovacao

    +

    A fila abaixo e lida da superficie web do painel e respeita o papel da sessao autenticada.

    +
    + Bootstrap +
    +
    +
    +

    Nenhum item carregado ainda

    +

    Clique em atualizar leitura para sincronizar a fila de revisao do painel.

    +
    +
    +
    +
    +
    + +
    +
    +
    +
    +

    Checklist de aprovacao

    +

    Playbook para a decisao humana

    +

    Aprovacao e ativacao continuam controladas pelo papel administrativo e pela leitura do contrato compartilhado.

    +
    + +
    +

    Antes de aprovar

    +
      + {approval_notes_markup} +
    +
    + +
    +

    Lifecycle disponivel

    +
    +
    Aguardando leitura do contrato compartilhado...
    +
    +
    + +
    +

    Tipos de parametro

    +
    + Aguardando +
    +
    +
    +
    +
    +
    + +
    +
    +
    +
    +

    Catalogo ativo

    +

    Ativacao e superficie publicada

    +

    Quando a sessao tem permissao de publicacao, o painel tambem exibe o catalogo conhecido de tools ativas.

    +
    + Catalogo +
    +
    +
    +
    +

    Catalogo ainda nao sincronizado

    +

    A leitura da ativacao aparece aqui assim que a sessao web carregar as publicacoes disponiveis.

    +
    +
    +
    +
    +
    +
    +
    +
    + + + + + +""" + + +def _render_tool_review_workflow(items: tuple[AdminToolReviewWorkflowStep, ...]) -> str: + cards: list[str] = [] + for item in items: + badge_classes = _BADGE_CLASS_MAP.get(item.status_variant, _BADGE_CLASS_MAP["secondary"]) + cards.append( + f""" +
    +
    +
    +
    +

    {escape(item.eyebrow)}

    +

    {escape(item.title)}

    +
    + {escape(item.status_label)} +
    +

    {escape(item.description)}

    +
    +
    +""".strip() + ) + return "\n".join(cards) + + + + + diff --git a/admin_app/view/router.py b/admin_app/view/router.py new file mode 100644 index 0000000..b7583de --- /dev/null +++ b/admin_app/view/router.py @@ -0,0 +1,401 @@ +from fastapi import APIRouter, Depends, Request +from fastapi.responses import HTMLResponse, RedirectResponse, Response + +from admin_app.api.dependencies import get_optional_panel_staff_context +from admin_app.core import AdminSettings, AuthenticatedStaffContext, get_admin_settings +from admin_app.view.assets import PANEL_STATIC_MOUNT_NAME +from admin_app.view.rendering import render_login_page, render_panel_home, render_tool_review_page +from admin_app.view.view_models import ( + AdminLoginPageView, + AdminPanelHomeView, + AdminPanelMetric, + AdminPanelModuleCard, + AdminPanelNavigationItem, + AdminPanelQuickAction, + AdminPanelRoadmapItem, + AdminPanelSurfaceLink, + AdminToolReviewPageView, + AdminToolReviewWorkflowStep, +) +from shared.contracts import AdminPermission, StaffRole + +panel_router = APIRouter(tags=["panel"]) + + +@panel_router.get("/panel", name="panel_entry") +def panel_entry( + request: Request, + current_context: AuthenticatedStaffContext | None = Depends(get_optional_panel_staff_context), +) -> RedirectResponse: + target_route_name = "panel_home" if current_context is not None else "admin_login_view" + return _redirect_to_route(request, target_route_name) + + +@panel_router.get("/panel/admin", response_class=HTMLResponse, name="panel_home") +def panel_home( + request: Request, + current_context: AuthenticatedStaffContext | None = Depends(get_optional_panel_staff_context), +) -> Response: + if current_context is None: + return _redirect_to_route(request, "admin_login_view") + + settings = _resolve_settings(request) + view = _build_home_view(request, settings) + css_href = str(request.url_for(PANEL_STATIC_MOUNT_NAME, path="styles/panel.css")) + js_href = str(request.url_for(PANEL_STATIC_MOUNT_NAME, path="scripts/panel.js")) + return HTMLResponse(render_panel_home(view, css_href=css_href, js_href=js_href)) + + +@panel_router.get("/login", response_class=HTMLResponse, name="admin_login_view") +def login_page( + request: Request, + current_context: AuthenticatedStaffContext | None = Depends(get_optional_panel_staff_context), +) -> Response: + if current_context is not None: + return _redirect_to_route(request, "panel_home") + + settings = _resolve_settings(request) + view = _build_login_view(request, settings) + css_href = str(request.url_for(PANEL_STATIC_MOUNT_NAME, path="styles/panel.css")) + js_href = str(request.url_for(PANEL_STATIC_MOUNT_NAME, path="scripts/panel.js")) + return HTMLResponse(render_login_page(view, css_href=css_href, js_href=js_href)) + + +@panel_router.get("/panel/tools/review", response_class=HTMLResponse, name="admin_tool_review_view") +def tool_review_page( + request: Request, + current_context: AuthenticatedStaffContext | None = Depends(get_optional_panel_staff_context), +) -> Response: + if current_context is None: + return _redirect_to_route(request, "admin_login_view") + + settings = _resolve_settings(request) + view = _build_tool_review_view(request, settings) + css_href = str(request.url_for(PANEL_STATIC_MOUNT_NAME, path="styles/panel.css")) + js_href = str(request.url_for(PANEL_STATIC_MOUNT_NAME, path="scripts/panel.js")) + return HTMLResponse(render_tool_review_page(view, css_href=css_href, js_href=js_href)) + + +def _build_home_view(request: Request, settings: AdminSettings) -> AdminPanelHomeView: + panel_href = str(request.url_for("panel_home")) + tool_review_view_href = str(request.url_for("admin_tool_review_view")) + system_configuration_href = _build_prefixed_path(settings.admin_api_prefix, "/system/configuration") + audit_href = _build_prefixed_path(settings.admin_api_prefix, "/audit/events") + + return AdminPanelHomeView( + service="orquestrador-admin", + app_name=settings.admin_app_name, + panel_title="Painel Administrativo", + panel_subtitle=( + "Area interna protegida para operar o admin com mais clareza, foco e navegacao orientada por fluxo." + ), + environment=settings.admin_environment, + version=settings.admin_version, + api_prefix=settings.admin_api_prefix or "/", + release_label="Bootstrap UI v1", + navigation=( + AdminPanelNavigationItem( + label="Dashboard", + href=panel_href, + description="Entrada principal do ambiente interno.", + badge="Ativo", + is_active=True, + ), + AdminPanelNavigationItem( + label="Revisao de tools", + href=tool_review_view_href, + description="Fluxo humano de revisao, aprovacao e ativacao.", + badge="Operacao", + ), + AdminPanelNavigationItem( + label="Areas do sistema", + href="#modules", + description="Mapa claro dos modulos internos disponiveis.", + badge="Painel", + ), + AdminPanelNavigationItem( + label="Fluxo recomendado", + href="#workflow", + description="Sequencia sugerida para operar o admin.", + badge="Guia", + ), + ), + quick_actions=( + AdminPanelQuickAction( + label="Revisar tools", + href=tool_review_view_href, + button_class="btn-dark", + ), + AdminPanelQuickAction( + label="Ver areas", + href="#modules", + button_class="btn-outline-dark", + ), + AdminPanelQuickAction( + label="Ver fluxo", + href="#workflow", + button_class="btn-outline-secondary", + ), + ), + metrics=( + AdminPanelMetric( + label="Runtimes independentes", + value="2", + description="Produto e admin seguem isolados para deploy e operacao.", + ), + AdminPanelMetric( + label="Perfis internos", + value=str(len(StaffRole)), + description="Hierarquia base com viewer, staff e admin.", + ), + AdminPanelMetric( + label="Permissoes administrativas", + value=str(len(AdminPermission)), + description="Camada pronta para crescer por modulo sem misturar contexto.", + ), + AdminPanelMetric( + label="Refresh token", + value=f"{settings.admin_auth_refresh_token_ttl_days} dias", + description="Sessao web persistida com renovacao controlada.", + ), + ), + modules=( + AdminPanelModuleCard( + eyebrow="Fluxo principal", + title="Revisao de tools", + description="A principal area operacional do painel para leitura da fila, aprovacao humana e ativacao controlada.", + status_label="Tela ativa", + status_variant="success", + highlights=( + "Fila protegida por sessao web", + "Catalogo ativo para comparacao", + "Leitura clara do workflow de aprovacao", + ), + cta_label="Abrir revisao", + href=tool_review_view_href, + is_available=True, + ), + AdminPanelModuleCard( + eyebrow="Acompanhamento", + title="Configuracao do sistema", + description="Snapshot do runtime administrativo, politicas de seguranca e dados de sessao do painel.", + status_label="API pronta", + status_variant="secondary", + highlights=( + "Runtime e banco monitorados", + "Politicas de credencial centralizadas", + "Base pronta para futura tela dedicada", + ), + ), + AdminPanelModuleCard( + eyebrow="Governanca", + title="Auditoria operacional", + description="Eventos de login, logout, aprovacao e publicacao continuam registrados para rastreabilidade.", + status_label="Auditavel", + status_variant="secondary", + highlights=( + "Historico de operacao interna", + "Base para filtros e timeline", + "Suporte a conformidade do fluxo administrativo", + ), + ), + AdminPanelModuleCard( + eyebrow="Seguranca", + title="Sessao administrativa", + description="Acesso ao painel protegido por StaffAccount, token assinado e refresh token rotacionado.", + status_label="Protegido", + status_variant="success", + highlights=( + "StaffAccount isolado do usuario final", + "Cookies httpOnly no navegador", + "Rotacao controlada da sessao web", + ), + ), + ), + surface_links=( + AdminPanelSurfaceLink( + method="Acesso", + label="Dashboard administrativa", + href=panel_href, + description="Entrada principal do time interno depois do login.", + ), + AdminPanelSurfaceLink( + method="Operacao", + label="Revisao de tools", + href=tool_review_view_href, + description="Area com fila, contrato e catalogo ativo para tomada de decisao.", + ), + AdminPanelSurfaceLink( + method="Runtime", + label="Configuracao do sistema", + href=system_configuration_href, + description="Snapshot tecnico do ambiente, mantido como superficie protegida enquanto a tela visual nao chega.", + ), + AdminPanelSurfaceLink( + method="Auditoria", + label="Eventos administrativos", + href=audit_href, + description="Consulta de eventos internos para rastrear operacoes sensiveis.", + ), + ), + roadmap=( + AdminPanelRoadmapItem( + step="01", + title="Entrar pelo login administrativo", + description="A sessao web libera o ambiente interno e evita navegacao confusa antes da autenticacao.", + status_label="Obrigatorio", + ), + AdminPanelRoadmapItem( + step="02", + title="Passar pela dashboard", + description="A home protegida organiza os modulos e mostra por onde comecar a operacao.", + status_label="Entrada", + ), + AdminPanelRoadmapItem( + step="03", + title="Abrir revisao de tools", + description="Use o hub de revisao para analisar fila, contrato e ativacao das tools.", + status_label="Principal", + ), + AdminPanelRoadmapItem( + step="04", + title="Consultar runtime e auditoria", + description="Quando necessario, acompanhe configuracao e eventos do admin para suportar a decisao operacional.", + status_label="Suporte", + ), + ), + ) + + +def _build_login_view(request: Request, settings: AdminSettings) -> AdminLoginPageView: + dashboard_href = str(request.url_for("panel_home")) + auth_endpoint = _build_prefixed_path(settings.admin_api_prefix, "/panel/auth/login") + session_endpoint = _build_prefixed_path(settings.admin_api_prefix, "/panel/auth/session") + logout_endpoint = _build_prefixed_path(settings.admin_api_prefix, "/panel/auth/logout") + password_requirements = [] + if settings.admin_auth_password_require_uppercase: + password_requirements.append("maiuscula") + if settings.admin_auth_password_require_lowercase: + password_requirements.append("minuscula") + if settings.admin_auth_password_require_digit: + password_requirements.append("digito") + if settings.admin_auth_password_require_symbol: + password_requirements.append("simbolo") + + password_policy_label = ( + f"Minimo de {settings.admin_auth_password_min_length} caracteres" + + (f" com {', '.join(password_requirements)}." if password_requirements else ".") + ) + + return AdminLoginPageView( + app_name=settings.admin_app_name, + title="Login administrativo", + subtitle=( + "Entre primeiro com sua conta interna. A dashboard e os modulos do sistema so aparecem depois da autenticacao." + ), + environment=settings.admin_environment, + version=settings.admin_version, + dashboard_href=dashboard_href, + auth_endpoint=auth_endpoint, + session_endpoint=session_endpoint, + logout_endpoint=logout_endpoint, + email_placeholder="voce@empresa.com", + password_placeholder="Sua senha administrativa", + access_token_ttl_label=f"{settings.admin_auth_access_token_ttl_minutes} minutos", + refresh_token_ttl_label=f"{settings.admin_auth_refresh_token_ttl_days} dias", + password_policy_label=password_policy_label, + security_highlights=( + "Identidade separada do usuario de atendimento", + "Rotacao de refresh token ja implementada", + "Trilha de auditoria para login e logout", + ), + integration_notes=( + "A dashboard administrativa so aparece depois da autenticacao do StaffAccount.", + "Revisao, configuracao e operacao interna ficam atras da sessao web do painel.", + "Cookies httpOnly e refresh token rotacionado mantem a sessao do navegador protegida.", + ), + ) + + +def _build_tool_review_view(request: Request, settings: AdminSettings) -> AdminToolReviewPageView: + dashboard_href = str(request.url_for("panel_home")) + login_href = str(request.url_for("admin_login_view")) + overview_endpoint = _build_prefixed_path(settings.admin_api_prefix, "/panel/tools/overview") + contracts_endpoint = _build_prefixed_path(settings.admin_api_prefix, "/panel/tools/contracts") + review_queue_endpoint = _build_prefixed_path(settings.admin_api_prefix, "/panel/tools/review-queue") + publications_endpoint = _build_prefixed_path(settings.admin_api_prefix, "/panel/tools/publications") + + return AdminToolReviewPageView( + app_name=settings.admin_app_name, + title="Revisao, aprovacao e ativacao", + subtitle=( + "Hub visual para o time interno acompanhar a fila de revisao, validar o contrato compartilhado e inspecionar o catalogo de tools ativas antes da ativacao." + ), + environment=settings.admin_environment, + version=settings.admin_version, + dashboard_href=dashboard_href, + login_href=login_href, + overview_endpoint=overview_endpoint, + contracts_endpoint=contracts_endpoint, + review_queue_endpoint=review_queue_endpoint, + publications_endpoint=publications_endpoint, + workflow=( + AdminToolReviewWorkflowStep( + eyebrow="Leitura inicial", + title="Revisar fila", + description="Carregar a fila de geracao e entender em que gate cada item se encontra.", + status_label="Revisao", + status_variant="info", + ), + AdminToolReviewWorkflowStep( + eyebrow="Decisao humana", + title="Aprovar com criterio", + description="Conferir contrato, parametros e prontidao tecnica antes de liberar a proxima etapa.", + status_label="Aprovacao", + status_variant="warning", + ), + AdminToolReviewWorkflowStep( + eyebrow="Publicacao", + title="Ativar no catalogo", + description="Usar o catalogo publicado como referencia para a versao que chega ao runtime de produto.", + status_label="Ativacao", + status_variant="success", + ), + ), + review_notes=( + "Conferir se o gate do item combina com o estado esperado do lifecycle.", + "Observar se a descricao e o objetivo operacional da tool estao claros para o time.", + "Usar o catalogo ativo como comparativo antes de promover uma nova versao.", + ), + approval_notes=( + "Verificar nome, descricao e semantica dos parametros antes da aprovacao.", + "Confirmar se a tool respeita a separacao entre admin e product definida nas ADRs.", + "Checar se a publicacao planejada e auditavel e segura para o runtime de produto.", + ), + activation_notes=( + "Publicacoes ativas exigem papel com permissao publish_tools.", + "A leitura do catalogo e feita via sessao web do painel para facilitar a operacao do navegador.", + "Sem permissao de publicacao, a tela continua util para revisao, mas bloqueia o catalogo ativo.", + ), + ) + + +def _redirect_to_route(request: Request, route_name: str) -> RedirectResponse: + return RedirectResponse(url=str(request.url_for(route_name)), status_code=302) + + +def _resolve_settings(request: Request) -> AdminSettings: + app_settings = getattr(request.app.state, "admin_settings", None) + if isinstance(app_settings, AdminSettings): + return app_settings + return get_admin_settings() + + +def _build_prefixed_path(api_prefix: str, path: str) -> str: + normalized_prefix = api_prefix.rstrip("/") + normalized_path = path if path.startswith("/") else f"/{path}" + if not normalized_prefix: + return normalized_path + if normalized_path == "/": + return f"{normalized_prefix}/" + return f"{normalized_prefix}{normalized_path}" diff --git a/admin_app/view/static/scripts/panel.js b/admin_app/view/static/scripts/panel.js new file mode 100644 index 0000000..6fceafa --- /dev/null +++ b/admin_app/view/static/scripts/panel.js @@ -0,0 +1,249 @@ +document.documentElement.dataset.panelReady = "true"; + +const loginForm = document.querySelector('[data-admin-login-form="true"]'); +const reviewBoard = document.querySelector('[data-admin-tool-review-board="true"]'); + +if (loginForm) { + mountLoginForm(loginForm); +} + +if (reviewBoard) { + mountToolReviewBoard(reviewBoard); +} + +function mountLoginForm(form) { + const feedback = document.getElementById("admin-login-feedback"); + const submitButton = form.querySelector('button[type="submit"]'); + const submitLabel = form.querySelector("[data-submit-label]"); + const submitSpinner = form.querySelector("[data-submit-spinner]"); + + form.addEventListener("submit", async (event) => { + event.preventDefault(); + toggleLoading(true); + clearFeedback(); + + const formData = new FormData(form); + const payload = { + email: String(formData.get("email") || "").trim(), + password: String(formData.get("password") || ""), + }; + + try { + const response = await fetch(form.dataset.authEndpoint, { + method: "POST", + credentials: "same-origin", + headers: { + "Content-Type": "application/json", + Accept: "application/json", + }, + body: JSON.stringify(payload), + }); + const authBody = await readJson(response); + if (!response.ok) { + throw new Error(authBody?.detail || "Nao foi possivel autenticar no admin."); + } + + showFeedback("success", authBody?.message || "Sessao administrativa web iniciada com sucesso."); + form.reset(); + + const redirectTo = authBody?.redirect_to || form.dataset.dashboardHref; + if (redirectTo) { + window.setTimeout(() => { + window.location.assign(redirectTo); + }, 250); + } + } catch (error) { + showFeedback("danger", error instanceof Error ? error.message : "Erro inesperado durante o login."); + } finally { + toggleLoading(false); + } + }); + + function toggleLoading(isLoading) { + submitButton.disabled = isLoading; + submitSpinner.classList.toggle("d-none", !isLoading); + submitLabel.textContent = isLoading ? "Validando acesso..." : "Entrar no painel"; + } + + function clearFeedback() { + feedback.className = "alert d-none mt-4 mb-0 rounded-4"; + feedback.textContent = ""; + } + + function showFeedback(variant, message) { + feedback.className = `alert alert-${variant} mt-4 mb-0 rounded-4`; + feedback.textContent = message; + } +} + +function mountToolReviewBoard(board) { + const refreshButton = board.querySelector("[data-admin-tool-refresh]"); + const refreshLabel = board.querySelector("[data-tool-refresh-label]"); + const refreshSpinner = board.querySelector("[data-tool-refresh-spinner]"); + const feedback = document.getElementById("admin-tool-review-feedback"); + const queueList = board.querySelector("[data-tool-review-queue-list]"); + const publicationList = board.querySelector("[data-tool-publication-list]"); + const lifecycleList = board.querySelector("[data-tool-contract-lifecycle]"); + const parameterTypes = board.querySelector("[data-tool-parameter-types]"); + + if (refreshButton) { + refreshButton.addEventListener("click", () => { + void loadBoard(); + }); + } + + void loadBoard(); + + async function loadBoard() { + toggleRefreshing(true); + clearFeedback(); + + const overviewResult = await fetchPanelJson(board.dataset.overviewEndpoint); + const contractsResult = await fetchPanelJson(board.dataset.contractsEndpoint); + const reviewQueueResult = await fetchPanelJson(board.dataset.reviewQueueEndpoint); + const publicationsResult = await fetchPanelJson(board.dataset.publicationsEndpoint); + + if (!overviewResult.ok && !contractsResult.ok && !reviewQueueResult.ok && !publicationsResult.ok) { + showFeedback("warning", overviewResult.message || "Entre com uma sessao administrativa web para carregar esta tela."); + } + + if (overviewResult.ok) { + renderOverview(overviewResult.body); + } + if (contractsResult.ok) { + renderContracts(contractsResult.body); + } else { + renderLockedLifecycle(contractsResult.message); + } + if (reviewQueueResult.ok) { + renderReviewQueue(reviewQueueResult.body); + } else { + renderLockedQueue(reviewQueueResult.message); + } + if (publicationsResult.ok) { + renderPublications(publicationsResult.body); + } else { + renderLockedPublications(publicationsResult.message); + } + + setText("[data-tool-review-last-sync]", formatNow()); + toggleRefreshing(false); + } + + function toggleRefreshing(isLoading) { + if (!refreshButton || !refreshLabel || !refreshSpinner) { + return; + } + refreshButton.disabled = isLoading; + refreshSpinner.classList.toggle("d-none", !isLoading); + refreshLabel.textContent = isLoading ? "Atualizando..." : "Atualizar leitura"; + } + + function clearFeedback() { + feedback.className = "alert d-none rounded-4 mb-4"; + feedback.textContent = ""; + } + + function showFeedback(variant, message) { + feedback.className = `alert alert-${variant} rounded-4 mb-4`; + feedback.textContent = message; + } + + function renderOverview(payload) { + const workflow = Array.isArray(payload?.workflow) ? payload.workflow : []; + const nextSteps = Array.isArray(payload?.next_steps) ? payload.next_steps : []; + setText("[data-tool-review-lifecycle-count]", String(workflow.length || 0)); + if (nextSteps.length > 0 && !feedback.textContent) { + showFeedback("info", `Proximos passos: ${nextSteps[0]}`); + } + } + + function renderContracts(payload) { + const lifecycle = Array.isArray(payload?.lifecycle_statuses) ? payload.lifecycle_statuses : []; + const parameterTypeList = Array.isArray(payload?.parameter_types) ? payload.parameter_types : []; + lifecycleList.innerHTML = lifecycle.length > 0 + ? lifecycle.map((item) => `
    ${escapeHtml(item.label)}
    ${escapeHtml(item.description)}
    `).join("") + : `
    Nenhuma etapa disponivel.
    `; + parameterTypes.innerHTML = parameterTypeList.length > 0 + ? parameterTypeList.map((item) => `${escapeHtml(item.label)}`).join("") + : `Sem tipos`; + } + + function renderLockedLifecycle(message) { + lifecycleList.innerHTML = `
    Leitura indisponivel
    ${escapeHtml(message || "A sessao atual nao pode ler o contrato compartilhado.")}
    `; + parameterTypes.innerHTML = `Bloqueado`; + } + + function renderReviewQueue(payload) { + const items = Array.isArray(payload?.items) ? payload.items : []; + setText("[data-tool-review-queue-count]", String(items.length)); + setText("[data-tool-review-queue-mode]", payload?.queue_mode || "Fila web"); + queueList.innerHTML = items.length > 0 + ? items.map((item) => `
    ${escapeHtml(item.gate || "revisao")}

    ${escapeHtml(item.display_name || item.tool_name || "Tool")}

    ${escapeHtml(item.tool_name || "")}
    ${escapeHtml(item.status || "pendente")}

    ${escapeHtml(item.summary || payload?.message || "Item aguardando analise do time.")}

    `).join("") + : `

    Fila sem itens no momento

    ${escapeHtml(payload?.message || "Nenhuma tool aguardando revisao agora.")}

    `; + } + + function renderLockedQueue(message) { + setText("[data-tool-review-queue-count]", "0"); + setText("[data-tool-review-queue-mode]", "Bloqueado"); + queueList.innerHTML = `

    Fila indisponivel

    ${escapeHtml(message || "A sessao atual nao pode acessar a fila de revisao.")}

    `; + } + + function renderPublications(payload) { + const items = Array.isArray(payload?.publications) ? payload.publications : []; + setText("[data-tool-review-publication-count]", String(items.length)); + setText("[data-tool-publication-source]", payload?.source || "Catalogo web"); + publicationList.innerHTML = items.length > 0 + ? items.slice(0, 9).map((item) => `
    ${escapeHtml(item.domain || "tool")}

    ${escapeHtml(item.display_name || item.tool_name || "Tool")}

    ${escapeHtml(item.tool_name || "")}
    v${escapeHtml(String(item.version || 1))}

    ${escapeHtml(item.description || "Publicacao ativa no catalogo do produto.")}

    ${escapeHtml(item.implementation_module || "")}
    `).join("") + : `

    Catalogo ativo vazio

    Nenhuma publicacao ativa retornada pela sessao web.

    `; + } + + function renderLockedPublications(message) { + setText("[data-tool-review-publication-count]", "0"); + setText("[data-tool-publication-source]", "Bloqueado"); + publicationList.innerHTML = `

    Catalogo protegido

    ${escapeHtml(message || "A sessao atual nao possui permissao para ler as publicacoes ativas.")}

    `; + } +} + +async function fetchPanelJson(url) { + const response = await fetch(url, { + credentials: "same-origin", + headers: { Accept: "application/json" }, + }); + const body = await readJson(response); + if (response.ok) { + return { ok: true, body }; + } + const defaultMessage = response.status === 401 + ? "Entre com uma sessao administrativa web para visualizar esta area." + : body?.detail || "Nao foi possivel carregar os dados desta superficie."; + return { ok: false, body, message: defaultMessage }; +} + +async function readJson(response) { + try { + return await response.json(); + } catch { + return null; + } +} + +function setText(selector, value) { + const target = document.querySelector(selector); + if (target) { + target.textContent = value; + } +} + +function formatNow() { + return new Date().toLocaleTimeString("pt-BR", { hour: "2-digit", minute: "2-digit" }); +} + +function escapeHtml(value) { + return String(value || "") + .replaceAll("&", "&") + .replaceAll("<", "<") + .replaceAll(">", ">") + .replaceAll('"', """) + .replaceAll("'", "'"); +} diff --git a/admin_app/view/static/styles/panel.css b/admin_app/view/static/styles/panel.css new file mode 100644 index 0000000..27bc66f --- /dev/null +++ b/admin_app/view/static/styles/panel.css @@ -0,0 +1,261 @@ +:root { + --admin-bg: #f6f1e8; + --admin-surface: rgba(255, 255, 255, 0.84); + --admin-surface-strong: rgba(255, 255, 255, 0.92); + --admin-ink: #20242f; + --admin-muted: #677084; + --admin-accent: #144d47; + --admin-accent-soft: rgba(20, 77, 71, 0.08); + --admin-line: rgba(32, 36, 47, 0.08); + --admin-shadow: 0 24px 60px rgba(56, 44, 23, 0.11); +} + +body.admin-view-body { + min-height: 100vh; + color: var(--admin-ink); + background: + radial-gradient(circle at top left, rgba(20, 77, 71, 0.18), transparent 28%), + radial-gradient(circle at top right, rgba(193, 106, 51, 0.16), transparent 24%), + linear-gradient(180deg, #fbf7f1 0%, var(--admin-bg) 100%); +} + +.admin-shell-card, +.admin-hero-card, +.admin-surface-card, +.admin-metric-card, +.admin-module-card, +.admin-login-card, +.admin-login-info-card { + background: var(--admin-surface); + backdrop-filter: blur(18px); + box-shadow: var(--admin-shadow); +} + +.admin-shell-card, +.admin-hero-card, +.admin-surface-card, +.admin-metric-card, +.admin-login-card, +.admin-login-info-card { + border-radius: 1.75rem; +} + +.admin-module-card, +.admin-roadmap-item, +.admin-runtime-block, +.admin-login-kpi, +.admin-login-note, +.admin-login-policy, +.admin-login-session-card { + border-radius: 1.35rem; +} + +.admin-sidebar-sticky { + position: sticky; + top: 1.5rem; +} + +.admin-runtime-block, +.admin-module-card, +.admin-roadmap-item, +.admin-surface-link, +.admin-login-kpi, +.admin-login-note, +.admin-login-policy, +.admin-login-session-card { + background: var(--admin-surface-strong); + border: 1px solid var(--admin-line); +} + +.admin-hero-card, +.admin-login-info-card, +.admin-login-card { + position: relative; + overflow: hidden; + border: 1px solid rgba(20, 77, 71, 0.08); +} + +.admin-hero-card::after, +.admin-login-info-card::after, +.admin-login-card::after { + content: ""; + position: absolute; + inset: auto -6rem -8rem auto; + width: 16rem; + height: 16rem; + border-radius: 50%; + background: radial-gradient(circle, rgba(20, 77, 71, 0.17), transparent 72%); +} + +.admin-nav-link { + background: rgba(255, 255, 255, 0.7); + border: 1px solid transparent; +} + +.admin-nav-link:hover { + background: rgba(255, 255, 255, 0.94); + border-color: rgba(20, 77, 71, 0.1); +} + +.admin-nav-link.active { + background: linear-gradient(135deg, #163f3a 0%, #215a53 100%); + color: #fff; +} + +.admin-nav-link.active .text-dark, +.admin-nav-link.active .text-secondary, +.admin-nav-link.active .badge { + color: #fff !important; +} + +.admin-nav-link.active .badge { + background: rgba(255, 255, 255, 0.14) !important; + border-color: rgba(255, 255, 255, 0.16) !important; +} + +.admin-metric-card { + border: 1px solid rgba(20, 77, 71, 0.06); +} + +.admin-module-card { + box-shadow: inset 0 1px 0 rgba(255, 255, 255, 0.5); +} + +.admin-surface-link { + padding: 1rem 1.1rem; + margin-bottom: 0.75rem; +} + +.admin-surface-link:hover { + background: rgba(20, 77, 71, 0.05); +} + +.admin-roadmap-item { + position: relative; + padding-left: 1.3rem !important; +} + +.admin-roadmap-item::before { + content: ""; + position: absolute; + top: 1rem; + bottom: 1rem; + left: 0; + width: 4px; + border-radius: 999px; + background: linear-gradient(180deg, rgba(20, 77, 71, 0.72), rgba(20, 77, 71, 0.18)); +} + +.admin-quick-actions .btn { + min-height: 3.25rem; +} + +.admin-login-form .form-control { + border: 1px solid rgba(20, 77, 71, 0.12); + background: rgba(255, 255, 255, 0.9); +} + +.admin-login-form .form-control:focus { + border-color: rgba(20, 77, 71, 0.32); + box-shadow: 0 0 0 0.25rem rgba(20, 77, 71, 0.12); +} + +.admin-login-policy, +.admin-login-kpi, +.admin-login-note, +.admin-login-session-card { + position: relative; + z-index: 1; +} + +#admin-login-feedback { + position: relative; + z-index: 1; +} + +[data-panel-ready="true"] .admin-hero-card, +[data-panel-ready="true"] .admin-shell-card, +[data-panel-ready="true"] .admin-surface-card, +[data-panel-ready="true"] .admin-metric-card, +[data-panel-ready="true"] .admin-login-card, +[data-panel-ready="true"] .admin-login-info-card, +[data-panel-ready="true"] .admin-login-session-card { + animation: admin-fade-up 520ms ease both; +} + +[data-panel-ready="true"] .admin-metric-card:nth-child(2), +[data-panel-ready="true"] .admin-surface-card:nth-of-type(2), +[data-panel-ready="true"] .admin-login-info-card { + animation-delay: 120ms; +} + +@keyframes admin-fade-up { + from { + opacity: 0; + transform: translateY(14px); + } + + to { + opacity: 1; + transform: translateY(0); + } +} + +@media (max-width: 1199px) { + .admin-sidebar-sticky { + position: static; + } +} + +@media (max-width: 767px) { + .admin-shell-card, + .admin-hero-card, + .admin-surface-card, + .admin-metric-card, + .admin-login-card, + .admin-login-info-card { + border-radius: 1.4rem; + } + + .admin-module-card, + .admin-roadmap-item, + .admin-runtime-block, + .admin-login-kpi, + .admin-login-note, + .admin-login-policy, + .admin-login-session-card { + border-radius: 1.1rem; + } +} + +.admin-tool-review-note, +.admin-tool-workflow-card, +.admin-tool-review-card, +.admin-tool-publication-card, +.admin-tool-inline-note, +.admin-tool-empty-state { + background: var(--admin-surface-strong); + border: 1px solid var(--admin-line); + border-radius: 1.35rem; +} + +.admin-tool-workflow-card, +.admin-tool-review-card, +.admin-tool-publication-card, +.admin-tool-empty-state, +.admin-tool-inline-note { + box-shadow: inset 0 1px 0 rgba(255, 255, 255, 0.5); +} + +.admin-tool-review-grid { + display: grid; + gap: 1rem; +} + +.admin-tool-inline-note { + padding: 0.9rem 1rem; +} + +.admin-tool-review-page .admin-hero-card::after { + background: radial-gradient(circle, rgba(20, 77, 71, 0.2), transparent 72%); +} diff --git a/admin_app/view/view_models.py b/admin_app/view/view_models.py new file mode 100644 index 0000000..bad5fc1 --- /dev/null +++ b/admin_app/view/view_models.py @@ -0,0 +1,109 @@ +from pydantic import BaseModel + + +class AdminPanelNavigationItem(BaseModel): + label: str + href: str + description: str + badge: str | None = None + is_active: bool = False + + +class AdminPanelQuickAction(BaseModel): + label: str + href: str + button_class: str = "btn-outline-dark" + + +class AdminPanelMetric(BaseModel): + label: str + value: str + description: str + + +class AdminPanelModuleCard(BaseModel): + eyebrow: str + title: str + description: str + status_label: str + status_variant: str = "secondary" + highlights: tuple[str, ...] = () + cta_label: str | None = None + href: str | None = None + is_available: bool = False + + +class AdminPanelSurfaceLink(BaseModel): + method: str + label: str + href: str + description: str + + +class AdminPanelRoadmapItem(BaseModel): + step: str + title: str + description: str + status_label: str + + +class AdminPanelHomeView(BaseModel): + service: str + app_name: str + panel_title: str + panel_subtitle: str + environment: str + version: str + api_prefix: str + release_label: str + navigation: tuple[AdminPanelNavigationItem, ...] + quick_actions: tuple[AdminPanelQuickAction, ...] + metrics: tuple[AdminPanelMetric, ...] + modules: tuple[AdminPanelModuleCard, ...] + surface_links: tuple[AdminPanelSurfaceLink, ...] + roadmap: tuple[AdminPanelRoadmapItem, ...] + + +class AdminLoginPageView(BaseModel): + app_name: str + title: str + subtitle: str + environment: str + version: str + dashboard_href: str + auth_endpoint: str + session_endpoint: str + logout_endpoint: str + email_placeholder: str + password_placeholder: str + access_token_ttl_label: str + refresh_token_ttl_label: str + password_policy_label: str + security_highlights: tuple[str, ...] + integration_notes: tuple[str, ...] + + +class AdminToolReviewWorkflowStep(BaseModel): + eyebrow: str + title: str + description: str + status_label: str + status_variant: str = "secondary" + + +class AdminToolReviewPageView(BaseModel): + app_name: str + title: str + subtitle: str + environment: str + version: str + dashboard_href: str + login_href: str + overview_endpoint: str + contracts_endpoint: str + review_queue_endpoint: str + publications_endpoint: str + workflow: tuple[AdminToolReviewWorkflowStep, ...] + review_notes: tuple[str, ...] + approval_notes: tuple[str, ...] + activation_notes: tuple[str, ...] diff --git a/tests/test_admin_app_bootstrap.py b/tests/test_admin_app_bootstrap.py index 9bf6b0d..142e407 100644 --- a/tests/test_admin_app_bootstrap.py +++ b/tests/test_admin_app_bootstrap.py @@ -1,4 +1,4 @@ -import unittest +import unittest from fastapi.testclient import TestClient @@ -9,7 +9,7 @@ from shared.contracts import StaffRole class AdminAppBootstrapTests(unittest.TestCase): - def test_admin_app_root_endpoint(self): + def test_admin_app_root_endpoint_returns_json_for_non_browser_requests(self): app = create_app(AdminSettings(admin_environment="staging")) client = TestClient(app) @@ -26,6 +26,15 @@ class AdminAppBootstrapTests(unittest.TestCase): }, ) + def test_admin_app_root_endpoint_redirects_browser_to_login(self): + app = create_app(AdminSettings()) + client = TestClient(app) + + response = client.get("/", headers={"accept": "text/html"}, follow_redirects=False) + + self.assertEqual(response.status_code, 302) + self.assertTrue(response.headers["location"].endswith("/login")) + def test_admin_app_health_endpoint(self): app = create_app(AdminSettings(admin_version="1.2.3")) client = TestClient(app) diff --git a/tests/test_admin_auth_service.py b/tests/test_admin_auth_service.py index 604b0f8..06ce45c 100644 --- a/tests/test_admin_auth_service.py +++ b/tests/test_admin_auth_service.py @@ -1,4 +1,4 @@ -import unittest +import unittest from datetime import datetime, timedelta, timezone from admin_app.core import AdminSecurityService, AdminSettings @@ -182,6 +182,21 @@ class AdminAuthServiceTests(unittest.TestCase): with self.assertRaises(ValueError): self.auth_service.get_authenticated_context(session.access_token) + def test_get_authenticated_context_accepts_naive_session_expiry_from_database(self): + session = self.auth_service.login( + email="admin@empresa.com", + password="SenhaMuitoSegura!123", + ip_address="127.0.0.1", + user_agent="unittest", + ) + stored_session = self.session_repository.get_by_id(session.session_id) + stored_session.expires_at = stored_session.expires_at.replace(tzinfo=None) + self.session_repository.save(stored_session) + + context = self.auth_service.get_authenticated_context(session.access_token) + + self.assertEqual(context.session_id, session.session_id) + self.assertEqual(context.principal.email, "admin@empresa.com") def test_refresh_session_rejects_expired_session(self): session = self.auth_service.login( email="admin@empresa.com", @@ -203,4 +218,4 @@ class AdminAuthServiceTests(unittest.TestCase): if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/tests/test_admin_panel_auth_web.py b/tests/test_admin_panel_auth_web.py new file mode 100644 index 0000000..6bf4130 --- /dev/null +++ b/tests/test_admin_panel_auth_web.py @@ -0,0 +1,163 @@ +import unittest + +from fastapi.testclient import TestClient + +from admin_app.api.dependencies import get_auth_service, get_current_panel_staff_context +from admin_app.api.panel_session import PANEL_ACCESS_COOKIE_NAME, PANEL_REFRESH_COOKIE_NAME +from admin_app.app_factory import create_app +from admin_app.core import ( + AdminAuthenticatedSession, + AdminSettings, + AuthenticatedStaffContext, + AuthenticatedStaffPrincipal, +) +from shared.contracts import StaffRole + + +class _FakePanelAuthService: + def login(self, email: str, password: str, *, ip_address: str | None, user_agent: str | None): + if email == "admin@empresa.com" and password == "SenhaMuitoSegura!123": + principal = AuthenticatedStaffPrincipal( + id=1, + email="admin@empresa.com", + display_name="Administrador", + role=StaffRole.ADMIN, + is_active=True, + ) + return AdminAuthenticatedSession( + session_id=77, + access_token="panel-access-token", + refresh_token="panel-refresh-token", + token_type="bearer", + expires_in_seconds=1800, + principal=principal, + ) + return None + + def refresh_session(self, refresh_token: str, *, ip_address: str | None, user_agent: str | None): + if refresh_token == "panel-refresh-token": + principal = AuthenticatedStaffPrincipal( + id=1, + email="admin@empresa.com", + display_name="Administrador", + role=StaffRole.ADMIN, + is_active=True, + ) + return AdminAuthenticatedSession( + session_id=77, + access_token="panel-access-token-next", + refresh_token="panel-refresh-token-next", + token_type="bearer", + expires_in_seconds=1800, + principal=principal, + ) + return None + + def get_authenticated_context(self, access_token: str) -> AuthenticatedStaffContext: + if access_token in {"panel-access-token", "panel-access-token-next"}: + return AuthenticatedStaffContext( + principal=AuthenticatedStaffPrincipal( + id=1, + email="admin@empresa.com", + display_name="Administrador", + role=StaffRole.ADMIN, + is_active=True, + ), + session_id=77, + ) + raise ValueError("invalid token") + + def logout( + self, + session_id: int, + *, + actor_staff_account_id: int | None, + ip_address: str | None, + user_agent: str | None, + ) -> bool: + return session_id == 77 + + def logout_by_refresh_token( + self, + refresh_token: str, + *, + ip_address: str | None, + user_agent: str | None, + ) -> int | None: + if refresh_token in {"panel-refresh-token", "panel-refresh-token-next"}: + return 77 + return None + + +class AdminPanelAuthWebTests(unittest.TestCase): + def setUp(self): + app = create_app(AdminSettings(admin_auth_token_secret="test-secret")) + app.dependency_overrides[get_auth_service] = lambda: _FakePanelAuthService() + self.client = TestClient(app) + self.app = app + + def tearDown(self): + self.app.dependency_overrides.clear() + + def test_panel_login_sets_http_only_cookies_and_returns_session_payload(self): + response = self.client.post( + "/panel/auth/login", + json={"email": "admin@empresa.com", "password": "SenhaMuitoSegura!123"}, + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["session_id"], 77) + self.assertEqual(response.json()["redirect_to"], "/panel/admin") + self.assertEqual(response.cookies.get(PANEL_ACCESS_COOKIE_NAME), "panel-access-token") + self.assertEqual(response.cookies.get(PANEL_REFRESH_COOKIE_NAME), "panel-refresh-token") + set_cookie_headers = response.headers.get_list("set-cookie") + self.assertTrue(any("HttpOnly" in header for header in set_cookie_headers)) + + def test_panel_refresh_rotates_cookie_backed_session(self): + self.client.cookies.set(PANEL_REFRESH_COOKIE_NAME, "panel-refresh-token") + + response = self.client.post("/panel/auth/refresh") + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["message"], "Sessao administrativa web renovada.") + self.assertEqual(response.cookies.get(PANEL_ACCESS_COOKIE_NAME), "panel-access-token-next") + self.assertEqual(response.cookies.get(PANEL_REFRESH_COOKIE_NAME), "panel-refresh-token-next") + + def test_panel_session_reads_authenticated_staff_from_cookie_context(self): + app = create_app(AdminSettings(admin_auth_token_secret="test-secret")) + app.dependency_overrides[get_current_panel_staff_context] = lambda: AuthenticatedStaffContext( + principal=AuthenticatedStaffPrincipal( + id=1, + email="admin@empresa.com", + display_name="Administrador", + role=StaffRole.ADMIN, + is_active=True, + ), + session_id=77, + ) + client = TestClient(app) + try: + response = client.get("/panel/auth/session") + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["staff_account"]["email"], "admin@empresa.com") + self.assertEqual(response.json()["session_id"], 77) + + def test_panel_logout_clears_cookies_and_returns_login_redirect(self): + self.client.cookies.set(PANEL_ACCESS_COOKIE_NAME, "panel-access-token") + self.client.cookies.set(PANEL_REFRESH_COOKIE_NAME, "panel-refresh-token") + + response = self.client.post("/panel/auth/logout") + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["redirect_to"], "/login") + set_cookie_headers = response.headers.get_list("set-cookie") + self.assertTrue(any(PANEL_ACCESS_COOKIE_NAME in header and "Max-Age=0" in header for header in set_cookie_headers)) + self.assertTrue(any(PANEL_REFRESH_COOKIE_NAME in header and "Max-Age=0" in header for header in set_cookie_headers)) + + +if __name__ == "__main__": + unittest.main() + diff --git a/tests/test_admin_panel_tools_web.py b/tests/test_admin_panel_tools_web.py new file mode 100644 index 0000000..37d8d60 --- /dev/null +++ b/tests/test_admin_panel_tools_web.py @@ -0,0 +1,83 @@ +import unittest + +from fastapi.testclient import TestClient + +from admin_app.api.dependencies import get_current_panel_staff_principal +from admin_app.app_factory import create_app +from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal +from shared.contracts import StaffRole + + +class AdminPanelToolsWebTests(unittest.TestCase): + def _build_client_with_role( + self, + role: StaffRole, + settings: AdminSettings | None = None, + ) -> tuple[TestClient, object]: + app = create_app( + settings + or AdminSettings( + admin_auth_token_secret="test-secret", + admin_api_prefix="/admin", + ) + ) + app.dependency_overrides[get_current_panel_staff_principal] = lambda: AuthenticatedStaffPrincipal( + id=21, + email="staff@empresa.com", + display_name="Equipe Web", + role=role, + is_active=True, + ) + return TestClient(app), app + + def test_panel_tools_overview_is_available_for_staff_session(self): + client, app = self._build_client_with_role(StaffRole.STAFF) + try: + response = client.get("/admin/panel/tools/overview") + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + payload = response.json() + self.assertEqual(payload["mode"], "bootstrap_catalog") + self.assertIn("/admin/panel/tools/contracts", [item["href"] for item in payload["actions"]]) + + def test_panel_tools_review_queue_is_available_for_staff_session(self): + client, app = self._build_client_with_role(StaffRole.STAFF) + try: + response = client.get("/admin/panel/tools/review-queue") + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["queue_mode"], "bootstrap_empty_state") + + def test_panel_tools_publications_require_admin_publication_permission(self): + client, app = self._build_client_with_role(StaffRole.STAFF) + try: + response = client.get("/admin/panel/tools/publications") + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 403) + self.assertEqual( + response.json()["detail"], + "Permissao administrativa insuficiente: 'publish_tools'.", + ) + + def test_panel_tools_publications_return_catalog_for_admin_session(self): + client, app = self._build_client_with_role(StaffRole.ADMIN) + try: + response = client.get("/admin/panel/tools/publications") + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + payload = response.json() + self.assertEqual(payload["target_service"], "product") + self.assertGreaterEqual(len(payload["publications"]), 10) + self.assertIn("consultar_estoque", [item["tool_name"] for item in payload["publications"]]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_admin_system_configuration_web.py b/tests/test_admin_system_configuration_web.py new file mode 100644 index 0000000..f160fe6 --- /dev/null +++ b/tests/test_admin_system_configuration_web.py @@ -0,0 +1,134 @@ +import unittest + +from fastapi.testclient import TestClient + +from admin_app.app_factory import create_app +from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal +from admin_app.api.dependencies import get_current_staff_principal +from shared.contracts import StaffRole + + +class AdminSystemConfigurationWebTests(unittest.TestCase): + def _build_client_with_role( + self, + role: StaffRole, + settings: AdminSettings | None = None, + ) -> tuple[TestClient, object]: + app = create_app( + settings + or AdminSettings( + admin_auth_token_secret="test-secret", + admin_api_prefix="/admin", + admin_environment="development", + admin_debug=True, + ) + ) + app.dependency_overrides[get_current_staff_principal] = lambda: AuthenticatedStaffPrincipal( + id=10, + email="staff@empresa.com", + display_name="Equipe Interna", + role=role, + is_active=True, + ) + return TestClient(app), app + + def test_configuration_routes_require_manage_settings_permission(self): + client, app = self._build_client_with_role(StaffRole.STAFF) + try: + response = client.get("/admin/system/configuration", headers={"Authorization": "Bearer token"}) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 403) + self.assertEqual( + response.json()["detail"], + "Permissao administrativa insuficiente: 'manage_settings'.", + ) + + def test_configuration_overview_returns_runtime_security_and_sources(self): + settings = AdminSettings( + admin_auth_token_secret="test-secret", + admin_app_name="Admin Interno", + admin_environment="development", + admin_version="0.9.0", + admin_api_prefix="/admin", + admin_debug=True, + admin_db_host="db.internal", + admin_db_port=3307, + admin_db_name="orquestrador_admin_dev", + admin_db_cloud_sql_connection_name="project:region:instance", + admin_auth_password_pepper="pepper", + admin_auth_access_token_ttl_minutes=45, + admin_auth_refresh_token_ttl_days=10, + admin_bootstrap_enabled=True, + admin_bootstrap_email="bootstrap@empresa.com", + admin_bootstrap_display_name="Bootstrap Admin", + admin_bootstrap_password="SenhaMuitoSegura!123", + admin_bootstrap_role="admin", + ) + client, app = self._build_client_with_role(StaffRole.ADMIN, settings) + try: + response = client.get("/admin/system/configuration", headers={"Authorization": "Bearer token"}) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + payload = response.json() + self.assertEqual(payload["service"], "orquestrador-admin") + self.assertEqual(payload["runtime"]["application"]["app_name"], "Admin Interno") + self.assertEqual(payload["runtime"]["database"]["host"], "db.internal") + self.assertTrue(payload["runtime"]["database"]["cloud_sql_configured"]) + self.assertEqual(payload["runtime"]["panel_session"]["cookie_path"], "/admin") + self.assertFalse(payload["runtime"]["panel_session"]["secure_cookies"]) + self.assertEqual(payload["security"]["tokens"]["access_token_ttl_minutes"], 45) + self.assertTrue(payload["security"]["password"]["pepper_configured"]) + self.assertTrue(payload["security"]["bootstrap"]["enabled"]) + self.assertTrue(payload["security"]["bootstrap"]["password_configured"]) + self.assertIn("panel_session", [item["key"] for item in payload["sources"]]) + + def test_runtime_configuration_route_exposes_panel_cookie_metadata(self): + settings = AdminSettings( + admin_auth_token_secret="test-secret", + admin_api_prefix="/admin", + admin_environment="production", + admin_debug=False, + ) + client, app = self._build_client_with_role(StaffRole.ADMIN, settings) + try: + response = client.get("/admin/system/configuration/runtime", headers={"Authorization": "Bearer token"}) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + runtime = response.json()["runtime"] + self.assertEqual(runtime["panel_session"]["access_cookie_name"], "orquestrador_admin_panel_access") + self.assertEqual(runtime["panel_session"]["refresh_cookie_name"], "orquestrador_admin_panel_refresh") + self.assertEqual(runtime["panel_session"]["same_site"], "lax") + self.assertTrue(runtime["panel_session"]["secure_cookies"]) + + def test_security_configuration_route_returns_credential_strategy_snapshot(self): + settings = AdminSettings( + admin_auth_token_secret="test-secret", + admin_api_prefix="/admin", + admin_auth_password_min_length=14, + admin_auth_token_issuer="admin-runtime", + admin_auth_refresh_token_bytes=48, + admin_bootstrap_enabled=True, + admin_bootstrap_role="admin", + ) + client, app = self._build_client_with_role(StaffRole.ADMIN, settings) + try: + response = client.get("/admin/system/configuration/security", headers={"Authorization": "Bearer token"}) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + security = response.json()["security"] + self.assertEqual(security["password"]["min_length"], 14) + self.assertEqual(security["tokens"]["issuer"], "admin-runtime") + self.assertEqual(security["tokens"]["refresh_token_bytes"], 48) + self.assertEqual(security["bootstrap"]["role"], "admin") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_admin_tools_web.py b/tests/test_admin_tools_web.py new file mode 100644 index 0000000..fd8ed3d --- /dev/null +++ b/tests/test_admin_tools_web.py @@ -0,0 +1,135 @@ +import unittest + +from fastapi.testclient import TestClient + +from admin_app.api.dependencies import get_current_staff_principal +from admin_app.app_factory import create_app +from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal +from shared.contracts import StaffRole + + +class AdminToolsWebTests(unittest.TestCase): + def _build_client_with_role( + self, + role: StaffRole, + settings: AdminSettings | None = None, + ) -> tuple[TestClient, object]: + app = create_app( + settings + or AdminSettings( + admin_auth_token_secret="test-secret", + admin_api_prefix="/admin", + ) + ) + app.dependency_overrides[get_current_staff_principal] = lambda: AuthenticatedStaffPrincipal( + id=11, + email="staff@empresa.com", + display_name="Equipe de Tools", + role=role, + is_active=True, + ) + return TestClient(app), app + + def test_tools_overview_requires_manage_tool_drafts_permission(self): + client, app = self._build_client_with_role(StaffRole.VIEWER) + try: + response = client.get("/admin/tools/overview", headers={"Authorization": "Bearer token"}) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 403) + self.assertEqual( + response.json()["detail"], + "Permissao administrativa insuficiente: 'manage_tool_drafts'.", + ) + + def test_tools_overview_returns_metrics_workflow_and_actions(self): + client, app = self._build_client_with_role(StaffRole.STAFF) + try: + response = client.get("/admin/tools/overview", headers={"Authorization": "Bearer token"}) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + payload = response.json() + self.assertEqual(payload["service"], "orquestrador-admin") + self.assertEqual(payload["mode"], "bootstrap_catalog") + self.assertEqual(payload["metrics"][0]["value"], "18") + self.assertIn("active", [item["code"] for item in payload["workflow"]]) + self.assertIn("/admin/tools/contracts", [item["href"] for item in payload["actions"]]) + self.assertIn("ToolDraft", payload["next_steps"][0]) + + def test_tools_contracts_return_shared_contract_snapshot(self): + client, app = self._build_client_with_role(StaffRole.STAFF) + try: + response = client.get("/admin/tools/contracts", headers={"Authorization": "Bearer token"}) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + payload = response.json() + self.assertEqual(payload["publication_source_service"], "admin") + self.assertEqual(payload["publication_target_service"], "product") + self.assertIn("draft", [item["code"] for item in payload["lifecycle_statuses"]]) + self.assertIn("string", [item["code"] for item in payload["parameter_types"]]) + self.assertIn("published_tool", payload["publication_fields"]) + + def test_tools_drafts_return_empty_state_until_persistence_exists(self): + client, app = self._build_client_with_role(StaffRole.STAFF) + try: + response = client.get("/admin/tools/drafts", headers={"Authorization": "Bearer token"}) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + payload = response.json() + self.assertEqual(payload["storage_status"], "pending_persistence") + self.assertEqual(payload["drafts"], []) + self.assertEqual(payload["supported_statuses"], ["draft"]) + + def test_tools_review_queue_is_available_for_staff(self): + client, app = self._build_client_with_role(StaffRole.STAFF) + try: + response = client.get("/admin/tools/review-queue", headers={"Authorization": "Bearer token"}) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + payload = response.json() + self.assertEqual(payload["queue_mode"], "bootstrap_empty_state") + self.assertEqual(payload["items"], []) + self.assertIn("validated", payload["supported_statuses"]) + + def test_tools_publications_require_publish_tools_permission(self): + client, app = self._build_client_with_role(StaffRole.STAFF) + try: + response = client.get("/admin/tools/publications", headers={"Authorization": "Bearer token"}) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 403) + self.assertEqual( + response.json()["detail"], + "Permissao administrativa insuficiente: 'publish_tools'.", + ) + + def test_tools_publications_return_bootstrap_catalog_for_admin(self): + client, app = self._build_client_with_role(StaffRole.ADMIN) + try: + response = client.get("/admin/tools/publications", headers={"Authorization": "Bearer token"}) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + payload = response.json() + self.assertEqual(payload["source"], "bootstrap_catalog") + self.assertEqual(payload["target_service"], "product") + self.assertGreaterEqual(len(payload["publications"]), 10) + self.assertIn("consultar_estoque", [item["tool_name"] for item in payload["publications"]]) + first = payload["publications"][0] + self.assertEqual(first["status"], "active") + self.assertEqual(first["implementation_module"], "app.services.tools.handlers") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_admin_view_bootstrap.py b/tests/test_admin_view_bootstrap.py new file mode 100644 index 0000000..e163a7f --- /dev/null +++ b/tests/test_admin_view_bootstrap.py @@ -0,0 +1,163 @@ +import unittest + +from fastapi.testclient import TestClient + +from admin_app.api.dependencies import get_optional_panel_staff_context +from admin_app.app_factory import create_app +from admin_app.core import AdminSettings, AuthenticatedStaffContext, AuthenticatedStaffPrincipal +from shared.contracts import StaffRole + + +def _build_panel_context() -> AuthenticatedStaffContext: + return AuthenticatedStaffContext( + principal=AuthenticatedStaffPrincipal( + id=7, + email="admin@empresa.com", + display_name="Administrador", + role=StaffRole.ADMIN, + is_active=True, + ), + session_id=77, + ) + + +class AdminViewBootstrapTests(unittest.TestCase): + def test_panel_entry_redirects_to_login_without_session(self): + app = create_app(AdminSettings(admin_app_name="Admin Interno", admin_version="1.4.0")) + client = TestClient(app) + + response = client.get("/panel", follow_redirects=False) + + self.assertEqual(response.status_code, 302) + self.assertTrue(response.headers["location"].endswith("/login")) + + def test_panel_entry_redirects_to_admin_dashboard_with_session(self): + app = create_app(AdminSettings(admin_app_name="Admin Interno", admin_version="1.4.0")) + app.dependency_overrides[get_optional_panel_staff_context] = lambda: _build_panel_context() + client = TestClient(app) + try: + response = client.get("/panel", follow_redirects=False) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 302) + self.assertTrue(response.headers["location"].endswith("/panel/admin")) + + def test_login_page_redirects_to_dashboard_when_session_exists(self): + app = create_app(AdminSettings(admin_app_name="Admin Interno", admin_version="1.4.0")) + app.dependency_overrides[get_optional_panel_staff_context] = lambda: _build_panel_context() + client = TestClient(app) + try: + response = client.get("/login", follow_redirects=False) + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 302) + self.assertTrue(response.headers["location"].endswith("/panel/admin")) + + def test_login_page_renders_focused_auth_experience(self): + app = create_app(AdminSettings(admin_app_name="Admin Interno", admin_version="1.4.0")) + client = TestClient(app) + + response = client.get("/login") + + self.assertEqual(response.status_code, 200) + self.assertIn("Login administrativo", response.text) + self.assertIn("Acesso restrito", response.text) + self.assertIn('data-admin-login-form="true"', response.text) + self.assertIn('data-auth-endpoint="/panel/auth/login"', response.text) + self.assertIn('data-dashboard-href="http://testserver/panel/admin"', response.text) + self.assertNotIn('data-session-endpoint=', response.text) + self.assertNotIn('data-logout-endpoint=', response.text) + self.assertNotIn("Voltar ao dashboard", response.text) + self.assertNotIn("/panel/tools/review", response.text) + + def test_admin_dashboard_renders_bootstrap_dashboard_when_session_exists(self): + app = create_app(AdminSettings(admin_app_name="Admin Interno", admin_version="1.4.0")) + app.dependency_overrides[get_optional_panel_staff_context] = lambda: _build_panel_context() + client = TestClient(app) + try: + response = client.get("/panel/admin") + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + self.assertIn("text/html", response.headers["content-type"]) + self.assertIn("Painel Administrativo", response.text) + self.assertIn("Dashboard do administrador", response.text) + self.assertIn("Areas do sistema", response.text) + self.assertIn("Entradas claras para as areas protegidas", response.text) + self.assertIn("Revisao de tools", response.text) + self.assertIn("/panel/tools/review", response.text) + self.assertIn("/panel/assets/styles/panel.css", response.text) + self.assertNotIn("API pronta para ser plugada na UI", response.text) + + def test_tool_review_page_redirects_to_login_without_session(self): + app = create_app(AdminSettings(admin_app_name="Admin Interno", admin_version="1.4.0")) + client = TestClient(app) + + response = client.get("/panel/tools/review", follow_redirects=False) + + self.assertEqual(response.status_code, 302) + self.assertTrue(response.headers["location"].endswith("/login")) + + def test_tool_review_page_renders_web_data_endpoints_when_session_exists(self): + app = create_app(AdminSettings(admin_app_name="Admin Interno", admin_version="1.4.0")) + app.dependency_overrides[get_optional_panel_staff_context] = lambda: _build_panel_context() + client = TestClient(app) + try: + response = client.get("/panel/tools/review") + finally: + app.dependency_overrides.clear() + + self.assertEqual(response.status_code, 200) + self.assertIn("Revisao, aprovacao e ativacao", response.text) + self.assertIn('data-admin-tool-review-board="true"', response.text) + self.assertIn('data-overview-endpoint="/panel/tools/overview"', response.text) + self.assertIn('data-contracts-endpoint="/panel/tools/contracts"', response.text) + self.assertIn('data-review-queue-endpoint="/panel/tools/review-queue"', response.text) + self.assertIn('data-publications-endpoint="/panel/tools/publications"', response.text) + self.assertNotIn("Abrir login administrativo", response.text) + + def test_prefixed_panel_routes_apply_auth_gate(self): + app = create_app(AdminSettings(admin_api_prefix="/admin")) + client = TestClient(app) + + panel_response = client.get("/admin/panel", follow_redirects=False) + login_response = client.get("/admin/login") + review_response = client.get("/admin/panel/tools/review", follow_redirects=False) + css_response = client.get("/admin/panel/assets/styles/panel.css") + + self.assertEqual(panel_response.status_code, 302) + self.assertEqual(login_response.status_code, 200) + self.assertEqual(review_response.status_code, 302) + self.assertEqual(css_response.status_code, 200) + self.assertTrue(panel_response.headers["location"].endswith("/admin/login")) + self.assertTrue(review_response.headers["location"].endswith("/admin/login")) + self.assertIn('data-auth-endpoint="/admin/panel/auth/login"', login_response.text) + self.assertIn('data-dashboard-href="http://testserver/admin/panel/admin"', login_response.text) + self.assertNotIn('data-session-endpoint=', login_response.text) + self.assertNotIn('data-logout-endpoint=', login_response.text) + self.assertIn("/admin/panel/assets/styles/panel.css", login_response.text) + self.assertIn("--admin-bg", css_response.text) + + def test_prefixed_admin_dashboard_and_review_render_when_session_exists(self): + app = create_app(AdminSettings(admin_api_prefix="/admin")) + app.dependency_overrides[get_optional_panel_staff_context] = lambda: _build_panel_context() + client = TestClient(app) + try: + panel_response = client.get("/admin/panel/admin") + review_response = client.get("/admin/panel/tools/review") + finally: + app.dependency_overrides.clear() + + self.assertEqual(panel_response.status_code, 200) + self.assertEqual(review_response.status_code, 200) + self.assertIn("Dashboard do administrador", panel_response.text) + self.assertIn("/admin/panel/tools/review", panel_response.text) + self.assertIn('data-overview-endpoint="/admin/panel/tools/overview"', review_response.text) + self.assertIn('data-publications-endpoint="/admin/panel/tools/publications"', review_response.text) + + +if __name__ == "__main__": + unittest.main()