diff --git a/api/tests/unit_tests/controllers/web/conftest.py b/api/tests/unit_tests/controllers/web/conftest.py index b7f3244c6cd..a7198334a07 100644 --- a/api/tests/unit_tests/controllers/web/conftest.py +++ b/api/tests/unit_tests/controllers/web/conftest.py @@ -2,9 +2,6 @@ from __future__ import annotations -from types import SimpleNamespace -from typing import Any - import pytest from flask import Flask @@ -15,69 +12,3 @@ def app() -> Flask: flask_app = Flask(__name__) flask_app.config["TESTING"] = True return flask_app - - -class FakeSession: - """Stand-in for db.session that returns pre-seeded objects by model class name.""" - - def __init__(self, mapping: dict[str, Any] | None = None): - self._mapping: dict[str, Any] = mapping or {} - - def get(self, model: type, _ident: object) -> Any: - return self._mapping.get(model.__name__) - - def scalar(self, stmt: Any) -> Any: - try: - model = stmt.column_descriptions[0]["entity"] - except (AttributeError, IndexError, KeyError, TypeError): - return None - return self._mapping.get(model.__name__) - - -class FakeDB: - """Minimal db stub exposing engine and session.""" - - def __init__(self, session: FakeSession | None = None): - self.session = session or FakeSession() - self.engine = object() - - -def make_app_model( - *, - app_id: str = "app-1", - tenant_id: str = "tenant-1", - mode: str = "chat", - enable_site: bool = True, - status: str = "normal", -) -> SimpleNamespace: - """Build a fake App model with common defaults.""" - tenant = SimpleNamespace( - id=tenant_id, - status="normal", - plan="basic", - custom_config_dict={}, - ) - return SimpleNamespace( - id=app_id, - tenant_id=tenant_id, - tenant=tenant, - mode=mode, - enable_site=enable_site, - status=status, - workflow=None, - app_model_config=None, - ) - - -def make_end_user( - *, - user_id: str = "end-user-1", - session_id: str = "session-1", - external_user_id: str = "ext-user-1", -) -> SimpleNamespace: - """Build a fake EndUser model with common defaults.""" - return SimpleNamespace( - id=user_id, - session_id=session_id, - external_user_id=external_user_id, - ) diff --git a/api/tests/unit_tests/controllers/web/test_web_passport.py b/api/tests/unit_tests/controllers/web/test_web_passport.py index 4e1a24a4da2..82ec7f1bd44 100644 --- a/api/tests/unit_tests/controllers/web/test_web_passport.py +++ b/api/tests/unit_tests/controllers/web/test_web_passport.py @@ -2,11 +2,14 @@ from __future__ import annotations +import uuid from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest from flask import Flask +from sqlalchemy import Engine, select +from sqlalchemy.orm import Session from werkzeug.exceptions import NotFound, Unauthorized from controllers.web.error import WebAppAuthRequiredError @@ -16,9 +19,62 @@ from controllers.web.passport import ( exchange_token_for_existing_web_user, generate_session_id, ) +from models.base import TypeBase +from models.enums import CustomizeTokenStrategy, EndUserType +from models.model import App, AppMode, EndUser, IconType, Site from services.webapp_auth_service import WebAppAuthType +@pytest.fixture +def database_session(sqlite_engine: Engine): + models = (App, Site, EndUser) + tables = [model.metadata.tables[model.__tablename__] for model in models] + TypeBase.metadata.create_all(sqlite_engine, tables=tables) + with Session(sqlite_engine, expire_on_commit=False) as session: + with patch("controllers.web.passport.db.session", session): + yield session + + +def _persist_webapp( + session: Session, + *, + app_code: str = "code1", + enable_site: bool = True, +) -> tuple[App, Site]: + app_model = App( + id=str(uuid.uuid4()), + tenant_id=str(uuid.uuid4()), + name="Web App", + mode=AppMode.CHAT, + icon_type=IconType.EMOJI, + icon="chat", + icon_background="#FFFFFF", + enable_site=enable_site, + enable_api=False, + ) + site = Site( + app_id=app_model.id, + title="Web App Site", + default_language="en-US", + customize_token_strategy=CustomizeTokenStrategy.UUID, + code=app_code, + ) + session.add_all([app_model, site]) + session.commit() + return app_model, site + + +def _end_user(app_model: App, *, session_id: str) -> EndUser: + return EndUser( + id=str(uuid.uuid4()), + tenant_id=app_model.tenant_id, + app_id=app_model.id, + type=EndUserType.BROWSER, + name="Web User", + session_id=session_id, + ) + + # --------------------------------------------------------------------------- # decode_enterprise_webapp_user_id # --------------------------------------------------------------------------- @@ -55,56 +111,48 @@ class TestDecodeEnterpriseWebappUserId: # generate_session_id # --------------------------------------------------------------------------- class TestGenerateSessionId: - @patch("controllers.web.passport.db") - def test_returns_unique_session_id(self, mock_db: MagicMock) -> None: - mock_db.session.scalar.return_value = 0 + def test_returns_unique_session_id(self, database_session: Session) -> None: sid = generate_session_id() assert isinstance(sid, str) assert len(sid) == 36 # UUID format - @patch("controllers.web.passport.db") - def test_retries_on_collision(self, mock_db: MagicMock) -> None: - # First call returns count=1 (collision), second returns 0 - mock_db.session.scalar.side_effect = [1, 0] - sid = generate_session_id() - assert isinstance(sid, str) - assert mock_db.session.scalar.call_count == 2 + def test_retries_on_collision(self, database_session: Session) -> None: + app_model, _ = _persist_webapp(database_session) + collision_id = str(uuid.uuid4()) + generated_id = str(uuid.uuid4()) + database_session.add(_end_user(app_model, session_id=collision_id)) + database_session.commit() + + with patch( + "controllers.web.passport.uuid.uuid4", + side_effect=[uuid.UUID(collision_id), uuid.UUID(generated_id)], + ): + sid = generate_session_id() + + assert sid == generated_id # --------------------------------------------------------------------------- # exchange_token_for_existing_web_user # --------------------------------------------------------------------------- class TestExchangeTokenForExistingWebUser: - @patch("controllers.web.passport.PassportService") - @patch("controllers.web.passport.db") - def test_external_auth_type_mismatch_raises(self, mock_db: MagicMock, mock_passport_cls: MagicMock) -> None: - site = SimpleNamespace(code="code1", app_id="app-1") - app_model = SimpleNamespace(id="app-1", status="normal", enable_site=True, tenant_id="t1") - mock_db.session.scalar.side_effect = [site, app_model] - + def test_external_auth_type_mismatch_raises(self, database_session: Session) -> None: + _persist_webapp(database_session) decoded = {"user_id": "u1", "auth_type": "internal"} # mismatch: expected "external" with pytest.raises(WebAppAuthRequiredError, match="external"): exchange_token_for_existing_web_user( app_code="code1", enterprise_user_decoded=decoded, auth_type=WebAppAuthType.EXTERNAL ) - @patch("controllers.web.passport.PassportService") - @patch("controllers.web.passport.db") - def test_internal_auth_type_mismatch_raises(self, mock_db: MagicMock, mock_passport_cls: MagicMock) -> None: - site = SimpleNamespace(code="code1", app_id="app-1") - app_model = SimpleNamespace(id="app-1", status="normal", enable_site=True, tenant_id="t1") - mock_db.session.scalar.side_effect = [site, app_model] - + def test_internal_auth_type_mismatch_raises(self, database_session: Session) -> None: + _persist_webapp(database_session) decoded = {"user_id": "u1", "auth_type": "external"} # mismatch: expected "internal" with pytest.raises(WebAppAuthRequiredError, match="internal"): exchange_token_for_existing_web_user( app_code="code1", enterprise_user_decoded=decoded, auth_type=WebAppAuthType.INTERNAL ) - @patch("controllers.web.passport.PassportService") - @patch("controllers.web.passport.db") - def test_site_not_found_raises(self, mock_db: MagicMock, mock_passport_cls: MagicMock) -> None: - mock_db.session.scalar.return_value = None + def test_site_not_found_raises(self, database_session: Session) -> None: decoded = {"user_id": "u1", "auth_type": "external"} with pytest.raises(NotFound): exchange_token_for_existing_web_user( @@ -125,69 +173,68 @@ class TestPassportResource: @patch("controllers.web.passport.PassportService") @patch("controllers.web.passport.generate_session_id", return_value="new-sess-id") - @patch("controllers.web.passport.db") @patch("controllers.web.passport.FeatureService.get_system_features") def test_creates_new_end_user_when_no_user_id( self, mock_features: MagicMock, - mock_db: MagicMock, mock_gen_session: MagicMock, mock_passport_cls: MagicMock, app: Flask, + database_session: Session, + sqlite_engine: Engine, ) -> None: mock_features.return_value = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False)) - site = SimpleNamespace(app_id="app-1", code="code1") - app_model = SimpleNamespace(id="app-1", status="normal", enable_site=True, tenant_id="t1") - mock_db.session.scalar.side_effect = [site, app_model] + app_model, _ = _persist_webapp(database_session) mock_passport_cls.return_value.issue.return_value = "issued-token" with app.test_request_context("/passport", headers={"X-App-Code": "code1"}): response = PassportResource().get() assert response["access_token"] == "issued-token" - mock_db.session.add.assert_called_once() - mock_db.session.commit.assert_called_once() + database_session.close() + with Session(sqlite_engine) as verification_session: + end_users = verification_session.scalars(select(EndUser)).all() + assert len(end_users) == 1 + assert end_users[0].session_id == "new-sess-id" + assert end_users[0].app_id == app_model.id + assert end_users[0].tenant_id == app_model.tenant_id @patch("controllers.web.passport.PassportService") - @patch("controllers.web.passport.db") @patch("controllers.web.passport.FeatureService.get_system_features") def test_reuses_existing_end_user_when_user_id_provided( self, mock_features: MagicMock, - mock_db: MagicMock, mock_passport_cls: MagicMock, app: Flask, + database_session: Session, ) -> None: mock_features.return_value = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False)) - site = SimpleNamespace(app_id="app-1", code="code1") - app_model = SimpleNamespace(id="app-1", status="normal", enable_site=True, tenant_id="t1") - existing_user = SimpleNamespace(id="eu-1", session_id="sess-existing") - mock_db.session.scalar.side_effect = [site, app_model, existing_user] + app_model, _ = _persist_webapp(database_session) + existing_user = _end_user(app_model, session_id="sess-existing") + database_session.add(existing_user) + database_session.commit() mock_passport_cls.return_value.issue.return_value = "reused-token" with app.test_request_context("/passport?user_id=sess-existing", headers={"X-App-Code": "code1"}): response = PassportResource().get() assert response["access_token"] == "reused-token" - # Should not create a new end user - mock_db.session.add.assert_not_called() + end_users = database_session.scalars(select(EndUser)).all() + assert [end_user.id for end_user in end_users] == [existing_user.id] - @patch("controllers.web.passport.db") @patch("controllers.web.passport.FeatureService.get_system_features") - def test_site_not_found_raises(self, mock_features: MagicMock, mock_db: MagicMock, app: Flask) -> None: + def test_site_not_found_raises(self, mock_features: MagicMock, app: Flask, database_session: Session) -> None: mock_features.return_value = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False)) - mock_db.session.scalar.return_value = None with app.test_request_context("/passport", headers={"X-App-Code": "code1"}): with pytest.raises(NotFound): PassportResource().get() - @patch("controllers.web.passport.db") @patch("controllers.web.passport.FeatureService.get_system_features") - def test_disabled_app_raises_not_found(self, mock_features: MagicMock, mock_db: MagicMock, app: Flask) -> None: + def test_disabled_app_raises_not_found( + self, mock_features: MagicMock, app: Flask, database_session: Session + ) -> None: mock_features.return_value = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False)) - site = SimpleNamespace(app_id="app-1", code="code1") - disabled_app = SimpleNamespace(id="app-1", status="normal", enable_site=False) - mock_db.session.scalar.side_effect = [site, disabled_app] + _persist_webapp(database_session, enable_site=False) with app.test_request_context("/passport", headers={"X-App-Code": "code1"}): with pytest.raises(NotFound): PassportResource().get()