From 56a31bab8d7e3ec0c360f8c312f93edcec62aed7 Mon Sep 17 00:00:00 2001 From: Eddy ZHANG Date: Mon, 7 Sep 2026 05:13:09 +0000 Subject: [PATCH] refactor(inner_api): dep-inject workspace payloads with @model_validate (#41540) --- .../inner_api/workspace/workspace.py | 17 +-- .../inner_api/workspace/test_workspace.py | 140 ++++++++++-------- 2 files changed, 89 insertions(+), 68 deletions(-) diff --git a/api/controllers/inner_api/workspace/workspace.py b/api/controllers/inner_api/workspace/workspace.py index e70a517b356..b1d5f82ed9b 100644 --- a/api/controllers/inner_api/workspace/workspace.py +++ b/api/controllers/inner_api/workspace/workspace.py @@ -5,7 +5,7 @@ from pydantic import BaseModel from sqlalchemy import select from controllers.common.schema import register_schema_models -from controllers.console.wraps import setup_required +from controllers.console.wraps import model_validate, setup_required from controllers.inner_api import inner_api_ns from controllers.inner_api.wraps import enterprise_inner_api_only from events.tenant_event import tenant_was_created @@ -54,9 +54,8 @@ class EnterpriseWorkspace(Resource): 404: "Owner account not found or service not available", } ) - def post(self): - args = WorkspaceCreatePayload.model_validate(inner_api_ns.payload or {}) - + @model_validate(WorkspaceCreatePayload) + def post(self, args: WorkspaceCreatePayload): account = db.session.scalar(select(Account).where(Account.email == args.owner_email).limit(1)) if account is None: return {"message": "owner account not found."}, 404 @@ -97,9 +96,8 @@ class EnterpriseWorkspaceNoOwnerEmail(Resource): 404: "Service not available", } ) - def post(self): - args = WorkspaceOwnerlessPayload.model_validate(inner_api_ns.payload or {}) - + @model_validate(WorkspaceOwnerlessPayload) + def post(self, args: WorkspaceOwnerlessPayload): tenant = TenantService.create_tenant(args.name, is_from_dashboard=True, session=db.session()) tenant_was_created.send(tenant) @@ -136,9 +134,8 @@ class EnterpriseWorkspaceMember(Resource): 404: "Workspace or account not found", } ) - def post(self): - args = WorkspaceMemberPayload.model_validate(inner_api_ns.payload or {}) - + @model_validate(WorkspaceMemberPayload) + def post(self, args: WorkspaceMemberPayload): try: role = TenantAccountRole(args.role) except ValueError: diff --git a/api/tests/unit_tests/controllers/inner_api/workspace/test_workspace.py b/api/tests/unit_tests/controllers/inner_api/workspace/test_workspace.py index c92349c59ad..f7c8e653a49 100644 --- a/api/tests/unit_tests/controllers/inner_api/workspace/test_workspace.py +++ b/api/tests/unit_tests/controllers/inner_api/workspace/test_workspace.py @@ -12,8 +12,10 @@ from unittest.mock import patch import pytest from flask import Flask +from flask_restx import Resource from pydantic import ValidationError from sqlalchemy.orm import Session, scoped_session, sessionmaker +from werkzeug.exceptions import UnprocessableEntity from controllers.inner_api.workspace.workspace import ( EnterpriseWorkspace, @@ -29,6 +31,7 @@ from services.account_service import ( EnterpriseWorkspaceMemberAccountNotFoundError, EnterpriseWorkspaceMemberWorkspaceNotFoundError, ) +from tests.unit_tests.config_override import config_overrides_context @pytest.fixture @@ -145,10 +148,9 @@ class TestEnterpriseWorkspace: # Act — unwrap to bypass auth/setup decorators (tested in test_auth_wraps.py) unwrapped_post = inspect.unwrap(api_instance.post) - with app.test_request_context(): - with patch("controllers.inner_api.workspace.workspace.inner_api_ns") as mock_ns: - mock_ns.payload = {"name": "My Workspace", "owner_email": "owner@example.com"} - result = unwrapped_post(api_instance) + payload = {"name": "My Workspace", "owner_email": "owner@example.com"} + with app.test_request_context(json=payload): + result = unwrapped_post(api_instance, WorkspaceCreatePayload.model_validate(payload)) # Assert assert result["message"] == "enterprise workspace created." @@ -168,10 +170,9 @@ class TestEnterpriseWorkspace: """Test that post() returns 404 when the owner account does not exist""" # Act unwrapped_post = inspect.unwrap(api_instance.post) - with app.test_request_context(): - with patch("controllers.inner_api.workspace.workspace.inner_api_ns") as mock_ns: - mock_ns.payload = {"name": "My Workspace", "owner_email": "missing@example.com"} - result = unwrapped_post(api_instance) + payload = {"name": "My Workspace", "owner_email": "missing@example.com"} + with app.test_request_context(json=payload): + result = unwrapped_post(api_instance, WorkspaceCreatePayload.model_validate(payload)) # Assert assert result == ({"message": "owner account not found."}, 404) @@ -214,10 +215,9 @@ class TestEnterpriseWorkspaceNoOwnerEmail: # Act — unwrap to bypass auth/setup decorators (tested in test_auth_wraps.py) unwrapped_post = inspect.unwrap(api_instance.post) - with app.test_request_context(): - with patch("controllers.inner_api.workspace.workspace.inner_api_ns") as mock_ns: - mock_ns.payload = {"name": "My Workspace"} - result = unwrapped_post(api_instance) + payload = {"name": "My Workspace"} + with app.test_request_context(json=payload): + result = unwrapped_post(api_instance, WorkspaceOwnerlessPayload.model_validate(payload)) # Assert assert result["message"] == "enterprise workspace created." @@ -251,16 +251,15 @@ class TestEnterpriseWorkspaceMember: mock_tenant_svc.join_enterprise_workspace_member.return_value = membership unwrapped_post = inspect.unwrap(api_instance.post) - with app.test_request_context(): - with patch("controllers.inner_api.workspace.workspace.inner_api_ns") as mock_ns: - mock_ns.payload = { - "workspace_id": "workspace-id", - "account_id": "account-id", - "email": "member@example.com", - "role": "normal", - "operator_account_id": "operator-id", - } - result = unwrapped_post(api_instance) + payload = { + "workspace_id": "workspace-id", + "account_id": "account-id", + "email": "member@example.com", + "role": "normal", + "operator_account_id": "operator-id", + } + with app.test_request_context(json=payload): + result = unwrapped_post(api_instance, WorkspaceMemberPayload.model_validate(payload)) assert result["message"] == "enterprise workspace member joined." assert result["member"] == { @@ -281,15 +280,14 @@ class TestEnterpriseWorkspaceMember: mock_tenant_svc.join_enterprise_workspace_member.side_effect = EnterpriseWorkspaceMemberWorkspaceNotFoundError unwrapped_post = inspect.unwrap(api_instance.post) - with app.test_request_context(): - with patch("controllers.inner_api.workspace.workspace.inner_api_ns") as mock_ns: - mock_ns.payload = { - "workspace_id": "missing-workspace", - "account_id": "account-id", - "email": "member@example.com", - "role": "normal", - } - result = unwrapped_post(api_instance) + payload = { + "workspace_id": "missing-workspace", + "account_id": "account-id", + "email": "member@example.com", + "role": "normal", + } + with app.test_request_context(json=payload): + result = unwrapped_post(api_instance, WorkspaceMemberPayload.model_validate(payload)) assert result == ({"message": "workspace not found."}, 404) mock_tenant_svc.join_enterprise_workspace_member.assert_called_once() @@ -299,15 +297,14 @@ class TestEnterpriseWorkspaceMember: mock_tenant_svc.join_enterprise_workspace_member.side_effect = EnterpriseWorkspaceMemberAccountNotFoundError unwrapped_post = inspect.unwrap(api_instance.post) - with app.test_request_context(): - with patch("controllers.inner_api.workspace.workspace.inner_api_ns") as mock_ns: - mock_ns.payload = { - "workspace_id": "workspace-id", - "account_id": "missing-account", - "email": "member@example.com", - "role": "normal", - } - result = unwrapped_post(api_instance) + payload = { + "workspace_id": "workspace-id", + "account_id": "missing-account", + "email": "member@example.com", + "role": "normal", + } + with app.test_request_context(json=payload): + result = unwrapped_post(api_instance, WorkspaceMemberPayload.model_validate(payload)) assert result == ({"message": "account not found."}, 404) mock_tenant_svc.join_enterprise_workspace_member.assert_called_once() @@ -315,29 +312,56 @@ class TestEnterpriseWorkspaceMember: @pytest.mark.usefixtures("database_session") def test_post_rejects_owner_role(self, api_instance, app: Flask): unwrapped_post = inspect.unwrap(api_instance.post) - with app.test_request_context(): - with patch("controllers.inner_api.workspace.workspace.inner_api_ns") as mock_ns: - mock_ns.payload = { - "workspace_id": "workspace-id", - "account_id": "account-id", - "email": "member@example.com", - "role": "owner", - } - result = unwrapped_post(api_instance) + payload = { + "workspace_id": "workspace-id", + "account_id": "account-id", + "email": "member@example.com", + "role": "owner", + } + with app.test_request_context(json=payload): + result = unwrapped_post(api_instance, WorkspaceMemberPayload.model_validate(payload)) assert result == ({"message": "cannot join workspace as owner."}, 400) @pytest.mark.usefixtures("database_session") def test_post_rejects_invalid_role(self, api_instance, app: Flask): unwrapped_post = inspect.unwrap(api_instance.post) - with app.test_request_context(): - with patch("controllers.inner_api.workspace.workspace.inner_api_ns") as mock_ns: - mock_ns.payload = { - "workspace_id": "workspace-id", - "account_id": "account-id", - "email": "member@example.com", - "role": "not-a-role", - } - result = unwrapped_post(api_instance) + payload = { + "workspace_id": "workspace-id", + "account_id": "account-id", + "email": "member@example.com", + "role": "not-a-role", + } + with app.test_request_context(json=payload): + result = unwrapped_post(api_instance, WorkspaceMemberPayload.model_validate(payload)) assert result == ({"message": "invalid workspace member role."}, 400) + + +class TestModelValidateDecorator: + """The handler tests unwrap the view, so this is what covers the decorators themselves.""" + + @pytest.mark.parametrize( + ("api_cls", "route"), + [ + (EnterpriseWorkspace, "/enterprise/workspace"), + (EnterpriseWorkspaceNoOwnerEmail, "/enterprise/workspace/ownerless"), + (EnterpriseWorkspaceMember, "/enterprise/workspace/member"), + ], + ) + def test_invalid_body_is_rejected_before_the_handler_runs( + self, app: Flask, api_cls: type[Resource], route: str + ) -> None: + api_instance = api_cls() + + with ( + config_overrides_context(INNER_API=True, INNER_API_KEY="inner-key"), + app.test_request_context(route, method="POST", json={}, headers={"X-Inner-Api-Key": "inner-key"}), + patch("controllers.console.wraps._is_setup_completed", return_value=True), + patch("controllers.inner_api.workspace.workspace.TenantService") as tenant_service, + ): + with pytest.raises(UnprocessableEntity) as exc_info: + api_instance.post() + + assert exc_info.value.code == 422 + tenant_service.assert_not_called()