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: move app import API coverage to unit tests (#38920)

Browse Source
This commit is contained in:
Asuka Minato 2026-07-21 16:54:32 +09:00 committed by GitHub
parent 7b188287da
commit 32d2594877
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 141 additions and 226 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

179
api/tests/test_containers_integration_tests/controllers/console/app/test_app_import_api.py
View File

@ -1,179 +0,0 @@
"""Testcontainers integration tests for controllers.console.app.app_import endpoints."""
from __future__ import annotations
from inspect import unwrap
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from flask import Flask
from controllers.console.app import app_import as app_import_module
from services.app_dsl_service import ImportStatus
class _Result:
def __init__(self, status: ImportStatus, app_id: str | None = "app-1"):
self.status = status
self.app_id = app_id
def model_dump(self, mode: str = "json"):
return {"status": self.status, "app_id": self.app_id}
def _install_features(monkeypatch: pytest.MonkeyPatch, enabled: bool) -> None:
features = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=enabled))
monkeypatch.setattr(app_import_module.FeatureService, "get_system_features", lambda: features)
class TestAppImportApi:
@pytest.fixture
def app(self, flask_app_with_containers: Flask):
return flask_app_with_containers
def test_import_post_returns_failed_status(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
api = app_import_module.AppImportApi()
method = unwrap(api.post)
_install_features(monkeypatch, enabled=False)
monkeypatch.setattr(
app_import_module.AppDslService,
"import_app",
lambda *_args, **_kwargs: _Result(ImportStatus.FAILED, app_id=None),
)
with app.test_request_context("/console/api/apps/imports", method="POST", json={"mode": "yaml-content"}):
response, status = method(api, SimpleNamespace(id="u1"))
assert status == 400
assert response["status"] == ImportStatus.FAILED
def test_import_post_returns_pending_status(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
api = app_import_module.AppImportApi()
method = unwrap(api.post)
_install_features(monkeypatch, enabled=False)
monkeypatch.setattr(
app_import_module.AppDslService,
"import_app",
lambda *_args, **_kwargs: _Result(ImportStatus.PENDING),
)
with app.test_request_context("/console/api/apps/imports", method="POST", json={"mode": "yaml-content"}):
response, status = method(api, SimpleNamespace(id="u1"))
assert status == 202
assert response["status"] == ImportStatus.PENDING
def test_import_post_updates_webapp_auth_when_enabled(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
api = app_import_module.AppImportApi()
method = unwrap(api.post)
_install_features(monkeypatch, enabled=True)
monkeypatch.setattr(
app_import_module.AppDslService,
"import_app",
lambda *_args, **_kwargs: _Result(ImportStatus.COMPLETED, app_id="app-123"),
)
update_access = MagicMock()
monkeypatch.setattr(app_import_module.EnterpriseService.WebAppAuth, "update_app_access_mode", update_access)
with app.test_request_context("/console/api/apps/imports", method="POST", json={"mode": "yaml-content"}):
response, status = method(api, SimpleNamespace(id="u1"))
update_access.assert_called_once_with("app-123", "private")
assert status == 200
assert response["status"] == ImportStatus.COMPLETED
def test_import_post_commits_session_on_success(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
api = app_import_module.AppImportApi()
method = unwrap(api.post)
_install_features(monkeypatch, enabled=False)
monkeypatch.setattr(
app_import_module.AppDslService,
"import_app",
lambda *_args, **_kwargs: _Result(ImportStatus.COMPLETED, app_id="app-123"),
)
fake_session = MagicMock()
fake_session.__enter__.return_value = fake_session
fake_session.__exit__.return_value = None
monkeypatch.setattr(app_import_module, "Session", lambda *_args, **_kwargs: fake_session)
with app.test_request_context("/console/api/apps/imports", method="POST", json={"mode": "yaml-content"}):
response, status = method(api, SimpleNamespace(id="u1"))
fake_session.commit.assert_called_once_with()
fake_session.rollback.assert_not_called()
assert status == 200
assert response["status"] == ImportStatus.COMPLETED
def test_import_post_rolls_back_session_on_failure(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
api = app_import_module.AppImportApi()
method = unwrap(api.post)
_install_features(monkeypatch, enabled=False)
monkeypatch.setattr(
app_import_module.AppDslService,
"import_app",
lambda *_args, **_kwargs: _Result(ImportStatus.FAILED, app_id=None),
)
fake_session = MagicMock()
fake_session.__enter__.return_value = fake_session
fake_session.__exit__.return_value = None
monkeypatch.setattr(app_import_module, "Session", lambda *_args, **_kwargs: fake_session)
with app.test_request_context("/console/api/apps/imports", method="POST", json={"mode": "yaml-content"}):
response, status = method(api, SimpleNamespace(id="u1"))
fake_session.rollback.assert_called_once_with()
fake_session.commit.assert_not_called()
assert status == 400
assert response["status"] == ImportStatus.FAILED
class TestAppImportConfirmApi:
@pytest.fixture
def app(self, flask_app_with_containers: Flask):
return flask_app_with_containers
def test_import_confirm_returns_failed_status(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
api = app_import_module.AppImportConfirmApi()
method = unwrap(api.post)
monkeypatch.setattr(
app_import_module.AppDslService,
"confirm_import",
lambda *_args, **_kwargs: _Result(ImportStatus.FAILED),
)
with app.test_request_context("/console/api/apps/imports/import-1/confirm", method="POST"):
response, status = method(api, SimpleNamespace(id="u1"), import_id="import-1")
assert status == 400
assert response["status"] == ImportStatus.FAILED
class TestAppImportCheckDependenciesApi:
@pytest.fixture
def app(self, flask_app_with_containers: Flask):
return flask_app_with_containers
def test_import_check_dependencies_returns_result(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
api = app_import_module.AppImportCheckDependenciesApi()
method = unwrap(api.get)
monkeypatch.setattr(
app_import_module.AppDslService,
"check_dependencies",
lambda *_args, **_kwargs: SimpleNamespace(model_dump=lambda mode="json": {"leaked_dependencies": []}),
)
with app.test_request_context("/console/api/apps/imports/app-1/check-dependencies", method="GET"):
response, status = method(api, app_model=SimpleNamespace(id="app-1"))
assert status == 200
assert response["leaked_dependencies"] == []

188
api/tests/unit_tests/controllers/console/app/test_app_import_api.py
View File

@ -2,15 +2,23 @@
from __future__ import annotations
from collections.abc import Iterator
from dataclasses import dataclass
from inspect import unwrap
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from flask import Flask
from sqlalchemy import event
from sqlalchemy.orm import Session
from controllers.console.app import app_import as app_import_module
from models.account import Account
from models.engine import db
from models.model import App
from services.app_dsl_service import ImportStatus
from services.entities.dsl_entities import CheckDependenciesResult
from services.feature_service import SystemFeatureModel, WebAppAuthModel
def _unwrap(func):
@ -38,17 +46,58 @@ class _Result:
def _install_features(monkeypatch: pytest.MonkeyPatch, enabled: bool) -> None:
features = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=enabled))
features = SystemFeatureModel(webapp_auth=WebAppAuthModel(enabled=enabled))
monkeypatch.setattr(app_import_module.FeatureService, "get_system_features", lambda: features)
def _mock_session(monkeypatch: pytest.MonkeyPatch) -> MagicMock:
fake_session = MagicMock()
fake_session.__enter__.return_value = fake_session
fake_session.__exit__.return_value = None
monkeypatch.setattr(app_import_module, "db", SimpleNamespace(engine=object()))
monkeypatch.setattr(app_import_module, "Session", lambda *_args, **_kwargs: fake_session)
return fake_session
def _make_account(account_id: str = "u1") -> Account:
account = Account(name="Test User", email="test@example.com")
account.id = account_id
return account
@pytest.fixture
def app() -> Iterator[Flask]:
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
db.init_app(app)
with app.app_context():
yield app
@dataclass
class TransactionEvents:
commits: int = 0
rollbacks: int = 0
@pytest.fixture
def transaction_events() -> TransactionEvents:
"""Observe transaction decisions while keeping the controller on a real SQLAlchemy session."""
observed = TransactionEvents()
def record_commit(_session: Session) -> None:
observed.commits += 1
def record_rollback(_session: Session) -> None:
observed.rollbacks += 1
event.listen(Session, "after_commit", record_commit)
event.listen(Session, "after_rollback", record_rollback)
try:
yield observed
finally:
event.remove(Session, "after_commit", record_commit)
event.remove(Session, "after_rollback", record_rollback)
def _failed_result_after_starting_transaction(
service: app_import_module.AppDslService, *, app_id: str | None = None
) -> _Result:
service._session.begin()
return _Result(ImportStatus.FAILED, app_id=app_id)
class TestAppImportApi:
@ -57,33 +106,39 @@ class TestAppImportApi:
return app_import_module.AppImportApi()
def test_import_post_returns_failed_status_and_rolls_back(
self, api, app: Flask, monkeypatch: pytest.MonkeyPatch
self,
api,
app: Flask,
monkeypatch: pytest.MonkeyPatch,
transaction_events: TransactionEvents,
) -> None:
method = unwrap(api.post)
_install_features(monkeypatch, enabled=False)
session = _mock_session(monkeypatch)
monkeypatch.setattr(
app_import_module.AppDslService,
"import_app",
lambda *_args, **_kwargs: _Result(ImportStatus.FAILED, app_id=None),
lambda service, *_args, **_kwargs: _failed_result_after_starting_transaction(service, app_id=None),
)
with app.test_request_context("/console/api/apps/imports", method="POST", json={"mode": "yaml-content"}):
response, status = method(api, SimpleNamespace(id="u1"))
response, status = method(api, _make_account())
session.rollback.assert_called_once_with()
session.commit.assert_not_called()
assert transaction_events.rollbacks == 1
assert transaction_events.commits == 0
assert status == 400
assert response["status"] == ImportStatus.FAILED
def test_import_post_returns_pending_status_and_commits(
self, api, app: Flask, monkeypatch: pytest.MonkeyPatch
self,
api,
app: Flask,
monkeypatch: pytest.MonkeyPatch,
transaction_events: TransactionEvents,
) -> None:
method = unwrap(api.post)
_install_features(monkeypatch, enabled=False)
session = _mock_session(monkeypatch)
monkeypatch.setattr(
app_import_module.AppDslService,
"import_app",
@ -91,20 +146,23 @@ class TestAppImportApi:
)
with app.test_request_context("/console/api/apps/imports", method="POST", json={"mode": "yaml-content"}):
response, status = method(api, SimpleNamespace(id="u1"))
response, status = method(api, _make_account())
session.commit.assert_called_once_with()
session.rollback.assert_not_called()
assert transaction_events.commits == 1
assert transaction_events.rollbacks == 0
assert status == 202
assert response["status"] == ImportStatus.PENDING
def test_import_post_updates_webapp_auth_when_enabled(
self, api, app: Flask, monkeypatch: pytest.MonkeyPatch
self,
api,
app: Flask,
monkeypatch: pytest.MonkeyPatch,
transaction_events: TransactionEvents,
) -> None:
method = unwrap(api.post)
_install_features(monkeypatch, enabled=True)
session = _mock_session(monkeypatch)
monkeypatch.setattr(
app_import_module.AppDslService,
"import_app",
@ -114,25 +172,28 @@ class TestAppImportApi:
monkeypatch.setattr(app_import_module.EnterpriseService.WebAppAuth, "update_app_access_mode", update_access)
with app.test_request_context("/console/api/apps/imports", method="POST", json={"mode": "yaml-content"}):
response, status = method(api, SimpleNamespace(id="u1"))
response, status = method(api, _make_account())
session.commit.assert_called_once_with()
session.rollback.assert_not_called()
assert transaction_events.commits == 1
assert transaction_events.rollbacks == 0
update_access.assert_called_once_with("app-123", "private")
assert status == 200
assert response["status"] == ImportStatus.COMPLETED
def test_import_post_attaches_permission_keys_when_creating_new_app_and_rbac_enabled(
self, api, app: Flask, monkeypatch: pytest.MonkeyPatch
self,
api,
app: Flask,
monkeypatch: pytest.MonkeyPatch,
transaction_events: TransactionEvents,
) -> None:
method = _unwrap(api.post)
_install_features(monkeypatch, enabled=False)
session = _mock_session(monkeypatch)
monkeypatch.setattr(
app_import_module,
"current_account_with_tenant",
lambda: (SimpleNamespace(id="u1"), "tenant-1"),
lambda: (_make_account(), "tenant-1"),
)
monkeypatch.setattr(app_import_module.dify_config, "RBAC_ENABLED", True)
monkeypatch.setattr(
@ -149,21 +210,24 @@ class TestAppImportApi:
with app.test_request_context("/console/api/apps/imports", method="POST", json={"mode": "yaml-content"}):
response, status = method()
session.commit.assert_called_once_with()
assert transaction_events.commits == 1
assert status == 200
assert response["permission_keys"] == ["app.acl.view_layout", "app.acl.edit"]
def test_import_post_does_not_attach_permission_keys_when_overwriting_existing_app(
self, api, app: Flask, monkeypatch: pytest.MonkeyPatch
self,
api,
app: Flask,
monkeypatch: pytest.MonkeyPatch,
transaction_events: TransactionEvents,
) -> None:
method = _unwrap(api.post)
_install_features(monkeypatch, enabled=False)
session = _mock_session(monkeypatch)
monkeypatch.setattr(
app_import_module,
"current_account_with_tenant",
lambda: (SimpleNamespace(id="u1"), "tenant-1"),
lambda: (_make_account(), "tenant-1"),
)
monkeypatch.setattr(app_import_module.dify_config, "RBAC_ENABLED", True)
monkeypatch.setattr(
@ -184,7 +248,7 @@ class TestAppImportApi:
):
response, status = method()
session.commit.assert_called_once_with()
assert transaction_events.commits == 1
assert status == 200
assert response["permission_keys"] == []
@ -195,35 +259,41 @@ class TestAppImportConfirmApi:
return app_import_module.AppImportConfirmApi()
def test_import_confirm_returns_failed_status_and_rolls_back(
self, api, app: Flask, monkeypatch: pytest.MonkeyPatch
self,
api,
app: Flask,
monkeypatch: pytest.MonkeyPatch,
transaction_events: TransactionEvents,
) -> None:
method = unwrap(api.post)
session = _mock_session(monkeypatch)
monkeypatch.setattr(
app_import_module.AppDslService,
"confirm_import",
lambda *_args, **_kwargs: _Result(ImportStatus.FAILED),
lambda service, *_args, **_kwargs: _failed_result_after_starting_transaction(service),
)
with app.test_request_context("/console/api/apps/imports/import-1/confirm", method="POST"):
response, status = method(api, SimpleNamespace(id="u1"), import_id="import-1")
response, status = method(api, _make_account(), import_id="import-1")
session.rollback.assert_called_once_with()
session.commit.assert_not_called()
assert transaction_events.rollbacks == 1
assert transaction_events.commits == 0
assert status == 400
assert response["status"] == ImportStatus.FAILED
def test_import_confirm_attaches_permission_keys_when_creating_new_app_and_rbac_enabled(
self, api, app: Flask, monkeypatch: pytest.MonkeyPatch
self,
api,
app: Flask,
monkeypatch: pytest.MonkeyPatch,
transaction_events: TransactionEvents,
) -> None:
method = _unwrap(api.post)
session = _mock_session(monkeypatch)
monkeypatch.setattr(
app_import_module,
"current_account_with_tenant",
lambda: (SimpleNamespace(id="u1"), "tenant-1"),
lambda: (_make_account(), "tenant-1"),
)
monkeypatch.setattr(
app_import_module.redis_client,
@ -248,20 +318,23 @@ class TestAppImportConfirmApi:
with app.test_request_context("/console/api/apps/imports/import-1/confirm", method="POST"):
response, status = method(import_id="import-1")
session.commit.assert_called_once_with()
assert transaction_events.commits == 1
assert status == 200
assert response["permission_keys"] == ["app.acl.view_layout", "app.acl.edit"]
def test_import_confirm_does_not_attach_permission_keys_when_overwriting_existing_app(
self, api, app: Flask, monkeypatch: pytest.MonkeyPatch
self,
api,
app: Flask,
monkeypatch: pytest.MonkeyPatch,
transaction_events: TransactionEvents,
) -> None:
method = _unwrap(api.post)
session = _mock_session(monkeypatch)
monkeypatch.setattr(
app_import_module,
"current_account_with_tenant",
lambda: (SimpleNamespace(id="u1"), "tenant-1"),
lambda: (_make_account(), "tenant-1"),
)
monkeypatch.setattr(
app_import_module.redis_client,
@ -286,6 +359,27 @@ class TestAppImportConfirmApi:
with app.test_request_context("/console/api/apps/imports/import-1/confirm", method="POST"):
response, status = method(import_id="import-1")
session.commit.assert_called_once_with()
assert transaction_events.commits == 1
assert status == 200
assert response["permission_keys"] == []
class TestAppImportCheckDependenciesApi:
def test_import_check_dependencies_returns_result(
self,
app: Flask,
monkeypatch: pytest.MonkeyPatch,
) -> None:
api = app_import_module.AppImportCheckDependenciesApi()
method = unwrap(api.get)
monkeypatch.setattr(
app_import_module.AppDslService,
"check_dependencies",
lambda *_args, **_kwargs: CheckDependenciesResult(leaked_dependencies=[]),
)
with app.test_request_context("/console/api/apps/imports/app-1/check-dependencies", method="GET"):
response, status = method(api, app_model=App(id="app-1"))
assert status == 200
assert response["leaked_dependencies"] == []