From 5fc4dfaf7ba182149079c1fa91a87056687ee33e Mon Sep 17 00:00:00 2001 From: YBoy Date: Mon, 30 Mar 2026 19:19:15 +0300 Subject: [PATCH] test: migrate web wraps controller tests to testcontainers (#34289) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> --- .../controllers/web/test_wraps.py | 215 ++++++++---------- 1 file changed, 95 insertions(+), 120 deletions(-) rename api/tests/{unit_tests => test_containers_integration_tests}/controllers/web/test_wraps.py (67%) diff --git a/api/tests/unit_tests/controllers/web/test_wraps.py b/api/tests/test_containers_integration_tests/controllers/web/test_wraps.py similarity index 67% rename from api/tests/unit_tests/controllers/web/test_wraps.py rename to api/tests/test_containers_integration_tests/controllers/web/test_wraps.py index 85049ae975..19833cc772 100644 --- a/api/tests/unit_tests/controllers/web/test_wraps.py +++ b/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_id": "app-1", - "end_user_id": "eu-1", + "app_code": site.code, + "app_id": app_model.id, + "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) - site = SimpleNamespace(code="code1") - end_user = SimpleNamespace(id="eu-1", session_id="sess-1") + with app.test_request_context("/", headers={"X-App-Code": site.code}): + result_app, result_user = decode_jwt_token() - # Configure session mock to return correct objects via scalar() - 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"}): - result_app, result_user = decode_jwt_token() - - assert result_app.id == "app-1" - assert result_user.id == "eu-1" + assert result_app.id == app_model.id + assert result_user.id == end_user.id @patch("controllers.web.wraps.FeatureService.get_system_features") @patch("controllers.web.wraps.extract_webapp_passport") - def test_missing_token_raises_unauthorized( - self, mock_extract: MagicMock, mock_features: MagicMock, app: Flask - ) -> None: + def test_missing_token_raises_unauthorized(self, mock_extract: MagicMock, mock_features: MagicMock, app) -> 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", - "end_user_id": "eu-1", + "app_id": non_existent_id, + "end_user_id": str(uuid4()), } mock_features.return_value = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False)) - session_mock = MagicMock() - session_mock.scalar.return_value = None # No app found - 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() + 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_id": "app-1", - "end_user_id": "eu-1", + "app_code": site.code, + "app_id": app_model.id, + "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) - - session_mock = MagicMock() - # 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() + with app.test_request_context("/", headers={"X-App-Code": site.code}): + 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_id": "app-1", - "end_user_id": "eu-1", + "app_code": site.code, + "app_id": app_model.id, + "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) - site = SimpleNamespace(code="code1") - - 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() + with app.test_request_context("/", headers={"X-App-Code": site.code}): + 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_id": "app-1", - "end_user_id": "eu-1", + "app_code": site.code, + "app_id": app_model.id, + "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) - site = SimpleNamespace(code="code1") - end_user = SimpleNamespace(id="eu-1", session_id="sess-1") - - 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") + with app.test_request_context("/", headers={"X-App-Code": site.code}): + with pytest.raises(Unauthorized, match="expired"): + decode_jwt_token(user_id="different-user")