opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-21 18:58:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: use SQLite sessions in unit misc (#39118)

Browse Source
This commit is contained in:
Asuka Minato 2026-07-21 16:27:13 +09:00 committed by GitHub
parent d7860571aa
commit 852169d99c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 104 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
api/tests/unit_tests/controllers/console/app/test_generator_api.py
View File

@ -7,10 +7,28 @@ from unittest.mock import MagicMock
import pytest
from flask import Flask
from sqlalchemy.orm import Session
from controllers.console.app import generator as generator_module
from controllers.console.app.error import ProviderNotInitializeError
from core.errors.error import ProviderTokenNotInitError
from models.model import App, AppMode
def _persist_app(session: Session, *, tenant_id: str = "t1") -> App:
app_model = App(
id="app-1",
tenant_id=tenant_id,
name="Workflow App",
description="",
mode=AppMode.WORKFLOW,
enable_site=False,
enable_api=False,
max_active_requests=None,
)
session.add(app_model)
session.commit()
return app_model
def _model_config_payload():
@ -66,12 +84,11 @@ def test_rule_code_generate_maps_token_error(app: Flask, monkeypatch: pytest.Mon
method(api, "t1")
def test_instruction_generate_app_not_found(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
@pytest.mark.parametrize("sqlite_session", [(App,)], indirect=True)
def test_instruction_generate_app_not_found(app: Flask, sqlite_session: Session) -> None:
api = generator_module.InstructionGenerateApi()
method = unwrap(api.post)
session = MagicMock()
session.scalar.return_value = None
_persist_app(sqlite_session, tenant_id="other-tenant")
with app.test_request_context(
"/console/api/instruction-generate",
@ -83,25 +100,21 @@ def test_instruction_generate_app_not_found(app: Flask, monkeypatch: pytest.Monk
"model_config": _model_config_payload(),
},
):
response, status = method(api, session, "t1")
response, status = method(api, sqlite_session, "t1")
assert status == 400
assert response["error"] == "app app-1 not found"
stmt = session.scalar.call_args.args[0]
compiled = stmt.compile()
statement = str(compiled)
assert "apps.id" in statement
assert "apps.tenant_id" in statement
assert "app-1" in compiled.params.values()
assert "t1" in compiled.params.values()
assert sqlite_session.get(App, "app-1") is not None
def test_instruction_generate_workflow_not_found(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
@pytest.mark.parametrize("sqlite_session", [(App,)], indirect=True)
def test_instruction_generate_workflow_not_found(
app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
) -> None:
api = generator_module.InstructionGenerateApi()
method = unwrap(api.post)
app_model = SimpleNamespace(id="app-1")
session = SimpleNamespace(scalar=lambda *_args, **_kwargs: app_model)
app_model = _persist_app(sqlite_session)
_install_workflow_service(monkeypatch, workflow=None)
with app.test_request_context(
@ -114,18 +127,20 @@ def test_instruction_generate_workflow_not_found(app: Flask, monkeypatch: pytest
"model_config": _model_config_payload(),
},
):
response, status = method(api, session, "t1")
response, status = method(api, sqlite_session, "t1")
assert status == 400
assert response["error"] == "workflow app-1 not found"
def test_instruction_generate_node_missing(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
@pytest.mark.parametrize("sqlite_session", [(App,)], indirect=True)
def test_instruction_generate_node_missing(
app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
) -> None:
api = generator_module.InstructionGenerateApi()
method = unwrap(api.post)
app_model = SimpleNamespace(id="app-1")
session = SimpleNamespace(scalar=lambda *_args, **_kwargs: app_model)
app_model = _persist_app(sqlite_session)
workflow = SimpleNamespace(graph_dict={"nodes": []})
_install_workflow_service(monkeypatch, workflow=workflow)
@ -140,18 +155,18 @@ def test_instruction_generate_node_missing(app: Flask, monkeypatch: pytest.Monke
"model_config": _model_config_payload(),
},
):
response, status = method(api, session, "t1")
response, status = method(api, sqlite_session, "t1")
assert status == 400
assert response["error"] == "node node-1 not found"
def test_instruction_generate_code_node(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
@pytest.mark.parametrize("sqlite_session", [(App,)], indirect=True)
def test_instruction_generate_code_node(app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session) -> None:
api = generator_module.InstructionGenerateApi()
method = unwrap(api.post)
app_model = SimpleNamespace(id="app-1")
session = SimpleNamespace(scalar=lambda *_args, **_kwargs: app_model)
app_model = _persist_app(sqlite_session)
workflow = SimpleNamespace(
graph_dict={
@ -173,18 +188,19 @@ def test_instruction_generate_code_node(app: Flask, monkeypatch: pytest.MonkeyPa
"model_config": _model_config_payload(),
},
):
response = method(api, session, "t1")
response = method(api, sqlite_session, "t1")
assert response == {"code": "x"}
assert workflow_service.app_model is app_model
assert workflow_service.session is session
assert workflow_service.session is sqlite_session
def test_instruction_generate_legacy_modify(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
@pytest.mark.parametrize("sqlite_session", [(App,)], indirect=True)
def test_instruction_generate_legacy_modify(
app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
) -> None:
api = generator_module.InstructionGenerateApi()
method = unwrap(api.post)
session = SimpleNamespace()
monkeypatch.setattr(
generator_module.LLMGenerator,
"instruction_modify_legacy",
@ -202,16 +218,15 @@ def test_instruction_generate_legacy_modify(app: Flask, monkeypatch: pytest.Monk
"model_config": _model_config_payload(),
},
):
response = method(api, session, "t1")
response = method(api, sqlite_session, "t1")
assert response == {"instruction": "ok"}
def test_instruction_generate_incompatible_params(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
@pytest.mark.parametrize("sqlite_session", [(App,)], indirect=True)
def test_instruction_generate_incompatible_params(app: Flask, sqlite_session: Session) -> None:
api = generator_module.InstructionGenerateApi()
method = unwrap(api.post)
session = SimpleNamespace()
with app.test_request_context(
"/console/api/instruction-generate",
method="POST",
@ -223,7 +238,7 @@ def test_instruction_generate_incompatible_params(app: Flask, monkeypatch: pytes
"model_config": _model_config_payload(),
},
):
response, status = method(api, session, "t1")
response, status = method(api, sqlite_session, "t1")
assert status == 400
assert response["error"] == "incompatible parameters"

88
api/tests/unit_tests/core/rag/extractor/test_notion_extractor.py
View File

@ -1,11 +1,51 @@
"""Unit tests for Notion extraction, HTTP parsing, and persisted metadata updates."""
import json
import uuid
from collections.abc import Iterator
from types import SimpleNamespace
from unittest import mock
import httpx
import pytest
from pytest_mock import MockerFixture
from sqlalchemy import Engine
from sqlalchemy.orm import Session, sessionmaker
from core.rag.extractor import notion_extractor
from core.rag.index_processor.constant.index_type import IndexStructureType
from models.base import TypeBase
from models.dataset import Document as DocumentModel
from models.enums import DataSourceType, DocumentCreatedFrom, IndexingStatus
@pytest.fixture
def persisted_document(
monkeypatch: pytest.MonkeyPatch,
sqlite_engine: Engine,
) -> Iterator[tuple[sessionmaker[Session], DocumentModel]]:
"""Persist a Notion document and bind the extractor's ``db.session`` to SQLite."""
TypeBase.metadata.create_all(sqlite_engine, tables=[DocumentModel.__table__])
session_maker = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
document = DocumentModel(
id=str(uuid.uuid4()),
tenant_id=str(uuid.uuid4()),
dataset_id=str(uuid.uuid4()),
position=1,
data_source_type=DataSourceType.NOTION_IMPORT,
data_source_info=json.dumps({"source": "notion", "last_edited_time": "2025-01-01T00:00:00.000Z"}),
batch="batch",
name="Notion page",
created_from=DocumentCreatedFrom.API,
created_by=str(uuid.uuid4()),
indexing_status=IndexingStatus.COMPLETED,
doc_form=IndexStructureType.PARAGRAPH_INDEX,
)
with session_maker() as sqlite_session:
sqlite_session.add(document)
sqlite_session.commit()
monkeypatch.setattr(notion_extractor, "db", SimpleNamespace(session=sqlite_session))
yield session_maker, document
def _mock_response(data, status_code: int = 200, text: str = ""):
@ -394,7 +434,11 @@ class TestNotionMetadataAndCredentialMethods:
assert extractor.update_last_edited_time(None) is None
def test_update_last_edited_time_updates_document_and_commits(self, monkeypatch: pytest.MonkeyPatch):
def test_update_last_edited_time_updates_document_and_commits(
self,
monkeypatch: pytest.MonkeyPatch,
persisted_document: tuple[sessionmaker[Session], DocumentModel],
):
extractor = notion_extractor.NotionExtractor(
notion_workspace_id="ws",
notion_obj_id="obj",
@ -402,40 +446,20 @@ class TestNotionMetadataAndCredentialMethods:
tenant_id="tenant",
notion_access_token="token",
)
class FakeDocumentModel:
data_source_info = "data_source_info"
id = "id"
execute_calls = []
class FakeUpdateStmt:
def where(self, *args):
return self
def values(self, **kwargs):
return self
class FakeSession:
committed = False
def execute(self, stmt):
execute_calls.append(stmt)
def commit(self):
self.committed = True
fake_db = SimpleNamespace(session=FakeSession())
monkeypatch.setattr(notion_extractor, "DocumentModel", FakeDocumentModel)
monkeypatch.setattr(notion_extractor, "update", lambda model: FakeUpdateStmt())
monkeypatch.setattr(notion_extractor, "db", fake_db)
monkeypatch.setattr(extractor, "get_notion_last_edited_time", lambda: "2026-01-01T00:00:00.000Z")
session_maker, document = persisted_document
doc_model = SimpleNamespace(id="doc-1", data_source_info_dict={"source": "notion"})
extractor.update_last_edited_time(doc_model)
extractor.update_last_edited_time(document)
assert execute_calls
assert fake_db.session.committed is True
# Closing the writer session rolls back an uncommitted update before the independent read below.
notion_extractor.db.session.close()
with session_maker() as verification_session:
stored_document = verification_session.get(DocumentModel, document.id)
assert stored_document is not None
assert stored_document.data_source_info_dict == {
"source": "notion",
"last_edited_time": "2026-01-01T00:00:00.000Z",
}
def test_get_notion_last_edited_time_uses_page_and_database_urls(self, mocker: MockerFixture):
extractor_page = notion_extractor.NotionExtractor(