opensource/dify
3
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-10-16 15:35:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: move plugin lifecycle coverage to SQLite (#40713)

Browse Source 
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Asuka Minato
2026-09-19 09:25:48 +00:00
committed by GitHub
co-authored by autofix-ci[bot]
parent d7630db953
commit b520bef0b1
1 changed files with 49 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/tests/integration_tests/services/plugin/test_plugin_lifecycle.py → api/tests/unit_tests/services/plugin/test_plugin_lifecycle_sqlite.py
+49 -63
View File
@@ -1,13 +1,13 @@
"""SQLite-backed lifecycle coverage for plugin permissions and auto-upgrade strategies."""
import pytest
from sqlalchemy import delete, func, select
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from core.db.session_factory import session_factory
from extensions.ext_database import db
from models import Tenant
from models.account import (
TenantPluginAutoUpgradeCategory,
TenantPluginAutoUpgradeMode,
TenantPluginAutoUpgradeStrategy,
TenantPluginAutoUpgradeStrategySetting,
TenantPluginDebugPermission,
TenantPluginInstallPermission,
@@ -20,74 +20,60 @@ PLUGIN_CATEGORY = TenantPluginAutoUpgradeCategory.TOOL
@pytest.fixture
def tenant(flask_req_ctx):
with session_factory.create_session() as session:
t = Tenant(name="plugin_it_tenant")
session.add(t)
session.commit()
tenant_id = t.id
yield tenant_id
with session_factory.create_session() as session:
session.execute(delete(TenantPluginPermission).where(TenantPluginPermission.tenant_id == tenant_id))
session.execute(
delete(TenantPluginAutoUpgradeStrategy).where(TenantPluginAutoUpgradeStrategy.tenant_id == tenant_id)
)
session.execute(delete(Tenant).where(Tenant.id == tenant_id))
session.commit()
def tenant(sqlite_session: Session) -> str:
tenant = Tenant(name="plugin_unit_tenant")
sqlite_session.add(tenant)
sqlite_session.commit()
return tenant.id
class TestPluginPermissionLifecycle:
def test_get_returns_none_for_new_tenant(self, tenant):
assert PluginPermissionService.get_permission(tenant, session=db.session()) is None
def test_get_returns_none_for_new_tenant(self, tenant: str, sqlite_session: Session) -> None:
assert PluginPermissionService.get_permission(tenant, session=sqlite_session) is None
def test_change_creates_row(self, tenant):
def test_change_creates_row(self, tenant: str, sqlite_session: Session) -> None:
result = PluginPermissionService.change_permission(
tenant,
TenantPluginInstallPermission.ADMINS,
TenantPluginDebugPermission.EVERYONE,
session=db.session,
session=sqlite_session,
)
assert result is True
perm = PluginPermissionService.get_permission(tenant, session=db.session())
perm = PluginPermissionService.get_permission(tenant, session=sqlite_session)
assert perm is not None
assert perm.install_permission == TenantPluginInstallPermission.ADMINS
assert perm.debug_permission == TenantPluginDebugPermission.EVERYONE
def test_change_updates_existing_row(self, tenant):
def test_change_updates_existing_row(self, tenant: str, sqlite_session: Session) -> None:
PluginPermissionService.change_permission(
tenant,
TenantPluginInstallPermission.ADMINS,
TenantPluginDebugPermission.NOBODY,
session=db.session,
session=sqlite_session,
)
PluginPermissionService.change_permission(
tenant,
TenantPluginInstallPermission.EVERYONE,
TenantPluginDebugPermission.ADMINS,
session=db.session,
session=sqlite_session,
)
perm = PluginPermissionService.get_permission(tenant, session=db.session())
perm = PluginPermissionService.get_permission(tenant, session=sqlite_session)
assert perm is not None
assert perm.install_permission == TenantPluginInstallPermission.EVERYONE
assert perm.debug_permission == TenantPluginDebugPermission.ADMINS
with session_factory.create_session() as session:
count = session.scalar(
select(func.count())
.select_from(TenantPluginPermission)
.where(TenantPluginPermission.tenant_id == tenant)
)
count = sqlite_session.scalar(
select(func.count()).select_from(TenantPluginPermission).where(TenantPluginPermission.tenant_id == tenant)
)
assert count == 1
class TestPluginAutoUpgradeLifecycle:
def test_get_returns_none_for_new_tenant(self, tenant):
assert PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=db.session()) is None
def test_get_returns_none_for_new_tenant(self, tenant: str, sqlite_session: Session) -> None:
assert PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=sqlite_session) is None
def test_change_creates_row(self, tenant):
def test_change_creates_row(self, tenant: str, sqlite_session: Session) -> None:
result = PluginAutoUpgradeService.change_strategy(
tenant,
strategy_setting=TenantPluginAutoUpgradeStrategySetting.LATEST,
@@ -96,16 +82,16 @@ class TestPluginAutoUpgradeLifecycle:
exclude_plugins=[],
include_plugins=[],
category=PLUGIN_CATEGORY,
session=db.session(),
session=sqlite_session,
)
assert result is True
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=db.session())
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=sqlite_session)
assert strategy is not None
assert strategy.strategy_setting == TenantPluginAutoUpgradeStrategySetting.LATEST
assert strategy.upgrade_time_of_day == 3
def test_change_updates_existing_row(self, tenant):
def test_change_updates_existing_row(self, tenant: str, sqlite_session: Session) -> None:
PluginAutoUpgradeService.change_strategy(
tenant,
strategy_setting=TenantPluginAutoUpgradeStrategySetting.FIX_ONLY,
@@ -114,7 +100,7 @@ class TestPluginAutoUpgradeLifecycle:
exclude_plugins=[],
include_plugins=[],
category=PLUGIN_CATEGORY,
session=db.session(),
session=sqlite_session,
)
PluginAutoUpgradeService.change_strategy(
tenant,
@@ -124,25 +110,25 @@ class TestPluginAutoUpgradeLifecycle:
exclude_plugins=[],
include_plugins=["plugin-a"],
category=PLUGIN_CATEGORY,
session=db.session(),
session=sqlite_session,
)
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=db.session())
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=sqlite_session)
assert strategy is not None
assert strategy.strategy_setting == TenantPluginAutoUpgradeStrategySetting.LATEST
assert strategy.upgrade_time_of_day == 12
assert strategy.upgrade_mode == TenantPluginAutoUpgradeMode.PARTIAL
assert strategy.include_plugins == ["plugin-a"]
def test_exclude_plugin_creates_strategy_when_none_exists(self, tenant):
PluginAutoUpgradeService.exclude_plugin(tenant, "my-plugin", PLUGIN_CATEGORY, session=db.session())
def test_exclude_plugin_creates_strategy_when_none_exists(self, tenant: str, sqlite_session: Session) -> None:
PluginAutoUpgradeService.exclude_plugin(tenant, "my-plugin", PLUGIN_CATEGORY, session=sqlite_session)
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=db.session())
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=sqlite_session)
assert strategy is not None
assert strategy.upgrade_mode == TenantPluginAutoUpgradeMode.EXCLUDE
assert "my-plugin" in strategy.exclude_plugins
def test_exclude_plugin_appends_in_exclude_mode(self, tenant):
def test_exclude_plugin_appends_in_exclude_mode(self, tenant: str, sqlite_session: Session) -> None:
PluginAutoUpgradeService.change_strategy(
tenant,
strategy_setting=TenantPluginAutoUpgradeStrategySetting.FIX_ONLY,
@@ -151,16 +137,16 @@ class TestPluginAutoUpgradeLifecycle:
exclude_plugins=["existing"],
include_plugins=[],
category=PLUGIN_CATEGORY,
session=db.session(),
session=sqlite_session,
)
PluginAutoUpgradeService.exclude_plugin(tenant, "new-plugin", PLUGIN_CATEGORY, session=db.session())
PluginAutoUpgradeService.exclude_plugin(tenant, "new-plugin", PLUGIN_CATEGORY, session=sqlite_session)
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=db.session())
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=sqlite_session)
assert strategy is not None
assert "existing" in strategy.exclude_plugins
assert "new-plugin" in strategy.exclude_plugins
def test_exclude_plugin_dedup_in_exclude_mode(self, tenant):
def test_exclude_plugin_dedup_in_exclude_mode(self, tenant: str, sqlite_session: Session) -> None:
PluginAutoUpgradeService.change_strategy(
tenant,
strategy_setting=TenantPluginAutoUpgradeStrategySetting.FIX_ONLY,
@@ -169,15 +155,15 @@ class TestPluginAutoUpgradeLifecycle:
exclude_plugins=["same-plugin"],
include_plugins=[],
category=PLUGIN_CATEGORY,
session=db.session(),
session=sqlite_session,
)
PluginAutoUpgradeService.exclude_plugin(tenant, "same-plugin", PLUGIN_CATEGORY, session=db.session())
PluginAutoUpgradeService.exclude_plugin(tenant, "same-plugin", PLUGIN_CATEGORY, session=sqlite_session)
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=db.session())
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=sqlite_session)
assert strategy is not None
assert strategy.exclude_plugins.count("same-plugin") == 1
def test_exclude_from_partial_mode_removes_from_include(self, tenant):
def test_exclude_from_partial_mode_removes_from_include(self, tenant: str, sqlite_session: Session) -> None:
PluginAutoUpgradeService.change_strategy(
tenant,
strategy_setting=TenantPluginAutoUpgradeStrategySetting.FIX_ONLY,
@@ -186,16 +172,16 @@ class TestPluginAutoUpgradeLifecycle:
exclude_plugins=[],
include_plugins=["p1", "p2"],
category=PLUGIN_CATEGORY,
session=db.session(),
session=sqlite_session,
)
PluginAutoUpgradeService.exclude_plugin(tenant, "p1", PLUGIN_CATEGORY, session=db.session())
PluginAutoUpgradeService.exclude_plugin(tenant, "p1", PLUGIN_CATEGORY, session=sqlite_session)
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=db.session())
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=sqlite_session)
assert strategy is not None
assert "p1" not in strategy.include_plugins
assert "p2" in strategy.include_plugins
def test_exclude_from_all_mode_switches_to_exclude(self, tenant):
def test_exclude_from_all_mode_switches_to_exclude(self, tenant: str, sqlite_session: Session) -> None:
PluginAutoUpgradeService.change_strategy(
tenant,
strategy_setting=TenantPluginAutoUpgradeStrategySetting.LATEST,
@@ -204,11 +190,11 @@ class TestPluginAutoUpgradeLifecycle:
exclude_plugins=[],
include_plugins=[],
category=PLUGIN_CATEGORY,
session=db.session(),
session=sqlite_session,
)
PluginAutoUpgradeService.exclude_plugin(tenant, "excluded-plugin", PLUGIN_CATEGORY, session=db.session())
PluginAutoUpgradeService.exclude_plugin(tenant, "excluded-plugin", PLUGIN_CATEGORY, session=sqlite_session)
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=db.session())
strategy = PluginAutoUpgradeService.get_strategy(tenant, PLUGIN_CATEGORY, session=sqlite_session)
assert strategy is not None
assert strategy.upgrade_mode == TenantPluginAutoUpgradeMode.EXCLUDE
assert "excluded-plugin" in strategy.exclude_plugins