From 9aad301f4cf69be0e689b9f76acbfe83042e4515 Mon Sep 17 00:00:00 2001 From: Cursx <33718736+Cursx@users.noreply.github.com> Date: Wed, 4 Feb 2026 17:44:17 +0800 Subject: [PATCH 01/11] Update ext_login.py --- api/extensions/ext_login.py | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/api/extensions/ext_login.py b/api/extensions/ext_login.py index 74299956c0..2867166e2a 100644 --- a/api/extensions/ext_login.py +++ b/api/extensions/ext_login.py @@ -48,7 +48,9 @@ def load_user_from_request(request_from_flask_login): account.current_tenant = tenant return account - if request.blueprint in {"console", "inner_api"}: + # Support both Blueprint-based routes and FastOpenAPI routes (which have no blueprint) + is_console_api = request.blueprint in {"console", "inner_api"} or request.path.startswith("/console/api/") + if is_console_api: if not auth_token: raise Unauthorized("Invalid Authorization token.") decoded = PassportService().verify(auth_token) @@ -115,7 +117,18 @@ def on_user_logged_in(_sender, user): @login_manager.unauthorized_handler def unauthorized_handler(): - """Handle unauthorized requests.""" + """Handle unauthorized requests. + + For FastOpenAPI routes (no blueprint), we raise Unauthorized exception + which will be caught and serialized properly by the framework. + For traditional Blueprint-based routes, we return a Response object. + """ + # Check if this is a FastOpenAPI route (no blueprint but console API path) + if request.blueprint is None and request.path.startswith("/console/api/"): + # Raise exception - FastOpenAPI will handle serialization + raise Unauthorized("Unauthorized.") + + # Traditional Blueprint routes - return Response object return Response( json.dumps({"code": "unauthorized", "message": "Unauthorized."}), status=401, From 3ebe5b0fc6cef2b3e92ab5987aa9b3458dbdcac9 Mon Sep 17 00:00:00 2001 From: Cursx <33718736+Cursx@users.noreply.github.com> Date: Wed, 4 Feb 2026 17:44:53 +0800 Subject: [PATCH 02/11] Update feature.py --- api/controllers/console/feature.py | 90 ++++++++++++++---------------- 1 file changed, 42 insertions(+), 48 deletions(-) diff --git a/api/controllers/console/feature.py b/api/controllers/console/feature.py index d3811e2d1b..f2385c4f71 100644 --- a/api/controllers/console/feature.py +++ b/api/controllers/console/feature.py @@ -1,60 +1,54 @@ -from flask_restx import Resource, fields from werkzeug.exceptions import Unauthorized +from controllers.fastopenapi import console_router from libs.login import current_account_with_tenant, current_user, login_required -from services.feature_service import FeatureService +from services.feature_service import FeatureModel, FeatureService, SystemFeatureModel -from . import console_ns from .wraps import account_initialization_required, cloud_utm_record, setup_required -@console_ns.route("/features") -class FeatureApi(Resource): - @console_ns.doc("get_tenant_features") - @console_ns.doc(description="Get feature configuration for current tenant") - @console_ns.response( - 200, - "Success", - console_ns.model("FeatureResponse", {"features": fields.Raw(description="Feature configuration object")}), - ) - @setup_required - @login_required - @account_initialization_required - @cloud_utm_record - def get(self): - """Get feature configuration for current tenant""" - _, current_tenant_id = current_account_with_tenant() - - return FeatureService.get_features(current_tenant_id).model_dump() +# NOTE: The original feature.py Swagger documentation incorrectly specified a wrapped format +# {"features": ...}, but the actual implementation returned a flat FeatureModel. +# The frontend has always used the flat format, so we maintain backward compatibility here. -@console_ns.route("/system-features") -class SystemFeatureApi(Resource): - @console_ns.doc("get_system_features") - @console_ns.doc(description="Get system-wide feature configuration") - @console_ns.response( - 200, - "Success", - console_ns.model( - "SystemFeatureResponse", {"features": fields.Raw(description="System feature configuration object")} - ), - ) - def get(self): - """Get system-wide feature configuration +@console_router.get( + "/features", + response_model=FeatureModel, + tags=["console"], +) +@setup_required +@login_required +@account_initialization_required +@cloud_utm_record +def get_tenant_features() -> FeatureModel: + """Get feature configuration for current tenant.""" + _, current_tenant_id = current_account_with_tenant() - NOTE: This endpoint is unauthenticated by design, as it provides system features - data required for dashboard initialization. + return FeatureService.get_features(current_tenant_id) - Authentication would create circular dependency (can't login without dashboard loading). - Only non-sensitive configuration data should be returned by this endpoint. - """ - # NOTE(QuantumGhost): ideally we should access `current_user.is_authenticated` - # without a try-catch. However, due to the implementation of user loader (the `load_user_from_request` - # in api/extensions/ext_login.py), accessing `current_user.is_authenticated` will - # raise `Unauthorized` exception if authentication token is not provided. - try: - is_authenticated = current_user.is_authenticated - except Unauthorized: - is_authenticated = False - return FeatureService.get_system_features(is_authenticated=is_authenticated).model_dump() +@console_router.get( + "/system-features", + response_model=SystemFeatureModel, + tags=["console"], +) +def get_system_features() -> SystemFeatureModel: + """Get system-wide feature configuration + + NOTE: This endpoint is unauthenticated by design, as it provides system features + data required for dashboard initialization. + + Authentication would create circular dependency (can't login without dashboard loading). + + Only non-sensitive configuration data should be returned by this endpoint. + """ + # NOTE(QuantumGhost): ideally we should access `current_user.is_authenticated` + # without a try-catch. However, due to the implementation of user loader (the `load_user_from_request` + # in api/extensions/ext_login.py), accessing `current_user.is_authenticated` will + # raise `Unauthorized` exception if authentication token is not provided. + try: + is_authenticated = current_user.is_authenticated + except Unauthorized: + is_authenticated = False + return FeatureService.get_system_features(is_authenticated=is_authenticated) From 88cdf5e66c52c0230f8845d28917d37792045140 Mon Sep 17 00:00:00 2001 From: Cursx <33718736+Cursx@users.noreply.github.com> Date: Wed, 4 Feb 2026 17:45:49 +0800 Subject: [PATCH 03/11] Create test_fastopenapi_feature.py --- .../console/test_fastopenapi_feature.py | 664 ++++++++++++++++++ 1 file changed, 664 insertions(+) create mode 100644 api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py diff --git a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py new file mode 100644 index 0000000000..fcebd42d58 --- /dev/null +++ b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py @@ -0,0 +1,664 @@ +import builtins +import contextlib +import importlib +import sys +from unittest.mock import MagicMock, PropertyMock, patch + +import pytest +from flask import Flask +from flask.views import MethodView +from werkzeug.exceptions import Unauthorized + +from extensions import ext_fastopenapi +from extensions.ext_database import db +from services.feature_service import FeatureModel, SystemFeatureModel + + +@pytest.fixture +def app(): + """ + Creates a Flask application instance configured for testing. + """ + app = Flask(__name__) + app.config["TESTING"] = True + app.config["SECRET_KEY"] = "test-secret" + app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:" + + # Initialize the database with the app + db.init_app(app) + + return app + + +@pytest.fixture(autouse=True) +def fix_method_view_issue(monkeypatch): + """ + Automatic fixture to patch 'builtins.MethodView'. + + Why this is needed: + The official legacy codebase contains a global patch in its initialization logic: + if not hasattr(builtins, "MethodView"): + builtins.MethodView = MethodView + + Some dependencies (like ext_fastopenapi or older Flask extensions) might implicitly + rely on 'MethodView' being available in the global builtins namespace. + + Refactoring Note: + While patching builtins is generally discouraged due to global side effects, + this fixture reproduces the production environment's state to ensure tests are realistic. + We use 'monkeypatch' to ensure that this change is undone after the test finishes, + keeping other tests isolated. + """ + if not hasattr(builtins, "MethodView"): + # 'raising=False' allows us to set an attribute that doesn't exist yet + monkeypatch.setattr(builtins, "MethodView", MethodView, raising=False) + + +# ------------------------------------------------------------------------------ +# Helper Functions for Fixture Complexity Reduction +# ------------------------------------------------------------------------------ + + +def _create_isolated_router(): + """ + Creates a fresh, isolated router instance to prevent route pollution. + """ + import controllers.fastopenapi + + # Dynamically get the class type (e.g., FlaskRouter) to avoid hardcoding dependencies + RouterClass = type(controllers.fastopenapi.console_router) + return RouterClass() + + +@contextlib.contextmanager +def _patch_auth_and_router(temp_router): + """ + Context manager that applies all necessary patches for: + 1. The console_router (redirecting to our isolated temp_router) + 2. Authentication decorators (disabling them with no-ops) + 3. User/Account loaders (mocking authenticated state) + """ + + def noop(f): + return f + + # We patch the SOURCE of the decorators/functions, not the destination module. + # This ensures that when 'controllers.console.feature' imports them, it gets the mocks. + with ( + patch("controllers.fastopenapi.console_router", temp_router), + patch("extensions.ext_fastopenapi.console_router", temp_router), + patch("controllers.console.wraps.setup_required", side_effect=noop), + patch("libs.login.login_required", side_effect=noop), + patch("controllers.console.wraps.account_initialization_required", side_effect=noop), + patch("controllers.console.wraps.cloud_utm_record", side_effect=noop), + patch("libs.login.current_account_with_tenant", return_value=(MagicMock(), "tenant-id")), + patch("libs.login.current_user", MagicMock(is_authenticated=True)), + ): + # Explicitly reload ext_fastopenapi to ensure it uses the patched console_router + import extensions.ext_fastopenapi + + importlib.reload(extensions.ext_fastopenapi) + + yield + + +def _force_reload_module(target_module: str, alias_module: str): + """ + Forces a reload of the specified module and handles sys.modules aliasing. + + Why reload? + Python decorators (like @route, @login_required) run at IMPORT time. + To apply our patches (mocks/no-ops) to these decorators, we must re-import + the module while the patches are active. + + Why alias? + If 'ext_fastopenapi' imports the controller as 'api.controllers...', but we import + it as 'controllers...', Python treats them as two separate modules. This causes: + 1. Double execution of decorators (registering routes twice -> AssertionError). + 2. Type mismatch errors (Class A from module X is not Class A from module Y). + + This function ensures both names point to the SAME loaded module instance. + """ + # 1. Clean existing entries to force re-import + if target_module in sys.modules: + del sys.modules[target_module] + if alias_module in sys.modules: + del sys.modules[alias_module] + + # 2. Import the module (triggering decorators with active patches) + module = importlib.import_module(target_module) + + # 3. Alias the module in sys.modules to prevent double loading + sys.modules[alias_module] = sys.modules[target_module] + + return module + + +def _cleanup_modules(target_module: str, alias_module: str): + """ + Removes the module and its alias from sys.modules to prevent side effects + on other tests. + """ + if target_module in sys.modules: + del sys.modules[target_module] + if alias_module in sys.modules: + del sys.modules[alias_module] + + +@pytest.fixture +def mock_feature_module_env(): + """ + Sets up a mocked environment for the feature module. + + This fixture orchestrates: + 1. Creating an isolated router. + 2. Patching authentication and global dependencies. + 3. Reloading the controller module to apply patches to decorators. + 4. cleaning up sys.modules afterwards. + """ + target_module = "controllers.console.feature" + alias_module = "api.controllers.console.feature" + + # 1. Prepare isolated router + temp_router = _create_isolated_router() + + # 2. Apply patches + try: + with _patch_auth_and_router(temp_router): + # 3. Reload module to register routes on the temp_router + feature_module = _force_reload_module(target_module, alias_module) + + yield feature_module + + finally: + # 4. Teardown: Clean up sys.modules + _cleanup_modules(target_module, alias_module) + + +# ------------------------------------------------------------------------------ +# Test Cases +# ------------------------------------------------------------------------------ + + +@pytest.mark.parametrize( + ("url", "service_mock_path", "mock_model_instance"), + [ + ( + "/console/api/features", + "controllers.console.feature.FeatureService.get_features", + FeatureModel(can_replace_logo=True), + ), + ( + "/console/api/system-features", + "controllers.console.feature.FeatureService.get_system_features", + SystemFeatureModel(enable_marketplace=True), + ), + ], +) +def test_console_features_success(app, mock_feature_module_env, url, service_mock_path, mock_model_instance): + """ + Tests that the feature APIs return a 200 OK status and correct JSON structure. + """ + # Patch the service layer to return our mock model instance + with patch(service_mock_path, return_value=mock_model_instance): + # Initialize the API extension + ext_fastopenapi.init_app(app) + + client = app.test_client() + response = client.get(url) + + # Assertions + assert response.status_code == 200, f"Request failed with status {response.status_code}: {response.text}" + + # Verify the JSON response matches the Pydantic model dump (flat format, not wrapped) + expected_data = mock_model_instance.model_dump(mode="json") + assert response.get_json() == expected_data + + +@pytest.mark.parametrize( + ("url", "service_mock_path"), + [ + ("/console/api/features", "controllers.console.feature.FeatureService.get_features"), + ("/console/api/system-features", "controllers.console.feature.FeatureService.get_system_features"), + ], +) +def test_console_features_service_error(app, mock_feature_module_env, url, service_mock_path): + """ + Tests how the application handles Service layer errors. + + Note: When an exception occurs in the view, it is typically caught by the framework + (Flask or the OpenAPI wrapper) and converted to a 500 error response. + This test verifies that the application returns a 500 status code. + """ + # Simulate a service failure + with patch(service_mock_path, side_effect=ValueError("Service Failure")): + ext_fastopenapi.init_app(app) + client = app.test_client() + + # When an exception occurs in the view, it is typically caught by the framework + # (Flask or the OpenAPI wrapper) and converted to a 500 error response. + response = client.get(url) + + assert response.status_code == 500 + # Check if the error details are exposed in the response (depends on error handler config) + # We accept either generic 500 or the specific error message + assert "Service Failure" in response.text or "Internal Server Error" in response.text + + +def test_system_features_unauthenticated(app, mock_feature_module_env): + """ + Tests that /console/api/system-features endpoint works without authentication. + + This test verifies the try-except block in get_system_features that handles + unauthenticated requests by passing is_authenticated=False to the service layer. + """ + feature_module = mock_feature_module_env + + # Override the behavior of the current_user mock + # The fixture patched 'libs.login.current_user', so 'controllers.console.feature.current_user' + # refers to that same Mock object. + mock_user = feature_module.current_user + + # Simulate property access raising Unauthorized + # Note: We must reset side_effect if it was set, or set it here. + # The fixture initialized it as MagicMock(is_authenticated=True). + # We want type(mock_user).is_authenticated to raise Unauthorized. + type(mock_user).is_authenticated = PropertyMock(side_effect=Unauthorized) + + # Patch the service layer for this specific test + with patch("controllers.console.feature.FeatureService.get_system_features") as mock_service: + # Setup mock service return value + mock_model = SystemFeatureModel(enable_marketplace=True) + mock_service.return_value = mock_model + + # Initialize app + ext_fastopenapi.init_app(app) + client = app.test_client() + + # Act + response = client.get("/console/api/system-features") + + # Assert + assert response.status_code == 200, f"Request failed: {response.text}" + + # Verify service was called with is_authenticated=False + mock_service.assert_called_once_with(is_authenticated=False) + + # Verify response body (flat format, not wrapped) + expected_data = mock_model.model_dump(mode="json") + assert response.get_json() == expected_data + + +# ------------------------------------------------------------------------------ +# FastOpenAPI Authentication Tests +# ------------------------------------------------------------------------------ +# These tests verify that our fixes for FastOpenAPI routing work correctly: +# 1. load_user_from_request supports FastOpenAPI routes (no blueprint) +# 2. unauthorized_handler returns serializable response for FastOpenAPI routes +# 3. Response format is flat (not wrapped in {"features": ...}) + + +class TestFastOpenAPIAuthenticationBehavior: + """ + Tests for FastOpenAPI-specific authentication behavior. + + Unlike the tests above that mock authentication decorators, + these tests verify the actual authentication flow works correctly + for FastOpenAPI routes where request.blueprint is None. + """ + + @pytest.fixture + def app_with_login_manager(self): + """ + Creates a Flask app with login manager configured, + simulating the production environment more closely. + """ + from flask_login import LoginManager + + app = Flask(__name__) + app.config["TESTING"] = True + app.config["SECRET_KEY"] = "test-secret" + app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:" + + # Initialize database + db.init_app(app) + + # Initialize login manager with unauthorized handler that matches our fix + login_manager = LoginManager() + login_manager.init_app(app) + + @login_manager.unauthorized_handler + def test_unauthorized_handler(): + """Simulates our fixed unauthorized_handler behavior.""" + from flask import request + # For FastOpenAPI routes, raise exception (serializable by orjson) + if request.blueprint is None and request.path.startswith("/console/api/"): + raise Unauthorized("Unauthorized.") + # For Blueprint routes, return Response (legacy behavior) + from flask import Response + import json + return Response( + json.dumps({"code": "unauthorized", "message": "Unauthorized."}), + status=401, + content_type="application/json", + ) + + return app + + def test_fastopenapi_route_has_no_blueprint(self, app_with_login_manager, fix_method_view_issue): + """ + Verify that FastOpenAPI routes have request.blueprint == None. + + This is the core assumption our authentication fix relies on. + """ + captured_blueprint = {} + + # Create a simple test route to capture request.blueprint + @app_with_login_manager.route("/console/api/test-blueprint") + def test_route(): + from flask import request + captured_blueprint["value"] = request.blueprint + return {"status": "ok"} + + client = app_with_login_manager.test_client() + response = client.get("/console/api/test-blueprint") + + assert response.status_code == 200 + # FastOpenAPI routes registered directly on app have no blueprint + assert captured_blueprint["value"] is None + + def test_unauthorized_response_is_serializable_json(self, app_with_login_manager, fix_method_view_issue): + """ + Verify that unauthorized response for FastOpenAPI routes is valid JSON. + + When unauthorized_handler raises Unauthorized exception for FastOpenAPI routes, + Flask/Werkzeug converts it to a proper HTTP 401 response that is serializable. + """ + @app_with_login_manager.route("/console/api/protected") + def protected_route(): + # Simulate login_required behavior when user is not authenticated + raise Unauthorized("Unauthorized.") + + client = app_with_login_manager.test_client() + response = client.get("/console/api/protected") + + assert response.status_code == 401 + # Response should be valid (either JSON or HTML error page, but not a serialization error) + assert response.data is not None + # Should not be a TypeError from orjson trying to serialize Response object + assert b"TypeError" not in response.data + + def test_response_format_is_flat_not_wrapped(self, app, mock_feature_module_env): + """ + Explicitly verify that response format is flat FeatureModel, + not wrapped in {"features": {...}}. + + This ensures backward compatibility with frontend expectations. + """ + mock_model = FeatureModel(can_replace_logo=True, billing=None) + + with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): + ext_fastopenapi.init_app(app) + client = app.test_client() + response = client.get("/console/api/features") + + assert response.status_code == 200 + json_data = response.get_json() + + # Should NOT be wrapped format + assert "features" not in json_data or not isinstance(json_data.get("features"), dict) + + # Should be flat format - top level keys are FeatureModel fields + assert "can_replace_logo" in json_data + + +# ------------------------------------------------------------------------------ +# Response Format and Content-Type Tests +# ------------------------------------------------------------------------------ +# These tests verify OpenAPI v3 migration requirements for response handling + + +class TestResponseFormatValidation: + """ + Tests for response format validation. + + Ensures FastOpenAPI produces correct Content-Type headers and JSON format + that matches OpenAPI 3.0 specification requirements. + """ + + def test_response_content_type_is_json(self, app, mock_feature_module_env): + """ + Verify response Content-Type is application/json. + + FastOpenAPI uses orjson for serialization, must produce correct Content-Type. + """ + mock_model = FeatureModel(can_replace_logo=True) + + with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): + ext_fastopenapi.init_app(app) + client = app.test_client() + response = client.get("/console/api/features") + + assert response.status_code == 200 + # Content-Type should be JSON + assert "application/json" in response.content_type + + def test_system_features_content_type_is_json(self, app, mock_feature_module_env): + """ + Verify /system-features response Content-Type is application/json. + """ + mock_model = SystemFeatureModel(enable_marketplace=True) + + with patch("controllers.console.feature.FeatureService.get_system_features", return_value=mock_model): + ext_fastopenapi.init_app(app) + client = app.test_client() + response = client.get("/console/api/system-features") + + assert response.status_code == 200 + assert "application/json" in response.content_type + + def test_feature_model_all_fields_serialized(self, app, mock_feature_module_env): + """ + Verify all FeatureModel fields are serialized in response. + + This ensures Pydantic model dump is complete for OpenAPI schema compliance. + """ + mock_model = FeatureModel( + can_replace_logo=True, + model_load_balancing_enabled=True, + ) + + with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): + ext_fastopenapi.init_app(app) + client = app.test_client() + response = client.get("/console/api/features") + + assert response.status_code == 200 + json_data = response.get_json() + + # Verify key fields are present + assert "can_replace_logo" in json_data + assert json_data["can_replace_logo"] is True + assert "model_load_balancing_enabled" in json_data + assert json_data["model_load_balancing_enabled"] is True + + +# ------------------------------------------------------------------------------ +# Authentication Behavior Tests with Realistic Mocking +# ------------------------------------------------------------------------------ +# These tests use more realistic mocking to verify authentication behavior + + +class TestAuthenticationWithRealisticMocking: + """ + Tests authentication behavior with more realistic mocking. + + Unlike tests that completely bypass decorators, these tests mock + at the user/account level to verify decorator behavior. + """ + + @pytest.fixture + def app_with_real_decorators(self, monkeypatch): + """ + Creates a Flask app where decorators execute but dependencies are mocked. + + This provides a middle ground between: + - Full integration tests (require database setup) + - Tests that completely bypass decorators (don't test auth flow) + """ + from flask import Flask + from flask_login import LoginManager + + app = Flask(__name__) + app.config["TESTING"] = True + app.config["SECRET_KEY"] = "test-secret" + app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:" + + # Initialize database + db.init_app(app) + + # Initialize login manager + login_manager = LoginManager() + login_manager.init_app(app) + + # Mock request loader to return None (unauthenticated) + @login_manager.request_loader + def mock_load_user(request): + # Check for test auth header + auth_header = request.headers.get("Authorization") + if auth_header and auth_header.startswith("Bearer test-valid-token"): + # Return a mock user + mock_user = MagicMock() + mock_user.is_authenticated = True + mock_user.current_tenant_id = "test-tenant-id" + return mock_user + return None + + @login_manager.unauthorized_handler + def handle_unauthorized(): + """Simulates production unauthorized_handler for FastOpenAPI routes.""" + from flask import request + if request.blueprint is None and request.path.startswith("/console/api/"): + raise Unauthorized("Unauthorized.") + from flask import Response + import json + return Response( + json.dumps({"code": "unauthorized", "message": "Unauthorized."}), + status=401, + content_type="application/json", + ) + + return app + + def test_features_endpoint_requires_authentication_concept(self, app_with_real_decorators, fix_method_view_issue): + """ + Conceptual test: verify that a protected route would return 401 without auth. + + Note: This test creates a simple protected route to verify the auth flow, + since the actual feature.py module loading is complex. + """ + from flask_login import login_required as flask_login_required + + @app_with_real_decorators.route("/console/api/test-protected") + @flask_login_required + def protected_test_route(): + return {"status": "authenticated"} + + client = app_with_real_decorators.test_client() + + # Without authentication - should return 401 + response = client.get("/console/api/test-protected") + assert response.status_code == 401 + + # With valid test token - should return 200 + response_with_auth = client.get( + "/console/api/test-protected", + headers={"Authorization": "Bearer test-valid-token"} + ) + assert response_with_auth.status_code == 200 + + def test_system_features_no_auth_decorator_concept(self, app_with_real_decorators, fix_method_view_issue): + """ + Conceptual test: verify that an unprotected route works without auth. + + This mirrors /system-features behavior which has no @login_required. + """ + @app_with_real_decorators.route("/console/api/test-public") + def public_test_route(): + return {"status": "public"} + + client = app_with_real_decorators.test_client() + + # Without authentication - should still work + response = client.get("/console/api/test-public") + assert response.status_code == 200 + assert response.get_json()["status"] == "public" + + +# ------------------------------------------------------------------------------ +# OpenAPI Schema Compliance Tests +# ------------------------------------------------------------------------------ +# These tests verify the generated OpenAPI schema meets requirements + + +class TestOpenAPISchemaCompliance: + """ + Tests for OpenAPI 3.0 schema compliance. + + Verifies that FastOpenAPI generates correct schema for endpoints. + """ + + def test_fastopenapi_registers_routes_correctly(self, app, mock_feature_module_env): + """ + Verify that FastOpenAPI registers routes with correct paths. + """ + ext_fastopenapi.init_app(app) + + # Check that routes are registered + rules = {rule.rule for rule in app.url_map.iter_rules()} + + # Feature endpoints should be registered + assert "/console/api/features" in rules + assert "/console/api/system-features" in rules + + def test_fastopenapi_routes_use_get_method(self, app, mock_feature_module_env): + """ + Verify that feature endpoints only accept GET method. + """ + mock_model = FeatureModel(can_replace_logo=True) + + with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): + ext_fastopenapi.init_app(app) + client = app.test_client() + + # GET should work + response_get = client.get("/console/api/features") + assert response_get.status_code == 200 + + # POST should return 405 Method Not Allowed + response_post = client.post("/console/api/features") + assert response_post.status_code == 405 + + def test_system_features_handles_both_auth_states(self, app, mock_feature_module_env): + """ + Verify /system-features correctly handles both authenticated and unauthenticated states. + + This is a critical test for the is_authenticated try-catch logic. + """ + feature_module = mock_feature_module_env + + # Test 1: When user is authenticated + with patch("controllers.console.feature.FeatureService.get_system_features") as mock_service: + mock_model = SystemFeatureModel(enable_marketplace=True) + mock_service.return_value = mock_model + + # Reset mock to authenticated state + type(feature_module.current_user).is_authenticated = PropertyMock(return_value=True) + + ext_fastopenapi.init_app(app) + client = app.test_client() + response = client.get("/console/api/system-features") + + assert response.status_code == 200 + # Service should be called with is_authenticated=True + mock_service.assert_called_with(is_authenticated=True) From 875dfbd0da2094596ccb06c0b19b4f3f67143c6b Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Wed, 4 Feb 2026 09:55:20 +0000 Subject: [PATCH 04/11] [autofix.ci] apply automated fixes --- api/controllers/console/feature.py | 1 - api/extensions/ext_login.py | 4 +- .../console/test_fastopenapi_feature.py | 126 ++++++++++-------- 3 files changed, 69 insertions(+), 62 deletions(-) diff --git a/api/controllers/console/feature.py b/api/controllers/console/feature.py index f2385c4f71..6d3bf739d7 100644 --- a/api/controllers/console/feature.py +++ b/api/controllers/console/feature.py @@ -6,7 +6,6 @@ from services.feature_service import FeatureModel, FeatureService, SystemFeature from .wraps import account_initialization_required, cloud_utm_record, setup_required - # NOTE: The original feature.py Swagger documentation incorrectly specified a wrapped format # {"features": ...}, but the actual implementation returned a flat FeatureModel. # The frontend has always used the flat format, so we maintain backward compatibility here. diff --git a/api/extensions/ext_login.py b/api/extensions/ext_login.py index 2867166e2a..b9f33c0de6 100644 --- a/api/extensions/ext_login.py +++ b/api/extensions/ext_login.py @@ -118,7 +118,7 @@ def on_user_logged_in(_sender, user): @login_manager.unauthorized_handler def unauthorized_handler(): """Handle unauthorized requests. - + For FastOpenAPI routes (no blueprint), we raise Unauthorized exception which will be caught and serialized properly by the framework. For traditional Blueprint-based routes, we return a Response object. @@ -127,7 +127,7 @@ def unauthorized_handler(): if request.blueprint is None and request.path.startswith("/console/api/"): # Raise exception - FastOpenAPI will handle serialization raise Unauthorized("Unauthorized.") - + # Traditional Blueprint routes - return Response object return Response( json.dumps({"code": "unauthorized", "message": "Unauthorized."}), diff --git a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py index fcebd42d58..51fa6e925d 100644 --- a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py +++ b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py @@ -301,7 +301,7 @@ def test_system_features_unauthenticated(app, mock_feature_module_env): class TestFastOpenAPIAuthenticationBehavior: """ Tests for FastOpenAPI-specific authentication behavior. - + Unlike the tests above that mock authentication decorators, these tests verify the actual authentication flow works correctly for FastOpenAPI routes where request.blueprint is None. @@ -314,55 +314,59 @@ class TestFastOpenAPIAuthenticationBehavior: simulating the production environment more closely. """ from flask_login import LoginManager - + app = Flask(__name__) app.config["TESTING"] = True app.config["SECRET_KEY"] = "test-secret" app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:" - + # Initialize database db.init_app(app) - + # Initialize login manager with unauthorized handler that matches our fix login_manager = LoginManager() login_manager.init_app(app) - + @login_manager.unauthorized_handler def test_unauthorized_handler(): """Simulates our fixed unauthorized_handler behavior.""" from flask import request + # For FastOpenAPI routes, raise exception (serializable by orjson) if request.blueprint is None and request.path.startswith("/console/api/"): raise Unauthorized("Unauthorized.") # For Blueprint routes, return Response (legacy behavior) - from flask import Response import json + + from flask import Response + return Response( json.dumps({"code": "unauthorized", "message": "Unauthorized."}), status=401, content_type="application/json", ) - + return app def test_fastopenapi_route_has_no_blueprint(self, app_with_login_manager, fix_method_view_issue): """ Verify that FastOpenAPI routes have request.blueprint == None. - + This is the core assumption our authentication fix relies on. """ captured_blueprint = {} - + # Create a simple test route to capture request.blueprint @app_with_login_manager.route("/console/api/test-blueprint") def test_route(): from flask import request + captured_blueprint["value"] = request.blueprint return {"status": "ok"} - + client = app_with_login_manager.test_client() response = client.get("/console/api/test-blueprint") - + assert response.status_code == 200 # FastOpenAPI routes registered directly on app have no blueprint assert captured_blueprint["value"] is None @@ -370,18 +374,19 @@ class TestFastOpenAPIAuthenticationBehavior: def test_unauthorized_response_is_serializable_json(self, app_with_login_manager, fix_method_view_issue): """ Verify that unauthorized response for FastOpenAPI routes is valid JSON. - + When unauthorized_handler raises Unauthorized exception for FastOpenAPI routes, Flask/Werkzeug converts it to a proper HTTP 401 response that is serializable. """ + @app_with_login_manager.route("/console/api/protected") def protected_route(): # Simulate login_required behavior when user is not authenticated raise Unauthorized("Unauthorized.") - + client = app_with_login_manager.test_client() response = client.get("/console/api/protected") - + assert response.status_code == 401 # Response should be valid (either JSON or HTML error page, but not a serialization error) assert response.data is not None @@ -392,22 +397,22 @@ class TestFastOpenAPIAuthenticationBehavior: """ Explicitly verify that response format is flat FeatureModel, not wrapped in {"features": {...}}. - + This ensures backward compatibility with frontend expectations. """ mock_model = FeatureModel(can_replace_logo=True, billing=None) - + with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): ext_fastopenapi.init_app(app) client = app.test_client() response = client.get("/console/api/features") - + assert response.status_code == 200 json_data = response.get_json() - + # Should NOT be wrapped format assert "features" not in json_data or not isinstance(json_data.get("features"), dict) - + # Should be flat format - top level keys are FeatureModel fields assert "can_replace_logo" in json_data @@ -421,7 +426,7 @@ class TestFastOpenAPIAuthenticationBehavior: class TestResponseFormatValidation: """ Tests for response format validation. - + Ensures FastOpenAPI produces correct Content-Type headers and JSON format that matches OpenAPI 3.0 specification requirements. """ @@ -429,16 +434,16 @@ class TestResponseFormatValidation: def test_response_content_type_is_json(self, app, mock_feature_module_env): """ Verify response Content-Type is application/json. - + FastOpenAPI uses orjson for serialization, must produce correct Content-Type. """ mock_model = FeatureModel(can_replace_logo=True) - + with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): ext_fastopenapi.init_app(app) client = app.test_client() response = client.get("/console/api/features") - + assert response.status_code == 200 # Content-Type should be JSON assert "application/json" in response.content_type @@ -448,34 +453,34 @@ class TestResponseFormatValidation: Verify /system-features response Content-Type is application/json. """ mock_model = SystemFeatureModel(enable_marketplace=True) - + with patch("controllers.console.feature.FeatureService.get_system_features", return_value=mock_model): ext_fastopenapi.init_app(app) client = app.test_client() response = client.get("/console/api/system-features") - + assert response.status_code == 200 assert "application/json" in response.content_type def test_feature_model_all_fields_serialized(self, app, mock_feature_module_env): """ Verify all FeatureModel fields are serialized in response. - + This ensures Pydantic model dump is complete for OpenAPI schema compliance. """ mock_model = FeatureModel( can_replace_logo=True, model_load_balancing_enabled=True, ) - + with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): ext_fastopenapi.init_app(app) client = app.test_client() response = client.get("/console/api/features") - + assert response.status_code == 200 json_data = response.get_json() - + # Verify key fields are present assert "can_replace_logo" in json_data assert json_data["can_replace_logo"] is True @@ -492,7 +497,7 @@ class TestResponseFormatValidation: class TestAuthenticationWithRealisticMocking: """ Tests authentication behavior with more realistic mocking. - + Unlike tests that completely bypass decorators, these tests mock at the user/account level to verify decorator behavior. """ @@ -501,26 +506,26 @@ class TestAuthenticationWithRealisticMocking: def app_with_real_decorators(self, monkeypatch): """ Creates a Flask app where decorators execute but dependencies are mocked. - + This provides a middle ground between: - Full integration tests (require database setup) - Tests that completely bypass decorators (don't test auth flow) """ from flask import Flask from flask_login import LoginManager - + app = Flask(__name__) app.config["TESTING"] = True app.config["SECRET_KEY"] = "test-secret" app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:" - + # Initialize database db.init_app(app) - + # Initialize login manager login_manager = LoginManager() login_manager.init_app(app) - + # Mock request loader to return None (unauthenticated) @login_manager.request_loader def mock_load_user(request): @@ -533,62 +538,65 @@ class TestAuthenticationWithRealisticMocking: mock_user.current_tenant_id = "test-tenant-id" return mock_user return None - + @login_manager.unauthorized_handler def handle_unauthorized(): """Simulates production unauthorized_handler for FastOpenAPI routes.""" from flask import request + if request.blueprint is None and request.path.startswith("/console/api/"): raise Unauthorized("Unauthorized.") - from flask import Response import json + + from flask import Response + return Response( json.dumps({"code": "unauthorized", "message": "Unauthorized."}), status=401, content_type="application/json", ) - + return app def test_features_endpoint_requires_authentication_concept(self, app_with_real_decorators, fix_method_view_issue): """ Conceptual test: verify that a protected route would return 401 without auth. - + Note: This test creates a simple protected route to verify the auth flow, since the actual feature.py module loading is complex. """ from flask_login import login_required as flask_login_required - + @app_with_real_decorators.route("/console/api/test-protected") @flask_login_required def protected_test_route(): return {"status": "authenticated"} - + client = app_with_real_decorators.test_client() - + # Without authentication - should return 401 response = client.get("/console/api/test-protected") assert response.status_code == 401 - + # With valid test token - should return 200 response_with_auth = client.get( - "/console/api/test-protected", - headers={"Authorization": "Bearer test-valid-token"} + "/console/api/test-protected", headers={"Authorization": "Bearer test-valid-token"} ) assert response_with_auth.status_code == 200 def test_system_features_no_auth_decorator_concept(self, app_with_real_decorators, fix_method_view_issue): """ Conceptual test: verify that an unprotected route works without auth. - + This mirrors /system-features behavior which has no @login_required. """ + @app_with_real_decorators.route("/console/api/test-public") def public_test_route(): return {"status": "public"} - + client = app_with_real_decorators.test_client() - + # Without authentication - should still work response = client.get("/console/api/test-public") assert response.status_code == 200 @@ -604,7 +612,7 @@ class TestAuthenticationWithRealisticMocking: class TestOpenAPISchemaCompliance: """ Tests for OpenAPI 3.0 schema compliance. - + Verifies that FastOpenAPI generates correct schema for endpoints. """ @@ -613,10 +621,10 @@ class TestOpenAPISchemaCompliance: Verify that FastOpenAPI registers routes with correct paths. """ ext_fastopenapi.init_app(app) - + # Check that routes are registered rules = {rule.rule for rule in app.url_map.iter_rules()} - + # Feature endpoints should be registered assert "/console/api/features" in rules assert "/console/api/system-features" in rules @@ -626,15 +634,15 @@ class TestOpenAPISchemaCompliance: Verify that feature endpoints only accept GET method. """ mock_model = FeatureModel(can_replace_logo=True) - + with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): ext_fastopenapi.init_app(app) client = app.test_client() - + # GET should work response_get = client.get("/console/api/features") assert response_get.status_code == 200 - + # POST should return 405 Method Not Allowed response_post = client.post("/console/api/features") assert response_post.status_code == 405 @@ -642,23 +650,23 @@ class TestOpenAPISchemaCompliance: def test_system_features_handles_both_auth_states(self, app, mock_feature_module_env): """ Verify /system-features correctly handles both authenticated and unauthenticated states. - + This is a critical test for the is_authenticated try-catch logic. """ feature_module = mock_feature_module_env - + # Test 1: When user is authenticated with patch("controllers.console.feature.FeatureService.get_system_features") as mock_service: mock_model = SystemFeatureModel(enable_marketplace=True) mock_service.return_value = mock_model - + # Reset mock to authenticated state type(feature_module.current_user).is_authenticated = PropertyMock(return_value=True) - + ext_fastopenapi.init_app(app) client = app.test_client() response = client.get("/console/api/system-features") - + assert response.status_code == 200 # Service should be called with is_authenticated=True mock_service.assert_called_with(is_authenticated=True) From 5cabf4c44227d131228e7a441dd12f88b533c8ce Mon Sep 17 00:00:00 2001 From: Cursx <33718736+Cursx@users.noreply.github.com> Date: Wed, 4 Feb 2026 18:09:40 +0800 Subject: [PATCH 05/11] Update test_fastopenapi_feature.py --- .../unit_tests/controllers/console/test_fastopenapi_feature.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py index 51fa6e925d..2c0aef0e96 100644 --- a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py +++ b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py @@ -400,7 +400,7 @@ class TestFastOpenAPIAuthenticationBehavior: This ensures backward compatibility with frontend expectations. """ - mock_model = FeatureModel(can_replace_logo=True, billing=None) + mock_model = FeatureModel(can_replace_logo=True) with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): ext_fastopenapi.init_app(app) From 8cf8b31826d9be6392ede82e9cb84a93cfe82652 Mon Sep 17 00:00:00 2001 From: Cursx <33718736+Cursx@users.noreply.github.com> Date: Wed, 4 Feb 2026 18:28:00 +0800 Subject: [PATCH 06/11] Update ext_fastopenapi.py --- api/extensions/ext_fastopenapi.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/api/extensions/ext_fastopenapi.py b/api/extensions/ext_fastopenapi.py index ab4d23a072..2ce60c3785 100644 --- a/api/extensions/ext_fastopenapi.py +++ b/api/extensions/ext_fastopenapi.py @@ -29,10 +29,11 @@ def init_app(app: DifyApp) -> None: # Ensure route decorators are evaluated. import controllers.console.init_validate as init_validate_module import controllers.console.ping as ping_module - from controllers.console import remote_files, setup + from controllers.console import feature, remote_files, setup _ = init_validate_module _ = ping_module + _ = feature _ = remote_files _ = setup From 378060882021bd4eb68c981d50f71e1b16aaf3f6 Mon Sep 17 00:00:00 2001 From: Cursx <33718736+Cursx@users.noreply.github.com> Date: Wed, 4 Feb 2026 19:19:56 +0800 Subject: [PATCH 07/11] Update test_fastopenapi_feature.py --- .../console/test_fastopenapi_feature.py | 672 +++--------------- 1 file changed, 96 insertions(+), 576 deletions(-) diff --git a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py index 2c0aef0e96..97ee6ddda5 100644 --- a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py +++ b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py @@ -1,7 +1,4 @@ import builtins -import contextlib -import importlib -import sys from unittest.mock import MagicMock, PropertyMock, patch import pytest @@ -13,175 +10,43 @@ from extensions import ext_fastopenapi from extensions.ext_database import db from services.feature_service import FeatureModel, SystemFeatureModel +if not hasattr(builtins, "MethodView"): + builtins.MethodView = MethodView # type: ignore[attr-defined] + @pytest.fixture -def app(): - """ - Creates a Flask application instance configured for testing. - """ +def app() -> Flask: + """Creates a Flask app configured for testing.""" app = Flask(__name__) app.config["TESTING"] = True app.config["SECRET_KEY"] = "test-secret" app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:" - - # Initialize the database with the app db.init_app(app) - return app -@pytest.fixture(autouse=True) -def fix_method_view_issue(monkeypatch): - """ - Automatic fixture to patch 'builtins.MethodView'. - - Why this is needed: - The official legacy codebase contains a global patch in its initialization logic: - if not hasattr(builtins, "MethodView"): - builtins.MethodView = MethodView - - Some dependencies (like ext_fastopenapi or older Flask extensions) might implicitly - rely on 'MethodView' being available in the global builtins namespace. - - Refactoring Note: - While patching builtins is generally discouraged due to global side effects, - this fixture reproduces the production environment's state to ensure tests are realistic. - We use 'monkeypatch' to ensure that this change is undone after the test finishes, - keeping other tests isolated. - """ - if not hasattr(builtins, "MethodView"): - # 'raising=False' allows us to set an attribute that doesn't exist yet - monkeypatch.setattr(builtins, "MethodView", MethodView, raising=False) - - -# ------------------------------------------------------------------------------ -# Helper Functions for Fixture Complexity Reduction -# ------------------------------------------------------------------------------ - - -def _create_isolated_router(): - """ - Creates a fresh, isolated router instance to prevent route pollution. - """ - import controllers.fastopenapi - - # Dynamically get the class type (e.g., FlaskRouter) to avoid hardcoding dependencies - RouterClass = type(controllers.fastopenapi.console_router) - return RouterClass() - - -@contextlib.contextmanager -def _patch_auth_and_router(temp_router): - """ - Context manager that applies all necessary patches for: - 1. The console_router (redirecting to our isolated temp_router) - 2. Authentication decorators (disabling them with no-ops) - 3. User/Account loaders (mocking authenticated state) - """ - - def noop(f): - return f - - # We patch the SOURCE of the decorators/functions, not the destination module. - # This ensures that when 'controllers.console.feature' imports them, it gets the mocks. +@pytest.fixture +def mock_auth(): + """Mocks authentication decorators and user context.""" + noop = lambda f: f with ( - patch("controllers.fastopenapi.console_router", temp_router), - patch("extensions.ext_fastopenapi.console_router", temp_router), patch("controllers.console.wraps.setup_required", side_effect=noop), patch("libs.login.login_required", side_effect=noop), patch("controllers.console.wraps.account_initialization_required", side_effect=noop), patch("controllers.console.wraps.cloud_utm_record", side_effect=noop), patch("libs.login.current_account_with_tenant", return_value=(MagicMock(), "tenant-id")), - patch("libs.login.current_user", MagicMock(is_authenticated=True)), + patch("libs.login.current_user", MagicMock(is_authenticated=True)) as mock_user, ): - # Explicitly reload ext_fastopenapi to ensure it uses the patched console_router - import extensions.ext_fastopenapi - - importlib.reload(extensions.ext_fastopenapi) - - yield - - -def _force_reload_module(target_module: str, alias_module: str): - """ - Forces a reload of the specified module and handles sys.modules aliasing. - - Why reload? - Python decorators (like @route, @login_required) run at IMPORT time. - To apply our patches (mocks/no-ops) to these decorators, we must re-import - the module while the patches are active. - - Why alias? - If 'ext_fastopenapi' imports the controller as 'api.controllers...', but we import - it as 'controllers...', Python treats them as two separate modules. This causes: - 1. Double execution of decorators (registering routes twice -> AssertionError). - 2. Type mismatch errors (Class A from module X is not Class A from module Y). - - This function ensures both names point to the SAME loaded module instance. - """ - # 1. Clean existing entries to force re-import - if target_module in sys.modules: - del sys.modules[target_module] - if alias_module in sys.modules: - del sys.modules[alias_module] - - # 2. Import the module (triggering decorators with active patches) - module = importlib.import_module(target_module) - - # 3. Alias the module in sys.modules to prevent double loading - sys.modules[alias_module] = sys.modules[target_module] - - return module - - -def _cleanup_modules(target_module: str, alias_module: str): - """ - Removes the module and its alias from sys.modules to prevent side effects - on other tests. - """ - if target_module in sys.modules: - del sys.modules[target_module] - if alias_module in sys.modules: - del sys.modules[alias_module] - - -@pytest.fixture -def mock_feature_module_env(): - """ - Sets up a mocked environment for the feature module. - - This fixture orchestrates: - 1. Creating an isolated router. - 2. Patching authentication and global dependencies. - 3. Reloading the controller module to apply patches to decorators. - 4. cleaning up sys.modules afterwards. - """ - target_module = "controllers.console.feature" - alias_module = "api.controllers.console.feature" - - # 1. Prepare isolated router - temp_router = _create_isolated_router() - - # 2. Apply patches - try: - with _patch_auth_and_router(temp_router): - # 3. Reload module to register routes on the temp_router - feature_module = _force_reload_module(target_module, alias_module) - - yield feature_module - - finally: - # 4. Teardown: Clean up sys.modules - _cleanup_modules(target_module, alias_module) + yield mock_user # ------------------------------------------------------------------------------ -# Test Cases +# Core Feature Endpoint Tests # ------------------------------------------------------------------------------ @pytest.mark.parametrize( - ("url", "service_mock_path", "mock_model_instance"), + ("url", "service_mock_path", "mock_model"), [ ( "/console/api/features", @@ -195,24 +60,15 @@ def mock_feature_module_env(): ), ], ) -def test_console_features_success(app, mock_feature_module_env, url, service_mock_path, mock_model_instance): - """ - Tests that the feature APIs return a 200 OK status and correct JSON structure. - """ - # Patch the service layer to return our mock model instance - with patch(service_mock_path, return_value=mock_model_instance): - # Initialize the API extension +def test_feature_endpoints_return_200_with_flat_json(app, mock_auth, url, service_mock_path, mock_model): + """Tests that feature endpoints return 200 with flat JSON format.""" + with patch(service_mock_path, return_value=mock_model): ext_fastopenapi.init_app(app) + response = app.test_client().get(url) - client = app.test_client() - response = client.get(url) - - # Assertions - assert response.status_code == 200, f"Request failed with status {response.status_code}: {response.text}" - - # Verify the JSON response matches the Pydantic model dump (flat format, not wrapped) - expected_data = mock_model_instance.model_dump(mode="json") - assert response.get_json() == expected_data + assert response.status_code == 200 + assert response.get_json() == mock_model.model_dump(mode="json") + assert "application/json" in response.content_type @pytest.mark.parametrize( @@ -222,451 +78,115 @@ def test_console_features_success(app, mock_feature_module_env, url, service_moc ("/console/api/system-features", "controllers.console.feature.FeatureService.get_system_features"), ], ) -def test_console_features_service_error(app, mock_feature_module_env, url, service_mock_path): - """ - Tests how the application handles Service layer errors. - - Note: When an exception occurs in the view, it is typically caught by the framework - (Flask or the OpenAPI wrapper) and converted to a 500 error response. - This test verifies that the application returns a 500 status code. - """ - # Simulate a service failure +def test_feature_endpoints_return_500_on_service_error(app, mock_auth, url, service_mock_path): + """Tests that service errors return 500.""" with patch(service_mock_path, side_effect=ValueError("Service Failure")): ext_fastopenapi.init_app(app) - client = app.test_client() + response = app.test_client().get(url) - # When an exception occurs in the view, it is typically caught by the framework - # (Flask or the OpenAPI wrapper) and converted to a 500 error response. - response = client.get(url) - - assert response.status_code == 500 - # Check if the error details are exposed in the response (depends on error handler config) - # We accept either generic 500 or the specific error message - assert "Service Failure" in response.text or "Internal Server Error" in response.text + assert response.status_code == 500 -def test_system_features_unauthenticated(app, mock_feature_module_env): - """ - Tests that /console/api/system-features endpoint works without authentication. - - This test verifies the try-except block in get_system_features that handles - unauthenticated requests by passing is_authenticated=False to the service layer. - """ - feature_module = mock_feature_module_env - - # Override the behavior of the current_user mock - # The fixture patched 'libs.login.current_user', so 'controllers.console.feature.current_user' - # refers to that same Mock object. - mock_user = feature_module.current_user - - # Simulate property access raising Unauthorized - # Note: We must reset side_effect if it was set, or set it here. - # The fixture initialized it as MagicMock(is_authenticated=True). - # We want type(mock_user).is_authenticated to raise Unauthorized. +def test_system_features_handles_unauthenticated_users(app, mock_auth): + """Tests /system-features passes is_authenticated=False when auth fails.""" + mock_user = mock_auth type(mock_user).is_authenticated = PropertyMock(side_effect=Unauthorized) + mock_model = SystemFeatureModel(enable_marketplace=True) - # Patch the service layer for this specific test - with patch("controllers.console.feature.FeatureService.get_system_features") as mock_service: - # Setup mock service return value - mock_model = SystemFeatureModel(enable_marketplace=True) - mock_service.return_value = mock_model - - # Initialize app + with patch("controllers.console.feature.FeatureService.get_system_features", return_value=mock_model) as svc: ext_fastopenapi.init_app(app) - client = app.test_client() + response = app.test_client().get("/console/api/system-features") - # Act - response = client.get("/console/api/system-features") + assert response.status_code == 200 + svc.assert_called_once_with(is_authenticated=False) - # Assert - assert response.status_code == 200, f"Request failed: {response.text}" - # Verify service was called with is_authenticated=False - mock_service.assert_called_once_with(is_authenticated=False) +def test_features_endpoint_rejects_post_method(app, mock_auth): + """Tests that feature endpoints only accept GET.""" + with patch("controllers.console.feature.FeatureService.get_features", return_value=FeatureModel()): + ext_fastopenapi.init_app(app) + response = app.test_client().post("/console/api/features") - # Verify response body (flat format, not wrapped) - expected_data = mock_model.model_dump(mode="json") - assert response.get_json() == expected_data + assert response.status_code == 405 + + +def test_routes_are_registered_correctly(app, mock_auth): + """Tests that FastOpenAPI registers routes with correct paths.""" + ext_fastopenapi.init_app(app) + rules = {rule.rule for rule in app.url_map.iter_rules()} + + assert "/console/api/features" in rules + assert "/console/api/system-features" in rules # ------------------------------------------------------------------------------ # FastOpenAPI Authentication Tests # ------------------------------------------------------------------------------ -# These tests verify that our fixes for FastOpenAPI routing work correctly: -# 1. load_user_from_request supports FastOpenAPI routes (no blueprint) -# 2. unauthorized_handler returns serializable response for FastOpenAPI routes -# 3. Response format is flat (not wrapped in {"features": ...}) -class TestFastOpenAPIAuthenticationBehavior: - """ - Tests for FastOpenAPI-specific authentication behavior. +@pytest.fixture +def app_with_login_manager(): + """Creates Flask app with login manager to test auth behavior.""" + from flask_login import LoginManager - Unlike the tests above that mock authentication decorators, - these tests verify the actual authentication flow works correctly - for FastOpenAPI routes where request.blueprint is None. - """ + app = Flask(__name__) + app.config["TESTING"] = True + app.config["SECRET_KEY"] = "test-secret" - @pytest.fixture - def app_with_login_manager(self): - """ - Creates a Flask app with login manager configured, - simulating the production environment more closely. - """ - from flask_login import LoginManager + login_manager = LoginManager() + login_manager.init_app(app) - app = Flask(__name__) - app.config["TESTING"] = True - app.config["SECRET_KEY"] = "test-secret" - app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:" + @login_manager.request_loader + def load_user(request): + if request.headers.get("Authorization") == "Bearer valid-token": + user = MagicMock() + user.is_authenticated = True + return user + return None - # Initialize database - db.init_app(app) - - # Initialize login manager with unauthorized handler that matches our fix - login_manager = LoginManager() - login_manager.init_app(app) - - @login_manager.unauthorized_handler - def test_unauthorized_handler(): - """Simulates our fixed unauthorized_handler behavior.""" - from flask import request - - # For FastOpenAPI routes, raise exception (serializable by orjson) - if request.blueprint is None and request.path.startswith("/console/api/"): - raise Unauthorized("Unauthorized.") - # For Blueprint routes, return Response (legacy behavior) - import json - - from flask import Response - - return Response( - json.dumps({"code": "unauthorized", "message": "Unauthorized."}), - status=401, - content_type="application/json", - ) - - return app - - def test_fastopenapi_route_has_no_blueprint(self, app_with_login_manager, fix_method_view_issue): - """ - Verify that FastOpenAPI routes have request.blueprint == None. - - This is the core assumption our authentication fix relies on. - """ - captured_blueprint = {} - - # Create a simple test route to capture request.blueprint - @app_with_login_manager.route("/console/api/test-blueprint") - def test_route(): - from flask import request - - captured_blueprint["value"] = request.blueprint - return {"status": "ok"} - - client = app_with_login_manager.test_client() - response = client.get("/console/api/test-blueprint") - - assert response.status_code == 200 - # FastOpenAPI routes registered directly on app have no blueprint - assert captured_blueprint["value"] is None - - def test_unauthorized_response_is_serializable_json(self, app_with_login_manager, fix_method_view_issue): - """ - Verify that unauthorized response for FastOpenAPI routes is valid JSON. - - When unauthorized_handler raises Unauthorized exception for FastOpenAPI routes, - Flask/Werkzeug converts it to a proper HTTP 401 response that is serializable. - """ - - @app_with_login_manager.route("/console/api/protected") - def protected_route(): - # Simulate login_required behavior when user is not authenticated + @login_manager.unauthorized_handler + def unauthorized(): + from flask import request + # FastOpenAPI routes: raise exception (serializable) + if request.blueprint is None and request.path.startswith("/console/api/"): raise Unauthorized("Unauthorized.") + # Blueprint routes: return Response + from flask import Response + import json + return Response(json.dumps({"code": "unauthorized"}), status=401, content_type="application/json") - client = app_with_login_manager.test_client() - response = client.get("/console/api/protected") + return app - assert response.status_code == 401 - # Response should be valid (either JSON or HTML error page, but not a serialization error) - assert response.data is not None - # Should not be a TypeError from orjson trying to serialize Response object - assert b"TypeError" not in response.data - def test_response_format_is_flat_not_wrapped(self, app, mock_feature_module_env): - """ - Explicitly verify that response format is flat FeatureModel, - not wrapped in {"features": {...}}. +def test_fastopenapi_route_has_no_blueprint(app_with_login_manager): + """Verifies FastOpenAPI routes have request.blueprint == None.""" + captured = {} - This ensures backward compatibility with frontend expectations. - """ - mock_model = FeatureModel(can_replace_logo=True) + @app_with_login_manager.route("/console/api/test") + def test_route(): + from flask import request + captured["blueprint"] = request.blueprint + return {"ok": True} - with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): - ext_fastopenapi.init_app(app) - client = app.test_client() - response = client.get("/console/api/features") + response = app_with_login_manager.test_client().get("/console/api/test") - assert response.status_code == 200 - json_data = response.get_json() + assert response.status_code == 200 + assert captured["blueprint"] is None - # Should NOT be wrapped format - assert "features" not in json_data or not isinstance(json_data.get("features"), dict) - # Should be flat format - top level keys are FeatureModel fields - assert "can_replace_logo" in json_data +def test_protected_route_returns_401_without_auth(app_with_login_manager): + """Tests that protected routes return 401 without authentication.""" + from flask_login import login_required + @app_with_login_manager.route("/console/api/protected") + @login_required + def protected(): + return {"status": "ok"} -# ------------------------------------------------------------------------------ -# Response Format and Content-Type Tests -# ------------------------------------------------------------------------------ -# These tests verify OpenAPI v3 migration requirements for response handling + client = app_with_login_manager.test_client() + # Without auth + assert client.get("/console/api/protected").status_code == 401 -class TestResponseFormatValidation: - """ - Tests for response format validation. - - Ensures FastOpenAPI produces correct Content-Type headers and JSON format - that matches OpenAPI 3.0 specification requirements. - """ - - def test_response_content_type_is_json(self, app, mock_feature_module_env): - """ - Verify response Content-Type is application/json. - - FastOpenAPI uses orjson for serialization, must produce correct Content-Type. - """ - mock_model = FeatureModel(can_replace_logo=True) - - with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): - ext_fastopenapi.init_app(app) - client = app.test_client() - response = client.get("/console/api/features") - - assert response.status_code == 200 - # Content-Type should be JSON - assert "application/json" in response.content_type - - def test_system_features_content_type_is_json(self, app, mock_feature_module_env): - """ - Verify /system-features response Content-Type is application/json. - """ - mock_model = SystemFeatureModel(enable_marketplace=True) - - with patch("controllers.console.feature.FeatureService.get_system_features", return_value=mock_model): - ext_fastopenapi.init_app(app) - client = app.test_client() - response = client.get("/console/api/system-features") - - assert response.status_code == 200 - assert "application/json" in response.content_type - - def test_feature_model_all_fields_serialized(self, app, mock_feature_module_env): - """ - Verify all FeatureModel fields are serialized in response. - - This ensures Pydantic model dump is complete for OpenAPI schema compliance. - """ - mock_model = FeatureModel( - can_replace_logo=True, - model_load_balancing_enabled=True, - ) - - with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): - ext_fastopenapi.init_app(app) - client = app.test_client() - response = client.get("/console/api/features") - - assert response.status_code == 200 - json_data = response.get_json() - - # Verify key fields are present - assert "can_replace_logo" in json_data - assert json_data["can_replace_logo"] is True - assert "model_load_balancing_enabled" in json_data - assert json_data["model_load_balancing_enabled"] is True - - -# ------------------------------------------------------------------------------ -# Authentication Behavior Tests with Realistic Mocking -# ------------------------------------------------------------------------------ -# These tests use more realistic mocking to verify authentication behavior - - -class TestAuthenticationWithRealisticMocking: - """ - Tests authentication behavior with more realistic mocking. - - Unlike tests that completely bypass decorators, these tests mock - at the user/account level to verify decorator behavior. - """ - - @pytest.fixture - def app_with_real_decorators(self, monkeypatch): - """ - Creates a Flask app where decorators execute but dependencies are mocked. - - This provides a middle ground between: - - Full integration tests (require database setup) - - Tests that completely bypass decorators (don't test auth flow) - """ - from flask import Flask - from flask_login import LoginManager - - app = Flask(__name__) - app.config["TESTING"] = True - app.config["SECRET_KEY"] = "test-secret" - app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:" - - # Initialize database - db.init_app(app) - - # Initialize login manager - login_manager = LoginManager() - login_manager.init_app(app) - - # Mock request loader to return None (unauthenticated) - @login_manager.request_loader - def mock_load_user(request): - # Check for test auth header - auth_header = request.headers.get("Authorization") - if auth_header and auth_header.startswith("Bearer test-valid-token"): - # Return a mock user - mock_user = MagicMock() - mock_user.is_authenticated = True - mock_user.current_tenant_id = "test-tenant-id" - return mock_user - return None - - @login_manager.unauthorized_handler - def handle_unauthorized(): - """Simulates production unauthorized_handler for FastOpenAPI routes.""" - from flask import request - - if request.blueprint is None and request.path.startswith("/console/api/"): - raise Unauthorized("Unauthorized.") - import json - - from flask import Response - - return Response( - json.dumps({"code": "unauthorized", "message": "Unauthorized."}), - status=401, - content_type="application/json", - ) - - return app - - def test_features_endpoint_requires_authentication_concept(self, app_with_real_decorators, fix_method_view_issue): - """ - Conceptual test: verify that a protected route would return 401 without auth. - - Note: This test creates a simple protected route to verify the auth flow, - since the actual feature.py module loading is complex. - """ - from flask_login import login_required as flask_login_required - - @app_with_real_decorators.route("/console/api/test-protected") - @flask_login_required - def protected_test_route(): - return {"status": "authenticated"} - - client = app_with_real_decorators.test_client() - - # Without authentication - should return 401 - response = client.get("/console/api/test-protected") - assert response.status_code == 401 - - # With valid test token - should return 200 - response_with_auth = client.get( - "/console/api/test-protected", headers={"Authorization": "Bearer test-valid-token"} - ) - assert response_with_auth.status_code == 200 - - def test_system_features_no_auth_decorator_concept(self, app_with_real_decorators, fix_method_view_issue): - """ - Conceptual test: verify that an unprotected route works without auth. - - This mirrors /system-features behavior which has no @login_required. - """ - - @app_with_real_decorators.route("/console/api/test-public") - def public_test_route(): - return {"status": "public"} - - client = app_with_real_decorators.test_client() - - # Without authentication - should still work - response = client.get("/console/api/test-public") - assert response.status_code == 200 - assert response.get_json()["status"] == "public" - - -# ------------------------------------------------------------------------------ -# OpenAPI Schema Compliance Tests -# ------------------------------------------------------------------------------ -# These tests verify the generated OpenAPI schema meets requirements - - -class TestOpenAPISchemaCompliance: - """ - Tests for OpenAPI 3.0 schema compliance. - - Verifies that FastOpenAPI generates correct schema for endpoints. - """ - - def test_fastopenapi_registers_routes_correctly(self, app, mock_feature_module_env): - """ - Verify that FastOpenAPI registers routes with correct paths. - """ - ext_fastopenapi.init_app(app) - - # Check that routes are registered - rules = {rule.rule for rule in app.url_map.iter_rules()} - - # Feature endpoints should be registered - assert "/console/api/features" in rules - assert "/console/api/system-features" in rules - - def test_fastopenapi_routes_use_get_method(self, app, mock_feature_module_env): - """ - Verify that feature endpoints only accept GET method. - """ - mock_model = FeatureModel(can_replace_logo=True) - - with patch("controllers.console.feature.FeatureService.get_features", return_value=mock_model): - ext_fastopenapi.init_app(app) - client = app.test_client() - - # GET should work - response_get = client.get("/console/api/features") - assert response_get.status_code == 200 - - # POST should return 405 Method Not Allowed - response_post = client.post("/console/api/features") - assert response_post.status_code == 405 - - def test_system_features_handles_both_auth_states(self, app, mock_feature_module_env): - """ - Verify /system-features correctly handles both authenticated and unauthenticated states. - - This is a critical test for the is_authenticated try-catch logic. - """ - feature_module = mock_feature_module_env - - # Test 1: When user is authenticated - with patch("controllers.console.feature.FeatureService.get_system_features") as mock_service: - mock_model = SystemFeatureModel(enable_marketplace=True) - mock_service.return_value = mock_model - - # Reset mock to authenticated state - type(feature_module.current_user).is_authenticated = PropertyMock(return_value=True) - - ext_fastopenapi.init_app(app) - client = app.test_client() - response = client.get("/console/api/system-features") - - assert response.status_code == 200 - # Service should be called with is_authenticated=True - mock_service.assert_called_with(is_authenticated=True) + # With valid token + assert client.get("/console/api/protected", headers={"Authorization": "Bearer valid-token"}).status_code == 200 From 6e6aa29b40722308b36e6678e9d48262db9ea41f Mon Sep 17 00:00:00 2001 From: Cursx <33718736+Cursx@users.noreply.github.com> Date: Wed, 4 Feb 2026 19:23:20 +0800 Subject: [PATCH 08/11] Update test_fastopenapi_feature.py --- .../unit_tests/controllers/console/test_fastopenapi_feature.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py index 97ee6ddda5..8ee01b6ea3 100644 --- a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py +++ b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py @@ -28,7 +28,8 @@ def app() -> Flask: @pytest.fixture def mock_auth(): """Mocks authentication decorators and user context.""" - noop = lambda f: f + def noop(f): + return f with ( patch("controllers.console.wraps.setup_required", side_effect=noop), patch("libs.login.login_required", side_effect=noop), From f60dc667afeb499f08d1560ed9904be30374ac6f Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Wed, 4 Feb 2026 11:27:23 +0000 Subject: [PATCH 09/11] [autofix.ci] apply automated fixes --- .../controllers/console/test_fastopenapi_feature.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py index 8ee01b6ea3..421fae986a 100644 --- a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py +++ b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py @@ -28,8 +28,10 @@ def app() -> Flask: @pytest.fixture def mock_auth(): """Mocks authentication decorators and user context.""" + def noop(f): return f + with ( patch("controllers.console.wraps.setup_required", side_effect=noop), patch("libs.login.login_required", side_effect=noop), @@ -148,12 +150,15 @@ def app_with_login_manager(): @login_manager.unauthorized_handler def unauthorized(): from flask import request + # FastOpenAPI routes: raise exception (serializable) if request.blueprint is None and request.path.startswith("/console/api/"): raise Unauthorized("Unauthorized.") # Blueprint routes: return Response - from flask import Response import json + + from flask import Response + return Response(json.dumps({"code": "unauthorized"}), status=401, content_type="application/json") return app @@ -166,6 +171,7 @@ def test_fastopenapi_route_has_no_blueprint(app_with_login_manager): @app_with_login_manager.route("/console/api/test") def test_route(): from flask import request + captured["blueprint"] = request.blueprint return {"ok": True} From c2b3eaa9cd14c6906439fea3f0c1ffa8f47f69da Mon Sep 17 00:00:00 2001 From: Cursx <33718736+Cursx@users.noreply.github.com> Date: Wed, 4 Feb 2026 20:11:27 +0800 Subject: [PATCH 10/11] Update test_fastopenapi_feature.py --- .../console/test_fastopenapi_feature.py | 244 +++++++++++------- 1 file changed, 154 insertions(+), 90 deletions(-) diff --git a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py index 421fae986a..7b7af1264f 100644 --- a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py +++ b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py @@ -1,4 +1,7 @@ import builtins +import contextlib +import importlib +import sys from unittest.mock import MagicMock, PropertyMock, patch import pytest @@ -10,12 +13,14 @@ from extensions import ext_fastopenapi from extensions.ext_database import db from services.feature_service import FeatureModel, SystemFeatureModel -if not hasattr(builtins, "MethodView"): - builtins.MethodView = MethodView # type: ignore[attr-defined] + +# ------------------------------------------------------------------------------ +# Fixtures +# ------------------------------------------------------------------------------ @pytest.fixture -def app() -> Flask: +def app(): """Creates a Flask app configured for testing.""" app = Flask(__name__) app.config["TESTING"] = True @@ -25,22 +30,73 @@ def app() -> Flask: return app -@pytest.fixture -def mock_auth(): - """Mocks authentication decorators and user context.""" +@pytest.fixture(autouse=True) +def fix_method_view_issue(monkeypatch): + """Patches builtins.MethodView for legacy compatibility.""" + if not hasattr(builtins, "MethodView"): + monkeypatch.setattr(builtins, "MethodView", MethodView, raising=False) + + +def _create_isolated_router(): + """Creates a fresh router instance to prevent route pollution.""" + import controllers.fastopenapi + + RouterClass = type(controllers.fastopenapi.console_router) + return RouterClass() + + +@contextlib.contextmanager +def _patch_auth_and_router(temp_router): + """Patches console_router and authentication decorators.""" def noop(f): return f with ( + patch("controllers.fastopenapi.console_router", temp_router), + patch("extensions.ext_fastopenapi.console_router", temp_router), patch("controllers.console.wraps.setup_required", side_effect=noop), patch("libs.login.login_required", side_effect=noop), patch("controllers.console.wraps.account_initialization_required", side_effect=noop), patch("controllers.console.wraps.cloud_utm_record", side_effect=noop), patch("libs.login.current_account_with_tenant", return_value=(MagicMock(), "tenant-id")), - patch("libs.login.current_user", MagicMock(is_authenticated=True)) as mock_user, + patch("libs.login.current_user", MagicMock(is_authenticated=True)), ): - yield mock_user + import extensions.ext_fastopenapi + + importlib.reload(extensions.ext_fastopenapi) + yield + + +def _force_reload_module(target_module: str, alias_module: str): + """Forces module reload to apply patches to decorators at import time.""" + if target_module in sys.modules: + del sys.modules[target_module] + if alias_module in sys.modules: + del sys.modules[alias_module] + + module = importlib.import_module(target_module) + sys.modules[alias_module] = sys.modules[target_module] + return module + + +@pytest.fixture +def mock_feature_module_env(): + """Sets up mocked environment for feature module with isolated router.""" + target_module = "controllers.console.feature" + alias_module = "api.controllers.console.feature" + + temp_router = _create_isolated_router() + + try: + with _patch_auth_and_router(temp_router): + feature_module = _force_reload_module(target_module, alias_module) + yield feature_module + finally: + if target_module in sys.modules: + del sys.modules[target_module] + if alias_module in sys.modules: + del sys.modules[alias_module] # ------------------------------------------------------------------------------ @@ -63,13 +119,13 @@ def mock_auth(): ), ], ) -def test_feature_endpoints_return_200_with_flat_json(app, mock_auth, url, service_mock_path, mock_model): - """Tests that feature endpoints return 200 with flat JSON format.""" +def test_console_features_success(app, mock_feature_module_env, url, service_mock_path, mock_model): + """Tests 200 response with flat JSON format and correct Content-Type.""" with patch(service_mock_path, return_value=mock_model): ext_fastopenapi.init_app(app) response = app.test_client().get(url) - assert response.status_code == 200 + assert response.status_code == 200, f"Failed: {response.text}" assert response.get_json() == mock_model.model_dump(mode="json") assert "application/json" in response.content_type @@ -81,7 +137,7 @@ def test_feature_endpoints_return_200_with_flat_json(app, mock_auth, url, servic ("/console/api/system-features", "controllers.console.feature.FeatureService.get_system_features"), ], ) -def test_feature_endpoints_return_500_on_service_error(app, mock_auth, url, service_mock_path): +def test_console_features_service_error(app, mock_feature_module_env, url, service_mock_path): """Tests that service errors return 500.""" with patch(service_mock_path, side_effect=ValueError("Service Failure")): ext_fastopenapi.init_app(app) @@ -90,110 +146,118 @@ def test_feature_endpoints_return_500_on_service_error(app, mock_auth, url, serv assert response.status_code == 500 -def test_system_features_handles_unauthenticated_users(app, mock_auth): +def test_system_features_unauthenticated(app, mock_feature_module_env): """Tests /system-features passes is_authenticated=False when auth fails.""" - mock_user = mock_auth - type(mock_user).is_authenticated = PropertyMock(side_effect=Unauthorized) - mock_model = SystemFeatureModel(enable_marketplace=True) + feature_module = mock_feature_module_env + type(feature_module.current_user).is_authenticated = PropertyMock(side_effect=Unauthorized) + mock_model = SystemFeatureModel(enable_marketplace=True) with patch("controllers.console.feature.FeatureService.get_system_features", return_value=mock_model) as svc: ext_fastopenapi.init_app(app) response = app.test_client().get("/console/api/system-features") assert response.status_code == 200 svc.assert_called_once_with(is_authenticated=False) - - -def test_features_endpoint_rejects_post_method(app, mock_auth): - """Tests that feature endpoints only accept GET.""" - with patch("controllers.console.feature.FeatureService.get_features", return_value=FeatureModel()): - ext_fastopenapi.init_app(app) - response = app.test_client().post("/console/api/features") - - assert response.status_code == 405 - - -def test_routes_are_registered_correctly(app, mock_auth): - """Tests that FastOpenAPI registers routes with correct paths.""" - ext_fastopenapi.init_app(app) - rules = {rule.rule for rule in app.url_map.iter_rules()} - - assert "/console/api/features" in rules - assert "/console/api/system-features" in rules + assert response.get_json() == mock_model.model_dump(mode="json") # ------------------------------------------------------------------------------ -# FastOpenAPI Authentication Tests +# FastOpenAPI Route Behavior Tests # ------------------------------------------------------------------------------ -@pytest.fixture -def app_with_login_manager(): - """Creates Flask app with login manager to test auth behavior.""" - from flask_login import LoginManager +class TestFastOpenAPIRouteBehavior: + """Tests for FastOpenAPI-specific routing behavior.""" - app = Flask(__name__) - app.config["TESTING"] = True - app.config["SECRET_KEY"] = "test-secret" + @pytest.fixture + def app_with_login_manager(self): + """Creates Flask app with login manager configured.""" + from flask_login import LoginManager - login_manager = LoginManager() - login_manager.init_app(app) + app = Flask(__name__) + app.config["TESTING"] = True + app.config["SECRET_KEY"] = "test-secret" + app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:" + db.init_app(app) - @login_manager.request_loader - def load_user(request): - if request.headers.get("Authorization") == "Bearer valid-token": - user = MagicMock() - user.is_authenticated = True - return user - return None + login_manager = LoginManager() + login_manager.init_app(app) - @login_manager.unauthorized_handler - def unauthorized(): - from flask import request + @login_manager.unauthorized_handler + def handle_unauthorized(): + from flask import request - # FastOpenAPI routes: raise exception (serializable) - if request.blueprint is None and request.path.startswith("/console/api/"): + if request.blueprint is None and request.path.startswith("/console/api/"): + raise Unauthorized("Unauthorized.") + import json + + from flask import Response + + return Response(json.dumps({"code": "unauthorized"}), status=401, content_type="application/json") + + return app + + def test_fastopenapi_route_has_no_blueprint(self, app_with_login_manager, fix_method_view_issue): + """Verifies FastOpenAPI routes have request.blueprint == None.""" + captured = {} + + @app_with_login_manager.route("/console/api/test") + def test_route(): + from flask import request + + captured["blueprint"] = request.blueprint + return {"ok": True} + + response = app_with_login_manager.test_client().get("/console/api/test") + assert response.status_code == 200 + assert captured["blueprint"] is None + + def test_unauthorized_raises_exception_not_response(self, app_with_login_manager, fix_method_view_issue): + """Verifies unauthorized handler raises Unauthorized (serializable by orjson).""" + + @app_with_login_manager.route("/console/api/protected") + def protected(): raise Unauthorized("Unauthorized.") - # Blueprint routes: return Response - import json - from flask import Response - - return Response(json.dumps({"code": "unauthorized"}), status=401, content_type="application/json") - - return app + response = app_with_login_manager.test_client().get("/console/api/protected") + assert response.status_code == 401 + assert b"TypeError" not in response.data # No serialization error -def test_fastopenapi_route_has_no_blueprint(app_with_login_manager): - """Verifies FastOpenAPI routes have request.blueprint == None.""" - captured = {} - - @app_with_login_manager.route("/console/api/test") - def test_route(): - from flask import request - - captured["blueprint"] = request.blueprint - return {"ok": True} - - response = app_with_login_manager.test_client().get("/console/api/test") - - assert response.status_code == 200 - assert captured["blueprint"] is None +# ------------------------------------------------------------------------------ +# OpenAPI Schema Compliance Tests +# ------------------------------------------------------------------------------ -def test_protected_route_returns_401_without_auth(app_with_login_manager): - """Tests that protected routes return 401 without authentication.""" - from flask_login import login_required +class TestOpenAPISchemaCompliance: + """Tests for route registration and HTTP method handling.""" - @app_with_login_manager.route("/console/api/protected") - @login_required - def protected(): - return {"status": "ok"} + def test_routes_registered_correctly(self, app, mock_feature_module_env): + """Verifies routes are registered with correct paths.""" + ext_fastopenapi.init_app(app) + rules = {rule.rule for rule in app.url_map.iter_rules()} - client = app_with_login_manager.test_client() + assert "/console/api/features" in rules + assert "/console/api/system-features" in rules - # Without auth - assert client.get("/console/api/protected").status_code == 401 + def test_routes_only_accept_get(self, app, mock_feature_module_env): + """Verifies feature endpoints reject non-GET methods with 405.""" + with patch("controllers.console.feature.FeatureService.get_features", return_value=FeatureModel()): + ext_fastopenapi.init_app(app) + client = app.test_client() - # With valid token - assert client.get("/console/api/protected", headers={"Authorization": "Bearer valid-token"}).status_code == 200 + assert client.get("/console/api/features").status_code == 200 + assert client.post("/console/api/features").status_code == 405 + + def test_system_features_handles_both_auth_states(self, app, mock_feature_module_env): + """Verifies /system-features handles authenticated state correctly.""" + feature_module = mock_feature_module_env + mock_model = SystemFeatureModel(enable_marketplace=True) + + with patch("controllers.console.feature.FeatureService.get_system_features", return_value=mock_model) as svc: + type(feature_module.current_user).is_authenticated = PropertyMock(return_value=True) + ext_fastopenapi.init_app(app) + response = app.test_client().get("/console/api/system-features") + + assert response.status_code == 200 + svc.assert_called_with(is_authenticated=True) From 253c16d4bf5b9d971f4df1c066b396884f0bd685 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Wed, 4 Feb 2026 12:15:59 +0000 Subject: [PATCH 11/11] [autofix.ci] apply automated fixes --- .../unit_tests/controllers/console/test_fastopenapi_feature.py | 1 - 1 file changed, 1 deletion(-) diff --git a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py index 7b7af1264f..4a1dc890b5 100644 --- a/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py +++ b/api/tests/unit_tests/controllers/console/test_fastopenapi_feature.py @@ -13,7 +13,6 @@ from extensions import ext_fastopenapi from extensions.ext_database import db from services.feature_service import FeatureModel, SystemFeatureModel - # ------------------------------------------------------------------------------ # Fixtures # ------------------------------------------------------------------------------