opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-09-08 11:04:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(api): decouple console tag management (#40843)

Browse Source 
Co-authored-by: Byron Wang <byron@dify.ai>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
GGbond 2026-08-20 12:31:01 +00:00 committed by GitHub
parent a7a30b20af
commit 6f70ca8d26
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 1012 additions and 490 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
api/.importlinter
View File

@ -237,6 +237,21 @@ forbidden_modules =
sqlalchemy
werkzeug
[importlinter:contract:tag-application-service-boundary]
name = Tag application service is framework and persistence neutral
type = forbidden
source_modules =
services.tag_application_service
forbidden_modules =
configs
controllers
extensions
flask
models
repositories
sqlalchemy
werkzeug
[importlinter:contract:recommended-app-query-service-boundary]
name = Recommended app query application service is framework and persistence neutral
type = forbidden

212
api/controllers/console/tag/tags.py
View File

@ -3,37 +3,32 @@ from uuid import UUID
from flask_restx import Resource
from pydantic import BaseModel, Field, RootModel, field_validator
from sqlalchemy import select
from werkzeug.exceptions import Forbidden
from werkzeug.exceptions import Forbidden, NotFound
from configs import dify_config
from controllers.common.fields import SimpleResultResponse
from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
from controllers.common.wraps import enforce_rbac_access
from controllers.console import console_ns
from controllers.console.flask_admission import console_account_admission
from controllers.console.wraps import (
RBACPermission,
RBACResourceScope,
account_initialization_required,
edit_permission_required,
model_validate,
setup_required,
with_current_tenant_id,
with_current_user,
)
from extensions.ext_database import db
from extensions.ext_application_services import application_services
from fields.base import ResponseModel
from libs.helper import dump_response
from libs.login import current_account_with_tenant, login_required
from models import Account
from libs.login import current_account_with_tenant
from machinery.context import RequestContext
from models.enums import TagType
from models.model import Tag
from services.tag_service import (
SaveTagPayload,
TagBindingCreatePayload,
TagBindingDeletePayload,
TagService,
UpdateTagPayload,
from services.tag_application_service import (
CreateTagInput,
TagBindingInput,
TagBindingTargetNotFoundError,
TagNameConflictError,
TagNotFoundError,
UpdateTagInput,
)
@ -59,7 +54,7 @@ class TagBindingRemovePayload(BaseModel):
class TagListQueryParam(BaseModel):
type: Literal["knowledge", "app", "snippet", ""] = Field("", description="Tag type filter")
type: Literal["knowledge", "app", "snippet"] = Field(description="Tag type filter")
keyword: str | None = Field(None, description="Search keyword")
@ -101,143 +96,158 @@ register_schema_models(
register_response_schema_models(console_ns, SimpleResultResponse, TagResponse, TagListResponse)
def _enforce_snippet_tag_rbac_if_needed(tag_type: TagType | str | None) -> None:
def _enforce_snippet_tag_rbac_if_needed(tag_type: TagType | str | None, context: RequestContext) -> None:
if tag_type != TagType.SNIPPET:
return
if not dify_config.RBAC_ENABLED:
return
current_user, current_tenant_id = current_account_with_tenant()
enforce_rbac_access(
tenant_id=current_tenant_id,
account_id=current_user.id,
tenant_id=_workspace_id(context),
account_id=context.account_id,
resource_type=RBACResourceScope.WORKSPACE,
scene=RBACPermission.SNIPPETS_CREATE_AND_MODIFY,
resource_required=False,
)
def _enforce_snippet_tag_rbac_by_tag_id(tag_id: str) -> None:
def _enforce_snippet_tag_rbac_by_tag_id(tag_id: str, context: RequestContext) -> None:
if not dify_config.RBAC_ENABLED:
return
_, current_tenant_id = current_account_with_tenant()
tag_type = db.session.scalar(select(Tag.type).where(Tag.id == tag_id, Tag.tenant_id == current_tenant_id).limit(1))
_enforce_snippet_tag_rbac_if_needed(tag_type)
tag_type = application_services().tags.get_tag_type(context, tag_id)
_enforce_snippet_tag_rbac_if_needed(tag_type, context)
def _workspace_id(context: RequestContext) -> str:
if context.active_workspace_id is None:
raise RuntimeError("Console account admission did not resolve an active workspace")
return context.active_workspace_id
def _require_tag_edit_permission(*, allow_dataset_editor: bool) -> None:
current_user, _ = current_account_with_tenant()
if current_user.has_edit_permission:
return
if allow_dataset_editor and current_user.is_dataset_editor:
return
raise Forbidden()
@console_ns.route("/tags")
class TagListApi(Resource):
@setup_required
@login_required
@account_initialization_required
@console_account_admission()
@console_ns.doc(params=query_params_from_model(TagListQueryParam))
@console_ns.response(200, "Success", console_ns.models[TagListResponse.__name__])
@with_current_tenant_id
@model_validate(TagListQueryParam)
def get(self, req_data: TagListQueryParam, current_tenant_id: str):
tags = TagService.get_tags(req_data.type, current_tenant_id, req_data.keyword, session=db.session())
def get(self, req_data: TagListQueryParam, request_context: RequestContext):
tags = application_services().tags.list_tags(request_context, req_data.type, req_data.keyword)
return dump_response(TagListResponse, tags), 200
@console_ns.expect(console_ns.models[TagBasePayload.__name__])
@console_ns.response(200, "Success", console_ns.models[TagResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@with_current_user
@console_account_admission()
@model_validate(TagBasePayload)
def post(self, req_data: TagBasePayload, current_user: Account):
def post(self, req_data: TagBasePayload, request_context: RequestContext):
# Allow users with edit permission, or dataset editors (including dataset operators).
if not (current_user.has_edit_permission or current_user.is_dataset_editor):
raise Forbidden()
_require_tag_edit_permission(allow_dataset_editor=True)
_enforce_snippet_tag_rbac_if_needed(req_data.type)
tag = TagService.save_tags(SaveTagPayload(name=req_data.name, type=req_data.type), db.session())
_enforce_snippet_tag_rbac_if_needed(req_data.type, request_context)
try:
tag = application_services().tags.create_tag(
request_context,
CreateTagInput(name=req_data.name, type=req_data.type.value),
)
except TagNameConflictError as error:
raise ValueError(str(error)) from None
return dump_response(TagResponse, {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}), 200
return dump_response(TagResponse, tag), 200
@console_ns.route("/tags/<uuid:tag_id>")
class TagUpdateDeleteApi(Resource):
@console_ns.expect(console_ns.models[TagUpdateRequestPayload.__name__])
@console_ns.response(200, "Success", console_ns.models[TagResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@with_current_user
@console_account_admission()
@model_validate(TagUpdateRequestPayload)
def patch(self, req_data: TagUpdateRequestPayload, current_user: Account, tag_id: UUID):
def patch(self, req_data: TagUpdateRequestPayload, request_context: RequestContext, tag_id: UUID):
tag_id_str = str(tag_id)
# The role of the current user in the ta table must be admin, owner, or editor
if not (current_user.has_edit_permission or current_user.is_dataset_editor):
raise Forbidden()
_require_tag_edit_permission(allow_dataset_editor=True)
_enforce_snippet_tag_rbac_by_tag_id(tag_id_str)
tag = TagService.update_tags(UpdateTagPayload(name=req_data.name), tag_id_str, db.session())
_enforce_snippet_tag_rbac_by_tag_id(tag_id_str, request_context)
try:
tag = application_services().tags.update_tag(
request_context,
tag_id_str,
UpdateTagInput(name=req_data.name),
)
except TagNameConflictError as error:
raise ValueError(str(error)) from None
except TagNotFoundError as error:
raise NotFound(str(error)) from None
binding_count = TagService.get_tag_binding_count(tag_id_str, db.session())
return dump_response(TagResponse, tag), 200
return (
dump_response(
TagResponse,
{"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count},
),
200,
)
@setup_required
@login_required
@account_initialization_required
@edit_permission_required
@console_account_admission()
@console_ns.response(204, "Tag deleted successfully")
def delete(self, tag_id: UUID):
def delete(self, request_context: RequestContext, tag_id: UUID):
tag_id_str = str(tag_id)
_enforce_snippet_tag_rbac_by_tag_id(tag_id_str)
TagService.delete_tag(tag_id_str, db.session())
_require_tag_edit_permission(allow_dataset_editor=False)
_enforce_snippet_tag_rbac_by_tag_id(tag_id_str, request_context)
try:
application_services().tags.delete_tag(request_context, tag_id_str)
except TagNotFoundError as error:
raise NotFound(str(error)) from None
return "", 204
def _require_tag_binding_edit_permission(current_user: Account) -> None:
def _require_tag_binding_edit_permission() -> None:
"""
Ensure the current account can edit tag bindings.
Tag binding operations are allowed for users who can edit resources (app/dataset) within the current tenant.
"""
# The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
if not (current_user.has_edit_permission or current_user.is_dataset_editor):
raise Forbidden()
_require_tag_edit_permission(allow_dataset_editor=True)
def _create_tag_bindings(current_user: Account, payload: TagBindingPayload) -> tuple[dict[str, str], int]:
_require_tag_binding_edit_permission(current_user)
def _create_tag_bindings(context: RequestContext, payload: TagBindingPayload) -> tuple[dict[str, str], int]:
_require_tag_binding_edit_permission()
_enforce_snippet_tag_rbac_if_needed(payload.type)
TagService.save_tag_binding(
TagBindingCreatePayload(
tag_ids=payload.tag_ids,
target_id=payload.target_id,
type=payload.type,
),
db.session(),
)
_enforce_snippet_tag_rbac_if_needed(payload.type, context)
try:
application_services().tags.create_bindings(
context,
TagBindingInput(
tag_ids=tuple(payload.tag_ids),
target_id=payload.target_id,
type=payload.type.value,
),
)
except TagBindingTargetNotFoundError as error:
raise NotFound(str(error)) from None
return {"result": "success"}, 200
def _remove_tag_bindings(current_user: Account, payload: TagBindingRemovePayload) -> tuple[dict[str, str], int]:
_require_tag_binding_edit_permission(current_user)
def _remove_tag_bindings(context: RequestContext, payload: TagBindingRemovePayload) -> tuple[dict[str, str], int]:
_require_tag_binding_edit_permission()
_enforce_snippet_tag_rbac_if_needed(payload.type)
TagService.delete_tag_binding(
TagBindingDeletePayload(
tag_ids=payload.tag_ids,
target_id=payload.target_id,
type=payload.type,
),
db.session(),
)
_enforce_snippet_tag_rbac_if_needed(payload.type, context)
try:
application_services().tags.delete_bindings(
context,
TagBindingInput(
tag_ids=tuple(payload.tag_ids),
target_id=payload.target_id,
type=payload.type.value,
),
)
except TagBindingTargetNotFoundError as error:
raise NotFound(str(error)) from None
return {"result": "success"}, 200
@ -248,13 +258,10 @@ class TagBindingCollectionApi(Resource):
@console_ns.doc("create_tag_binding")
@console_ns.expect(console_ns.models[TagBindingPayload.__name__])
@console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@with_current_user
@console_account_admission()
@model_validate(TagBindingPayload)
def post(self, req_data: TagBindingPayload, current_user: Account):
return _create_tag_bindings(current_user, req_data)
def post(self, req_data: TagBindingPayload, request_context: RequestContext):
return _create_tag_bindings(request_context, req_data)
@console_ns.route("/tag-bindings/remove")
@ -265,10 +272,7 @@ class TagBindingRemoveApi(Resource):
@console_ns.doc(description="Remove one or more tag bindings from a target.")
@console_ns.expect(console_ns.models[TagBindingRemovePayload.__name__])
@console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@with_current_user
@console_account_admission()
@model_validate(TagBindingRemovePayload)
def post(self, req_data: TagBindingRemovePayload, current_user: Account):
return _remove_tag_bindings(current_user, req_data)
def post(self, req_data: TagBindingRemovePayload, request_context: RequestContext):
return _remove_tag_bindings(request_context, req_data)

6
api/extensions/ext_application_services.py
View File

@ -21,6 +21,7 @@ from repositories.app_definition_query_repository import AppDefinitionQueryRepos
from repositories.data_source_api_key_auth_repository import SQLAlchemyDataSourceApiKeyAuthBindingRepository
from repositories.explore_banner_query_repository import ExploreBannerQueryRepository
from repositories.installation_state_repository import InstallationStateRepository
from repositories.tag_repository import TagRepository
from repositories.trial_app_query_repository import TrialAppQueryRepository
from repositories.webapp_access_query_repository import WebAppAccessQueryRepository
from repositories.workspace_member_query_repository import WorkspaceMemberQueryRepository
@ -53,6 +54,7 @@ from services.recommended_app_service import RecommendedAppService
from services.schema_definition_service import SchemaDefinitionService
from services.setup_adapters import RedisSetupLock, RegisterServiceAccountProvisioner
from services.setup_service import SetupService
from services.tag_application_service import TagApplicationService
from services.web_app_runtime_query_service import WebAppRuntimeQueryService
from services.webapp_access_query_service import (
WebAppAccessQueryService,
@ -105,6 +107,7 @@ class ApplicationServices:
recommended_app_queries: RecommendedAppQueryService
workspace_queries: WorkspaceQueryService
workspace_member_queries: WorkspaceMemberQueryService
tags: TagApplicationService
def build_application_services(
@ -194,6 +197,9 @@ def build_application_services(
),
roles=DeploymentWorkspaceMemberRoleResolver(),
),
tags=TagApplicationService(
tags=TagRepository(session_factory=database_client),
),
)

4
api/openapi/markdown/console-openapi.md
View File

@ -9264,7 +9264,7 @@ Remove one or more tag bindings from a target.
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| keyword | query | Search keyword | No | string |
| type | query | Tag type filter | No | string, <br>**Available values:** "", "app", "knowledge", "snippet" |
| type | query | Tag type filter | Yes | string, <br>**Available values:** "app", "knowledge", "snippet" |
#### Responses
@ -21862,7 +21862,7 @@ Non-sensitive bootstrap snapshot exposed before Console or Web authentication.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| keyword | string | Search keyword | No |
| type | string, <br>**Available values:** "", "app", "knowledge", "snippet" | Tag type filter<br>*Enum:* `""`, `"app"`, `"knowledge"`, `"snippet"` | No |
| type | string, <br>**Available values:** "app", "knowledge", "snippet" | Tag type filter<br>*Enum:* `"app"`, `"knowledge"`, `"snippet"` | Yes |
#### TagListResponse

214
api/repositories/tag_repository.py Normal file
View File

@ -0,0 +1,214 @@
"""SQLAlchemy persistence adapter for Console tag management."""
import uuid
from typing import override
import sqlalchemy as sa
from sqlalchemy import delete, func, select
from sqlalchemy.orm import Session, sessionmaker
from libs.helper import escape_like_pattern
from models.dataset import Dataset
from models.enums import TagType
from models.model import App, Tag, TagBinding
from models.snippet import CustomizedSnippet
from services.tag_application_service import (
CreateTagInput,
InvalidTagBindingTypeError,
TagBindingInput,
TagBindingTargetNotFoundError,
TagNameConflictError,
TagNotFoundError,
TagStore,
TagSummary,
UpdateTagInput,
)
class TagRepository(TagStore):
def __init__(self, session_factory: sessionmaker[Session]) -> None:
self._session_factory = session_factory
@override
def list_tags(self, workspace_id: str, tag_type: str, keyword: str | None) -> tuple[TagSummary, ...]:
stmt = (
select(Tag.id, Tag.name, Tag.type, func.count(TagBinding.id))
.outerjoin(
TagBinding,
sa.and_(TagBinding.tag_id == Tag.id, TagBinding.tenant_id == workspace_id),
)
.where(Tag.type == tag_type, Tag.tenant_id == workspace_id)
)
if keyword:
escaped_keyword = escape_like_pattern(keyword)
stmt = stmt.where(Tag.name.ilike(f"%{escaped_keyword}%", escape="\\"))
stmt = stmt.group_by(Tag.id, Tag.name, Tag.type, Tag.created_at).order_by(Tag.created_at.desc())
with self._session_factory() as session:
return tuple(
TagSummary(
id=tag_id,
name=name,
type=tag_kind.value,
binding_count=binding_count,
)
for tag_id, name, tag_kind, binding_count in session.execute(stmt).all()
)
@override
def get_tag_type(self, workspace_id: str, tag_id: str) -> str | None:
with self._session_factory() as session:
tag_type = session.scalar(select(Tag.type).where(Tag.id == tag_id, Tag.tenant_id == workspace_id).limit(1))
return tag_type.value if tag_type is not None else None
@override
def create_tag(self, workspace_id: str, actor_id: str, tag: CreateTagInput) -> TagSummary:
with self._session_factory.begin() as session:
existing = session.scalar(
select(Tag.id).where(Tag.name == tag.name, Tag.tenant_id == workspace_id, Tag.type == tag.type).limit(1)
)
if existing is not None:
raise TagNameConflictError
model = Tag(
name=tag.name,
type=TagType(tag.type),
created_by=actor_id,
tenant_id=workspace_id,
)
model.id = str(uuid.uuid4())
session.add(model)
session.flush()
return self._summary(model, binding_count=0)
@override
def update_tag(self, workspace_id: str, tag_id: str, tag: UpdateTagInput) -> TagSummary:
with self._session_factory.begin() as session:
model = session.scalar(select(Tag).where(Tag.id == tag_id, Tag.tenant_id == workspace_id).limit(1))
if model is None:
raise TagNotFoundError
if tag.name != model.name:
existing = session.scalar(
select(Tag.id)
.where(
Tag.name == tag.name,
Tag.tenant_id == workspace_id,
Tag.type == model.type,
Tag.id != tag_id,
)
.limit(1)
)
if existing is not None:
raise TagNameConflictError
model.name = tag.name
binding_count = (
session.scalar(
select(func.count(TagBinding.id)).where(
TagBinding.tag_id == tag_id,
TagBinding.tenant_id == workspace_id,
)
)
or 0
)
return self._summary(model, binding_count=binding_count)
@override
def delete_tag(self, workspace_id: str, tag_id: str) -> None:
with self._session_factory.begin() as session:
model = session.scalar(select(Tag).where(Tag.id == tag_id, Tag.tenant_id == workspace_id).limit(1))
if model is None:
raise TagNotFoundError
session.execute(
delete(TagBinding).where(
TagBinding.tag_id == tag_id,
TagBinding.tenant_id == workspace_id,
)
)
session.delete(model)
@override
def create_bindings(self, workspace_id: str, actor_id: str, binding: TagBindingInput) -> None:
with self._session_factory.begin() as session:
self._ensure_target_exists(session, workspace_id, binding)
requested_tag_ids = tuple(dict.fromkeys(binding.tag_ids))
if not requested_tag_ids:
return
valid_tag_ids = tuple(
session.scalars(
select(Tag.id).where(
Tag.id.in_(requested_tag_ids),
Tag.tenant_id == workspace_id,
Tag.type == binding.type,
)
).all()
)
if not valid_tag_ids:
return
existing_tag_ids = set(
session.scalars(
select(TagBinding.tag_id).where(
TagBinding.tag_id.in_(valid_tag_ids),
TagBinding.target_id == binding.target_id,
TagBinding.tenant_id == workspace_id,
)
).all()
)
session.add_all(
TagBinding(
tag_id=tag_id,
target_id=binding.target_id,
tenant_id=workspace_id,
created_by=actor_id,
)
for tag_id in valid_tag_ids
if tag_id not in existing_tag_ids
)
@override
def delete_bindings(self, workspace_id: str, binding: TagBindingInput) -> None:
with self._session_factory.begin() as session:
self._ensure_target_exists(session, workspace_id, binding)
session.execute(
delete(TagBinding).where(
TagBinding.target_id == binding.target_id,
TagBinding.tag_id.in_(binding.tag_ids),
TagBinding.tenant_id == workspace_id,
TagBinding.tag_id.in_(
select(Tag.id).where(
Tag.tenant_id == workspace_id,
Tag.type == binding.type,
)
),
)
)
@staticmethod
def _summary(tag: Tag, *, binding_count: int) -> TagSummary:
return TagSummary(
id=tag.id,
name=tag.name,
type=tag.type.value,
binding_count=binding_count,
)
@staticmethod
def _ensure_target_exists(session: Session, workspace_id: str, binding: TagBindingInput) -> None:
if binding.type == "knowledge":
stmt = select(Dataset.id).where(Dataset.tenant_id == workspace_id, Dataset.id == binding.target_id)
elif binding.type == "app":
stmt = select(App.id).where(App.tenant_id == workspace_id, App.id == binding.target_id)
elif binding.type == "snippet":
stmt = select(CustomizedSnippet.id).where(
CustomizedSnippet.tenant_id == workspace_id,
CustomizedSnippet.id == binding.target_id,
)
else:
raise InvalidTagBindingTypeError
if session.scalar(stmt.limit(1)) is None:
raise TagBindingTargetNotFoundError(binding.type)

103
api/services/tag_application_service.py Normal file
View File

@ -0,0 +1,103 @@
"""Application boundary for Console tag management."""
from collections.abc import Sequence
from typing import Literal, NamedTuple, Protocol
from machinery.context import RequestContext
type TagKind = Literal["knowledge", "app", "snippet"]
class TagSummary(NamedTuple):
id: str
name: str
type: str
binding_count: int
class CreateTagInput(NamedTuple):
name: str
type: TagKind
class UpdateTagInput(NamedTuple):
name: str
class TagBindingInput(NamedTuple):
tag_ids: tuple[str, ...]
target_id: str
type: TagKind
class TagStore(Protocol):
def list_tags(self, workspace_id: str, tag_type: str, keyword: str | None) -> Sequence[TagSummary]: ...
def get_tag_type(self, workspace_id: str, tag_id: str) -> str | None: ...
def create_tag(self, workspace_id: str, actor_id: str, tag: CreateTagInput) -> TagSummary: ...
def update_tag(self, workspace_id: str, tag_id: str, tag: UpdateTagInput) -> TagSummary: ...
def delete_tag(self, workspace_id: str, tag_id: str) -> None: ...
def create_bindings(self, workspace_id: str, actor_id: str, binding: TagBindingInput) -> None: ...
def delete_bindings(self, workspace_id: str, binding: TagBindingInput) -> None: ...
class TagApplicationError(Exception):
"""Base class for framework-neutral tag failures."""
class TagNotFoundError(TagApplicationError):
def __init__(self) -> None:
super().__init__("Tag not found")
class TagNameConflictError(TagApplicationError):
def __init__(self) -> None:
super().__init__("Tag name already exists")
class TagBindingTargetNotFoundError(TagApplicationError):
def __init__(self, target_type: TagKind) -> None:
target_name = {"knowledge": "Dataset", "app": "App", "snippet": "Snippet"}[target_type]
super().__init__(f"{target_name} not found")
class InvalidTagBindingTypeError(TagApplicationError):
def __init__(self) -> None:
super().__init__("Invalid binding type")
class TagApplicationService:
def __init__(self, *, tags: TagStore) -> None:
self._tags = tags
def list_tags(self, context: RequestContext, tag_type: str, keyword: str | None = None) -> tuple[TagSummary, ...]:
return tuple(self._tags.list_tags(self._workspace_id(context), tag_type, keyword))
def get_tag_type(self, context: RequestContext, tag_id: str) -> str | None:
return self._tags.get_tag_type(self._workspace_id(context), tag_id)
def create_tag(self, context: RequestContext, tag: CreateTagInput) -> TagSummary:
return self._tags.create_tag(self._workspace_id(context), context.account_id, tag)
def update_tag(self, context: RequestContext, tag_id: str, tag: UpdateTagInput) -> TagSummary:
return self._tags.update_tag(self._workspace_id(context), tag_id, tag)
def delete_tag(self, context: RequestContext, tag_id: str) -> None:
self._tags.delete_tag(self._workspace_id(context), tag_id)
def create_bindings(self, context: RequestContext, binding: TagBindingInput) -> None:
self._tags.create_bindings(self._workspace_id(context), context.account_id, binding)
def delete_bindings(self, context: RequestContext, binding: TagBindingInput) -> None:
self._tags.delete_bindings(self._workspace_id(context), binding)
@staticmethod
def _workspace_id(context: RequestContext) -> str:
if context.active_workspace_id is None:
raise RuntimeError("Console account admission did not resolve an active workspace")
return context.active_workspace_id

748
api/tests/unit_tests/controllers/console/tag/test_tags.py
View File

@ -1,12 +1,9 @@
from collections.abc import Iterator
from types import SimpleNamespace
from unittest.mock import PropertyMock, patch
from unittest.mock import MagicMock, patch
import pytest
from flask import Flask
from sqlalchemy import Engine
from sqlalchemy.orm import Session, scoped_session, sessionmaker
from werkzeug.exceptions import Forbidden
from werkzeug.exceptions import Forbidden, NotFound, UnprocessableEntity
import controllers.console.tag.tags as module
from controllers.console import console_ns
@ -21,194 +18,146 @@ from controllers.console.tag.tags import (
TagUpdateDeleteApi,
TagUpdateRequestPayload,
)
from machinery.context import RequestContext
from models import Account
from models.account import AccountStatus, TenantAccountRole
from models.base import TypeBase
from models.enums import TagType
from models.model import Tag
from services.tag_service import UpdateTagPayload
from services.tag_application_service import (
TagApplicationError,
TagBindingInput,
TagBindingTargetNotFoundError,
TagNameConflictError,
TagNotFoundError,
TagSummary,
UpdateTagInput,
)
def unwrap(func):
"""
Recursively unwrap decorated functions.
"""
while hasattr(func, "__wrapped__"):
func = func.__wrapped__
return func
@pytest.fixture
def app():
def app() -> Flask:
app = Flask("test_tag")
app.config["TESTING"] = True
return app
@pytest.fixture(autouse=True)
def sqlite_db_session(
sqlite_engine: Engine,
monkeypatch: pytest.MonkeyPatch,
) -> Iterator[scoped_session[Session]]:
TypeBase.metadata.create_all(sqlite_engine, tables=[TypeBase.metadata.tables[Tag.__tablename__]])
session_registry = scoped_session(sessionmaker(bind=sqlite_engine, expire_on_commit=False))
monkeypatch.setattr(module.db, "session", session_registry)
try:
yield session_registry
finally:
session_registry.remove()
def _assert_sqlite_session(session: object, sqlite_engine: Engine) -> None:
assert isinstance(session, Session)
assert session.get_bind() is sqlite_engine
assert session.is_active
@pytest.fixture
def admin_user():
def request_context() -> RequestContext:
return RequestContext(
request_id="request-1",
trace_id=None,
account_id="user-1",
active_workspace_id="tenant-1",
)
def _account(role: TenantAccountRole) -> Account:
account = Account(
name="Admin User",
email="admin@example.com",
name="Tag User",
email=f"{role.value}@example.com",
status=AccountStatus.ACTIVE,
)
account.id = "user-1"
account.role = TenantAccountRole.OWNER
account.role = role
return account
@pytest.fixture
def readonly_user():
account = Account(
name="Readonly User",
email="readonly@example.com",
status=AccountStatus.ACTIVE,
)
account.id = "user-2"
account.role = TenantAccountRole.NORMAL
return account
@pytest.fixture
def tag(sqlite_db_session: scoped_session[Session]):
tag = Tag(
tenant_id="tenant-1",
name="test-tag",
type=TagType.KNOWLEDGE,
created_by="user-1",
)
tag.id = "tag-1"
sqlite_db_session.add(tag)
sqlite_db_session.commit()
return tag
@pytest.fixture
def payload_patch():
def _patch(payload):
return patch.object(
type(console_ns),
"payload",
new_callable=PropertyMock,
return_value=payload,
)
return _patch
def tags_service() -> MagicMock:
tags = MagicMock()
with patch.object(module, "application_services", return_value=SimpleNamespace(tags=tags)):
yield tags
class TestTagListApi:
def test_get_success(self, app: Flask):
api = TagListApi()
method = unwrap(api.get)
@pytest.mark.parametrize("url", ["/", "/?type="])
def test_get_requires_non_empty_type(self, app: Flask, url: str) -> None:
class Handler:
@module.model_validate(TagListQueryParam)
def get(self, req_data: TagListQueryParam) -> TagListQueryParam:
return req_data
with app.test_request_context(url, method="GET"):
with pytest.raises(UnprocessableEntity):
Handler().get()
def test_get_uses_application_service(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
tags_service.list_tags.return_value = (TagSummary("tag-1", "Tag", "knowledge", 2),)
with app.test_request_context("/?type=knowledge"):
with (
patch(
"controllers.console.tag.tags.TagService.get_tags",
return_value=[
SimpleNamespace(
id="1",
name="tag",
type=TagType.KNOWLEDGE,
binding_count=1,
)
],
),
):
result, status = method(api, TagListQueryParam(type="knowledge"), "tenant-1")
result, status = unwrap(TagListApi().get)(
TagListApi(),
TagListQueryParam(type="knowledge"),
request_context,
)
tags_service.list_tags.assert_called_once_with(request_context, "knowledge", None)
assert status == 200
assert result == [{"id": "1", "name": "tag", "type": "knowledge", "binding_count": "1"}]
assert result == [{"id": "tag-1", "name": "Tag", "type": "knowledge", "binding_count": "2"}]
def test_get_snippet_tags(self, app: Flask, sqlite_engine: Engine):
api = TagListApi()
method = unwrap(api.get)
def test_get_snippet_tags_uses_same_query_boundary(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
tags_service.list_tags.return_value = (TagSummary("tag-1", "Snippet", "snippet", 1),)
with app.test_request_context("/?type=snippet"):
with (
patch(
"controllers.console.tag.tags.TagService.get_tags",
return_value=[
SimpleNamespace(
id="1",
name="snippet-tag",
type=TagType.SNIPPET,
binding_count=1,
)
],
) as get_tags_mock,
):
result, status = method(api, TagListQueryParam(type="snippet"), "tenant-1")
result, status = unwrap(TagListApi().get)(
TagListApi(),
TagListQueryParam(type="snippet"),
request_context,
)
get_tags_mock.assert_called_once()
assert get_tags_mock.call_args.args == ("snippet", "tenant-1", None)
_assert_sqlite_session(get_tags_mock.call_args.kwargs["session"], sqlite_engine)
tags_service.list_tags.assert_called_once_with(request_context, "snippet", None)
assert status == 200
assert result == [{"id": "1", "name": "snippet-tag", "type": "snippet", "binding_count": "1"}]
assert result[0]["type"] == "snippet"
def test_post_success(self, app: Flask, admin_user, tag):
api = TagListApi()
method = unwrap(api.post)
def test_post_preserves_dataset_editor_permission(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
tags_service.create_tag.return_value = TagSummary("tag-1", "Tag", "knowledge", 0)
dataset_operator = _account(TenantAccountRole.DATASET_OPERATOR)
payload = {"name": "test-tag", "type": "knowledge"}
req_data = TagBasePayload.model_validate(payload)
with app.test_request_context("/", json=payload):
with (
patch(
"controllers.console.tag.tags.TagService.save_tags",
return_value=tag,
),
):
result, status = method(api, req_data, admin_user)
with (
app.test_request_context("/", json={"name": "Tag", "type": "knowledge"}),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(dataset_operator, "tenant-1")),
):
result, status = unwrap(TagListApi().post)(
TagListApi(),
TagBasePayload(name="Tag", type=TagType.KNOWLEDGE),
request_context,
)
assert status == 200
assert result["name"] == "test-tag"
assert result["binding_count"] == "0"
tags_service.create_tag.assert_called_once()
def test_post_snippet_tag_checks_snippet_rbac_when_enabled(self, app: Flask, admin_user, tag):
api = TagListApi()
method = unwrap(api.post)
def test_post_snippet_tag_checks_rbac(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
tags_service.create_tag.return_value = TagSummary("tag-1", "Snippet", "snippet", 0)
owner = _account(TenantAccountRole.OWNER)
payload = {"name": "snippet-tag", "type": "snippet"}
req_data = TagBasePayload.model_validate(payload)
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", True),
patch.object(module, "current_account_with_tenant", return_value=(owner, "tenant-1")),
patch.object(module, "enforce_rbac_access") as enforce_rbac_access,
):
unwrap(TagListApi().post)(
TagListApi(),
TagBasePayload(name="Snippet", type=TagType.SNIPPET),
request_context,
)
with app.test_request_context("/", json=payload):
with (
patch("controllers.console.tag.tags.dify_config.RBAC_ENABLED", True),
patch(
"controllers.console.tag.tags.current_account_with_tenant",
return_value=(admin_user, "tenant-1"),
),
patch("controllers.console.tag.tags.enforce_rbac_access") as enforce_mock,
patch(
"controllers.console.tag.tags.TagService.save_tags",
return_value=tag,
),
):
method(api, req_data, admin_user)
enforce_mock.assert_called_once_with(
enforce_rbac_access.assert_called_once_with(
tenant_id="tenant-1",
account_id="user-1",
resource_type=module.RBACResourceScope.WORKSPACE,
@ -216,256 +165,305 @@ class TestTagListApi:
resource_required=False,
)
def test_post_forbidden(self, app: Flask, readonly_user):
api = TagListApi()
method = unwrap(api.post)
def test_post_rejects_read_only_member(self, app: Flask, request_context: RequestContext) -> None:
readonly = _account(TenantAccountRole.NORMAL)
with app.test_request_context("/"):
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(readonly, "tenant-1")),
):
with pytest.raises(Forbidden):
method(api, TagBasePayload(name="test", type=TagType.KNOWLEDGE), readonly_user)
unwrap(TagListApi().post)(
TagListApi(),
TagBasePayload(name="Tag", type=TagType.KNOWLEDGE),
request_context,
)
def test_post_maps_name_conflict_to_legacy_value_error(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
tags_service.create_tag.side_effect = TagNameConflictError()
owner = _account(TenantAccountRole.OWNER)
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(owner, "tenant-1")),
):
with pytest.raises(ValueError, match="Tag name already exists") as exc_info:
unwrap(TagListApi().post)(
TagListApi(),
TagBasePayload(name="Tag", type=TagType.KNOWLEDGE),
request_context,
)
assert exc_info.value.__cause__ is None
assert exc_info.value.__suppress_context__ is True
def test_post_does_not_coerce_unknown_application_error_to_transport_error(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
tags_service.create_tag.side_effect = TagApplicationError("unexpected")
owner = _account(TenantAccountRole.OWNER)
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(owner, "tenant-1")),
):
with pytest.raises(TagApplicationError, match="unexpected"):
unwrap(TagListApi().post)(
TagListApi(),
TagBasePayload(name="Tag", type=TagType.KNOWLEDGE),
request_context,
)
class TestTagUpdateDeleteApi:
def test_patch_success(self, app: Flask, admin_user, tag, sqlite_engine: Engine):
api = TagUpdateDeleteApi()
method = unwrap(api.patch)
payload = {"name": "updated"}
req_data = TagUpdateRequestPayload.model_validate(payload)
with app.test_request_context("/", json=payload):
with (
patch(
"controllers.console.tag.tags.TagService.update_tags",
return_value=tag,
) as update_tags_mock,
patch(
"controllers.console.tag.tags.TagService.get_tag_binding_count",
return_value=3,
),
):
result, status = method(api, req_data, admin_user, "tag-1")
assert status == 200
update_payload, tag_id, session = update_tags_mock.call_args.args
assert update_payload == UpdateTagPayload(name="updated")
assert tag_id == "tag-1"
_assert_sqlite_session(session, sqlite_engine)
assert result["binding_count"] == "3"
def test_patch_forbidden(self, app: Flask, readonly_user):
api = TagUpdateDeleteApi()
method = unwrap(api.patch)
with app.test_request_context("/"):
with pytest.raises(Forbidden):
method(api, TagUpdateRequestPayload(name="test"), readonly_user, "tag-1")
def test_delete_success(self, app: Flask, admin_user, sqlite_engine: Engine):
api = TagUpdateDeleteApi()
method = unwrap(api.delete)
def test_patch_authorizes_snippet_before_update(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
tags_service.get_tag_type.return_value = "snippet"
tags_service.update_tag.return_value = TagSummary("tag-1", "Updated", "snippet", 3)
owner = _account(TenantAccountRole.OWNER)
with (
app.test_request_context("/"),
patch("controllers.console.tag.tags.TagService.delete_tag") as delete_mock,
patch.object(module.dify_config, "RBAC_ENABLED", True),
patch.object(module, "current_account_with_tenant", return_value=(owner, "tenant-1")),
patch.object(module, "enforce_rbac_access") as enforce_rbac_access,
):
result, status = method(api, "tag-1")
result, status = unwrap(TagUpdateDeleteApi().patch)(
TagUpdateDeleteApi(),
TagUpdateRequestPayload(name="Updated"),
request_context,
"tag-1",
)
delete_mock.assert_called_once()
tag_id, session = delete_mock.call_args.args
assert tag_id == "tag-1"
_assert_sqlite_session(session, sqlite_engine)
assert status == 204
def test_delete_snippet_tag_checks_type_in_current_tenant(
self,
app: Flask,
admin_user,
sqlite_db_session: scoped_session[Session],
sqlite_engine: Engine,
):
api = TagUpdateDeleteApi()
method = unwrap(api.delete)
tag = Tag(
tenant_id="tenant-1",
name="snippet-tag",
type=TagType.SNIPPET,
created_by="user-1",
)
tag.id = "tag-1"
sqlite_db_session.add(tag)
sqlite_db_session.commit()
with (
app.test_request_context("/"),
patch("controllers.console.tag.tags.dify_config.RBAC_ENABLED", True),
patch(
"controllers.console.tag.tags.current_account_with_tenant",
return_value=(admin_user, "tenant-1"),
),
patch("controllers.console.tag.tags.enforce_rbac_access") as enforce_mock,
patch("controllers.console.tag.tags.TagService.delete_tag") as delete_mock,
):
result, status = method(api, "tag-1")
enforce_mock.assert_called_once_with(
enforce_rbac_access.assert_called_once_with(
tenant_id="tenant-1",
account_id="user-1",
resource_type=module.RBACResourceScope.WORKSPACE,
scene=module.RBACPermission.SNIPPETS_CREATE_AND_MODIFY,
resource_required=False,
)
delete_mock.assert_called_once()
tag_id, session = delete_mock.call_args.args
assert tag_id == "tag-1"
_assert_sqlite_session(session, sqlite_engine)
assert result == ""
assert status == 204
tags_service.update_tag.assert_called_once_with(request_context, "tag-1", UpdateTagInput(name="Updated"))
assert status == 200
assert result["binding_count"] == "3"
def test_delete_does_not_apply_snippet_rbac_to_tag_from_another_tenant(
self,
app: Flask,
admin_user,
sqlite_db_session: scoped_session[Session],
sqlite_engine: Engine,
):
api = TagUpdateDeleteApi()
method = unwrap(api.delete)
tag = Tag(
tenant_id="other-tenant",
name="other-tenant-snippet-tag",
type=TagType.SNIPPET,
created_by="other-user",
)
tag.id = "tag-1"
sqlite_db_session.add(tag)
sqlite_db_session.commit()
def test_patch_rejects_read_only_member(self, app: Flask, request_context: RequestContext) -> None:
readonly = _account(TenantAccountRole.NORMAL)
with (
app.test_request_context("/"),
patch("controllers.console.tag.tags.dify_config.RBAC_ENABLED", True),
patch(
"controllers.console.tag.tags.current_account_with_tenant",
return_value=(admin_user, "tenant-1"),
),
patch("controllers.console.tag.tags.enforce_rbac_access") as enforce_mock,
patch("controllers.console.tag.tags.TagService.delete_tag") as delete_mock,
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(readonly, "tenant-1")),
):
result, status = method(api, "tag-1")
with pytest.raises(Forbidden):
unwrap(TagUpdateDeleteApi().patch)(
TagUpdateDeleteApi(),
TagUpdateRequestPayload(name="Updated"),
request_context,
"tag-1",
)
enforce_mock.assert_not_called()
delete_mock.assert_called_once()
tag_id, session = delete_mock.call_args.args
assert tag_id == "tag-1"
_assert_sqlite_session(session, sqlite_engine)
assert result == ""
assert status == 204
def test_patch_maps_missing_tag_to_not_found(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
tags_service.update_tag.side_effect = TagNotFoundError()
owner = _account(TenantAccountRole.OWNER)
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(owner, "tenant-1")),
):
with pytest.raises(NotFound, match="Tag not found") as exc_info:
unwrap(TagUpdateDeleteApi().patch)(
TagUpdateDeleteApi(),
TagUpdateRequestPayload(name="Updated"),
request_context,
"tag-1",
)
assert exc_info.value.__cause__ is None
assert exc_info.value.__suppress_context__ is True
def test_delete_does_not_grant_dataset_operator_legacy_edit_permission(
self, app: Flask, request_context: RequestContext
) -> None:
dataset_operator = _account(TenantAccountRole.DATASET_OPERATOR)
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(dataset_operator, "tenant-1")),
):
with pytest.raises(Forbidden):
unwrap(TagUpdateDeleteApi().delete)(TagUpdateDeleteApi(), request_context, "tag-1")
def test_delete_calls_application_service(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
owner = _account(TenantAccountRole.OWNER)
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(owner, "tenant-1")),
):
result, status = unwrap(TagUpdateDeleteApi().delete)(TagUpdateDeleteApi(), request_context, "tag-1")
tags_service.delete_tag.assert_called_once_with(request_context, "tag-1")
assert (result, status) == ("", 204)
def test_delete_snippet_tag_checks_type_in_current_workspace(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
tags_service.get_tag_type.return_value = "snippet"
owner = _account(TenantAccountRole.OWNER)
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", True),
patch.object(module, "current_account_with_tenant", return_value=(owner, "tenant-1")),
patch.object(module, "enforce_rbac_access") as enforce_rbac_access,
):
unwrap(TagUpdateDeleteApi().delete)(TagUpdateDeleteApi(), request_context, "tag-1")
tags_service.get_tag_type.assert_called_once_with(request_context, "tag-1")
enforce_rbac_access.assert_called_once()
def test_delete_does_not_authorize_tag_outside_current_workspace(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
tags_service.get_tag_type.return_value = None
tags_service.delete_tag.side_effect = TagNotFoundError()
owner = _account(TenantAccountRole.OWNER)
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", True),
patch.object(module, "current_account_with_tenant", return_value=(owner, "tenant-1")),
patch.object(module, "enforce_rbac_access") as enforce_rbac_access,
):
with pytest.raises(NotFound):
unwrap(TagUpdateDeleteApi().delete)(TagUpdateDeleteApi(), request_context, "tag-1")
enforce_rbac_access.assert_not_called()
class TestTagBindingCollectionApi:
def test_create_success(self, app: Flask, admin_user, payload_patch):
api = TagBindingCollectionApi()
method = unwrap(api.post)
class TestTagBindings:
def test_create_passes_stable_binding_input(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
owner = _account(TenantAccountRole.OWNER)
payload = TagBindingPayload(
tag_ids=["tag-1", "tag-2"],
target_id="snippet-1",
type=TagType.SNIPPET,
)
payload = {
"tag_ids": ["tag-1"],
"target_id": "target-1",
"type": "knowledge",
}
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(owner, "tenant-1")),
):
result, status = unwrap(TagBindingCollectionApi().post)(TagBindingCollectionApi(), payload, request_context)
with app.test_request_context("/", json=payload):
with (
payload_patch(payload),
patch("controllers.console.tag.tags.TagService.save_tag_binding") as save_mock,
):
result, status = method(api, TagBindingPayload.model_validate(payload), admin_user)
tags_service.create_bindings.assert_called_once_with(
request_context,
TagBindingInput(("tag-1", "tag-2"), "snippet-1", "snippet"),
)
assert (result, status) == ({"result": "success"}, 200)
save_mock.assert_called_once()
assert status == 200
assert result["result"] == "success"
def test_create_maps_missing_target_to_not_found(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
tags_service.create_bindings.side_effect = TagBindingTargetNotFoundError("app")
owner = _account(TenantAccountRole.OWNER)
payload = TagBindingPayload(tag_ids=["tag-1"], target_id="missing", type=TagType.APP)
def test_create_snippet_binding_success(self, app: Flask, admin_user, payload_patch):
api = TagBindingCollectionApi()
method = unwrap(api.post)
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(owner, "tenant-1")),
):
with pytest.raises(NotFound, match="App not found") as exc_info:
unwrap(TagBindingCollectionApi().post)(TagBindingCollectionApi(), payload, request_context)
payload = {
"tag_ids": ["tag-1"],
"target_id": "snippet-1",
"type": "snippet",
}
assert exc_info.value.__cause__ is None
assert exc_info.value.__suppress_context__ is True
with app.test_request_context("/", json=payload):
with (
payload_patch(payload),
patch("controllers.console.tag.tags.TagService.save_tag_binding") as save_mock,
):
result, status = method(api, TagBindingPayload.model_validate(payload), admin_user)
def test_create_rejects_read_only_member(self, app: Flask, request_context: RequestContext) -> None:
readonly = _account(TenantAccountRole.NORMAL)
payload = TagBindingPayload(tag_ids=["tag-1"], target_id="app-1", type=TagType.APP)
save_mock.assert_called_once()
binding_payload = save_mock.call_args.args[0]
assert binding_payload.type == TagType.SNIPPET
assert binding_payload.target_id == "snippet-1"
assert status == 200
assert result["result"] == "success"
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(readonly, "tenant-1")),
):
with pytest.raises(Forbidden):
unwrap(TagBindingCollectionApi().post)(TagBindingCollectionApi(), payload, request_context)
def test_create_forbidden(self, app: Flask, readonly_user, payload_patch):
api = TagBindingCollectionApi()
method = unwrap(api.post)
def test_remove_passes_stable_binding_input(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
owner = _account(TenantAccountRole.OWNER)
payload = TagBindingRemovePayload(
tag_ids=["tag-1"],
target_id="app-1",
type=TagType.APP,
)
with app.test_request_context("/", json={}):
with (
payload_patch({}),
):
with pytest.raises(Forbidden):
method(
api,
TagBindingPayload(tag_ids=["tag-1"], target_id="target-1", type=TagType.KNOWLEDGE),
readonly_user,
)
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(owner, "tenant-1")),
):
result, status = unwrap(TagBindingRemoveApi().post)(TagBindingRemoveApi(), payload, request_context)
tags_service.delete_bindings.assert_called_once_with(
request_context,
TagBindingInput(("tag-1",), "app-1", "app"),
)
assert (result, status) == ({"result": "success"}, 200)
def test_remove_maps_missing_target_to_not_found(
self, app: Flask, request_context: RequestContext, tags_service: MagicMock
) -> None:
tags_service.delete_bindings.side_effect = TagBindingTargetNotFoundError("knowledge")
owner = _account(TenantAccountRole.OWNER)
payload = TagBindingRemovePayload(tag_ids=["tag-1"], target_id="missing", type=TagType.KNOWLEDGE)
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(owner, "tenant-1")),
):
with pytest.raises(NotFound, match="Dataset not found") as exc_info:
unwrap(TagBindingRemoveApi().post)(TagBindingRemoveApi(), payload, request_context)
assert exc_info.value.__cause__ is None
assert exc_info.value.__suppress_context__ is True
def test_remove_rejects_read_only_member(self, app: Flask, request_context: RequestContext) -> None:
readonly = _account(TenantAccountRole.NORMAL)
payload = TagBindingRemovePayload(tag_ids=["tag-1"], target_id="app-1", type=TagType.APP)
with (
app.test_request_context("/"),
patch.object(module.dify_config, "RBAC_ENABLED", False),
patch.object(module, "current_account_with_tenant", return_value=(readonly, "tenant-1")),
):
with pytest.raises(Forbidden):
unwrap(TagBindingRemoveApi().post)(TagBindingRemoveApi(), payload, request_context)
class TestTagBindingRemoveApi:
def test_remove_success(self, app: Flask, admin_user, payload_patch):
api = TagBindingRemoveApi()
method = unwrap(api.post)
payload = {
"tag_ids": ["tag-1", "tag-2"],
"target_id": "target-1",
"type": "knowledge",
}
with app.test_request_context("/", json=payload):
with (
payload_patch(payload),
patch("controllers.console.tag.tags.TagService.delete_tag_binding") as delete_mock,
):
result, status = method(api, TagBindingRemovePayload.model_validate(payload), admin_user)
delete_mock.assert_called_once()
delete_payload = delete_mock.call_args.args[0]
assert delete_payload.tag_ids == ["tag-1", "tag-2"]
assert status == 200
assert result["result"] == "success"
def test_remove_forbidden(self, app: Flask, readonly_user, payload_patch):
api = TagBindingRemoveApi()
method = unwrap(api.post)
with app.test_request_context("/", json={}):
with (
payload_patch({}),
):
with pytest.raises(Forbidden):
method(
api,
TagBindingRemovePayload(tag_ids=["tag-1"], target_id="target-1", type=TagType.KNOWLEDGE),
readonly_user,
)
class TestTagResponseModel:
def test_tag_response_normalizes_enum_type(self):
class TestTagResponseAndRoutes:
def test_tag_response_normalizes_enum_type(self) -> None:
payload = module.TagResponse.model_validate(
{"id": "tag-1", "name": "tag", "type": TagType.KNOWLEDGE, "binding_count": 1}
).model_dump(mode="json")
@ -473,32 +471,22 @@ class TestTagResponseModel:
assert payload["type"] == "knowledge"
assert payload["binding_count"] == "1"
class TestTagBindingRouteMetadata:
def test_write_routes_are_not_deprecated(self):
def test_binding_routes_keep_contract(self) -> None:
assert TagBindingCollectionApi.post.__apidoc__["id"] == "create_tag_binding"
assert TagBindingRemoveApi.post.__apidoc__["id"] == "remove_tag_bindings"
assert TagBindingCollectionApi.post.__apidoc__.get("deprecated") is not True
assert TagBindingRemoveApi.post.__apidoc__.get("deprecated") is not True
def test_write_routes_have_stable_operation_ids(self):
assert TagBindingCollectionApi.post.__apidoc__["id"] == "create_tag_binding"
assert TagBindingRemoveApi.post.__apidoc__["id"] == "remove_tag_bindings"
def test_write_routes_are_registered(self):
route_map = {
resource.__name__: urls
for resource, urls, _route_doc, _kwargs in console_ns.resources
if resource.__name__
in {
"TagBindingCollectionApi",
"TagBindingRemoveApi",
}
if resource.__name__ in {"TagBindingCollectionApi", "TagBindingRemoveApi"}
}
assert route_map == {
"TagBindingCollectionApi": ("/tag-bindings",),
"TagBindingRemoveApi": ("/tag-bindings/remove",),
}
assert route_map["TagBindingCollectionApi"] == ("/tag-bindings",)
assert route_map["TagBindingRemoveApi"] == ("/tag-bindings/remove",)
def test_legacy_write_routes_are_not_registered(self):
urls = {url for _resource, resource_urls, _route_doc, _kwargs in console_ns.resources for url in resource_urls}
assert "/tag-bindings/create" not in urls
assert "/tag-bindings/<uuid:id>" not in urls

14
api/tests/unit_tests/extensions/test_ext_application_services.py
View File

@ -26,6 +26,7 @@ from services.auth.data_source_api_key_auth_service import DataSourceApiKeyAuthS
from services.enterprise.enterprise_service import WebAppSettings
from services.errors.enterprise import EnterpriseAPIError, EnterpriseAPINotFoundError
from services.init_validation_service import InvalidInitializationPasswordError
from services.tag_application_service import TagApplicationService
from services.webapp_access_query_service import WebAppAccessUnavailableError
@ -152,6 +153,19 @@ def test_build_application_services_does_not_construct_schema_manager(
schema_manager.assert_not_called()
def test_build_application_services_wires_tag_boundary(
sqlite_session_factory: sessionmaker[Session],
) -> None:
services = ext_application_services.build_application_services(
database_client=sqlite_session_factory,
deployment_edition=DeploymentEdition.COMMUNITY,
initialization_password="",
redis=MagicMock(spec=RedisClientWrapper),
)
assert isinstance(services.tags, TagApplicationService)
def test_build_application_services_wires_account_profile_repository(
sqlite_session_factory: sessionmaker[Session],
) -> None:

129
api/tests/unit_tests/repositories/test_tag_repository.py Normal file
View File

@ -0,0 +1,129 @@
import pytest
from sqlalchemy import select
from sqlalchemy.orm import Session, sessionmaker
from models.enums import TagType
from models.model import Tag, TagBinding
from models.snippet import CustomizedSnippet, SnippetType
from repositories.tag_repository import TagRepository
from services.tag_application_service import (
CreateTagInput,
TagBindingInput,
TagBindingTargetNotFoundError,
TagNameConflictError,
TagNotFoundError,
UpdateTagInput,
)
def _tag(tag_id: str, *, workspace_id: str, tag_type: TagType, name: str) -> Tag:
tag = Tag(tenant_id=workspace_id, type=tag_type, name=name, created_by="account-1")
tag.id = tag_id
return tag
def _snippet(snippet_id: str, *, workspace_id: str) -> CustomizedSnippet:
snippet = CustomizedSnippet(
tenant_id=workspace_id,
name="Snippet",
description="",
type=SnippetType.NODE.value,
created_by="account-1",
updated_by="account-1",
)
snippet.id = snippet_id
return snippet
def test_list_tags_scopes_binding_counts_and_escapes_keyword(
sqlite_session_factory: sessionmaker[Session],
) -> None:
with sqlite_session_factory.begin() as session:
session.add_all(
[
_tag("tag-1", workspace_id="workspace-1", tag_type=TagType.APP, name="50% discount"),
_tag("tag-2", workspace_id="workspace-1", tag_type=TagType.APP, name="500 discount"),
_tag("tag-3", workspace_id="workspace-2", tag_type=TagType.APP, name="50% other"),
TagBinding(tenant_id="workspace-1", tag_id="tag-1", target_id="app-1", created_by="account-1"),
TagBinding(tenant_id="workspace-2", tag_id="tag-1", target_id="app-2", created_by="account-2"),
]
)
result = TagRepository(sqlite_session_factory).list_tags("workspace-1", "app", "50%")
assert result == (("tag-1", "50% discount", "app", 1),)
def test_tag_lifecycle_uses_owned_transactions_and_workspace_scope(
sqlite_session_factory: sessionmaker[Session],
) -> None:
repository = TagRepository(sqlite_session_factory)
created = repository.create_tag("workspace-1", "account-1", CreateTagInput("Original", "knowledge"))
with sqlite_session_factory.begin() as session:
session.add(
TagBinding(
tenant_id="workspace-1",
tag_id=created.id,
target_id="dataset-1",
created_by="account-1",
)
)
assert repository.get_tag_type("workspace-1", created.id) == "knowledge"
assert repository.get_tag_type("workspace-2", created.id) is None
updated = repository.update_tag("workspace-1", created.id, UpdateTagInput("Updated"))
assert updated.name == "Updated"
assert updated.binding_count == 1
with pytest.raises(TagNameConflictError):
repository.create_tag("workspace-1", "account-1", CreateTagInput("Updated", "knowledge"))
with pytest.raises(TagNotFoundError):
repository.update_tag("workspace-2", created.id, UpdateTagInput("Leaked"))
repository.delete_tag("workspace-1", created.id)
with sqlite_session_factory() as session:
assert session.scalar(select(Tag.id).where(Tag.id == created.id)) is None
assert session.scalar(select(TagBinding.id).where(TagBinding.tag_id == created.id)) is None
def test_binding_mutations_validate_target_type_and_workspace(
sqlite_session_factory: sessionmaker[Session],
) -> None:
with sqlite_session_factory.begin() as session:
session.add_all(
[
_snippet("snippet-1", workspace_id="workspace-1"),
_tag("tag-1", workspace_id="workspace-1", tag_type=TagType.SNIPPET, name="Valid"),
_tag("tag-2", workspace_id="workspace-1", tag_type=TagType.APP, name="Wrong type"),
_tag("tag-3", workspace_id="workspace-2", tag_type=TagType.SNIPPET, name="Wrong workspace"),
]
)
repository = TagRepository(sqlite_session_factory)
binding = TagBindingInput(("tag-1", "tag-1", "tag-2", "tag-3"), "snippet-1", "snippet")
repository.create_bindings("workspace-1", "account-1", binding)
repository.create_bindings("workspace-1", "account-1", binding)
with sqlite_session_factory() as session:
bindings = session.scalars(select(TagBinding).where(TagBinding.target_id == "snippet-1")).all()
assert len(bindings) == 1
assert bindings[0].tag_id == "tag-1"
assert bindings[0].tenant_id == "workspace-1"
repository.delete_bindings("workspace-1", binding)
with sqlite_session_factory() as session:
assert session.scalars(select(TagBinding).where(TagBinding.target_id == "snippet-1")).all() == []
def test_binding_mutation_rejects_missing_target(sqlite_session_factory: sessionmaker[Session]) -> None:
repository = TagRepository(sqlite_session_factory)
with pytest.raises(TagBindingTargetNotFoundError, match="Snippet not found"):
repository.create_bindings(
"workspace-1",
"account-1",
TagBindingInput(("tag-1",), "missing", "snippet"),
)

49
api/tests/unit_tests/services/test_tag_application_service.py Normal file
View File

@ -0,0 +1,49 @@
from unittest.mock import MagicMock
import pytest
from machinery.context import RequestContext
from services.tag_application_service import (
CreateTagInput,
TagApplicationService,
TagBindingInput,
TagSummary,
UpdateTagInput,
)
@pytest.fixture
def context() -> RequestContext:
return RequestContext("request-1", None, "account-1", "workspace-1")
def test_service_passes_stable_identity_to_store(context: RequestContext) -> None:
store = MagicMock()
store.list_tags.return_value = [TagSummary("tag-1", "Tag", "app", 1)]
store.create_tag.return_value = TagSummary("tag-2", "New", "app", 0)
store.update_tag.return_value = TagSummary("tag-2", "Updated", "app", 0)
service = TagApplicationService(tags=store)
assert service.list_tags(context, "app", "search") == (TagSummary("tag-1", "Tag", "app", 1),)
service.create_tag(context, CreateTagInput("New", "app"))
service.update_tag(context, "tag-2", UpdateTagInput("Updated"))
service.delete_tag(context, "tag-2")
service.create_bindings(context, TagBindingInput(("tag-1",), "app-1", "app"))
service.delete_bindings(context, TagBindingInput(("tag-1",), "app-1", "app"))
store.list_tags.assert_called_once_with("workspace-1", "app", "search")
store.create_tag.assert_called_once_with("workspace-1", "account-1", CreateTagInput("New", "app"))
store.update_tag.assert_called_once_with("workspace-1", "tag-2", UpdateTagInput("Updated"))
store.delete_tag.assert_called_once_with("workspace-1", "tag-2")
store.create_bindings.assert_called_once_with(
"workspace-1", "account-1", TagBindingInput(("tag-1",), "app-1", "app")
)
store.delete_bindings.assert_called_once_with("workspace-1", TagBindingInput(("tag-1",), "app-1", "app"))
def test_service_rejects_context_without_active_workspace() -> None:
context = RequestContext("request-1", None, "account-1", None)
service = TagApplicationService(tags=MagicMock())
with pytest.raises(RuntimeError, match="active workspace"):
service.list_tags(context, "app")

2
packages/contracts/generated/api/console/tags/orpc.gen.ts
View File

@ -50,7 +50,7 @@ export const get = oc
path: '/tags',
tags: ['console'],
})
.input(z.object({ query: zGetTagsQuery.optional() }))
.input(z.object({ query: zGetTagsQuery }))
.output(zGetTagsResponse)
export const post = oc

4
packages/contracts/generated/api/console/tags/types.gen.ts
View File

@ -27,9 +27,9 @@ export type TagType = 'app' | 'knowledge' | 'snippet'
export type GetTagsData = {
body?: never
path?: never
query?: {
query: {
keyword?: string
type?: '' | 'app' | 'knowledge' | 'snippet'
type: 'app' | 'knowledge' | 'snippet'
}
url: '/tags'
}

2
packages/contracts/generated/api/console/tags/zod.gen.ts
View File

@ -41,7 +41,7 @@ export const zTagBasePayload = z.object({
export const zGetTagsQuery = z.object({
keyword: z.string().optional(),
type: z.enum(['', 'app', 'knowledge', 'snippet']).optional().default(''),
type: z.enum(['app', 'knowledge', 'snippet']),
})
/**