opensource/dify
3
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-09-24 23:22:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: use SQLite sessions in web controller fixtures (#39052)

Browse Source
This commit is contained in:
Asuka Minato
2026-08-03 15:41:56 +00:00
committed by GitHub
parent a77236e564
commit 730794426a
2 changed files with 98 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/tests/unit_tests/controllers/web/conftest.py
-69
View File
@@ -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,
)
api/tests/unit_tests/controllers/web/test_web_passport.py
+98 -51
View File
@@ -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()