fix(ci): satisfy Python boundary guards and MCP checks

api/controllers/mcp/mcp.py

@@ -4,14 +4,14 @@ from flask import Response, request
 from flask_restx import Resource
 from pydantic import BaseModel, Field, RootModel, ValidationError
 from sqlalchemy import select
-from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.orm import Session
 
 from controllers.common.access_response import mcp_server_not_found_response
 from controllers.common.schema import register_response_schema_models, register_schema_model
 from controllers.mcp import mcp_ns
+from core.db.session_factory import session_factory
 from core.mcp import types as mcp_types
 from core.mcp.server.streamable_http import handle_mcp_request, negotiate_protocol_version
-from extensions.ext_database import db
 from graphon.variables.input_entities import VariableEntity, VariableEntityType
 from libs import helper
 from models.enums import EndUserType
@@ -76,7 +76,7 @@ class MCPAppApi(Resource):
             ValidationError: Invalid request format or parameters
         """
         # response-contract:ignore MCP route returns Flask Response from JSON-RPC handler
-        sessions = sessionmaker(db.engine, expire_on_commit=False)
+        sessions = session_factory.get_session_maker()
         try:
             # Resolve identity before parsing the body: missing identities have
             # one bounded response even for malformed or oversized JSON. Only
@@ -220,8 +220,7 @@ class MCPAppApi(Resource):
             app_model_config = app.app_model_config_with_session(session=session)
             if app_model_config is None:
                 raise MCPRequestError(mcp_types.INVALID_REQUEST, "App is unavailable")
-            features_dict = app_model_config.to_dict()
-            raw_user_input_form = features_dict.get("user_input_form", [])
+            raw_user_input_form = app_model_config.user_input_form_list
 
         # Convert to VariableEntity objects
         try:

api/controllers/web/wraps.py

@@ -5,7 +5,6 @@ from typing import Concatenate
 
 from flask import request
 from flask_restx import Resource
-from sqlalchemy import select
 from werkzeug.exceptions import NotFound, Unauthorized
 
 from constants import HEADER_NAME_APP_CODE
@@ -15,14 +14,12 @@ from controllers.web.error import (
     WebAppAuthRequiredError,
     WebAppNotFoundError,
 )
-from core.app.public_runtime import published_app_filter
-from core.db.session_factory import session_factory
 from core.logging.context import set_identity_context
 from extensions.ext_application_services import application_services
 from libs.passport import PassportService
 from libs.token import extract_webapp_passport
-from models.enums import AppStatus
-from models.model import App, EndUser, Site
+from models.model import App, EndUser
+from repositories.web_passport_repository import get_web_passport_identity
 from services.enterprise.enterprise_service import EnterpriseService, WebAppAccessMode, WebAppSettings
 from services.system_feature_service import SystemFeatureService
 from services.web_passport_gateways import resolve_web_app_auth_type
@@ -70,36 +67,18 @@ def decode_jwt_token(app_code: str | None = None, user_id: str | None = None) ->
             raise Unauthorized("App token is missing.")
         decoded = PassportService().verify(tk)
         app_code = decoded.get("app_code")
+        if not isinstance(app_code, str) or not app_code:
+            raise WebAppNotFoundError()
         app_id = decoded.get("app_id")
-        with session_factory.create_session() as session:
-            # A previously issued passport cannot expose an App after its
-            # published artifact disappears. Console/Debug never use this gate.
-            app_model = session.scalar(
-                select(App).where(App.id == app_id, App.status == AppStatus.NORMAL, published_app_filter())
-            )
-            if not app_model:
-                raise WebAppNotFoundError()
-            site = session.scalar(
-                select(Site).where(Site.code == app_code, Site.app_id == app_model.id, Site.status == AppStatus.NORMAL)
-            )
-            if not app_code or not site:
-                raise WebAppNotFoundError()
-            if app_model.enable_site is False:
-                raise WebAppNotFoundError()
-            end_user_id = decoded.get("end_user_id")
-            end_user = session.scalar(
-                select(EndUser).where(
-                    EndUser.id == end_user_id,
-                    EndUser.app_id == app_model.id,
-                    EndUser.tenant_id == app_model.tenant_id,
-                )
-            )
-            if not end_user:
-                raise NotFound()
+        app_model, end_user = get_web_passport_identity(app_id, app_code, decoded.get("end_user_id"))
+        if app_model is None:
+            raise WebAppNotFoundError()
+        if end_user is None:
+            raise NotFound()
 
-            # Validate user_id against end_user's session_id if provided
-            if user_id is not None and end_user.session_id != user_id:
-                raise Unauthorized("Authentication has expired.")
+        # Validate user_id against end_user's session_id if provided.
+        if user_id is not None and end_user.session_id != user_id:
+            raise Unauthorized("Authentication has expired.")
 
         # for enterprise webapp auth
         app_web_auth_enabled = False

api/repositories/web_passport_repository.py

@@ -6,11 +6,37 @@ from sqlalchemy import func, select
 from sqlalchemy.orm import Session, sessionmaker
 
 from core.app.public_runtime import published_app_filter
+from core.db.session_factory import session_factory as global_session_factory
 from models.enums import AppStatus, EndUserType
 from models.model import App, EndUser, Site
 from services.entities.passport_entities import EndUserRecord, WebAppRecord, WebPassportEndUserResolution
 
 
+def get_web_passport_identity(
+    app_id: str | None, app_code: str | None, end_user_id: str | None
+) -> tuple[App | None, EndUser | None]:
+    """Resolve a previously issued passport against the current published app and owner chain."""
+    with global_session_factory.create_session() as session:
+        app = session.scalar(
+            select(App).where(App.id == app_id, App.status == AppStatus.NORMAL, published_app_filter())
+        )
+        if app is None:
+            return None, None
+        site = session.scalar(
+            select(Site).where(Site.code == app_code, Site.app_id == app.id, Site.status == AppStatus.NORMAL)
+        )
+        if app_code is None or site is None or app.enable_site is False:
+            return None, None
+        end_user = session.scalar(
+            select(EndUser).where(
+                EndUser.id == end_user_id,
+                EndUser.app_id == app.id,
+                EndUser.tenant_id == app.tenant_id,
+            )
+        )
+        return app, end_user
+
+
 class WebPassportRepository:
     def __init__(
         self,

api/tests/unit_tests/controllers/common/test_access_response.py

@@ -6,13 +6,14 @@ import logging
 from collections.abc import Buffer, Generator
 from contextlib import contextmanager
 from types import SimpleNamespace
-from typing import Literal, Never, override
+from typing import Literal, Never, cast, override
 from unittest.mock import MagicMock
 
 import pytest
 from flask import Blueprint, Flask, Response, jsonify, request
 from flask.testing import FlaskClient
 from flask_restx import Namespace, Resource
+from sqlalchemy import Table
 from werkzeug.exceptions import NotFound, Unauthorized
 
 import controllers.service_api.wraps as token_wraps
@@ -29,6 +30,7 @@ from controllers.trigger import webhook
 from extensions import ext_request_logging
 from libs.external_api import ExternalApi
 from models.engine import db
+from models.model import ApiToken
 from services.app.mcp_server_service import AppMCPServerStatus
 from tests.unit_tests.config_override import apply_config_overrides
 
@@ -121,18 +123,14 @@ def test_invalid_token_final_response_and_no_usage(token_client: TokenClient) ->
 
 
 def test_invalid_token_database_path_does_not_record_usage(http_app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
-    session = MagicMock()
-    session.scalar.return_value = None
-    maker = MagicMock()
-    maker.return_value.__enter__.return_value = session
     usage = MagicMock()
-    monkeypatch.setattr(token_service, "Session", maker)
     monkeypatch.setattr(token_service.ApiTokenCache, "set", MagicMock())
     monkeypatch.setattr(token_service, "record_token_usage", usage)
-    with http_app.app_context(), pytest.raises(Unauthorized, match="Access token is invalid"):
-        token_service.query_token_from_db("missing-token", "app")
+    with http_app.app_context():
+        cast(Table, ApiToken.__table__).create(db.engine)
+        with pytest.raises(Unauthorized, match="Access token is invalid"):
+            token_service.query_token_from_db("missing-token", "app")
     usage.assert_not_called()
-    session.add.assert_not_called()
 
 
 def test_success_token_and_other_surface_are_not_rewritten(token_client: TokenClient) -> None:
@@ -158,10 +156,10 @@ def mcp_client(http_app: Flask, monkeypatch: pytest.MonkeyPatch) -> MCPClient:
     api.add_namespace(namespace)
     http_app.register_blueprint(bp)
     monkeypatch.setattr(mcp, "mcp_ns", namespace)
-    session = MagicMock()
+    session = MagicMock()  # guard-ignore: no-new-session-mock -- controls HTTP query results
     maker = MagicMock()
-    maker.return_value.begin.return_value.__enter__.return_value = session
-    monkeypatch.setattr(mcp, "sessionmaker", maker)
+    maker.begin.return_value.__enter__.return_value = session
+    monkeypatch.setattr(mcp.session_factory, "get_session_maker", lambda: maker)
     execute = MagicMock()
     end_user = MagicMock()
     monkeypatch.setattr(mcp, "handle_mcp_request", execute)
@@ -371,7 +369,7 @@ def test_mcp_reads_body_outside_sessions_and_rechecks_identity_before_execution(
         finally:
             active_sessions.pop()
 
-    monkeypatch.setattr(mcp, "sessionmaker", lambda *_args, **_kwargs: SimpleNamespace(begin=begin))
+    monkeypatch.setattr(mcp.session_factory, "get_session_maker", lambda: SimpleNamespace(begin=begin))
     form = MagicMock(return_value=[])
     process = MagicMock(return_value=Response("ok", status=200))
     monkeypatch.setattr(mcp.MCPAppApi, "_get_user_input_form", form)

api/tests/unit_tests/controllers/mcp/test_mcp.py

@@ -13,7 +13,7 @@ import pytest
 from flask import Flask, Response
 from pydantic import ValidationError
 from sqlalchemy import inspect
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, sessionmaker
 
 import controllers.mcp.mcp as module
 from models.engine import db
@@ -39,7 +39,8 @@ def app() -> Iterator[Flask]:
             Workflow.__table__,
         ):
             table.create(db.engine)
-        yield app
+        with patch.object(module.session_factory, "get_session_maker", return_value=sessionmaker(db.engine)):
+            yield app
 
 
 @pytest.fixture(autouse=True)
@@ -602,7 +603,7 @@ def test_identity_query_does_not_accept_a_server_bound_to_another_tenants_app()
     server.tenant_id = str(uuid4())
     db.session.add_all([app_model, server])
     db.session.commit()
-    with module.sessionmaker(db.engine).begin() as session:
+    with sessionmaker(db.engine).begin() as session:
         with pytest.raises(module.MCPServerNotFoundError):
             module.MCPAppApi()._get_mcp_server_and_app("server-1", session)
 

api/tests/unit_tests/controllers/web/test_public_app_admission.py

@@ -268,7 +268,9 @@ def test_real_database_failure_is_not_reclassified_as_unpublished(
     failing_factory = MagicMock(return_value=failing_session)
     h.access_repo._session_factory = failing_factory
     h.passport_repo._session_factory = failing_factory
-    monkeypatch.setattr(wraps.session_factory, "create_session", failing_factory)
+    from repositories import web_passport_repository
+
+    monkeypatch.setattr(web_passport_repository.global_session_factory, "create_session", failing_factory)
     response = h.client.get("/api" + route, headers={"X-App-Code": "fixture-code"})
     assert response.status_code == (503 if route.startswith(("/webapp", "/login")) else 500)
     assert response.get_json()["code"] != "app_not_found"

api/tests/unit_tests/repositories/test_webapp_access_query_repository.py

@@ -1,6 +1,7 @@
 from unittest.mock import MagicMock
 
 import pytest
+from sqlalchemy import Engine, event
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.orm import Session, sessionmaker
 
@@ -82,13 +83,22 @@ def test_malformed_app_id_is_not_a_database_error() -> None:
     factory.assert_not_called()
 
 
-def test_app_id_database_failure_remains_unavailable() -> None:
+def test_app_id_database_failure_remains_unavailable(
+    sqlite_session_factory: sessionmaker[Session], sqlite_engine: Engine
+) -> None:
     database_error = OperationalError("select", {}, RuntimeError("connection failed"))
-    session = MagicMock()
-    session.__enter__.return_value.scalar.side_effect = database_error
-    repository = WebAppAccessQueryRepository(session_factory=MagicMock(return_value=session))
-    with pytest.raises(WebAppAccessUnavailableError):
-        repository.is_app_available(_APP_ID)
+
+    def fail_query(*_args: object) -> None:
+        raise database_error
+
+    event.listen(sqlite_engine, "before_cursor_execute", fail_query)
+    try:
+        repository = WebAppAccessQueryRepository(session_factory=sqlite_session_factory)
+        with pytest.raises(WebAppAccessUnavailableError) as raised:
+            repository.is_app_available(_APP_ID)
+        assert raised.value.__cause__ is database_error
+    finally:
+        event.remove(sqlite_engine, "before_cursor_execute", fail_query)
 
 
 def test_find_app_id_by_code_returns_none_for_missing_code(

api/tests/unit_tests/services/test_webapp_access_query_service.py

@@ -321,6 +321,9 @@ class _BatchQueries:
     def find_app_id_by_code(self, app_code: str) -> str | None:
         raise AssertionError(f"Batch queries must not resolve app codes: {app_code}")
 
+    def is_app_available(self, app_id: str) -> bool:
+        raise AssertionError(f"Batch queries must not check app availability: {app_id}")
+
     def get_access_mode(self, app_id: str) -> WebAppAccessMode:
         raise AssertionError(f"Batch queries must not use single-app lookups: {app_id}")