test: migrate web wraps controller tests to testcontainers (#34289)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

api/tests/test_containers_integration_tests/controllers/web/test_wraps.py

@@ -1,13 +1,14 @@
-"""Unit tests for controllers.web.wraps — JWT auth decorator and validation helpers."""
+"""Testcontainers integration tests for controllers.web.wraps — JWT auth decorator and validation helpers."""
 
 from __future__ import annotations
 
 from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
+from uuid import uuid4
 
 import pytest
-from flask import Flask
+from sqlalchemy.orm import Session
 from werkzeug.exceptions import BadRequest, NotFound, Unauthorized
 
 from controllers.web.error import WebAppAuthAccessDeniedError, WebAppAuthRequiredError
@@ -18,12 +19,8 @@ from controllers.web.wraps import (
 )
 
 
-# ---------------------------------------------------------------------------
-# _validate_webapp_token
-# ---------------------------------------------------------------------------
 class TestValidateWebappToken:
     def test_enterprise_enabled_and_app_auth_requires_webapp_source(self) -> None:
-        """When both flags are true, a non-webapp source must raise."""
         decoded = {"token_source": "other"}
         with pytest.raises(WebAppAuthRequiredError):
             _validate_webapp_token(decoded, app_web_auth_enabled=True, system_webapp_auth_enabled=True)
@@ -38,7 +35,6 @@ class TestValidateWebappToken:
             _validate_webapp_token(decoded, app_web_auth_enabled=True, system_webapp_auth_enabled=True)
 
     def test_public_app_rejects_webapp_source(self) -> None:
-        """When auth is not required, a webapp-sourced token must be rejected."""
         decoded = {"token_source": "webapp"}
         with pytest.raises(Unauthorized):
             _validate_webapp_token(decoded, app_web_auth_enabled=False, system_webapp_auth_enabled=False)
@@ -52,18 +48,13 @@ class TestValidateWebappToken:
         _validate_webapp_token(decoded, app_web_auth_enabled=False, system_webapp_auth_enabled=False)
 
     def test_system_enabled_but_app_public(self) -> None:
-        """system_webapp_auth_enabled=True but app is public — webapp source rejected."""
         decoded = {"token_source": "webapp"}
         with pytest.raises(Unauthorized):
             _validate_webapp_token(decoded, app_web_auth_enabled=False, system_webapp_auth_enabled=True)
 
 
-# ---------------------------------------------------------------------------
-# _validate_user_accessibility
-# ---------------------------------------------------------------------------
 class TestValidateUserAccessibility:
     def test_skips_when_auth_disabled(self) -> None:
-        """No checks when system or app auth is disabled."""
         _validate_user_accessibility(
             decoded={},
             app_code="code",
@@ -123,7 +114,6 @@ class TestValidateUserAccessibility:
     def test_external_auth_type_checks_sso_update_time(
         self, mock_perm_check: MagicMock, mock_sso_time: MagicMock
     ) -> None:
-        # granted_at is before SSO update time → denied
         mock_sso_time.return_value = datetime.now(UTC)
         old_granted = int((datetime.now(UTC) - timedelta(hours=1)).timestamp())
         decoded = {"user_id": "u1", "auth_type": "external", "granted_at": old_granted}
@@ -164,7 +154,6 @@ class TestValidateUserAccessibility:
         recent_granted = int(datetime.now(UTC).timestamp())
         decoded = {"user_id": "u1", "auth_type": "external", "granted_at": recent_granted}
         settings = SimpleNamespace(access_mode="public")
-        # Should not raise
         _validate_user_accessibility(
             decoded=decoded,
             app_code="code",
@@ -191,10 +180,49 @@ class TestValidateUserAccessibility:
             )
 
 
-# ---------------------------------------------------------------------------
-# decode_jwt_token
-# ---------------------------------------------------------------------------
 class TestDecodeJwtToken:
+    @pytest.fixture
+    def app(self, flask_app_with_containers):
+        return flask_app_with_containers
+
+    def _create_app_site_enduser(self, db_session: Session, *, enable_site: bool = True):
+        from models.model import App, AppMode, CustomizeTokenStrategy, EndUser, Site
+
+        tenant_id = str(uuid4())
+        app_model = App(
+            tenant_id=tenant_id,
+            mode=AppMode.CHAT.value,
+            name="test-app",
+            enable_site=enable_site,
+            enable_api=True,
+        )
+        db_session.add(app_model)
+        db_session.commit()
+        db_session.expire_all()
+
+        site = Site(
+            app_id=app_model.id,
+            title="test-site",
+            default_language="en-US",
+            customize_token_strategy=CustomizeTokenStrategy.NOT_ALLOW,
+            code="code1",
+        )
+        db_session.add(site)
+        db_session.commit()
+        db_session.expire_all()
+
+        end_user = EndUser(
+            tenant_id=tenant_id,
+            app_id=app_model.id,
+            type="browser",
+            session_id="sess-1",
+        )
+        db_session.add(end_user)
+        db_session.commit()
+        db_session.expire_all()
+
+        return app_model, site, end_user
+
     @patch("controllers.web.wraps._validate_user_accessibility")
     @patch("controllers.web.wraps._validate_webapp_token")
     @patch("controllers.web.wraps.EnterpriseService.WebAppAuth.get_app_access_mode_by_id")
@@ -202,10 +230,8 @@ class TestDecodeJwtToken:
     @patch("controllers.web.wraps.FeatureService.get_system_features")
     @patch("controllers.web.wraps.PassportService")
     @patch("controllers.web.wraps.extract_webapp_passport")
-    @patch("controllers.web.wraps.db")
     def test_happy_path(
         self,
-        mock_db: MagicMock,
         mock_extract: MagicMock,
         mock_passport_cls: MagicMock,
         mock_features: MagicMock,
@@ -213,40 +239,28 @@ class TestDecodeJwtToken:
         mock_access_mode: MagicMock,
         mock_validate_token: MagicMock,
         mock_validate_user: MagicMock,
-        app: Flask,
+        app,
+        db_session_with_containers: Session,
     ) -> None:
+        app_model, site, end_user = self._create_app_site_enduser(db_session_with_containers)
+
         mock_extract.return_value = "jwt-token"
         mock_passport_cls.return_value.verify.return_value = {
-            "app_code": "code1",
+            "app_code": site.code,
-            "app_id": "app-1",
+            "app_id": app_model.id,
-            "end_user_id": "eu-1",
+            "end_user_id": end_user.id,
         }
         mock_features.return_value = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False))
 
-        app_model = SimpleNamespace(id="app-1", enable_site=True)
+        with app.test_request_context("/", headers={"X-App-Code": site.code}):
-        site = SimpleNamespace(code="code1")
+            result_app, result_user = decode_jwt_token()
-        end_user = SimpleNamespace(id="eu-1", session_id="sess-1")
 
-        # Configure session mock to return correct objects via scalar()
+        assert result_app.id == app_model.id
-        session_mock = MagicMock()
+        assert result_user.id == end_user.id
-        session_mock.scalar.side_effect = [app_model, site, end_user]
-        session_ctx = MagicMock()
-        session_ctx.__enter__ = MagicMock(return_value=session_mock)
-        session_ctx.__exit__ = MagicMock(return_value=False)
-        mock_db.engine = "engine"
-
-        with patch("controllers.web.wraps.Session", return_value=session_ctx):
-            with app.test_request_context("/", headers={"X-App-Code": "code1"}):
-                result_app, result_user = decode_jwt_token()
-
-        assert result_app.id == "app-1"
-        assert result_user.id == "eu-1"
 
     @patch("controllers.web.wraps.FeatureService.get_system_features")
     @patch("controllers.web.wraps.extract_webapp_passport")
-    def test_missing_token_raises_unauthorized(
+    def test_missing_token_raises_unauthorized(self, mock_extract: MagicMock, mock_features: MagicMock, app) -> None:
-        self, mock_extract: MagicMock, mock_features: MagicMock, app: Flask
-    ) -> None:
         mock_features.return_value = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False))
         mock_extract.return_value = None
 
@@ -257,137 +271,98 @@ class TestDecodeJwtToken:
     @patch("controllers.web.wraps.FeatureService.get_system_features")
     @patch("controllers.web.wraps.PassportService")
     @patch("controllers.web.wraps.extract_webapp_passport")
-    @patch("controllers.web.wraps.db")
     def test_missing_app_raises_not_found(
         self,
-        mock_db: MagicMock,
         mock_extract: MagicMock,
         mock_passport_cls: MagicMock,
         mock_features: MagicMock,
-        app: Flask,
+        app,
     ) -> None:
+        non_existent_id = str(uuid4())
         mock_extract.return_value = "jwt-token"
         mock_passport_cls.return_value.verify.return_value = {
             "app_code": "code1",
-            "app_id": "app-1",
+            "app_id": non_existent_id,
-            "end_user_id": "eu-1",
+            "end_user_id": str(uuid4()),
         }
         mock_features.return_value = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False))
 
-        session_mock = MagicMock()
+        with app.test_request_context("/", headers={"X-App-Code": "code1"}):
-        session_mock.scalar.return_value = None  # No app found
+            with pytest.raises(NotFound):
-        session_ctx = MagicMock()
+                decode_jwt_token()
-        session_ctx.__enter__ = MagicMock(return_value=session_mock)
-        session_ctx.__exit__ = MagicMock(return_value=False)
-        mock_db.engine = "engine"
-
-        with patch("controllers.web.wraps.Session", return_value=session_ctx):
-            with app.test_request_context("/", headers={"X-App-Code": "code1"}):
-                with pytest.raises(NotFound):
-                    decode_jwt_token()
 
     @patch("controllers.web.wraps.FeatureService.get_system_features")
     @patch("controllers.web.wraps.PassportService")
     @patch("controllers.web.wraps.extract_webapp_passport")
-    @patch("controllers.web.wraps.db")
     def test_disabled_site_raises_bad_request(
         self,
-        mock_db: MagicMock,
         mock_extract: MagicMock,
         mock_passport_cls: MagicMock,
         mock_features: MagicMock,
-        app: Flask,
+        app,
+        db_session_with_containers: Session,
     ) -> None:
+        app_model, site, end_user = self._create_app_site_enduser(db_session_with_containers, enable_site=False)
+
         mock_extract.return_value = "jwt-token"
         mock_passport_cls.return_value.verify.return_value = {
-            "app_code": "code1",
+            "app_code": site.code,
-            "app_id": "app-1",
+            "app_id": app_model.id,
-            "end_user_id": "eu-1",
+            "end_user_id": end_user.id,
         }
         mock_features.return_value = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False))
 
-        app_model = SimpleNamespace(id="app-1", enable_site=False)
+        with app.test_request_context("/", headers={"X-App-Code": site.code}):
-
+            with pytest.raises(BadRequest, match="Site is disabled"):
-        session_mock = MagicMock()
+                decode_jwt_token()
-        # scalar calls: app_model, site (code found), then end_user
-        session_mock.scalar.side_effect = [app_model, SimpleNamespace(code="code1"), None]
-        session_ctx = MagicMock()
-        session_ctx.__enter__ = MagicMock(return_value=session_mock)
-        session_ctx.__exit__ = MagicMock(return_value=False)
-        mock_db.engine = "engine"
-
-        with patch("controllers.web.wraps.Session", return_value=session_ctx):
-            with app.test_request_context("/", headers={"X-App-Code": "code1"}):
-                with pytest.raises(BadRequest, match="Site is disabled"):
-                    decode_jwt_token()
 
     @patch("controllers.web.wraps.FeatureService.get_system_features")
     @patch("controllers.web.wraps.PassportService")
     @patch("controllers.web.wraps.extract_webapp_passport")
-    @patch("controllers.web.wraps.db")
     def test_missing_end_user_raises_not_found(
         self,
-        mock_db: MagicMock,
         mock_extract: MagicMock,
         mock_passport_cls: MagicMock,
         mock_features: MagicMock,
-        app: Flask,
+        app,
+        db_session_with_containers: Session,
     ) -> None:
+        app_model, site, _ = self._create_app_site_enduser(db_session_with_containers)
+        non_existent_eu = str(uuid4())
+
         mock_extract.return_value = "jwt-token"
         mock_passport_cls.return_value.verify.return_value = {
-            "app_code": "code1",
+            "app_code": site.code,
-            "app_id": "app-1",
+            "app_id": app_model.id,
-            "end_user_id": "eu-1",
+            "end_user_id": non_existent_eu,
         }
         mock_features.return_value = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False))
 
-        app_model = SimpleNamespace(id="app-1", enable_site=True)
+        with app.test_request_context("/", headers={"X-App-Code": site.code}):
-        site = SimpleNamespace(code="code1")
+            with pytest.raises(NotFound):
-
+                decode_jwt_token()
-        session_mock = MagicMock()
-        session_mock.scalar.side_effect = [app_model, site, None]  # end_user is None
-        session_ctx = MagicMock()
-        session_ctx.__enter__ = MagicMock(return_value=session_mock)
-        session_ctx.__exit__ = MagicMock(return_value=False)
-        mock_db.engine = "engine"
-
-        with patch("controllers.web.wraps.Session", return_value=session_ctx):
-            with app.test_request_context("/", headers={"X-App-Code": "code1"}):
-                with pytest.raises(NotFound):
-                    decode_jwt_token()
 
     @patch("controllers.web.wraps.FeatureService.get_system_features")
     @patch("controllers.web.wraps.PassportService")
     @patch("controllers.web.wraps.extract_webapp_passport")
-    @patch("controllers.web.wraps.db")
     def test_user_id_mismatch_raises_unauthorized(
         self,
-        mock_db: MagicMock,
         mock_extract: MagicMock,
         mock_passport_cls: MagicMock,
         mock_features: MagicMock,
-        app: Flask,
+        app,
+        db_session_with_containers: Session,
     ) -> None:
+        app_model, site, end_user = self._create_app_site_enduser(db_session_with_containers)
+
         mock_extract.return_value = "jwt-token"
         mock_passport_cls.return_value.verify.return_value = {
-            "app_code": "code1",
+            "app_code": site.code,
-            "app_id": "app-1",
+            "app_id": app_model.id,
-            "end_user_id": "eu-1",
+            "end_user_id": end_user.id,
         }
         mock_features.return_value = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False))
 
-        app_model = SimpleNamespace(id="app-1", enable_site=True)
+        with app.test_request_context("/", headers={"X-App-Code": site.code}):
-        site = SimpleNamespace(code="code1")
+            with pytest.raises(Unauthorized, match="expired"):
-        end_user = SimpleNamespace(id="eu-1", session_id="sess-1")
+                decode_jwt_token(user_id="different-user")
-
-        session_mock = MagicMock()
-        session_mock.scalar.side_effect = [app_model, site, end_user]
-        session_ctx = MagicMock()
-        session_ctx.__enter__ = MagicMock(return_value=session_mock)
-        session_ctx.__exit__ = MagicMock(return_value=False)
-        mock_db.engine = "engine"
-
-        with patch("controllers.web.wraps.Session", return_value=session_ctx):
-            with app.test_request_context("/", headers={"X-App-Code": "code1"}):
-                with pytest.raises(Unauthorized, match="expired"):
-                    decode_jwt_token(user_id="different-user")